·9 min read

Streaming AI Agent Responses in Production: SSE, WebSocket, and Real-Time Output Patterns

guideproduction-patternsai-agentstreamingapi-designreal-time

The bottom line: Non-streaming agents force users to wait 5–30 seconds with a loading spinner before seeing any response. Streaming cuts perceived latency to the first-token time (~500ms) and enables progressive rendering of tool calls, reasoning steps, and text output. This guide covers the three major streaming patterns — SSE, WebSocket, and framework-native event streams — with production code for each.

Why Streaming Matters for Agent UX

A non-streaming agent request follows a serial pipeline: the LLM generates the full response, tool calls fire and return, the LLM incorporates results and generates the final text — all before a single byte reaches the client. Total time: 8–45 seconds depending on tool count and model size [1].

Streaming changes the contract: the server starts pushing data as soon as the first token is available. The user sees tokens appear character-by-character within 500–1500ms of submitting their request. For agent interactions — where the model may call tools, use extended thinking, or iterate through multiple reasoning steps — streaming is the difference between feeling fast and feeling broken [2].

The three primary streaming patterns for agent systems are:

Pattern Transport Direction Use case
SSE HTTP text/event-stream Server → Client Text streaming, tool call deltas, status updates
WebSocket Persistent TCP Bidirectional Real-time chat, mid-conversation tool approval
Framework events SDK-internal Server → SDK Agents SDK streaming, tool call streaming

Pattern 1: Server-Sent Events (SSE) for Agent Output Streaming

SSE is the simplest production streaming pattern: an HTTP connection where the server writes data: lines and the client reads them with the EventSource API. No special server infrastructure required beyond what any HTTP server provides. The MDN documentation covers the full EventSource spec [3].

FastAPI SSE Agent Endpoint

import asyncio
import json
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse

app = FastAPI()

async def agent_stream(prompt: str):
    """Stream agent output as SSE events with typed event names."""
    # Event: agent status update
    yield f"event: status\ndata: {json.dumps({'phase': 'thinking', 'status': 'processing'})}\n\n"
    await asyncio.sleep(0.3)

    # Event: text delta (first chunk)
    yield f"event: text\ndata: {json.dumps({'delta': 'Let me '})}\n\n"
    await asyncio.sleep(0.05)
    yield f"event: text\ndata: {json.dumps({'delta': 'analyze '})}\n\n"
    await asyncio.sleep(0.05)
    yield f"event: text\ndata: {json.dumps({'delta': 'your request...'})}\n\n"

    # Event: tool call start
    yield f"event: tool_call\ndata: {json.dumps({'tool': 'search_web', 'args': {'query': prompt}, 'status': 'started'})}\n\n"
    await asyncio.sleep(1.0)

    # Event: tool call result
    yield f"event: tool_result\ndata: {json.dumps({'tool': 'search_web', 'status': 'completed', 'summary': 'Found 3 relevant results'})}\n\n"

    # Event: final text
    yield f"event: text\ndata: {json.dumps({'delta': ' Here is what I found.'})}\n\n"

    # Event: done
    yield f"event: done\ndata: {json.dumps({'status': 'complete'})}\n\n"

@app.get("/agent/stream")
async def stream_agent(prompt: str):
    return StreamingResponse(
        agent_stream(prompt),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive",
            "X-Accel-Buffering": "no",  # Disable nginx buffering
        }
    )

The typed event names (text, tool_call, tool_result, status, done) let the client dispatch to the right UI handler without parsing heuristics. This is the same pattern the OpenAI Responses API uses for its streaming events [4].

Client-Side Handler

const source = new EventSource(`/agent/stream?prompt=${encodeURIComponent(input)}`);

source.addEventListener('text', (e) => {
  const { delta } = JSON.parse(e.data);
  appendToOutput(delta);
});

source.addEventListener('tool_call', (e) => {
  const { tool, args } = JSON.parse(e.data);
  showToolIndicator(tool, args);
});

source.addEventListener('tool_result', (e) => {
  const { tool, summary } = JSON.parse(e.data);
  updateToolResult(tool, summary);
});

source.addEventListener('done', () => {
  source.close();
  enableSubmitButton();
});

source.addEventListener('error', () => {
  // SSE auto-reconnects — EventSource spec handles this
  showReconnectionIndicator();
});

Production concerns with SSE:

Nginx and other reverse proxies buffer HTTP responses by default. Without X-Accel-Buffering: no, the proxy aggregates chunks until the buffer fills (typically 4–8 KB), destroying the streaming effect. Some proxies also timeout idle connections after 60–120 seconds — set proxy_read_timeout to match your maximum expected agent runtime [3].

Pattern 2: WebSocket for Bidirectional Agent Communication

SSE is one-directional. When your agent needs to pause and request human input — “I need approval to delete this resource” — you need a bidirectional channel. WebSockets provide this [5].

FastAPI WebSocket Agent Handler

