Sessions and Event Streaming

A session is a running instance of an agent. It has a lifecycle, emits events as it works, and terminates in a known state. Understanding how to start sessions, consume their events, and handle their lifecycle correctly is not optional background knowledge — it is the core operational skill for building on Managed Agents.

Session Lifecycle

Every session passes through a defined set of states:

rescheduling — the infrastructure is provisioning or re-provisioning the sandbox. Sessions start here and can return here between turns.

running — the agent is actively executing. Events are being emitted.

idle — the agent has stopped and is waiting. The accompanying stop_reason tells you why: end_turn (the agent finished its turn), requires_action (the session is blocked waiting on you — for example a tool confirmation), or retries_exhausted (the platform gave up retrying a failing operation).

terminated — the session is over. No further events will be emitted and it cannot be resumed.

The key difference from a request/response API: idle is not terminal. A session bounces between running and idle as turns complete and new input arrives. Your consumer must check the stop_reason on idle — an end_turn means results are ready; a requires_action means the session is waiting on you and will sit there until you respond.

Starting a Session

A session is created against an agent and an environment (the environment ID is required), then you send the task as a user.message event:

import anthropic

client = anthropic.Anthropic()

# Create the session
session = client.beta.sessions.create(
    agent={"type": "agent", "id": "agent_01abc123"},
    environment_id="env_01def456"
)

# Send the task as a user.message event
client.beta.sessions.events.send(
    session_id=session.id,
    event={
        "type": "user.message",
        "content": "Research Stripe's Q1 2026 product launches and write a competitor intelligence report."
    }
)

print(f"Session ID: {session.id}")

The session is created immediately. Execution starts asynchronously once the message arrives. The session.id is what you use to track and retrieve results.

Streaming Events

For sessions where you want to observe execution as it happens, stream the events over Server-Sent Events (SSE) — a one-way HTTP stream the server pushes events down as the agent works, so you read them as they happen rather than polling for them. Open the stream before you send the message — events emitted while no stream is attached are not replayed to a stream you open later:

# Stream events from an active session
with client.beta.sessions.events.stream(session_id=session.id) as stream:
    for event in stream:
        event_type = event.type

        if event_type == "agent.message":
            print(event.content)

        elif event_type == "agent.tool_use":
            print(f"[Tool call: {event.tool_name}]")

        elif event_type == "session.status_idle":
            print(f"[Idle — stop reason: {event.stop_reason}]")
            if event.stop_reason == "end_turn":
                break

        elif event_type == "session.error":
            print(f"[Error: {event.error}]")
            break

The streaming connection delivers events as they are emitted. If your client disconnects, the session keeps running — but the stream has no replay, so reconnection requires the consolidation pattern described below.

Event Types

Events are dot-namespaced by source. The ones your consumer will handle most:

agent.message — the agent produced a message. This is the agent's visible output.

agent.tool_use — the agent is calling a tool. Contains the tool name and input, and an evaluated_permission field indicating whether the call proceeds automatically or waits for confirmation.

agent.custom_tool_use — the agent is calling one of your custom tools. Your code executes it and returns the result as a user.custom_tool_result event.

session.status_idle — the session went idle. Carries the stop_reason (end_turn, requires_action, retries_exhausted).

session.error — an error occurred. The session may or may not continue depending on the error type.

span.model_request_end — a model inference call inside the session finished. Carries model_usage (token counts) — this is your billing and cost-attribution surface.

Events you send back into the session are namespaced under user.*: user.message (input), user.tool_confirmation (approve or deny a pending tool call), and user.custom_tool_result (return a custom tool's output).

Polling vs Streaming

Streaming is appropriate when:

• The session runs for minutes and you want to show progress to a user
• You need to log events in real time for observability
• You want to react to specific events (like a tool call) while the session is in progress

Polling is appropriate when:

• The session runs for hours and holding a streaming connection open is not practical
• Your integration is fire-and-forget — you just need the final result
• You are building a batch pipeline that checks session statuses on a schedule

import anthropic
import time

client = anthropic.Anthropic()

def wait_for_turn_end(session_id: str, poll_interval: int = 30, timeout: int = None) -> dict:
    """Poll a session until it goes idle or terminates.
    
    Args:
        session_id: The session to poll.
        poll_interval: Seconds between polls (default 30).
        timeout: Maximum seconds to wait before raising TimeoutError (default None = no limit).
    """
    import time as _time
    elapsed = 0
    while True:
        session = client.beta.sessions.retrieve(session_id)
        
        if session.status == "idle":
            # stop_reason tells you WHY it stopped:
            # end_turn = done; requires_action = waiting on you;
            # retries_exhausted = the platform gave up retrying
            return {"status": "idle", "stop_reason": session.stop_reason}
        elif session.status == "terminated":
            return {"status": "terminated"}
        
        if timeout is not None and elapsed >= timeout:
            raise TimeoutError(f"Session {session_id} did not finish within {timeout}s")
        
        print(f"Session {session_id}: {session.status} — waiting {poll_interval}s")
        _time.sleep(poll_interval)
        elapsed += poll_interval

One polling trap specific to this lifecycle: a session idle with stop_reason: requires_action is waiting for you — a pending tool confirmation, for example. If your poller only watches for end_turn, that session will sit blocked forever. Handle requires_action explicitly.

For multi-hour pipelines, poll every 60-120 seconds. For shorter tasks, streaming is cleaner.

Session Threads

In multi-agent sessions where an orchestrator dispatches to subagents, events are organized into threads. Each thread corresponds to one agent's execution within the broader session.

The session_thread_id field on events tells you which thread (and therefore which agent) produced each event:

with client.beta.sessions.events.stream(session_id=session.id) as stream:
    for event in stream:
        thread_id = getattr(event, 'session_thread_id', 'main')
        
        if event.type == "agent.message":
            print(f"[{thread_id}] {event.content}")

This separation is critical for production observability. When an orchestrator delegates to three parallel subagents, you need to know which agent produced which output — not just that output exists.

Reconnection: Stream Plus List, Dedupe by ID

The SSE stream has no replay. If your connection drops, the session keeps running, but a reconnected stream only delivers events emitted after the reconnection. Any events emitted while you were disconnected never appear on the new stream — naively reconnecting produces silent event loss, and if one of the missed events was a pending tool confirmation, a session that deadlocks waiting for an answer you never saw.

The correct pattern on every connect and reconnect is consolidation:

def consume_with_consolidation(session_id: str, seen_event_ids: set):
    # 1. Open the stream FIRST, so nothing slips through the gap
    with client.beta.sessions.events.stream(session_id=session_id) as stream:
        # 2. Fetch the full event history to recover anything missed
        for event in client.beta.sessions.events.list(session_id=session_id):
            if event.id not in seen_event_ids:
                seen_event_ids.add(event.id)
                handle(event)

        # 3. Continue with live events, deduping by event ID —
        #    an event can appear in both the list and the stream
        for event in stream:
            if event.id not in seen_event_ids:
                seen_event_ids.add(event.id)
                handle(event)

The same ordering rule applies at session start: open the stream before you send the user.message, or fetch the event list after sending — otherwise the agent's first events can be emitted before your stream is attached and you will never see them.

Always track seen event IDs in your consumer. The stream and the list overlap by design; dedupe by ID is what makes the combination lossless without reprocessing events you already handled.