Session-Event-Stream

Die Kommunikation mit Claude Managed Agents ist event-basiert. Du sendest User-Events an den Agent und erhältst Agent- und Session-Events zurück, um den Status zu verfolgen.



Event-Typen

Events fließen in zwei Richtungen.

• User-Events und System-Events sind das, was du an den Agent sendest: user.*-Events starten eine Session und steuern sie während ihres Verlaufs; system.message aktualisiert den System-Prompt des Agents zwischen den Turns.
• Session-Events, Span-Events und Agent-Events werden an dich gesendet, um dir Einblick in deinen Session-Zustand und den Fortschritt des Agents zu geben. Stream-Verbindungen, die sich dafür anmelden, erhalten zusätzlich Event-Deltas.

Session-, Span-, Agent-, User- und System-Event-Typ-Strings folgen einer {domain}.{action}-Namenskonvention. Die nur im Stream verfügbaren Delta-Preview-Events (event_start, event_delta) sind die Ausnahme. Siehe Event-Typen in der Referenz für den vollständigen Katalog.

Jedes persistierte Event enthält einen processed_at-Zeitstempel, der angibt, wann das Event serverseitig aufgezeichnet wurde. Wenn processed_at null ist, bedeutet das, dass das Event vom Harness in die Warteschlange gestellt wurde und verarbeitet wird, nachdem vorangehende Events abgeschlossen sind.



Events integrieren



Event-Deltas

Standardmäßig erreicht der Antworttext des Agents den Stream als gepufferte agent.message-Events, die jeweils erst ausgegeben werden, nachdem die Modellanfrage, die sie erzeugt hat, abgeschlossen ist. Event-Deltas ermöglichen es dir, diesen Text inkrementell als Live-Vorschau zu rendern, während das Modell ihn noch generiert. Eine Vorschau ist nicht die Antwort: Vorschauen sind eine Best-Effort-Anzeigehilfe, und die gepufferte agent.message ist immer der maßgebliche Datensatz. Ein Client, der Vorschauen ignoriert, erhält trotzdem einen vollständigen, korrekten Stream.



Vorschauen aktivieren

Vorschauen sind pro Stream-Verbindung opt-in. Füge den Query-Parameter event_deltas[] zu GET /v1/sessions/{session_id}/events/stream hinzu und wiederhole ihn einmal für jeden Event-Typ, für den du eine Vorschau möchtest. Die akzeptierten Werte sind agent.message und agent.thinking; jeder andere Wert gibt einen 400-Fehler zurück. Nur der Event-Stream auf Session-Ebene unterstützt diesen Parameter. Session-Thread-Event-Streams lehnen ihn ab.

Wenn ein Event mit Vorschau beginnt, gibt der Stream ein event_start aus, das den Typ und die id des kommenden Events enthält:

{
  "type": "event_start",
  "event": {
    "type": "agent.message",
    "id": "sevt_01abc..."
  }
}

Für agent.message folgen auf den Start event_delta-Events, die inkrementellen Text enthalten. Jedes Delta benennt das Event, das es erweitert, in event_id und den Content-Block, den es erweitert, in delta.index:

{
  "type": "event_delta",
  "event_id": "sevt_01abc...",
  "delta": {
    "type": "content_delta",
    "index": 0,
    "content": {
      "type": "text",
      "text": "Here is the summary"
    }
  }
}

Wenn ein agent.thinking-Event als Vorschau ausgegeben wird, wird nur das event_start ausgegeben. Es folgen keine event_delta-Events, und der Inhalt kommt wie üblich im gepufferten agent.thinking-Event an.

Im Gegensatz zu persistierten Events haben event_start und event_delta keine eigene id oder processed_at. Der einzige Identifier, den sie tragen, ist die id des Events, das sie als Vorschau anzeigen.



Akkumulieren und abgleichen

Die Python-, TypeScript- und Go-SDKs enthalten einen Accumulator-Helper, der die Vorschau anhand der id des Events indiziert und die index-Buchführung für dich übernimmt. Das manuelle Muster funktioniert in jeder Sprache: Wende es in den anderen SDKs auf die generierten Event-Typen an.