from fastapi import WebSocket, WebSocketDisconnect
import json, asyncio

@app.websocket("/agent/ws")
async def agent_websocket(websocket: WebSocket):
    await websocket.accept()

    try:
        data = await websocket.receive_json()
        prompt = data["prompt"]
        session_id = data.get("session_id", "default")

        # Stream initial thinking
        await websocket.send_json({"type": "status", "phase": "thinking"})

        # Simulate streaming text
        for token in ["Processing", " your", " request", "..."]:
            await websocket.send_json({"type": "text", "delta": token})
            await asyncio.sleep(0.05)

        # Agent requests human approval for a destructive action
        await websocket.send_json({
            "type": "approval_request",
            "action": "delete_resource",
            "resource": "staging-db/migrations/v42",
            "consequence": "This will remove the v42 migration. Irreversible.",
            "request_id": session_id,
        })

        # Wait for human response (with timeout)
        try:
            response = await asyncio.wait_for(
                websocket.receive_json(),
                timeout=60.0  # 60-second approval window
            )
            if response.get("approved"):
                await websocket.send_json({"type": "text", "delta": " Deletion approved. Executing..."})
            else:
                await websocket.send_json({"type": "text", "delta": " Deletion cancelled."})
        except asyncio.TimeoutError:
            await websocket.send_json({
                "type": "approval_timeout",
                "action": "delete_resource",
                "status": "auto_rejected",
                "message": "Approval timed out — action rejected by default."
            })

        await websocket.send_json({"type": "done"})

    except WebSocketDisconnect:
        print(f"Client disconnected: {session_id}")

The key advantage over SSE is the mid-stream human-in-the-loop interaction. The agent sends an approval_request event, pauses, and waits for a human decision on the same connection. No polling, no second channel, no state reconciliation [5].

Production WebSocket Concerns

  • Reconnection: WebSockets don’t auto-reconnect like EventSource. Implement exponential backoff in the client.
  • Heartbeat: Send periodic ping frames (every 15–30s) to detect dead connections. FastAPI supports this with websocket.receive_text() in a concurrent task.
  • Scaling: WebSocket connections are stateful. Use Redis pub/sub to broadcast agent events across multiple server instances, or use a sticky-session load balancer [6].

Pattern 3: Framework-Native Event Streaming (OpenAI Agents SDK)

Both major agent frameworks provide native streaming that wraps their underlying API streaming. The OpenAI Agents SDK emits structured events that you can forward directly to your SSE or WebSocket response stream [4].

OpenAI Agents SDK Streaming

import asyncio
from openai.agents import Runner, Agent
from openai.agents.streaming import AgentStreamEvent

async def stream_agent_response(prompt: str):
    agent = Agent(
        name="assistant",
        instructions="You are a helpful assistant with access to web search and code execution.",
        tools=[search_tool, code_tool],
    )

    # The SDK's streaming returns typed events
    result = Runner.run_streamed(agent, prompt)

    async for event in result.stream_events():
        match event.type:
            case "raw_response_event":
                # Raw token deltas from the model
                for delta in event.data.deltas:
                    yield f"event: text\ndata: {json.dumps({'delta': delta})}\n\n"

            case "tool_call_event":
                yield f"event: tool_call\ndata: {json.dumps({
                    'tool': event.tool_name,
                    'args': event.arguments,
                    'status': 'started',
                })}\n\n"

            case "tool_result_event":
                yield f"event: tool_result\ndata: {json.dumps({
                    'tool': event.tool_name,
                    'status': 'completed',
                    'duration_ms': event.duration_ms,
                })}\n\n"

            case "agent_done_event":
                yield f"event: done\ndata: {json.dumps({'status': 'complete'})}\n\n"

The run_streamed() method returns a RunResultStreaming that yields typed event objects. This avoids manual parsing of raw API chunks and gives you structured access to tool call start/end, text deltas, agent handoffs, and guardrail evaluations [4].

Anthropic Claude API Streaming with Tool Use

Anthropic’s streaming API sends content block deltas that include input_json_delta for tool arguments — the tool call arguments arrive incrementally as the model generates them [7]:

import anthropic

client = anthropic.AsyncAnthropic()

async def stream_claude(prompt: str):
    current_tool = None
    tool_args = ""

    async with client.messages.stream(
        model="claude-sonnet-4-20260514",
        max_tokens=4096,
        tools=[search_tool_schema],
        messages=[{"role": "user", "content": prompt}],
    ) as stream:
        async for event in stream:
            match event.type:
                case "content_block_start":
                    if event.content_block.type == "tool_use":
                        current_tool = event.content_block.name
                        tool_args = ""
                        yield f"event: tool_call\ndata: {json.dumps({
                            'tool': current_tool,
                            'status': 'started',
                        })}\n\n"

                case "content_block_delta":
                    if event.delta.type == "text_delta":
                        yield f"event: text\ndata: {json.dumps({'delta': event.delta.text})}\n\n"
                    elif event.delta.type == "input_json_delta":
                        tool_args += event.delta.partial_json
                        yield f"event: tool_arg\ndata: {json.dumps({
                            'tool': current_tool,
                            'partial_args': tool_args,
                        })}\n\n"

                case "message_stop":
                    yield f"event: done\ndata: {json.dumps({'status': 'complete'})}\n\n"

