LLM Streaming with SSE: Token-by-Token Responses in Python and FastAPI
TL;DR — LLM streaming with SSE: token-by-token responses. Jatin Bansal: "Provider pushes decode loop output incrementally over open HTTP response. Every major provider: stream=true, receive SSE events until terminal event. Backpressure inverted — slow client pauses GPU, billing ticks. stream.close() stops decoding within 1-2 tokens." Folarin: "SSE designed for one-way token streaming in 2012. All providers use it. Plain HTTP, works through proxies, reconnects automatically. FastAPI StreamingResponse with async generator. X-Accel-Buffering: no for nginx." Dr Williams: "stream=True yields ChatCompletionChunk with delta. AsyncGenerator pattern. X-Accel-Buffering: no critical — without it Nginx buffers entire response." MachineLearningPlus: "stream=True twice — JSON body and requests.post(). Anthropic uses typed events (content_block_delta). asyncio.Queue with maxsize=50 for backpressure." LLM Stream Parser: "OpenAI: no event: line, data: carries delta.content. Anthropic: event: line with content_block_delta, text_delta inside JSON." Learn more with LLM gateway, LiteLLM tutorial, OpenRouter tutorial, and switch provider.
Jatin Bansal defines streaming: "Streaming, at the API level, is the provider pushing the decode loop's output incrementally over a single open HTTP response instead of buffering the full message and returning it on close. Every major provider — Anthropic, OpenAI, Google, Mistral, the OSS serving stacks behind vLLM/TGI — exposes the same primitive: set stream=true, receive a sequence of server-sent events (SSE), each carrying a typed event name and a JSON payload, until a terminal event closes the stream."
Dr Williams explains the impact: "Streaming is the difference between an AI product that feels fast and one that feels slow. Instead of waiting 10–30 seconds for a completed response, a streaming API delivers each token as it is generated."
LLM Streaming Architecture
consumes SSE stream
appends tokens to UI"] end subgraph Server["FastAPI Server"] Endpoint["POST /chat endpoint
StreamingResponse
media_type: text/event-stream"] SSEGen["SSE Generator
async generator
yields data: {token}\\n\\n
terminates with data: [DONE]"] Headers["Headers
Cache-Control: no-cache
Connection: keep-alive
X-Accel-Buffering: no"] end subgraph Provider["LLM Provider"] StreamCall["stream=True API call
yields chunks incrementally"] OpenAIStream["OpenAI Stream
choices[0].delta.content
no event: line"] AnthropicStream["Anthropic Stream
event: content_block_delta
text_delta in JSON"] GoogleStream["Google Stream
chunk text
stream=True"] end subgraph Backpressure["Backpressure Handling"] Queue["asyncio.Queue
maxsize=50
bounded buffer
pauses producer when full"] Drain["Node.js drain
response.write() returns false
await drain when buffer full"] Close["stream.close()
closes HTTP socket
stops decoding in 1-2 tokens"] end Fetch --> Endpoint Endpoint --> SSEGen SSEGen --> Headers SSEGen --> StreamCall StreamCall --> OpenAIStream StreamCall --> AnthropicStream StreamCall --> GoogleStream OpenAIStream --> Queue AnthropicStream --> Queue GoogleStream --> Queue Queue --> Drain Drain --> Close
SSE Event Formats by Provider
| Provider | Event Format | Text Delta Location | Terminal Event |
|---|---|---|---|
| OpenAI | data: {json} (no event: line) |
choices[0].delta.content |
data: [DONE] |
| Anthropic | event: content_block_delta\ndata: {json} |
delta.text (type: text_delta) |
event: message_stop |
data: {json} |
candidates[0].content.parts[0].text |
Stream end | |
| OpenAI Responses | event: response.output_text.delta\ndata: {json} |
delta string |
event: response.completed |
Implementation
from dataclasses import dataclass
from typing import Optional
from enum import Enum
class Provider(Enum):
OPENAI = "openai"
ANTHROPIC = "anthropic"
GOOGLE = "google"
@dataclass
class LLMStreamingGuide:
"""LLM streaming with SSE implementation guide."""
def get_sse_format(self) -> str:
"""SSE format explanation."""
return (
"SSE (Server-Sent Events) format:\n"
"- Each event: data: <payload>\\n\\n\n"
"- Event type (optional): "
"event: <type>\\n\n"
"- Terminal: data: [DONE]\\n\\n\n"
"- Error: event: error\\n"
" data: <message>\\n\\n\n"
"\n"
"SSE rides on plain HTTP:\n"
"- Works through proxies and "
" load balancers\n"
"- Reconnects on its own\n"
"- No second protocol needed\n"
"- No separate connection\n"
"- media_type: text/event-stream"
)
def get_openai_streaming(self) -> str:
"""OpenAI streaming in Python."""
return (
"from openai import OpenAI\n"
"\n"
"client = OpenAI()\n"
"\n"
"# stream=True switches from\n"
"# ChatCompletion to\n"
"# ChatCompletionChunk objects\n"
"with client.chat.completions\\\n"
" .create(\n"
" model='gpt-4o',\n"
" messages=[\n"
" {'role': 'user',\n"
" 'content': 'Hello!'}\n"
" ],\n"
" stream=True,\n"
") as stream:\n"
" for chunk in stream:\n"
" delta = chunk.choices[0]\\\n"
" .delta.content\n"
" if delta is not None:\n"
" print(delta,\n"
" end='',\n"
" flush=True)\n"
"\n"
"# Async version\n"
"from openai import AsyncOpenAI\n"
"\n"
"async def stream_async():\n"
" client = AsyncOpenAI()\n"
" stream = await client\\\n"
" .chat.completions.create(\n"
" model='gpt-4o',\n"
" messages=[...],\n"
" stream=True)\n"
" async for chunk in stream:\n"
" delta = chunk.choices[0]\\\n"
" .delta.content\n"
" if delta:\n"
" yield delta"
)
def get_anthropic_streaming(self) -> str:
"""Anthropic streaming in Python."""
return (
"from anthropic import "
"AsyncAnthropic\n"
"\n"
"client = AsyncAnthropic()\n"
"\n"
"# Anthropic uses typed events:\n"
"# message_start,\n"
"# content_block_start,\n"
"# content_block_delta,\n"
"# message_stop\n"
"# Text in content_block_delta\n"
"\n"
"async def stream_anthropic():\n"
" async with client.messages\\\n"
" .stream(\n"
" model='claude-3-5-sonnet',\n"
" max_tokens=1024,\n"
" messages=[\n"
" {'role': 'user',\n"
" 'content': 'Hello!'}\n"
" ],\n"
" ) as stream:\n"
" async for text in \\\n"
" stream.text_stream:\n"
" yield text\n"
"\n"
"# Raw event iteration\n"
"async with client.messages\\\n"
" .stream(...) as stream:\n"
" async for event in stream:\n"
" if (event.type ==\n"
" 'content_block_delta'):\n"
" if event.delta.type == \\\n"
" 'text_delta':\n"
" print(\n"
" event.delta.text,\n"
" end='',\n"
" flush=True)"
)
def get_fastapi_streaming(self) -> str:
"""FastAPI StreamingResponse with SSE."""
return (
"from fastapi import FastAPI\n"
"from fastapi.responses import \\\n"
" StreamingResponse\n"
"from openai import AsyncOpenAI\n"
"import json\n"
"\n"
"app = FastAPI()\n"
"client = AsyncOpenAI()\n"
"\n"
"@app.post('/chat')\n"
"async def chat(body: dict):\n"
" async def event_stream():\n"
" try:\n"
" stream = await client\\\n"
" .chat.completions\\\n"
" .create(\n"
" model='gpt-4o',\n"
" messages=[\n"
" {'role': 'user',\n"
" 'content': body[\n"
" 'message']}\n"
" ],\n"
" stream=True)\n"
" async for chunk in stream:\n"
" delta = chunk\\\n"
" .choices[0]\\\n"
" .delta.content\n"
" if delta:\n"
" yield (\n"
" f'data: '\n"
" f'{json.dumps('\n"
" f'{\"token\": delta})}'\n"
" f'\\n\\n')\n"
" yield 'data: [DONE]\\n\\n'\n"
" except Exception as e:\n"
" yield (\n"
" f'event: error\\n'\n"
" f'data: {str(e)}\\n\\n')\n"
"\n"
" return StreamingResponse(\n"
" event_stream(),\n"
" media_type=\n"
" 'text/event-stream',\n"
" headers={\n"
" 'Cache-Control':\n"
" 'no-cache',\n"
" 'Connection':\n"
" 'keep-alive',\n"
" 'X-Accel-Buffering':\n"
" 'no',\n"
" },\n"
" )"
)
def get_backpressure(self) -> str:
"""Backpressure handling with asyncio.Queue."""
return (
"import asyncio\n"
"import json\n"
"\n"
"async def producer(\n"
" prompt: str,\n"
" queue: asyncio.Queue\n"
"):\n"
" '''Produce tokens from LLM\n"
" into bounded queue.'''\n"
" stream = await client\\\n"
" .chat.completions.create(\n"
" model='gpt-4o',\n"
" messages=[\n"
" {'role': 'user',\n"
" 'content': prompt}\n"
" ],\n"
" stream=True)\n"
" async for chunk in stream:\n"
" delta = chunk.choices[0]\\\n"
" .delta.content\n"
" if delta:\n"
" # Blocks if queue full\n"
" # (backpressure)\n"
" await queue.put(delta)\n"
" await queue.put(None)\n"
"\n"
"async def consumer(\n"
" queue: asyncio.Queue\n"
"):\n"
" '''Consume tokens from\n"
" queue as SSE.'''\n"
" while True:\n"
" token = await queue.get()\n"
" if token is None:\n"
" yield 'data: [DONE]\\n\\n'\n"
" break\n"
" yield (\n"
" f'data: '\n"
" f'{json.dumps('\n"
" f'{\"token\": token})}'\n"
" f'\\n\\n')\n"
"\n"
"@app.post('/chat')\n"
"async def chat(body: dict):\n"
" queue = asyncio.Queue(\n"
" maxsize=50) # bounded\n"
" # Start producer\n"
" task = asyncio.create_task(\n"
" producer(\n"
" body['message'],\n"
" queue))\n"
" return StreamingResponse(\n"
" consumer(queue),\n"
" media_type=\n"
" 'text/event-stream',\n"
" headers={\n"
" 'X-Accel-Buffering':\n"
" 'no',\n"
" },\n"
" )"
)
def get_react_client(self) -> str:
"""React client for SSE consumption."""
return (
"// React client consuming SSE\n"
"async function streamChat(\n"
" message: string\n"
") {\n"
" const response = await fetch(\n"
" '/chat',\n"
" {\n"
" method: 'POST',\n"
" headers: {\n"
" 'Content-Type':\n"
" 'application/json',\n"
" },\n"
" body: JSON.stringify({\n"
" message,\n"
" }),\n"
" }\n"
" );\n"
"\n"
" const reader = response\n"
" .body!.getReader();\n"
" const decoder = new TextDecoder();\n"
" let buffer = '';\n"
"\n"
" while (true) {\n"
" const { done, value } =\n"
" await reader.read();\n"
" if (done) break;\n"
"\n"
" buffer += decoder.decode(\n"
" value,\n"
" { stream: true }\n"
" );\n"
"\n"
" const lines = buffer\n"
" .split('\\n\\n');\n"
" buffer = lines.pop() || '';\n"
"\n"
" for (const line of lines) {\n"
" if (line.startsWith(\n"
" 'data: '\n"
" )) {\n"
" const data = line\n"
" .slice(6);\n"
" if (data === '[DONE]')\n"
" return;\n"
" const { token } =\n"
" JSON.parse(data);\n"
" // Append token to UI\n"
" setMessages(prev =>\n"
" [...prev, token]\n"
" );\n"
" }\n"
" }\n"
" }\n"
"}"
)
def get_error_handling(self) -> dict:
"""Error handling in SSE streams."""
return {
"error_event": (
"Send errors as SSE events so "
"client can show them instead "
"of connection dying silently: "
"event: error\\ndata: <message>\\n\\n"
),
"stream_close": (
"stream.close() closes HTTP "
"socket — provider stops "
"decoding within 1-2 tokens. "
"Without close, you keep "
"paying for tokens user never saw."
),
"partial_json": (
"input_json_delta chunks are not "
"individually parseable JSON. "
"Use partial JSON parser for "
"incremental tool args rendering. "
"Only json.loads after "
"content_block_stop."
),
"buffer_split": (
"Deltas can split across chunk "
"boundaries and multibyte chars. "
"Buffer partial events until "
"separator arrives."
),
"malformed_payload": (
"If non-[DONE] JSON payload is "
"malformed or truncated, throw "
"SSEPayloadError rather than "
"silently returning empty."
),
}
def get_production_tips(self) -> dict:
"""Production tips."""
return {
"nginx": (
"X-Accel-Buffering: no header "
"critical — without it, Nginx "
"buffers entire response before "
"forwarding, defeating streaming."
),
"backpressure_signal": (
"Watch server-side throughput "
"per stream dropping below "
"50-60 tokens/sec on known-fast "
"model — indicates slow client."
),
"node_js": (
"response.write() returns false "
"when client buffer full. Must "
"await drain rather than buffer "
"in memory. Skipping this OOMs "
"Node worker on 100k-token response."
),
"queue_size": (
"asyncio.Queue maxsize=50. "
"When full, put() pauses producer "
"until consumer catches up. "
"No token dropped. No memory explodes."
),
"sse_vs_websockets": (
"SSE for one-way token streaming. "
"WebSockets for full duplex "
"(live collaboration, voice). "
"SSE is the right default for chat."
),
"vercel_ai_sdk": (
"useChat hook handles SSE parsing, "
"message state, loading/error states, "
"and cancellation. Set "
"x-vercel-ai-ui-message-stream: v1 "
"header for custom backends."
),
}
LLM Streaming SSE Checklist
- [ ] LLM streaming uses Server-Sent Events (SSE) to deliver tokens incrementally
- [ ] SSE: provider pushes decode loop output over open HTTP response instead of buffering
- [ ] Every major provider supports streaming: OpenAI, Anthropic, Google, Mistral, vLLM/TGI
- [ ] Set stream=true in API call, receive SSE events until terminal event closes stream
- [ ] SSE format: data:
\n\n for each event, data: [DONE]\n\n to terminate - [ ] SSE rides on plain HTTP — works through proxies, load balancers, reconnects automatically
- [ ] No second protocol or separate connection needed — media_type: text/event-stream
- [ ] SSE designed for one-way server-to-client streaming in 2012 — right default for chat
- [ ] WebSockets for full duplex (live collaboration, voice) — SSE for token output
- [ ] OpenAI streaming: stream=True yields ChatCompletionChunk objects with delta
- [ ] OpenAI: each chunk contains delta (difference from previous), not full message
- [ ] OpenAI: text in choices[0].delta.content, no event: line, data: [DONE] terminates
- [ ] OpenAI: stream=True appears twice — JSON body (tells OpenAI) and requests.post() (tells library)
- [ ] Skip either stream=True and streaming breaks — requests buffers full response
- [ ] Anthropic streaming: typed events with event: line before data: line
- [ ] Anthropic events: message_start, content_block_start, content_block_delta, message_stop
- [ ] Anthropic: text in content_block_delta, delta type text_delta inside JSON
- [ ] Anthropic: only content_block_delta has actual text, rest carry metadata
- [ ] OpenAI Responses API: response.output_text.delta carries text, response.completed closes
- [ ] Tool calls: response.function_call_arguments.delta for streaming tool args
- [ ] Tool calls have progress events (response.function_call.in_progress, .completed)
- [ ] FastAPI StreamingResponse takes async generator, sends each yielded chunk to client
- [ ] SSE generator: yield data: {json} for each token, then data: [DONE] to close
- [ ] Headers: Cache-Control: no-cache, Connection: keep-alive, X-Accel-Buffering: no
- [ ] X-Accel-Buffering: no critical — without it Nginx buffers entire response, defeats streaming
- [ ] AsyncGenerator[str, None] pattern — yields values asynchronously, consumed with async for
- [ ] Error handling: send errors as SSE events (event: error\ndata:
) not silent death - [ ] stream.close() closes HTTP socket — provider stops decoding within 1-2 tokens
- [ ] Without stream.close(), you keep paying for tokens user never saw
- [ ] Backpressure exists but inverted — slow client pauses GPU, billing meter still ticks
- [ ] Backpressure signal: server-side throughput dropping below 50-60 tokens/sec on fast model
- [ ] Node.js: response.write() returns false when buffer full — await drain, do not buffer in memory
- [ ] Skipping drain OOMs Node worker handling 100k-token response from slow mobile client
- [ ] asyncio.Queue with maxsize=50 between producer (LLM stream) and consumer (SSE output)
- [ ] Queue.put() pauses producer when full — no token dropped, no memory explodes
- [ ] input_json_delta chunks not individually parseable — use partial JSON parser for incremental rendering
- [ ] Only json.loads after content_block_stop — fragments are partial JSON string
- [ ] Deltas can split across chunk boundaries and multibyte chars — buffer partial events
- [ ] Malformed or truncated payload: throw SSEPayloadError, do not silently return empty
- [ ] Text and tool-call JSON arrive interleaved under different event.index values — separate buffers
- [ ] React client: fetch() with ReadableStream, TextDecoder, split on \n\n, parse data: lines
- [ ] Vercel AI SDK: useChat hook handles SSE parsing, state, cancellation — set x-vercel-ai-ui-message-stream: v1
- [ ] LLM Stream Parser: auto-detects Anthropic vs OpenAI shape, yields visible text
- [ ] SSEParser: low-level incremental framer, buffers partial events, handles chunk boundaries
- [ ] Read LLM gateway guide for gateway concepts
- [ ] Read LiteLLM tutorial for proxy with streaming
- [ ] Read OpenRouter tutorial for managed streaming
- [ ] Read switch LLM provider for provider migration
- [ ] Test: OpenAI streaming yields tokens with delta.content
- [ ] Test: Anthropic streaming yields text from content_block_delta events
- [ ] Test: FastAPI StreamingResponse sends SSE to client without buffering
- [ ] Test: X-Accel-Buffering: no prevents Nginx buffering
- [ ] Test: asyncio.Queue backpressure pauses producer when consumer is slow
- [ ] Test: error event sent to client on exception, not silent connection death
- [ ] Test: stream.close() stops decoding and billing
- [ ] Document SSE format, provider differences, backpressure strategy, and error handling
FAQ
What is LLM streaming with SSE?
LLM streaming uses Server-Sent Events to deliver tokens incrementally as they are generated, instead of buffering the full response. Jatin Bansal: "Streaming at the API level is the provider pushing the decode loop output incrementally over a single open HTTP response instead of buffering the full message. Every major provider — Anthropic, OpenAI, Google, Mistral, vLLM/TGI — exposes the same primitive: set stream=true, receive a sequence of SSE events, each carrying a typed event name and JSON payload, until a terminal event closes the stream." Folarin: "Token streaming is a one-way job. SSE was designed for exactly this in 2012. Every major provider streams over SSE: OpenAI, Anthropic, Google Gemini all use it. SSE rides on plain HTTP, works through proxies and load balancers, reconnects on its own, no second protocol needed." Dr Williams: "Streaming is the difference between an AI product that feels fast and one that feels slow. Instead of waiting 10-30 seconds, streaming delivers each token as generated."
How do you stream LLM responses with OpenAI in Python?
Set stream=True in the API call and iterate over chunks with delta.content. Dr Williams: "The stream=True parameter switches from returning a completed ChatCompletion to returning a context manager that yields ChatCompletionChunk objects. Each chunk contains a delta — the difference from the previous chunk — rather than the full message." MachineLearningPlus: "Two changes from a normal API call: set stream: true in the request body, and iterate over lines instead of calling .json(). Each line holds a delta object with a content field. stream=True appears twice — once in JSON body (tells OpenAI to stream) and once in requests.post() (tells requests not to buffer). Skip either and streaming breaks." Code: with client.chat.completions.create(model='gpt-4o', messages=[...], stream=True) as stream: for chunk in stream: if chunk.choices[0].delta.content is not None: print(chunk.choices[0].delta.content, end='', flush=True)
How do you stream LLM responses with FastAPI?
Use StreamingResponse with an async generator that yields SSE-formatted data. Folarin: "FastAPI StreamingResponse takes an async generator and sends each yielded chunk to the client. SSE frame: data:
How does Anthropic streaming differ from OpenAI streaming?
OpenAI uses delta.content in choices array. Anthropic uses typed events with event: line before data: line. MachineLearningPlus: "Anthropic format differs in one key way. Instead of a single delta.content field, Claude sends typed events. Token text arrives inside events of type content_block_delta. Each SSE event includes an event: field before the data: line. Filter for the event type you care about. Anthropic sends: message_start, content_block_start, content_block_delta, message_stop. Only content_block_delta has actual text." Jatin Bansal: "Responses API emits semantic events with response.* prefix: response.created opens, response.output_text.delta carries text chunks, response.function_call_arguments.delta carries tool-call arguments, response.completed closes. Tool calls have their own progress events which older Chat Completions streaming did not." LLM Stream Parser: "OpenAI events have no event: line; each data: line carries choices[0].delta.content. Anthropic events carry event: line (message_start, content_block_delta); delta type lives inside JSON (text_delta for text, input_json_delta for tool-call args)."
How do you handle backpressure in LLM streaming?
Use asyncio.Queue with bounded buffer between producer and consumer, and handle Node.js drain events. Jatin Bansal: "Backpressure exists but inverted. TCP receive-window pressure on slow client causes upstream to pause decoding — GPU sits idle, billing meter ticks. Signal to watch: server-side throughput per stream dropping below 50-60 tokens/sec. In Node.js, response.write() returns false when client buffer full; you must await drain rather than buffer in memory. Skipping this is how a slow mobile client OOMs a Node worker handling 100k-token response." MachineLearningPlus: "Put asyncio.Queue between producer (LLM stream) and consumer. Queue decouples speeds with bounded buffer. maxsize=50 is the lever. When queue fills, queue.put() pauses producer until consumer catches up. No token dropped. No memory explodes." Jatin Bansal: "stream.close() closes the underlying HTTP socket — provider sees disconnect and stops decoding within one or two tokens. Without close, you keep paying for tokens the user never saw."
Want a self-hosted AI company brain that does all of this out of the box?
Book a demo →