LLM Streaming Responses with FastAPI and Anthropic SDK: SSE in Production
How to implement real-time streaming LLM responses using FastAPI Server-Sent Events (SSE) and the Anthropic SDK — with proper error handling, client reconnection, and production deployment patterns.
Non-streaming LLM responses feel slow. You send a request, wait 5-15 seconds while nothing happens, then get the entire response dumped at once. For anything conversational or long-form, this is a bad user experience.
Streaming fixes this — the response starts appearing token by token within milliseconds of the first output, making the interface feel responsive even when the total generation time is the same.
Here is how to implement production-grade streaming with FastAPI and the Anthropic SDK.
How Streaming Works
Instead of waiting for the complete response, the Anthropic API sends tokens as they are generated, using Server-Sent Events (SSE) — a simple HTTP protocol where the server pushes data to the client over a persistent connection.
The client receives events that look like:
data: {"type": "content_block_delta", "delta": {"text": "Hello"}}
data: {"type": "content_block_delta", "delta": {"text": " world"}}
data: {"type": "message_stop"}
FastAPI's StreamingResponse lets you write a generator that yields these events.
Basic Streaming Endpoint
import anthropic
import json
import os
from fastapi import FastAPI, HTTPException
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
from typing import AsyncGenerator
app = FastAPI()
client = anthropic.Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))
class ChatRequest(BaseModel):
messages: list[dict]
system: str = ""
model: str = "claude-sonnet-5"
max_tokens: int = 2000
def stream_anthropic_response(request: ChatRequest) -> Generator[str, None, None]:
"""
Synchronous generator that yields SSE-formatted events from Anthropic streaming.
"""
with client.messages.stream(
model=request.model,
max_tokens=request.max_tokens,
system=request.system,
messages=request.messages,
) as stream:
for text in stream.text_stream:
# Format as SSE
event_data = json.dumps({"type": "text", "content": text})
yield f"data: {event_data}\n\n"
# Send final message metadata
final_message = stream.get_final_message()
metadata = {
"type": "done",
"usage": {
"input_tokens": final_message.usage.input_tokens,
"output_tokens": final_message.usage.output_tokens,
},
"stop_reason": final_message.stop_reason,
}
yield f"data: {json.dumps(metadata)}\n\n"
yield "data: [DONE]\n\n"
@app.post("/chat/stream")
async def chat_stream(request: ChatRequest):
"""Streaming chat endpoint using SSE."""
return StreamingResponse(
stream_anthropic_response(request),
media_type="text/event-stream",
headers={
"Cache-Control": "no-cache",
"Connection": "keep-alive",
"X-Accel-Buffering": "no", # Disable Nginx buffering
}
)Async Streaming (Better for Concurrent Requests)
The sync approach above works but blocks a thread per stream. For production with many concurrent users, use the async client:
import anthropic
from fastapi import FastAPI, HTTPException
from fastapi.responses import StreamingResponse
from typing import AsyncGenerator
import json
import os
app = FastAPI()
async_client = anthropic.AsyncAnthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))
async def async_stream_response(request: ChatRequest) -> AsyncGenerator[str, None]:
"""Async generator — doesn't block the event loop."""
try:
async with async_client.messages.stream(
model=request.model,
max_tokens=request.max_tokens,
system=request.system,
messages=request.messages,
) as stream:
async for text in stream.text_stream:
event_data = json.dumps({"type": "text", "content": text})
yield f"data: {event_data}\n\n"
final_message = await stream.get_final_message()
yield f"data: {json.dumps({'type': 'done', 'usage': {'input_tokens': final_message.usage.input_tokens, 'output_tokens': final_message.usage.output_tokens}})}\n\n"
yield "data: [DONE]\n\n"
except anthropic.APIStatusError as e:
error_data = json.dumps({"type": "error", "message": str(e.message), "status": e.status_code})
yield f"data: {error_data}\n\n"
except anthropic.APIConnectionError:
error_data = json.dumps({"type": "error", "message": "Connection to Anthropic API failed"})
yield f"data: {error_data}\n\n"
@app.post("/chat/stream")
async def chat_stream_async(request: ChatRequest):
return StreamingResponse(
async_stream_response(request),
media_type="text/event-stream",
headers={
"Cache-Control": "no-cache",
"Connection": "keep-alive",
"X-Accel-Buffering": "no",
}
)Streaming with Tool Use
Tool use (function calling) changes the streaming pattern because you need to handle tool call events and potentially do multiple passes:
async def stream_with_tool_support(
messages: list[dict],
tools: list[dict],
tool_executor # Callable that runs a tool and returns a string
) -> AsyncGenerator[str, None]:
"""
Handle streaming with potential tool use loops.
Yields text tokens to the client and handles tool calls invisibly.
"""
current_messages = messages.copy()
for iteration in range(5): # Max tool use iterations
async with async_client.messages.stream(
model="claude-sonnet-5",
max_tokens=2000,
tools=tools,
messages=current_messages,
) as stream:
tool_calls = []
current_tool = None
current_input_str = ""
async for event in stream:
if event.type == "content_block_start":
if event.content_block.type == "tool_use":
current_tool = {
"id": event.content_block.id,
"name": event.content_block.name,
}
# Inform client a tool is being called
yield f"data: {json.dumps({'type': 'tool_call', 'tool': event.content_block.name})}\n\n"
current_input_str = ""
elif event.type == "content_block_delta":
if hasattr(event.delta, "text") and event.delta.text:
# Regular text token
yield f"data: {json.dumps({'type': 'text', 'content': event.delta.text})}\n\n"
elif hasattr(event.delta, "partial_json"):
current_input_str += event.delta.partial_json
elif event.type == "content_block_stop":
if current_tool and current_input_str:
current_tool["input"] = json.loads(current_input_str)
tool_calls.append(current_tool)
current_tool = None
current_input_str = ""
final_message = await stream.get_final_message()
if final_message.stop_reason == "end_turn":
yield "data: [DONE]\n\n"
return
if final_message.stop_reason == "tool_use" and tool_calls:
# Execute all tool calls
tool_results = []
for call in tool_calls:
result = await tool_executor(call["name"], call["input"])
tool_results.append({
"type": "tool_result",
"tool_use_id": call["id"],
"content": result
})
# Add assistant response and tool results to message history
current_messages.append({"role": "assistant", "content": final_message.content})
current_messages.append({"role": "user", "content": tool_results})
yield "data: [DONE]\n\n"Frontend JavaScript Client
async function streamChat(messages) {
const response = await fetch('/chat/stream', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ messages })
});
const reader = response.body.getReader();
const decoder = new TextDecoder();
let buffer = '';
while (true) {
const { done, value } = await reader.read();
if (done) break;
buffer += decoder.decode(value, { stream: true });
const lines = buffer.split('\n\n');
buffer = lines.pop(); // Keep incomplete last chunk
for (const line of lines) {
if (!line.startsWith('data: ')) continue;
const data = line.slice(6).trim();
if (data === '[DONE]') return;
try {
const event = JSON.parse(data);
if (event.type === 'text') {
// Append text to UI
appendToChat(event.content);
} else if (event.type === 'error') {
console.error('Stream error:', event.message);
} else if (event.type === 'done') {
console.log('Tokens used:', event.usage);
}
} catch (e) {
// Ignore parse errors on partial chunks
}
}
}
}Production Deployment Notes
Nginx config for SSE (critical — Nginx buffers SSE by default):
location /chat/stream {
proxy_pass http://fastapi:8000;
proxy_http_version 1.1;
proxy_set_header Connection '';
proxy_buffering off; # This is critical
proxy_cache off;
proxy_read_timeout 300s; # Allow long streams
chunked_transfer_encoding on;
}Kubernetes deployment considerations:
# Set appropriate timeouts for your Ingress
apiVersion: networking.k8s.io/v1
kind: Ingress
metadata:
annotations:
nginx.ingress.kubernetes.io/proxy-read-timeout: "300"
nginx.ingress.kubernetes.io/proxy-send-timeout: "300"
nginx.ingress.kubernetes.io/proxy-buffering: "off"Track streaming costs: The Anthropic SDK returns usage after get_final_message(). Log it per request — streaming costs the same as non-streaming, but unconstrained max_tokens can balloon costs if users send long conversations.
More LLMOps content? Read our LLM prompt caching for cost reduction and LLM context window management in production.
Today I Fixed
Short real fixes from production — posted daily
Stay ahead of the curve
Get the latest DevOps, Kubernetes, AWS, and AI/ML guides delivered straight to your inbox. No spam — just practical engineering content.
Related Articles
LLM Context Window Management in Production: What Nobody Tells You
How to manage LLM context windows in production systems — token budgeting, conversation compression, RAG vs context stuffing, and real strategies for keeping your LLM application fast and cheap at scale.
LLM Evaluation with LLM-as-Judge: How to Measure AI Quality in Production
How to evaluate LLM output quality in production using LLM-as-Judge — building automated evaluation pipelines, scoring rubrics, and golden dataset testing with Claude API. With real code examples.
LLM Observability with OpenTelemetry: Tracing LLM Calls in Production
How to add OpenTelemetry tracing to LLM applications in production — instrumenting Anthropic SDK calls, tracking token usage and latency, connecting LLM traces to your existing observability stack.