使用服务器发送事件在 FastAPI 中实现流式传输
构建一个 FastAPI 端点，通过 StreamingResponse 和 text/event-stream 内容类型，将 LLM 的流式响应代理到浏览器客户端。
使用服务器发送事件在 FastAPI 中实现流式传输 是 CoddyKit 上的免费 AI Engineering Academy 课时。 这是第 3 节课，共 4 节。 你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。 这是 AI Engineering Academy 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 AI Engineering Academy 课程共包含 4 节课。
本课时的部分内容尚未翻译，以英文显示。
Why Server-Sent Events for LLM Streaming
Server-Sent Events (SSE) is a W3C standard that lets a server push a stream of text events to a browser client over a single long-lived HTTP connection. Unlike WebSockets, SSE is one-directional (server to client), works over standard HTTP/1.1, automatically reconnects on disconnection, and requires no special browser library. These properties make it the ideal transport for streaming LLM tokens from a FastAPI backend to a web frontend.
SSE Wire Format
SSE sends text data formatted as a series of fields separated by newlines. Each event contains an optional event type field, a data field with the payload, and an optional id for reconnection. Events are separated by a blank line. For LLM streaming, send each token as a data: token_text\n\n line and a special data: [DONE]\n\n event at the end to signal stream completion.
# SSE wire format example
'''
data: The\n\n
data: capital\n\n
data: of\n\n
data: France\n\n
data: is\n\n
data: Paris\n\n
data: [DONE]\n\n
'''
# Each 'data:' line is one event.
# The double newline (\n\n) terminates each event.
# The client receives these as EventSource message events.
# The content-type must be 'text/event-stream'.StreamingResponse in FastAPI
FastAPI's StreamingResponse accepts an async generator that yields strings and streams them to the client. By setting the media_type to 'text/event-stream' and formatting each yielded string as an SSE event, you turn any async generator into a proper SSE stream. FastAPI handles connection lifecycle, flushing, and HTTP headers automatically.
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from openai import AsyncOpenAI
import asyncio
app = FastAPI()
async_client = AsyncOpenAI()
async def token_generator(prompt: str):
stream = await async_client.chat.completions.create(
model='gpt-4o-mini',
messages=[{'role': 'user', 'content': prompt}],
stream=True,
)
async for chunk in stream:
delta = chunk.choices[0].delta.content
if delta:
yield f'data: {delta}\n\n' # SSE format
yield 'data: [DONE]\n\n'
@app.get('/stream')
async def stream_endpoint(prompt: str):
return StreamingResponse(
token_generator(prompt),
media_type='text/event-stream',
headers={'Cache-Control': 'no-cache', 'X-Accel-Buffering': 'no'},
)Important HTTP Headers for SSE
Three HTTP headers are critical for SSE to work correctly through proxies and CDNs. Cache-Control: no-cache prevents intermediaries from caching the stream. Connection: keep-alive keeps the TCP connection open. X-Accel-Buffering: no disables Nginx's response buffering, which would otherwise batch chunks and defeat the streaming effect. Without this last header, Nginx will buffer all output before forwarding to the browser.
from fastapi.responses import StreamingResponse
SSE_HEADERS = {
'Cache-Control': 'no-cache',
'Connection': 'keep-alive',
'X-Accel-Buffering': 'no', # disable nginx buffering
'Access-Control-Allow-Origin': '*', # CORS for cross-origin clients
}
@app.get('/chat')
async def chat_stream(prompt: str):
return StreamingResponse(
token_generator(prompt),
media_type='text/event-stream',
headers=SSE_HEADERS,
)Structured SSE Events with JSON Payloads
For richer streaming APIs, encode each event payload as JSON rather than raw text. This lets you include metadata alongside the token — for example, the token type (content vs tool call), a message ID, or a latency timestamp. The browser client parses each event's JSON and routes different event types to different UI components.
import json
import time
async def json_token_generator(prompt: str, session_id: str):
t_start = time.perf_counter()
stream = await async_client.chat.completions.create(
model='gpt-4o-mini',
messages=[{'role': 'user', 'content': prompt}],
stream=True,
)
async for chunk in stream:
delta = chunk.choices[0].delta.content
if delta:
payload = json.dumps({
'type': 'token',
'content': delta,
'session_id': session_id,
't_ms': round((time.perf_counter() - t_start) * 1000),
})
yield f'data: {payload}\n\n'
# Send completion event
yield f'data: {json.dumps({"type": "done", "session_id": session_id})}\n\n'Consuming SSE in a Browser (JavaScript)
The browser-side EventSource API connects to an SSE endpoint and fires events as they arrive. For token streaming, listen for the default message event, parse the data as JSON or treat it as a raw string, and append each token to the DOM. Handle the [DONE] sentinel by closing the EventSource connection.
// Browser-side JavaScript
const prompt = 'Explain hybrid search in one paragraph.';
const url = '/stream?prompt=' + encodeURIComponent(prompt);
const source = new EventSource(url);
const output = document.getElementById('output');
source.onmessage = (event) => {
if (event.data === '[DONE]') {
source.close(); // stop listening
return;
}
output.textContent += event.data; // append each token
};
source.onerror = (err) => {
console.error('SSE error:', err);
source.close();
};POST Requests with fetch for Streaming
EventSource only supports GET requests, which is limiting for complex prompts. For POST requests (sending a JSON body with conversation history), use the browser's fetch API with the Streams API to read the response body incrementally. This pattern is used by ChatGPT's web interface and most production LLM chat UIs.
// Browser-side: POST with fetch and ReadableStream
async function streamPost(messages) {
const response = await fetch('/chat', {
method: 'POST',
headers: {'Content-Type': 'application/json'},
body: JSON.stringify({messages}),
});
const reader = response.body.getReader();
const decoder = new TextDecoder();
const output = document.getElementById('output');
while (true) {
const {done, value} = await reader.read();
if (done) break;
const text = decoder.decode(value, {stream: true});
// Parse SSE lines
for (const line of text.split('\n')) {
if (line.startsWith('data: ') && line !== 'data: [DONE]') {
output.textContent += line.slice(6);
}
}
}
}FastAPI POST Endpoint for Chat Streaming
For POST-based chat streaming, define a Pydantic model for the request body, accept a list of messages, and stream the LLM response. This enables passing full conversation history with each request, supporting multi-turn chat applications. The pattern is identical to GET streaming except you extract the prompt from the request body.
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
class ChatRequest(BaseModel):
messages: list[dict]
model: str = 'gpt-4o-mini'
@app.post('/chat')
async def chat_post(request: ChatRequest):
async def generate():
stream = await async_client.chat.completions.create(
model=request.model,
messages=request.messages,
stream=True,
)
async for chunk in stream:
delta = chunk.choices[0].delta.content
if delta:
yield f'data: {delta}\n\n'
yield 'data: [DONE]\n\n'
return StreamingResponse(
generate(),
media_type='text/event-stream',
headers=SSE_HEADERS,
)Handling Client Disconnects
When a browser user navigates away or closes the tab, the HTTP connection closes and FastAPI raises asyncio.CancelledError in the streaming generator. Always handle this to avoid leaving LLM streaming requests open and incurring unnecessary API costs. Wrap your generator in a try/except for CancelledError and cancel the OpenAI stream when detected.
from fastapi import Request
@app.get('/stream')
async def stream_with_disconnect(prompt: str, request: Request):
async def generate_with_cancel():
try:
stream = await async_client.chat.completions.create(
model='gpt-4o-mini',
messages=[{'role': 'user', 'content': prompt}],
stream=True,
)
async for chunk in stream:
if await request.is_disconnected():
break # client gone, stop generating
delta = chunk.choices[0].delta.content
if delta:
yield f'data: {delta}\n\n'
except asyncio.CancelledError:
pass # client disconnected
finally:
yield 'data: [DONE]\n\n'
return StreamingResponse(generate_with_cancel(), media_type='text/event-stream')Adding Request Authentication
Production streaming endpoints must authenticate requests to prevent unauthorized LLM usage. Use FastAPI's Depends with an API key or JWT header check. Authentication happens before the generator starts, so the overhead is minimal and the stream only begins after the user is verified.
from fastapi import Header, HTTPException, Depends
VALID_API_KEYS = {'sk-demo-key-1', 'sk-demo-key-2'}
async def verify_api_key(x_api_key: str = Header(None)):
if x_api_key not in VALID_API_KEYS:
raise HTTPException(status_code=401, detail='Invalid API key')
return x_api_key
@app.post('/chat')
async def authenticated_chat(
request: ChatRequest,
api_key: str = Depends(verify_api_key),
):
async def generate():
stream = await async_client.chat.completions.create(
model=request.model,
messages=request.messages,
stream=True,
)
async for chunk in stream:
delta = chunk.choices[0].delta.content
if delta:
yield f'data: {delta}\n\n'
yield 'data: [DONE]\n\n'
return StreamingResponse(generate(), media_type='text/event-stream', headers=SSE_HEADERS)Testing SSE Endpoints
Test streaming endpoints with FastAPI's TestClient in streaming mode. Use with client.stream('GET', '/stream', params={...}) as r and iterate over r.iter_lines() to receive SSE events. This lets you verify that tokens are formatted correctly, the DONE sentinel is sent, and error cases produce appropriate SSE error events.
from fastapi.testclient import TestClient
def test_sse_endpoint():
with TestClient(app) as client:
with client.stream('GET', '/stream', params={'prompt': 'Say hi'}) as r:
assert r.status_code == 200
assert 'text/event-stream' in r.headers['content-type']
events = []
for line in r.iter_lines():
if line.startswith('data: '):
events.append(line[6:])
assert events[-1] == '[DONE]'
full_text = ''.join(e for e in events if e != '[DONE]')
assert len(full_text) > 0Quick Check
Test your understanding of FastAPI streaming with SSE from this lesson.
Lesson Recap
In this lesson you learned: Server-Sent Events is the standard HTTP transport for streaming LLM tokens to browser clients, StreamingResponse with text/event-stream turns any async generator into an SSE stream in FastAPI, and critical headers including X-Accel-Buffering and Cache-Control are required for correct behavior behind proxies. Handle client disconnects to avoid orphaned LLM API calls. Next up we tackle streaming responses that contain tool calls.
常见问题解答
「使用服务器发送事件在 FastAPI 中实现流式传输」课时是免费的吗？
是的 — 「使用服务器发送事件在 FastAPI 中实现流式传输」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 AI Engineering Academy 课程的其余内容，请升级到 CoddyKit PRO。 AI Engineering Academy 课程共包含 4 节课。
「使用服务器发送事件在 FastAPI 中实现流式传输」这节课中我会学到什么？
构建一个 FastAPI 端点，通过 StreamingResponse 和 text/event-stream 内容类型，将 LLM 的流式响应代理到浏览器客户端。 你通过在浏览器中直接运行的动手代码来练习 AI Engineering Academy，全天候 AI 导师会在你学习这节课的过程中回答你的问题。
学习 AI Engineering Academy 需要有经验吗？
无需任何先前经验。CoddyKit 上的 AI Engineering Academy 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。 这是第 3 节课，共 4 节。
「使用服务器发送事件在 FastAPI 中实现流式传输」课时需要多长时间？
大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。
我能在这节 AI Engineering Academy 课中编写并运行代码吗？
能。每节 AI Engineering Academy 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。
此课程中的所有课时
- 理解令牌流式传输
- 使用 Python SDK 消费数据流
- 使用服务器发送事件在 FastAPI 中实现流式传输
- 处理流式响应中的工具调用