# Orchestrator Pattern for Handling Agent Event Streams Server-Side: Implementation Guide

> Learn the orchestrator pattern for server-side agent event stream handling. Ingest, parse, route, and consolidate agent tasks for efficient workflow management.

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

---

**The orchestrator pattern is a server-side coordination layer that ingests asynchronous agent event streams, parses natural language outputs into structured data, routes tasks to specialized sub-agents using correlation IDs, and emits consolidated responses.**

The `anthropics/cwc-workshops` repository demonstrates this architecture through a FastAPI-based implementation that manages autonomous agents emitting continuous event streams. This pattern solves the fundamental challenge of transforming free-form LLM prose into reliable, structured pipelines while maintaining request context across distributed operations.

## How the Orchestrator Pattern Works

The implementation in [`agent_decomposition/agents/before/stockpilot.py`](https://github.com/anthropics/cwc-workshops/blob/main/agent_decomposition/agents/before/stockpilot.py) establishes a five-stage pipeline that processes events from ingestion to final output. Each stage handles specific failure modes inherent in LLM-driven systems, ensuring robust server-side coordination.

### 1. Ingesting the Asynchronous Event Stream

The orchestrator subscribes to streaming endpoints using asynchronous I/O to prevent blocking during high-volume agent communication. In [`agent_decomposition/agents/before/stockpilot.py`](https://github.com/anthropics/cwc-workshops/blob/main/agent_decomposition/agents/before/stockpilot.py), the FastAPI router receives raw JSON events via HTTP streaming:

```python

# agent_decomposition/agents/before/stockpilot.py

from fastapi import APIRouter, Request
import json

router = APIRouter()

@router.post("/events")
async def receive_events(request: Request):
    async for line in request.stream():
        event = json.loads(line)
        await orchestrate(event)          # hand-off to the orchestrator core

    return {"status": "ok"}

```

This coroutine-based approach ensures the server can handle concurrent agent streams without resource exhaustion.

### 2. Parsing Prose into Structured Data

Because agents return natural language rather than machine-readable formats, the orchestrator extracts structured payloads using either deterministic parsing or secondary LLM prompts. The [`agent_decomposition/agents/before/subagents.py`](https://github.com/anthropics/cwc-workshops/blob/main/agent_decomposition/agents/before/subagents.py) file contains regex-based extraction for predictable formats:

```python

# agent_decomposition/agents/before/subagents.py

import re
import json

def extract_json(prose: str) -> dict:
    # Simple regex-based extraction used in the workshop

    match = re.search(r"\{.*\}", prose, re.DOTALL)
    if match:
        return json.loads(match.group())
    raise ValueError("No JSON found")

```

When patterns vary, the orchestrator can prompt a secondary LLM to reformat the text into JSON, bridging the gap between conversational agent outputs and structured API requirements.

### 3. Routing to Specialized Sub-Agents

Once the orchestrator identifies the intent (e.g., `forecast_demand`), it dispatches to dedicated sub-agent modules rather than handling business logic internally. The [`agent_decomposition/agents/starter/agent.py`](https://github.com/anthropics/cwc-workshops/blob/main/agent_decomposition/agents/starter/agent.py) demonstrates this delegation pattern:

```python

# agent_decomposition/agents/starter/agent.py

from agent_decomposition.agents.before.stockpilot import forecast_demand

async def orchestrate(event: dict):
    intent = event["intent"]
    if intent == "forecast_demand":
        raw_output = await forecast_demand(event["payload"])
        structured = extract_json(raw_output)
        # …continue processing or forward downstream

```

This design keeps the orchestrator as a thin routing layer while isolating domain-specific logic in specialized sub-agents.

### 4. Managing State with Correlation IDs

Event streams often deliver messages out of order or duplicate transmissions during network retries. The orchestrator maintains idempotency through a lightweight state machine keyed by correlation IDs. The shared utilities in [`agent_decomposition/agents/common.py`](https://github.com/anthropics/cwc-workshops/blob/main/agent_decomposition/agents/common.py) implement this tracking:

```python

# agent_decomposition/agents/common.py

state = {}

def store_state(corr_id: str, data: dict):
    state[corr_id] = data          # simple in-memory cache for demo purposes

def retrieve_state(corr_id: str) -> dict:
    return state.get(corr_id, {})

```

Each incoming event carries a unique identifier that allows the orchestrator to match sub-agent responses with originating requests, even when multiple agents process events concurrently.

### 5. Emitting Consolidated Results

After sub-agents complete processing, the orchestrator constructs standardized responses for downstream consumption. The implementation in [`agent_decomposition/agents/before/stockpilot.py`](https://github.com/anthropics/cwc-workshops/blob/main/agent_decomposition/agents/before/stockpilot.py) finalizes the pipeline:

```python

# agent_decomposition/agents/before/stockpilot.py

async def send_response(corr_id: str, result: dict):
    response = {"correlation_id": corr_id, "result": result}
    # This could be sent back over SSE, HTTP, or a message queue

    await some_output_sink.send(json.dumps(response))

```

This emission stage transforms heterogeneous agent outputs into consistent, schema-validated payloads suitable for UI rendering or further pipeline processing.

## Key Architectural Principles

The orchestrator pattern in `anthropics/cwc-workshops` relies on several core design decisions that ensure reliability at scale:

- **Event-Driven Async Design**: Utilizes FastAPI coroutines to process streams without blocking the main thread, maximizing throughput during agent spikes.
- **Prose-to-Structure Extraction**: Implements dual-mode parsing (deterministic regex + LLM reformatting) to handle both rigid and flexible agent output formats.
- **Delegation Over Monoliths**: Routes specific capabilities to isolated sub-agent modules, preventing the orchestrator from becoming bloated with business logic.
- **Correlation ID Tracking**: Guarantees exactly-once processing semantics through unique identifiers, essential for financial or inventory management workflows where duplicate operations cause data corruption.

## Critical Source Files

Understanding the orchestrator pattern requires examining these specific modules:

- **[`agent_decomposition/agents/before/stockpilot.py`](https://github.com/anthropics/cwc-workshops/blob/main/agent_decomposition/agents/before/stockpilot.py)**: Contains the top-level orchestrator implementing event ingestion, sub-agent delegation, and response emission.
- **[`agent_decomposition/agents/before/subagents.py`](https://github.com/anthropics/cwc-workshops/blob/main/agent_decomposition/agents/before/subagents.py)**: Houses specialized agent implementations and demonstrates prose-to-JSON extraction techniques.
- **[`agent_decomposition/agents/starter/agent.py`](https://github.com/anthropics/cwc-workshops/blob/main/agent_decomposition/agents/starter/agent.py)**: Provides a concrete routing example showing how `forecast_demand` intents are processed.
- **[`agent_decomposition/agents/common.py`](https://github.com/anthropics/cwc-workshops/blob/main/agent_decomposition/agents/common.py)**: Offers shared utilities for state management and correlation ID handling across the agent ecosystem.
- **[`ship-your-first-managed-agent/app.py`](https://github.com/anthropics/cwc-workshops/blob/main/ship-your-first-managed-agent/app.py)**: Demonstrates FastAPI application wiring, showing how the orchestrator integrates into a deployable server.

## Summary

- The orchestrator pattern serves as an async coordination layer between LLM-driven agents and downstream services, implemented in FastAPI within the `anthropics/cwc-workshops` repository.
- **Stream ingestion** happens via coroutine-based endpoints in [`stockpilot.py`](https://github.com/anthropics/cwc-workshops/blob/main/stockpilot.py), ensuring non-blocking processing of high-volume agent events.
- **Prose parsing** extracts structured data from natural language outputs using either regex patterns or secondary LLM prompts, as shown in [`subagents.py`](https://github.com/anthropics/cwc-workshops/blob/main/subagents.py).
- **Intent routing** delegates specific tasks to specialized sub-agents while maintaining a thin orchestrator core, preventing architectural bloat.
- **Correlation IDs** provide state management and idempotency guarantees, allowing the system to handle out-of-order or duplicate events gracefully.
- **Consolidated emission** transforms heterogeneous agent outputs into standardized JSON payloads suitable for API consumption or UI rendering.

## Frequently Asked Questions

### What makes the orchestrator pattern necessary for agent systems?

LLM-driven agents generate free-form text rather than structured API responses, and they operate asynchronously with unpredictable latency. The orchestrator pattern provides the server-side infrastructure to parse these natural language outputs, maintain request context across distributed operations, and route tasks appropriately without coupling agent logic to transport protocols.

### How does the orchestrator handle unstructured LLM outputs?

The implementation uses a two-tier extraction strategy. First, it attempts deterministic parsing via regex patterns (as shown in [`agent_decomposition/agents/before/subagents.py`](https://github.com/anthropics/cwc-workshops/blob/main/agent_decomposition/agents/before/subagents.py)). If the format is irregular, the orchestrator falls back to prompting a secondary LLM to reformat the prose into valid JSON, ensuring downstream services always receive structured data regardless of the agent's initial output style.

### What role do correlation IDs play in event stream processing?

Correlation IDs act as unique fingerprints for each request flowing through the system. The orchestrator stores these IDs in a lightweight state cache (implemented in [`agent_decomposition/agents/common.py`](https://github.com/anthropics/cwc-workshops/blob/main/agent_decomposition/agents/common.py)) to track which sub-agents have processed specific events. This mechanism prevents duplicate operations during network retries and enables the reassembly of multi-step agent workflows that arrive out of chronological order.

### Can this pattern scale beyond in-memory state management?

While the workshop examples use an in-memory dictionary for demonstration, the correlation ID pattern supports production scaling through external state stores like Redis or DynamoDB. The `store_state()` and `retrieve_state()` functions in [`common.py`](https://github.com/anthropics/cwc-workshops/blob/main/common.py) abstract the storage mechanism, allowing operators to swap the dictionary for distributed caches without modifying the orchestrator's core routing logic.