Managed Agents Events

Driving a session through events

A Managed Agents session is an event stream. You send user.* events; Anthropic emits agent.*, session.*, and span.* events back. Get four things right and the rest follows.

1. Stream-first ordering

The SSE stream only delivers events that occur after it opens. If you send the user message and then open the stream, the agent may have already emitted its first events — they arrive buffered in a single batch and you lose real-time visibility into the early turn.

# ✅ Correct
session = client.beta.sessions.create(...)
with client.beta.sessions.events.stream(session_id=session.id) as stream:
    client.beta.sessions.events.send(
        session_id=session.id,
        events=[{"type": "user.message", "content": [{"type": "text", "text": "..."}]}],
    )
    for event in stream:
        ...

# ❌ Wrong — first events arrive as one buffered batch
session = client.beta.sessions.create(...)
client.beta.sessions.events.send(...)
with client.beta.sessions.events.stream(...) as stream:  # opened too late
    for event in stream:
        ...

The nuvel scaffold's orchestrator.py does this correctly — the with stream: block is entered before the kickoff events.send().

2. The idle-break gate

The naive break — if event.type == "session.status_idle": break — is wrong. The session goes idle transiently while waiting for:

• A user.tool_confirmation (when an always_ask permission fires)
• A user.custom_tool_result (when the agent calls one of your custom tools)

In both cases the session is idle but not done. The terminal vs transient distinction lives in event.stop_reason.type:

Plus session.status_terminated (terminal error or archive) — always break.

for event in stream:
    handle(event)
    if event.type == "session.status_terminated":
        break
    if event.type == "session.status_idle":
        rtype = event.stop_reason.type
        if rtype != "requires_action":
            break  # end_turn or retries_exhausted — terminal
        # else: fall through, handle the pending tool/confirmation

Single rule: break on terminated, or on idle with a non-requires_action stop reason.

3. Custom-tool round-trip

When the agent calls one of your custom tools, the SDK emits agent.custom_tool_use and the session goes idle. Reply with user.custom_tool_result carrying the event id as custom_tool_use_id (not the agent's internal toolu_... id):

if event.type == "agent.custom_tool_use":
    try:
        result = dispatch_custom_tool(event.name, event.input)
        is_error = False
    except Exception as exc:
        result = f"Tool {event.name} failed: {exc}"
        is_error = True

    client.beta.sessions.events.send(
        session_id=session.id,
        events=[{
            "type": "user.custom_tool_result",
            "custom_tool_use_id": event.id,   # NOT a toolu_ id
            "content": [{"type": "text", "text": str(result)}],
            "is_error": is_error,
        }],
    )

For permission prompts, the same pattern with user.tool_confirmation / tool_use_id (also event.id) and result: "allow" | "deny".

4. Lossless reconnect after a dropped stream

The SSE stream has no replay. If your connection drops (network blip, deploy restart, idle timeout) and you reconnect naively, you only get events emitted after reconnection — anything in the gap is lost from the live stream.

The fix: on every (re)connect, fetch the full event history before consuming the live stream, and dedupe by event ID.

seen = set()
with client.beta.sessions.events.stream(session_id=session.id) as stream:
    # 1. Drain history first — covers any gap before the stream opened.
    history = client.beta.sessions.events.list(session_id=session.id)
    for event in history.data:
        seen.add(event.id)
        handle(event)

    # 2. Tail the live stream, skipping anything we already saw.
    for event in stream:
        if event.id in seen:
            continue
        seen.add(event.id)
        handle(event)
        # ... idle-break gate ...

Without this, a session can deadlock: the agent emitted agent.custom_tool_use during your gap, the session is idle waiting on you, your reconnected stream never sees the request, and nobody resolves it.

Event types worth knowing

The stream also echoes user-sent events with processed_at: null initially, then again with a timestamp once processed. Useful for "queued" → "acknowledged" UI states.

A complete loop

def run(client, agent_id, env_id, prompt):
    session = client.beta.sessions.create(agent=agent_id, environment_id=env_id)

    with client.beta.sessions.events.stream(session_id=session.id) as stream:
        client.beta.sessions.events.send(
            session_id=session.id,
            events=[{"type": "user.message", "content": [{"type": "text", "text": prompt}]}],
        )

        for event in stream:
            yield event  # to caller / UI

            if event.type == "session.status_terminated":
                return

            if event.type == "session.status_idle":
                if event.stop_reason.type != "requires_action":
                    return  # terminal idle

            if event.type == "agent.custom_tool_use":
                result = dispatch_custom_tool(event.name, event.input)
                client.beta.sessions.events.send(
                    session_id=session.id,
                    events=[{
                        "type": "user.custom_tool_result",
                        "custom_tool_use_id": event.id,
                        "content": [{"type": "text", "text": str(result)}],
                    }],
                )

This is essentially what orchestrator.py does in the nuvel scaffold. Read the generated code if you need the slightly more defensive version with serialization and error handling.

Common bugs

• Breaking on session.status_idle alone — sessions look like they "complete" right before the agent calls a tool. Always check stop_reason.type.
• Sending tool results to the wrong ID — custom_tool_use_id is the event id (e.g. sevt_...), not the underlying toolu_... id you might see embedded.
• Stream after send — symptoms: first 1-2 messages of every session feel "skipped" in the UI; they actually arrived in a buffered batch.
• No reconnect plan — works in dev, breaks the first time the dyno reboots mid-session. Always add the consolidation pattern before deploying.