Event Stream Pattern Using SSE for Streaming Agent Responses in Anthropics CWC Workshops

The repository implements a resilient Server-Sent Events (SSE) streaming pattern that opens a long-lived HTTP connection to receive incremental agent outputs, processes each JSON event with a _process_sse_event handler, and automatically reconnects on transient network failures while skipping duplicate events.

The anthropics/cwc-workshops codebase demonstrates how to build real-time interactions with Anthropic Managed Agents using the event stream pattern using SSE for streaming agent responses. This architecture enables low-latency delivery of agent steps without polling overhead, handling network interruptions gracefully through a structured retry mechanism.

How the SSE Streaming Pattern Works

The implementation follows the standard Server-Sent Events protocol with application-specific extensions for agent lifecycle management.

Initiating the SSE Connection

The client initiates a streaming run by sending a POST request to the Managed Agent endpoint with the Accept: text/event-stream header. According to agent-battle/my_agent.py, the connection is opened with streaming enabled and an extended timeout to accommodate long-running agent operations.

import httpx
import json

def stream_agent_run(agent_id, payload):
    url = f"https://api.anthropic.com/v1/agents/{agent_id}/runs"
    with httpx.Client(timeout=None) as client:
        with client.stream(
            "POST", 
            url, 
            json=payload,
            headers={"Accept": "text/event-stream"},
            stream=True
        ) as response:
            yield from _iter_sse_events(response)

The server responds with a stream of events, each prefixed with data: followed by a JSON payload.

Processing Events with _process_sse_event

In agent-battle/my_agent.py around line 500, the core event processing logic resides in the _process_sse_event function. The docstring explicitly states: """Process one SSE event. Returns True to keep streaming."""

The function parses the JSON event and branches based on the type field:

• run_step – Contains incremental agent output; the function processes the step content and returns True to continue streaming
• error – Signals a server-side failure; logs the error and returns False to terminate the stream
• run_complete – Indicates the agent finished execution; records the final outcome and returns False

def _process_sse_event(event: dict) -> bool:
    """Process one SSE event. Returns True to keep streaming."""
    event_type = event.get("type")
    
    if event_type == "run_step":
        print(f"Agent output: {event['output']}")
        return True
    elif event_type == "error":
        print(f"Stream error: {event['error']}")
        return False
    elif event_type == "run_complete":
        print(f"Run finished: {event['outcome']}")
        return False
    return True

Handling Reconnections and Error Recovery

The pattern implements robust fault tolerance through a predefined _STREAM_RETRY tuple in agent-battle/my_agent.py. When exceptions matching this set occur (including httpx.RemoteProtocolError, httpx.ReadError, and anthropic.APIConnectionError), the client does not abort but instead reopens the SSE stream.

The client maintains event ID tracking to enable exactly-once processing semantics across reconnections. After re-establishing the connection, it skips events already processed based on cached event IDs, ensuring continuity even during network instability.

_STREAM_RETRY = (
    httpx.RemoteProtocolError,
    httpx.ReadError,
    httpx.ReadTimeout,
    httpx.ConnectError,
    httpx.ConnectTimeout,
    anthropic.APIConnectionError,
)

def resilient_stream(agent_id, payload, last_event_id=None):
    while True:
        try:
            for event in stream_agent_run(agent_id, payload):
                if not _process_sse_event(event):
                    return
                last_event_id = event.get("id")
        except _STREAM_RETRY as exc:
            # Reconnect logic with last_event_id for continuity

            print(f"Reconnecting after {exc}")
            continue
        except Exception:
            raise

Implementation Details from the Source Code

The SSE streaming architecture spans several files in the agent-battle directory:

• agent-battle/my_agent.py – Contains the high-level orchestration, including the _process_sse_event function and _STREAM_RETRY exception tuple. This file implements the core event loop and reconnection logic.
• agent-battle/harness/agent.py – Encapsulates the low-level client interactions and HTTP stream management used by the SSE consumer.
• agent-battle/harness/client.py – Provides the HTTP client configuration and request building utilities used to initiate the SSE connection.
• agent-battle/harness/logging_.py – Records run progress, costs, and timestamps for each received event to support the workshop leaderboard functionality.