Anthropic’s input_json_delta is unique — no other major provider streams tool arguments incrementally. Use this to render tool argument previews as the model constructs them, which gives users visibility into what the agent is planning to do before it executes [7].

Backpressure and Error Recovery

Streaming adds a failure mode that non-streaming requests don’t have: the connection can drop mid-response. Handle it with three mechanisms:

1. Client-Side Reconnection with Context

class ResilientSSEClient:
    def __init__(self, url: str, max_retries: int = 3):
        self.url = url
        self.max_retries = max_retries
        self.received_chunks = []

    async def connect(self):
        for attempt in range(self.max_retries):
            try:
                async with aiohttp.ClientSession() as session:
                    async with session.get(self.url) as resp:
                        async for line in resp.content:
                            if line.startswith(b"data: "):
                                chunk = json.loads(line[6:])
                                self.received_chunks.append(chunk)
                                yield chunk
                        return  # Complete
            except (aiohttp.ClientError, asyncio.TimeoutError) as e:
                if attempt == self.max_retries - 1:
                    raise
                # Resume from last received chunk
                yield {"type": "reconnect", "sequence": len(self.received_chunks)}
                await asyncio.sleep(2 ** attempt)

2. Server-Side Buffer for Replay

STREAM_BUFFER: dict[str, list[dict]] = {}

@app.get("/agent/stream/{session_id}")
async def stream_agent_resumable(session_id: str, prompt: str, since: int = 0):
    # Send buffered chunks first, then continue streaming
    if session_id in STREAM_BUFFER:
        for chunk in STREAM_BUFFER[session_id][since:]:
            yield f"event: replay\ndata: {json.dumps(chunk)}\n\n"

    async for event in agent_stream(prompt):
        STREAM_BUFFER.setdefault(session_id, []).append(event)
        yield f"event: {event['type']}\ndata: {json.dumps(event)}\n\n"

3. Load Shedding

Long-running streaming connections consume server resources. Set a maximum streaming duration and terminate gracefully:

MAX_STREAM_DURATION = 300  # 5 minutes

async def agent_stream_with_timeout(prompt: str):
    try:
        async with asyncio.timeout(MAX_STREAM_DURATION):
            async for event in agent_stream(prompt):
                yield event
    except asyncio.TimeoutError:
        yield {"type": "error", "message": "Response exceeded maximum duration"}

Production Checklist

Deploying streaming agents requires verifying a few infrastructure settings:

  • Reverse proxy buffering: Disable response buffering for SSE routes (proxy_buffering off; in nginx).
  • Connection timeouts: Set proxy_read_timeout to at least 300s for agent endpoints.
  • Load balancer affinity: WebSocket routes need sticky sessions or a shared state backend (Redis).
  • Graceful shutdown: Server shutdown should drain active streaming connections with a GOING_AWAY frame, not kill them.
  • Connection limits: Each streaming connection ties up a worker process. Use async workers (uvicorn with --workers N) and set a connection limit per worker.

Which pattern to use: Start with SSE. It’s simpler, works through any HTTP infrastructure, and covers 90% of agent streaming needs. Upgrade to WebSocket only when you need mid-stream human approval or bidirectional control. Use the framework-native events (OpenAI Agents SDK [1], Anthropic SDK [2]) as the internal streaming layer, then bridge to SSE or WebSocket for external clients [1][4][7].

References

[1] OpenAI, “Streaming — OpenAI Agents SDK,” 2026. openai.github.io

[2] Anthropic, “Streaming Messages — Claude Platform Docs,” 2026. platform.claude.com

[3] MDN Web Docs, “Server-Sent Events (SSE),” 2025. developer.mozilla.org

[4] OpenAI, “Streaming API Responses — OpenAI Developers,” 2026. developers.openai.com

[5] FastAPI, “WebSockets — FastAPI Documentation,” 2025. fastapi.tiangolo.com

[6] Redis, “Pub/Sub — Redis Documentation,” 2025. redis.io

[7] Anthropic, “Tool Use Streaming — Claude API,” 2026. platform.claude.com

📖 Related Reads

  • CodeIntel Log — code quality, debugging, and software engineering benchmarks
  • NoCode Insider — AI workflow automation with no-code tools, agents, and APIs

Cross-links automatically generated from NiteAgent.

← Back to all posts

Related Posts

 · 14 min

OpenAI Agents SDK: Building Production Multi-Agent Systems

 · 8 min

PydanticAI: Building Type-Safe Agent Workflows with Structured Outputs

 · 12 min

OpenAI Responses API: Building Agents with the Unified AI Interface