Im manuellen Muster behandelst du die Vorschau als Scratch-Buffer und das gepufferte Event als den Datensatz. Indiziere den Buffer nach (event_id, index). Gleiche pro Modellanfrage ab: Ein Turn beginnt mit einem einzelnen session.status_running-Event, dann erzeugt bei einem normal abgeschlossenen Turn jede Modellanfrage in dieser Reihenfolge span.model_request_start, event_start, die event_delta-Events, die gepufferte agent.message und schließlich span.model_request_end (im Tab Span-Events). Verarbeite jedes Event, sobald es eintrifft:

1. Bei event_start merkst du dir die angekündigte id. Die Identifier stimmen immer überein: event_start.event.id, jede event_delta.event_id und die id der gepufferten agent.message sind derselbe Wert.
2. Bei jedem event_delta hängst du delta.content.text an den Eintrag bei (event_id, delta.index) an und renderst den laufenden Text. Das erste Delta für einen index erstellt diesen Eintrag.
3. Wenn die gepufferte agent.message eintrifft, gleichst du sie anhand der id ab, verwirfst die akkumulierte Vorschau und renderst stattdessen den Inhalt der Nachricht.
4. Bei span.model_request_end schließt du jede Vorschau, die nicht durch ihr gepuffertes Event abgeglichen wurde. Es kommen keine weiteren Deltas mehr dafür. Wenn der Turn einen Fehler auslöst oder unterbrochen wird, kommt das gepufferte Event möglicherweise nie an; span.model_request_end kommt trotzdem.

# Vorschau-Snapshots, indiziert nach Event-ID. accumulate_managed_agents_event faltet jedes
# event_start / event_delta in einen agent.message-Snapshot; die gepufferte
# agent.message ersetzt ihn.
previews: dict[str, BetaManagedAgentsAgentMessageEvent] = {}

# Aktiviere agent.message-Vorschauen für diese Verbindung
with client.beta.sessions.events.stream(
    session.id, event_deltas=["agent.message"]
) as stream:
    client.beta.sessions.events.send(
        session.id,
        events=[
            {
                "type": "user.message",
                "content": [{"type": "text", "text": "Describe the repo in one sentence."}],
            },
        ],
    )

    for event in stream:
        match event.type:
            case "event_start":
                snapshot = accumulate_managed_agents_event(None, event)
                if snapshot is not None:
                    previews[event.event.id] = snapshot
                print(f"event_start             {event.event.type} {event.event.id}")
            case "event_delta":
                preview = accumulate_managed_agents_event(previews.get(event.event_id), event)
                if preview is not None:
                    previews[event.event_id] = preview
                    text = "".join(block.text for block in preview.content)
                    print(f"event_delta             preview: {text!r}")
            case "agent.message":
                # Das gepufferte Event ist der Datensatz: es ersetzt und schließt die Vorschau
                preview = accumulate_managed_agents_event(previews.pop(event.id, None), event)
                text = "".join(block.text for block in preview.content)
                print(f"agent.message           {event.id} {text!r}")
            case "span.model_request_end":
                # Es kommen keine Deltas mehr. Schließe jede Vorschau, deren
                # gepuffertes Event nie angekommen ist.
                for event_id in previews:
                    print(f"span.model_request_end  closing preview for {event_id}")
                previews.clear()
            case "session.status_idle":
                break



Einschränkungen

Vorschauen sind auf Reaktionsfähigkeit optimiert. Baue unter Berücksichtigung dieser Einschränkungen:

• Best Effort: Unter Last kann der Server Deltas für ein Event verwerfen. In diesem Fall erhältst du ein zusammenhängendes Präfix des Textes und dann keine weiteren Deltas für dieses Event. Die gepufferte agent.message kommt trotzdem vollständig an. Behandle eine akkumulierte Vorschau niemals als final.
• Keine Wiederholung bei Reconnect: Deltas werden nur an die Verbindung zugestellt, die sich dafür angemeldet hat, solange sie offen ist. Wenn der Stream abbricht, folge dem Reconnect-Verfahren im Tab „Events streamen": Öffne den Stream erneut und liste den Event-Verlauf auf. Der Verlauf enthält alle gepufferten Events, die ausgegeben wurden, während du getrennt warst, einschließlich der agent.message, auf die deine Vorschau gewartet hat. Es gibt keine Möglichkeit, verpasste Deltas erneut anzufordern.
• Nur primärer Thread, nur Text: Vorschauen decken Assistant-Text auf dem primären Thread der Session ab. Tool-Nutzung, Tool-Ergebnisse, MCP-Ergebnisse und Aktivität auf anderen Session-Threads werden nie als Vorschau ausgegeben.
• Nur Start bei agent.thinking: Eine agent.thinking-Vorschau gibt nur das event_start als Signal aus, dass ein Thinking-Block begonnen hat; es folgen keine event_delta-Events.
• Nie persistiert: event_start und event_delta existieren nur im Live-Stream. Sie erscheinen nicht im Event-Verlauf der Session (GET /v1/sessions/{session_id}/events).



Weitere Szenarien



Custom-Tool-Aufrufe verarbeiten

Wenn der Agent ein Custom Tool aufruft:

1. Die Session gibt ein agent.custom_tool_use-Event aus, das den Tool-Namen und die Eingabe enthält.
2. Die Session pausiert mit einem session.status_idle-Event, das stop_reason: requires_action enthält. Die blockierenden Event-IDs befinden sich im Array stop_reason.event_ids.
3. Führe das Tool in deinem System aus und sende für jedes ein user.custom_tool_result-Event, wobei du die Event-ID im Parameter custom_tool_use_id zusammen mit dem Ergebnisinhalt übergibst.
4. Sobald alle blockierenden Events aufgelöst sind, wechselt die Session zurück zu running.

with client.beta.sessions.events.stream(session.id) as stream:
    for event in stream:
        if event.type == "session.status_idle" and (stop_reason := event.stop_reason):
            match stop_reason.type:
                case "requires_action":
                    for event_id in stop_reason.event_ids:
                        # Suche das Custom-Tool-Use-Event und führe es aus
                        tool_event = events_by_id[event_id]
                        result = call_tool(tool_event.name, tool_event.input)

                        # Sende das Ergebnis zurück
                        client.beta.sessions.events.send(
                            session.id,
                            events=[
                                {
                                    "type": "user.custom_tool_result",
                                    "custom_tool_use_id": event_id,
                                    "content": [{"type": "text", "text": result}],
                                },
                            ],
                        )
                case "end_turn":
                    break



Tool-Bestätigung

Wenn eine Permission Policy eine Bestätigung erfordert, bevor ein Tool ausgeführt wird:

1. Die Session gibt ein agent.tool_use- oder agent.mcp_tool_use-Event aus.
2. Die Session pausiert mit einem session.status_idle-Event, das stop_reason: requires_action enthält. Die blockierenden Event-IDs befinden sich im Array stop_reason.event_ids.
3. Sende für jedes ein user.tool_confirmation-Event, wobei du die Event-ID im Parameter tool_use_id übergibst. Setze result auf "allow" oder "deny". Verwende deny_message, um eine Ablehnung zu erklären.
4. Sobald alle blockierenden Events aufgelöst sind, wechselt die Session zurück zu running.

with client.beta.sessions.events.stream(session.id) as stream:
    for event in stream:
        if event.type == "session.status_idle" and (stop_reason := event.stop_reason):
            match stop_reason.type:
                case "requires_action":
                    for event_id in stop_reason.event_ids:
                        # Genehmige den ausstehenden Tool-Aufruf
                        client.beta.sessions.events.send(
                            session.id,
                            events=[
                                {
                                    "type": "user.tool_confirmation",
                                    "tool_use_id": event_id,
                                    "result": "allow",
                                },
                            ],
                        )
                case "end_turn":
                    break



Eine Idle-Session fortsetzen

