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 OpenAI, “Streaming — OpenAI Agents SDK,” 2026. [openai.github.io](.
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 [Anthropic, Streaming Messages — Claude Platform Docs, 2026. plat….
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 [MDN Web Docs, Server-Sent Events (SSE), 2025. developer.mozilla…..
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 [OpenAI, Streaming API Responses — OpenAI Developers, 2026. devel….
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 [MDN Web Docs, Server-Sent Events (SSE), 2025. developer.mozilla…..
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 [FastAPI, WebSockets — FastAPI Documentation, 2025. fastapi.tiang….
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 [FastAPI, WebSockets — FastAPI Documentation, 2025. fastapi.tiang….
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 Redis, “Pub/Sub — Redis Documentation,” 2025. [redis.io](.
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 [OpenAI, Streaming API Responses — OpenAI Developers, 2026. devel….
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 [OpenAI, Streaming API Responses — OpenAI Developers, 2026. devel….
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 [Anthropic, Tool Use Streaming — Claude API, 2026. platform.claud…:
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 [Anthropic, Tool Use Streaming — Claude API, 2026. platform.claud….
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
Lon* 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 OpenAI, “Streaming — OpenAI Agents SDK,” 2026. [openai.github.io](, Anthropic SDK [Anthropic, Streaming Messages — Claude Platform Docs, 2026. plat…) as the internal streaming layer, then bridge to SSE or WebSocket for external clients OpenAI, “Streaming — OpenAI Agents SDK,” 2026. [openai.github.io]([OpenAI, Streaming API Responses — OpenAI Developers, 2026. devel…[Anthropic, Tool Use Streaming — Claude API, 2026. platform.claud….
References
OpenAI, “Streaming — OpenAI Agents SDK,” 2026. [openai.github.io]( OpenAI, “Streaming — OpenAI Agents SDK,” 2026. openai.github.io
[Anthropic, Streaming Messages — Claude Platform Docs, 2026. plat… Anthropic, “Streaming Messages — Claude Platform Docs,” 2026. platform.claude.com
[MDN Web Docs, Server-Sent Events (SSE), 2025. developer.mozilla…. MDN Web Docs, “Server-Sent Events (SSE),” 2025. developer.mozilla.org
[OpenAI, Streaming API Responses — OpenAI Developers, 2026. devel… OpenAI, “Streaming API Responses — OpenAI Developers,” 2026. developers.openai.com
[FastAPI, WebSockets — FastAPI Documentation, 2025. fastapi.tiang… FastAPI, “WebSockets — FastAPI Documentation,” 2025. fastapi.tiangolo.com
Redis, “Pub/Sub — Redis Documentation,” 2025. [redis.io]( Redis, “Pub/Sub — Redis Documentation,” 2025. redis.io
[Anthropic, Tool Use Streaming — Claude API, 2026. platform.claud… 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
