Session-Ereignisstrom

Sitzungs-Ereignisstrom

Die Kommunikation mit Claude Managed Agents ist ereignisbasiert. Sie senden Benutzerereignisse an den Agent und erhalten Agent- und Sitzungsereignisse zurück, um den Status zu verfolgen.

Alle Managed Agents API-Anfragen erfordern den managed-agents-2026-04-01 Beta-Header. Das SDK setzt den Beta-Header automatisch.

Ereignistypen

Ereignisse fließen in zwei Richtungen.

Benutzerereignisse sind das, was Sie an den Agent senden, um eine Sitzung zu starten und sie während des Fortschritts zu steuern.
Sitzungsereignisse, Span-Ereignisse und Agent-Ereignisse werden an Sie gesendet, um Einblick in Ihren Sitzungsstatus und Agentfortschritt zu erhalten.

Ereignistypzeichenfolgen folgen einer {domain}.{action} Benennungskonvention.

Jedes Ereignis enthält einen processed_at Zeitstempel, der angibt, wann das Ereignis serverseitig aufgezeichnet wurde. Wenn processed_at null ist, bedeutet dies, dass das Ereignis vom Harness in die Warteschlange eingereiht wurde und nach Abschluss der vorherigen Ereignisse verarbeitet wird.

Ereignisse integrieren

Zusätzliche Szenarien

Umgang mit benutzerdefinierten Toolaufrufen

Wenn der Agent ein benutzerdefiniertes Tool aufruft:

Die Sitzung gibt ein agent.custom_tool_use Ereignis aus, das den Toolnamen und die Eingabe enthält.
Die Sitzung pausiert mit einem session.status_idle Ereignis, das stop_reason: requires_action enthält. Die blockierenden Ereignis-IDs befinden sich im Array stop_reason.requires_action.event_ids.
Führen Sie das Tool in Ihrem System aus und senden Sie für jedes ein user.custom_tool_result Ereignis, wobei Sie die Ereignis-ID im Parameter custom_tool_use_id zusammen mit dem Ergebnis-Inhalt übergeben.
Sobald alle blockierenden Ereignisse gelöst sind, wechselt die Sitzung zurück zu running.

Werkzeugbestätigung

Wenn eine Berechtigungsrichtlinie eine Bestätigung vor der Ausführung eines Werkzeugs erfordert:

Die Sitzung gibt ein agent.tool_use- oder agent.mcp_tool_use-Ereignis aus.
Die Sitzung wird mit einem session.status_idle-Ereignis angehalten, das stop_reason: requires_action enthält. Die blockierenden Ereignis-IDs befinden sich im Array stop_reason.requires_action.event_ids.
Senden Sie für jedes ein user.tool_confirmation-Ereignis und übergeben Sie die Ereignis-ID im Parameter tool_use_id. Setzen Sie result auf "allow" oder "deny". Verwenden Sie deny_message, um eine Ablehnung zu erklären.
Sobald alle blockierenden Ereignisse gelöst sind, wechselt die Sitzung zurück zu running.

Nutzung verfolgen

Das Sitzungsobjekt enthält ein usage-Feld mit kumulativen Token-Statistiken. Rufen Sie die Sitzung ab, nachdem sie untätig wird, um die neuesten Gesamtwerte zu lesen, und verwenden Sie diese, 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 zwischengespeicherte Eingabe-Token und output_tokens meldet die Gesamtzahl der Ausgabe-Token über alle Modellaufrufe in der Sitzung. Die Felder cache_creation_input_tokens und cache_read_input_tokens spiegeln die Aktivität des Prompt-Cachings wider. Cache-Einträge verwenden eine TTL von 5 Minuten, daher profitieren aufeinanderfolgende Turns innerhalb dieses Fensters von Cache-Lesevorgängen, die die Kosten pro Token reduzieren.

with client.beta.sessions.events.stream(session.id) as stream:
    for event in stream:
        if event.type == "session.status_idle" and (stop := event.stop_reason):
            match stop.type:
                case "requires_action":
                    for event_id in stop.event_ids:
                        # Look up the custom tool use event and execute it
                        tool_event = events_by_id[event_id]
                        result = call_tool(tool_event.name, tool_event.input)

                        # Send the result back
                        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

with client.beta.sessions.events.stream(session.id) as stream:
    for event in stream:
        if event.type == "session.status_idle" and (stop := event.stop_reason):
            match stop.type:
                case "requires_action":
                    for event_id in stop.event_ids:
                        # Approve the pending tool call
                        client.beta.sessions.events.send(
                            session.id,
                            events=[
                                {
                                    "type": "user.tool_confirmation",
                                    "tool_use_id": event_id,
                                    "result": "allow",
                                },
                            ],
                        )
                case "end_turn":
                    break