# Event Stream Pattern Using SSE for Streaming Agent Responses in Anthropics CWC Workshops > Learn the event stream pattern with SSE for agent responses. Discover how Anthropics CWC workshops use SSE for resilient, reconnected streaming and duplicate event skipping. - Repository: [Anthropic/cwc-workshops](https://github.com/anthropics/cwc-workshops) - Tags: deep-dive - Published: 2026-07-18 --- **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`](https://github.com/anthropics/cwc-workshops/blob/main/agent-battle/my_agent.py), the connection is opened with streaming enabled and an extended timeout to accommodate long-running agent operations. ```python 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`](https://github.com/anthropics/cwc-workshops/blob/main/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` ```python 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`](https://github.com/anthropics/cwc-workshops/blob/main/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. ```python _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`](https://github.com/anthropics/cwc-workshops/blob/main/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`](https://github.com/anthropics/cwc-workshops/blob/main/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`](https://github.com/anthropics/cwc-workshops/blob/main/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`](https://github.com/anthropics/cwc-workshops/blob/main/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: ```python 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`](https://github.com/anthropics/cwc-workshops/blob/main/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`](https://github.com/anthropics/cwc-workshops/blob/main/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`](https://github.com/anthropics/cwc-workshops/blob/main/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`](https://github.com/anthropics/cwc-workshops/blob/main/agent-battle/harness/agent.py) and [`agent-battle/harness/client.py`](https://github.com/anthropics/cwc-workshops/blob/main/agent-battle/harness/client.py), while [`agent-battle/harness/logging_.py`](https://github.com/anthropics/cwc-workshops/blob/main/agent-battle/harness/logging_.py) records the telemetry for each successfully received event.