Code Example: Consuming the Agent Event Stream

This complete example demonstrates the event stream pattern using SSE for streaming agent responses with automatic reconnection and event deduplication:

import httpx
import json
import anthropic

_STREAM_RETRY = (
    httpx.RemoteProtocolError,
    httpx.ReadError,
    httpx.ReadTimeout,
    httpx.ConnectError,
    httpx.ConnectTimeout,
    anthropic.APIConnectionError,
)

def run_managed_agent(agent_id, prompt):
    """Stream agent responses using SSE with automatic reconnection."""
    url = f"https://api.anthropic.com/v1/agents/{agent_id}/runs"
    payload = {"input": prompt}
    processed_ids = set()
    
    while True:
        try:
            with httpx.Client(timeout=None) as client:
                headers = {"Accept": "text/event-stream"}
                with client.stream("POST", url, json=payload, 
                                 headers=headers) as response:
                    
                    for line in response.iter_lines():
                        if not line.startswith(b"data:"):
                            continue
                            
                        event = json.loads(line[5:])
                        event_id = event.get("id")
                        
                        # Skip duplicates on reconnection

                        if event_id in processed_ids:
                            continue
                        processed_ids.add(event_id)
                        
                        if not _process_sse_event(event):
                            return event
                            
        except _STREAM_RETRY:
            continue  # Reconnect and resume from last event

        except Exception as e:
            raise RuntimeError(f"Fatal stream error: {e}")

def _process_sse_event(event):
    """Process one SSE event. Returns True to keep streaming."""
    event_type = event.get("type")
    
    if event_type == "run_step":
        print(event["output"], end="")
        return True
    elif event_type in ("error", "run_complete"):
        return False
    return True

# Usage

result = run_managed_agent("agent-123", "Analyze this code")

Summary

• The event stream pattern using SSE for streaming agent responses opens a long-lived HTTP POST connection with Accept: text/event-stream to receive real-time agent outputs.
• The _process_sse_event function in agent-battle/my_agent.py handles event types (run_step, error, run_complete) and returns a boolean indicating whether to continue streaming.
• A _STREAM_RETRY tuple defines retriable network exceptions that trigger automatic reconnection rather than failure.
• The client implements event ID tracking to prevent duplicate processing when reconnecting after network interruptions.

Frequently Asked Questions

What is the purpose of the _process_sse_event function?

The _process_sse_event function serves as the central event dispatcher in agent-battle/my_agent.py. It receives each parsed SSE event as a dictionary, inspects the type field, and performs appropriate actions such as displaying agent output or terminating the stream. The function returns True to signal the main loop should continue listening for additional events, or False when the run completes or encounters an error.

How does the client handle network interruptions during streaming?

When network errors listed in the _STREAM_RETRY tuple occur—including httpx.RemoteProtocolError, httpx.ReadError, or anthropic.APIConnectionError—the client catches these exceptions and enters a reconnection loop. It reopens the SSE stream and uses cached event IDs to skip messages already processed before the interruption, ensuring the agent conversation resumes seamlessly without data loss.

What HTTP headers are required to initiate the SSE stream?

The client must include the Accept: text/event-stream header in the POST request to the /v1/agents/{agent_id}/runs endpoint. This signals the server to return responses using the Server-Sent Events protocol rather than a standard JSON response, enabling the chunked transfer of incremental agent outputs.

Where is the retry logic implemented in the codebase?

The retry logic and exception tuple _STREAM_RETRY are defined in agent-battle/my_agent.py, which orchestrates the high-level streaming workflow. The actual HTTP client configuration and stream iteration utilities reside in agent-battle/harness/agent.py and agent-battle/harness/client.py, while agent-battle/harness/logging_.py records the telemetry for each successfully received event.

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