How to Stream Session Events from Claude Agents in Real-Time

Use the client.beta.sessions.events.stream() context manager to open a Server-Sent Events (SSE) connection that yields real-time events, then iterate over the stream to handle agent messages, tool calls, and status updates.

Claude Managed Agents expose a Server-Sent Events (SSE) stream that delivers every event generated by a session—including user messages, agent replies, tool uses, and status updates. The anthropics/cwc-workshops repository demonstrates how to consume these events using the Anthropic Python SDK, enabling low-latency interactive applications.

Core Architecture and Components

The streaming implementation relies on several key components working together to maintain a persistent connection between your application and Claude's agent infrastructure.

Anthropic Client
The anthropic.Anthropic() client authenticates requests and provides access to the beta.sessions.events namespace. According to the source in ship-your-first-managed-agent/agent_complete.py at line 15, this client is the entry point for all session operations.

The stream() Context Manager
Located in agent_complete.py (lines 52-66), client.beta.sessions.events.stream(session_id) opens an SSE connection to the session endpoint. The returned iterator yields Event objects as they arrive from the server, blocking until new data is available.

The send() Method
Also in agent_complete.py (lines 54-57), this method transmits events into the active session. You use this to push user messages, tool results, and interrupt signals.

Event Loop
The consumer logic appears in agent-battle/my_agent.py (lines 664-672). This loop inspects ev.type to branch between message rendering, tool execution, and session termination.

Tool-Result Handling
When the agent emits agent.custom_tool_use, the client must execute a local handler and return the result via a user.custom_tool_result event, as shown in agent_complete.py (lines 59-65).

Implementing the Event Stream Handler

To stream session events from Claude agents, follow this five-step pattern derived from the reference implementations:

1. Create or retrieve a session using client.beta.sessions.create().

2. Open the stream by entering the client.beta.sessions.events.stream(session_id) context manager.

3. Push the initial user message using client.beta.sessions.events.send() with type user.message.

4. Iterate over the stream, inspecting each event's type property:

   • agent.message contains the agent's text response.
   • agent.custom_tool_use signals that the agent is requesting a local tool execution.
   • session.status_idle or session.status_terminated indicates the session has finished.

5. Reconnect on transport errors by re-entering the context manager if the connection drops.

Handling Interactive Tool Calls

Real-time streaming enables bidirectional tool use. When the agent requires external data, it emits an agent.custom_tool_use event containing the tool name and input parameters.

Your handler must:

• Execute the requested function locally.
• Format the result as a user.custom_tool_result event.
• Reference the original request via the custom_tool_use_id field.

This pattern appears in agent_complete.py (lines 59-65), where the implementation calls a local handle_tool() function and immediately feeds the result back into the session stream.

Resilient Streaming with Reconnection Logic

Production deployments require handling network instability. The agent-battle/my_agent.py file (lines 660-676) demonstrates a robust pattern that catches specific exceptions and re-establishes the connection:

• Wrap the stream context in a while loop.
• Catch anthropic.APIConnectionError and anthropic.APIStatusError.
• Apply exponential backoff before retrying.
• Maintain session state across reconnections using the persistent session_id.

Practical Code Examples

Minimal Event Consumer

This basic implementation from my_agent.py (lines 664-672) prints agent messages until the session idles:

import anthropic

client = anthropic.Anthropic()  # ANTHROPIC_API_KEY from environment

session_id = "ses_123..."       # Obtained from client.beta.sessions.create()

with client.beta.sessions.events.stream(session_id) as stream:
    # Send initial prompt

    client.beta.sessions.events.send(
        session_id,
        events=[{
            "type": "user.message",
            "content": [{"type": "text", "text": "Analyze this dataset"}]
        }],
    )
    
    # Consume events

    for ev in stream:
        if ev.type == "agent.message":
            text = "".join(b.text for b in ev.content if hasattr(b, "text"))
            print(f"Agent: {text}")
        elif ev.type == "session.status_idle":
            print("Session complete")
            break

Streamlit Integration with Tool Handling

Adapted from ship-your-first-managed-agent/agent_complete.py (lines 52-66), this example handles interactive tool calls in a web UI:

import streamlit as st
import json

def chat(session_id: str, user_input: str):
    with client.beta.sessions.events.stream(session_id) as stream:
        client.beta.sessions.events.send(
            session_id,
            events=[{
                "type": "user.message",
                "content": [{"type": "text", "text": user_input}]
            }],
        )
        
        for ev in stream:
            if ev.type == "agent.message":
                text = "".join(b.text for b in ev.content if hasattr(b, "text"))
                st.write(f"**Claude:** {text}")
            
            elif ev.type == "agent.custom_tool_use":
                # Execute local tool logic

                result = run_local_tool(ev.name, ev.input)
                client.beta.sessions.events.send(
                    session_id,
                    events=[{
                        "type": "user.custom_tool_result",
                        "custom_tool_use_id": ev.id,
                        "content": [{"type": "text", "text": json.dumps(result)}]
                    }],
                )
            
            elif ev.type.startswith("session.status"):
                break

Robust Reconnection Handler

This pattern from agent-battle/my_agent.py (lines 660-676) survives transient network failures:

import time

_STREAM_RETRY = (
    anthropic.APIConnectionError,
    anthropic.APIStatusError,
)

def robust_stream(session_id: str, prompt: str):
    while True:
        try:
            with client.beta.sessions.events.stream(session_id) as stream:
                client.beta.sessions.events.send(
                    session_id,
                    events=[{
                        "type": "user.message",
                        "content": [{"type": "text", "text": prompt}]
                    }],
                )
                
                for ev in stream:
                    if not handle_event(ev):  # Your processing logic

                        return
                break  # Clean exit

                
        except _STREAM_RETRY as exc:
            st.warning(f"Connection lost: {exc}. Reconnecting...")
            time.sleep(2)  # Back-off before retry

Summary

• Server-Sent Events: Claude Managed Agents use SSE streams accessible via client.beta.sessions.events.stream(session_id).
• Bidirectional Communication: Use send() to transmit user messages and tool results while iterating over the stream to receive agent outputs.
• Tool Integration: Handle agent.custom_tool_use events by executing local functions and returning user.custom_tool_result events with matching IDs.
• Production Resilience: Wrap streams in retry loops to handle APIConnectionError and APIStatusError without losing session context.
• Key Reference Files: Study ship-your-first-managed-agent/agent_complete.py for basic loops and agent-battle/my_agent.py for production-grade implementations.

Frequently Asked Questions

What event types does the Claude agent stream emit?

The stream emits several distinct event types including agent.message for text responses, agent.custom_tool_use when the agent invokes a local tool, user.custom_tool_result for returning tool outputs, and status events like session.status_idle or session.status_terminated indicating session completion. The user.interrupt event allows clients to stop processing immediately.

How do I handle network interruptions during streaming?

Wrap the client.beta.sessions.events.stream() context manager in a retry loop that catches anthropic.APIConnectionError and anthropic.APIStatusError. As demonstrated in agent-battle/my_agent.py, re-enter the stream context after a brief delay to resume receiving events from the same session without data loss.

Can I send multiple messages while the stream is active?

Yes. The client.beta.sessions.events.send() method operates independently of the streaming iterator. You can send additional user.message events or user.custom_tool_result responses at any time while consuming the stream, enabling fully interactive conversational flows where the agent reacts to real-time inputs.

What is the difference between session.status_idle and session.status_terminated?

session.status_idle indicates the agent has finished processing the current turn and is awaiting further input, allowing you to continue the session by sending new events. session.status_terminated signals that the session has ended permanently, either through completion or error, and no further interaction is possible with that session ID.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}