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

> Stream session events from Claude agents in real-time using client.beta.sessions.events.stream(). Handle agent messages, tool calls, and status updates as they happen.

- Repository: [Anthropic/cwc-workshops](https://github.com/anthropics/cwc-workshops)
- Tags: how-to-guide
- Published: 2026-07-21

---

**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`](https://github.com/anthropics/cwc-workshops/blob/main/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`](https://github.com/anthropics/cwc-workshops/blob/main/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`](https://github.com/anthropics/cwc-workshops/blob/main/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`](https://github.com/anthropics/cwc-workshops/blob/main/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`](https://github.com/anthropics/cwc-workshops/blob/main/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`](https://github.com/anthropics/cwc-workshops/blob/main/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`](https://github.com/anthropics/cwc-workshops/blob/main/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`](https://github.com/anthropics/cwc-workshops/blob/main/my_agent.py) (lines 664-672) prints agent messages until the session idles:

```python
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`](https://github.com/anthropics/cwc-workshops/blob/main/ship-your-first-managed-agent/agent_complete.py) (lines 52-66), this example handles interactive tool calls in a web UI:

```python
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`](https://github.com/anthropics/cwc-workshops/blob/main/agent-battle/my_agent.py) (lines 660-676) survives transient network failures:

```python
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`](https://github.com/anthropics/cwc-workshops/blob/main/ship-your-first-managed-agent/agent_complete.py) for basic loops and [`agent-battle/my_agent.py`](https://github.com/anthropics/cwc-workshops/blob/main/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`](https://github.com/anthropics/cwc-workshops/blob/main/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.