Step-by-step: stream LLM tokens to your frontend in real time.
Users expect to see AI responses appear word by word, not wait for the entire response to generate. This guide covers implementing Server-Sent Events (SSE) streaming in FastAPI for real-time LLM token delivery, with proper error handling, timeout management, and client-side integration patterns.
Set up the StreamingResponse
FastAPI's StreamingResponse wraps an async generator that yields SSE-formatted chunks. Each chunk contains one or more tokens from the LLM response.
from fastapi.responses import StreamingResponse
@router.post("/v1/chat/stream")
async def chat_stream(body: ChatRequest):
return StreamingResponse(
stream_llm_response(body.messages),
media_type="text/event-stream",
headers={
"Cache-Control": "no-cache",
"Connection": "keep-alive",
"X-Accel-Buffering": "no", # disable Nginx buffering
},
)Create the async generator for OpenAI streaming
Call OpenAI's API with stream=True and yield each token as an SSE event. Track total tokens for billing and handle errors gracefully.
async def stream_llm_response(messages: list[dict]) -> AsyncGenerator[str, None]:
total_tokens = 0
try:
stream = await client.chat.completions.create(
model="gpt-4o", messages=messages, stream=True,
)
async for chunk in stream:
if chunk.choices[0].delta.content:
token = chunk.choices[0].delta.content
total_tokens += 1
yield f"data: {json.dumps({'token': token})}\n\n"
yield f"data: {json.dumps({'done': True, 'total_tokens': total_tokens})}\n\n"
except Exception as e:
yield f"data: {json.dumps({'error': str(e)})}\n\n"Handle Anthropic's streaming format
Anthropic's streaming API uses a different event format than OpenAI. Abstract the difference so your endpoint works with either provider.
async def stream_anthropic(messages: list[dict]) -> AsyncGenerator[str, None]:
async with anthropic_client.messages.stream(
model="claude-sonnet-4-20250514", messages=messages, max_tokens=4096,
) as stream:
async for text in stream.text_stream:
yield f"data: {json.dumps({'token': text})}\n\n"
yield f"data: {json.dumps({'done': True})}\n\n"Add timeout and error handling
LLM API calls can hang. Add timeouts, connection error recovery, and clean shutdown to prevent stuck SSE connections.
import asyncio
async def stream_with_timeout(messages, timeout=60):
try:
async with asyncio.timeout(timeout):
async for chunk in stream_llm_response(messages):
yield chunk
except asyncio.TimeoutError:
yield f"data: {json.dumps({'error': 'Response timed out'})}\n\n"
except Exception as e:
yield f"data: {json.dumps({'error': f'Stream error: {str(e)}'})}\n\n"Client-side integration
Connect to the SSE endpoint from your frontend using the EventSource API or fetch with ReadableStream. Parse each event and append tokens to the UI.
// Frontend JavaScript — works with any framework
const response = await fetch("/v1/chat/stream", {
method: "POST",
headers: { "Content-Type": "application/json", "Authorization": "Bearer TOKEN" },
body: JSON.stringify({ messages }),
});
const reader = response.body.getReader();
const decoder = new TextDecoder();
while (true) {
const { done, value } = await reader.read();
if (done) break;
const lines = decoder.decode(value).split("\n");
for (const line of lines) {
if (line.startsWith("data: ")) {
const data = JSON.parse(line.slice(6));
if (data.token) appendToUI(data.token);
if (data.done) onComplete(data.total_tokens);
}
}
}Skip the setup, start building
Everything in this guide is already implemented and configured in FastAPI AI Kit. Clone the repo and start building your product features immediately.
Ready to ship your AI backend this weekend?
Join developers who skipped weeks of boilerplate and went straight to building.