Sessions bleiben zwischen Interaktionen bestehen. Der Gesprächsverlauf wird beibehalten, es sei denn, die Session wird explizit gelöscht. Wenn eine Session in den Idle-Zustand wechselt, wird ihre Sandbox als Checkpoint gespeichert, wodurch der vollständige Sandbox-Zustand erhalten bleibt, einschließlich des Dateisystems, installierter Pakete und aller Dateien, die der Agent erstellt hat. Dadurch kannst du nach Inaktivität sauber fortsetzen.

Um eine Session fortzusetzen, sende wie gewohnt ein user.message-Event an sie:

# In Produktion übergib die gespeicherte ID der Session, die du fortsetzen willst.
ant beta:sessions:events send --session-id "$SESSION_ID" <<'YAML'
events:
  - type: user.message
    content:
      - type: text
        text: Now run the tests against the changes you made earlier.
YAML



System-Nachrichten senden

Sende ein system.message-Event, um den System-Prompt des Agents zwischen den Turns zu aktualisieren. Im Gegensatz zum system-Feld in der Agent-Definition (das bei der Session-Erstellung festgelegt wird) ermöglicht dir system.message, den System-Prompt im Verlauf der Session zu ändern. Verwende es, wenn der Agent mitten in der Session aktualisierte Anweisungen auf Systemebene benötigt: eine andere Persona, überarbeitete Einschränkungen oder zur Laufzeit abgerufenen Kontext, der das Verhalten des Modells von nun an prägen soll.

ant beta:sessions:events send --session-id "$SESSION_ID" <<'YAML'
events:
  - type: system.message
    content:
      - type: text
        text: "The user's current timezone is America/New_York."
YAML

system.message kann nicht gesendet werden, während die Session mit stop_reason: requires_action im Idle-Zustand ist. content akzeptiert 1–1000 Text-Elemente.



Nutzung verfolgen

Das Session-Objekt enthält ein usage-Feld mit kumulativen Token-Statistiken. Rufe die Session ab, nachdem sie in den Idle-Zustand gewechselt ist, um die aktuellen Gesamtwerte zu lesen, und verwende sie, um Kosten zu verfolgen, Budgets durchzusetzen oder den Verbrauch zu überwachen.

{
  "id": "sesn_01...",
  "status": "idle",
  "usage": {
    "input_tokens": 5000,
    "output_tokens": 3200,
    "cache_creation_input_tokens": 2000,
    "cache_read_input_tokens": 20000
  }
}

input_tokens meldet nicht gecachte Input-Tokens und output_tokens meldet die gesamten Output-Tokens über alle Modellaufrufe in der Session. Die Felder cache_creation_input_tokens und cache_read_input_tokens spiegeln die Prompt-Caching-Aktivität wider. Cache-Einträge verwenden eine TTL von 5 Minuten, sodass aufeinanderfolgende Turns innerhalb dieses Zeitfensters von Cache-Reads profitieren, die die Kosten pro Token reduzieren.



Console-Observability

Die Console bietet eine visuelle Timeline-Ansicht deiner Agent-Sessions. Navigiere zum Abschnitt Claude Managed Agents in der Console, um Folgendes zu sehen:

• Session-Liste: Alle Sessions mit ihrem Status, Erstellungszeitpunkt und Modell
• Tracing-Ansicht: Eine chronologische Ansicht der Events (Inhalt, Zeitstempel, Token-Nutzung) innerhalb einer Session. Tracing-Ansichten sind nur für Developer und Admins zugänglich.
• Tool-Ausführung: Details zu jedem Tool-Aufruf und seinem Ergebnis



Debugging-Tipps

• Session-Events prüfen: Session-Fehler werden über das session.error-Event übermittelt
• Tool-Ergebnisse überprüfen: Fehler bei der Tool-Ausführung erklären oft unerwartetes Agent-Verhalten
• Token-Nutzung verfolgen: Überwache den Token-Verbrauch, um Prompts zu optimieren und Kosten zu reduzieren
• System-Prompts verwenden: Füge Logging-Anweisungen zum System-Prompt hinzu, damit der Agent seine Überlegungen erklärt