FastAPI SSE 流式推送完全指南：一个 yield 搞定 AI 对话、日志与实时通知

【免费下载链接】fastapiFastAPI framework, high performance, easy to learn, fast to code, ready for production项目地址: https://gitcode.com/GitHub_Trending/fa/fastapi

AI 对话接口要等模型把整段回答生成完才返回，用户盯着空白的等待体验很差。FastAPI 从 0.135.0 起内置了 Server-Sent Events（SSE）流式推送：用yield生成器加EventSourceResponse就能把 token、日志行、订单状态逐条推给浏览器，浏览器原生EventSource直接消费，断线自动重连。读完本文，你可以独立搭出带 Pydantic 校验、可断点续传的生产级 SSE 接口。

SSE 适合解决什么问题

SSE 建立在普通 HTTP 之上：客户端发起一次请求，服务器不是一次性返回响应体，而是保持连接打开、持续写入事件。方向是单向的——服务器推、客户端收，客户端在这条连接上不能回传数据。

它和另外两种流式方案各有边界：

一句话选型：只要数据是服务器单向流出、前端跑在浏览器里，SSE 的"零配置重连"就是它相对 JSONL 的最大优势；需要双向通信才上 WebSocket。

⚠️ SSE 走text/event-stream，本质是 UTF-8 文本，二进制数据（音频、视频帧）不该走 SSE。那类需求用StreamingResponse做通用流式传输，参考仓库内的 Stream Data 进阶指南。

一个 SSE 事件在线上长什么样

SSE 的报文规则很朴素：每个事件由若干字段名: 值行组成，字段之间没有分隔符，事件与事件之间用一个空行（\n\n）切分。以订单状态流为例，网络上实际传输的字节长这样：

event: state data: {"order_id": "ord-1001", "state": "packed"} id: 1 event: state data: {"order_id": "ord-1001", "state": "shipped"} id: 2 : ping

各字段的含义与作用：

流结束的哨兵值

很多协议（包括 OpenAI 风格的对话接口）用一条特殊标记表示"流结束"。配合raw_data可以原样发出：

event: done data: [DONE]

客户端收到event === "done"或data === "[DONE]"即可收流，不需要靠"连接被关闭"这种不可靠的信号判断结束。

data与raw_data：编码与否的区别

ServerSentEvent（fastapi/sse.py L52-L156）提供两条负载通道：

• data：总是做 JSON 序列化。哪怕传字符串，线上也会带引号——data="hello"实际发出data: "hello"（test_string_data_json_encoded 钉死了这一行为）；
• raw_data：原样写进data:字段，不做任何编码。发预格式化文本、HTML 片段、CSV 行、日志行时用这个（test_raw_data_sent_without_json_encoding 验证了引号不会出现）。

⚠️ 两者互斥。同一事件同时设置data和raw_data，model_validator在构造时就抛ValueError（见 fastapi/sse.py L148-L156，佐证测试 test_data_and_raw_data_mutually_exclusive）。

非法值在构造期就抛错

ServerSentEvent基于 Pydantic 实现，非法输入不会拖到响应阶段才暴露，而是在ServerSentEvent(...)构造那一刻就抛ValueError：

三种生成器写法怎么选

异步生成器 + 返回类型声明（最小完整示例）

启用 SSE 只需两步：端点函数里yield，并声明response_class=EventSourceResponse（从fastapi.sse导入）。以订单状态流为例：

from collections.abc import AsyncIterable from fastapi import FastAPI from fastapi.sse import EventSourceResponse from pydantic import BaseModel app = FastAPI() class OrderStatus(BaseModel): order_id: str state: str order_events = [ OrderStatus(order_id="ord-1001", state="paid"), OrderStatus(order_id="ord-1001", state="packed"), OrderStatus(order_id="ord-1001", state="shipped"), ] @app.get("/orders/stream", response_class=EventSourceResponse) async def stream_order_events() -> AsyncIterable[OrderStatus]: for event in order_events: yield event

这段代码的关键是：路由注册时 FastAPI 会检测"生成器 +EventSourceResponse"的组合并把路由标记为 SSE 流（fastapi/routing.py L1124-L1130）；运行期每个yield出的OrderStatus都先过 Pydantic 再编码成data:字段。返回值校验失败会在流中途抛ResponseValidationError，而不是静默发出脏数据。

同步def：声明Iterable[T]，线程池迭代

阻塞式的数据源（读文件、查同步数据库）不用改造成异步，普通def生成器同样可以：

from collections.abc import Iterable @app.get("/orders/stream-sync", response_class=EventSourceResponse) def stream_order_events_sync() -> Iterable[OrderStatus]: for event in order_events: yield event

注意两点：返回类型标注要写Iterable而非AsyncIterable；路由层会用iterate_in_threadpool在线程池里迭代它，保证阻塞调用不卡事件循环（fastapi/routing.py L592-L595）。

省略返回类型：jsonable_encoder兜底

@app.get("/orders/stream-plain", response_class=EventSourceResponse) async def stream_order_events_plain(): for event in order_events: yield event

没有类型标注时，每个产出对象走 jsonable_encoder 转成可 JSON 化结构再json.dumps（序列化分支见 fastapi/routing.py L549-L553）。能跑，但丢掉了校验和类型化文档。

三行对比与选型结论

结论：能声明返回类型就声明，同步/异步只由数据源性质决定。类型化端点在 OpenAPI 中会呈现itemSchema/contentSchema结构（断言见 test_sse_router_typed_openapi_schema），客户端代码生成器可以直接消费。另外，同一端点里普通对象与ServerSentEvent可以混合yield，ServerSentEvent会跳过逐项校验由你控制结构（test_mixed_plain_and_sse_events）。

全局默认：default_response_class

一个应用或一组路由都以 SSE 输出时，不用每个端点重复标注：

app = FastAPI(default_response_class=EventSourceResponse) # 或只作用于某组路由 router = APIRouter(default_response_class=EventSourceResponse)

路由下的生成器端点默认按 SSE 处理，OpenAPI schema 同样生成（test_default_response_class_on_app_stream、test_default_response_class_on_parent_router_stream）。

SSE over POST：MCP 等协议的流式输出

浏览器EventSource只支持GET，但FastAPI 的 SSE 端点不限方法。MCP 这类把 SSE 流承载在POST上的协议可以直接写：

from collections.abc import AsyncIterable from fastapi.sse import ServerSentEvent from pydantic import BaseModel class ChatRequest(BaseModel): message: str @app.post("/chat/stream", response_class=EventSourceResponse) async def stream_chat(request: ChatRequest) -> AsyncIterable[ServerSentEvent]: for word in request.message.split(): yield ServerSentEvent(data=word, event="token") yield ServerSentEvent(raw_data="[DONE]", event="done")

POST 请求体经过正常 Pydantic 校验后，响应体是标准 SSE 流（test_post_method_sse覆盖了该行为）。消费端因为不能用EventSource，需要基于fetch的 SSE 解析器或各协议的官方 SDK。

浏览器如何消费 SSE 流

GET 端点的前端消费代码极简，全程不需要第三方库：

const es = new EventSource("/orders/stream"); es.addEventListener("state", (event) => { const status = JSON.parse(event.data); console.log(`order ${status.order_id} -> ${status.state}`); }); es.onmessage = (event) => { // 没有 event 字段的事件派发到这里 console.log(event.data); }; es.onerror = () => { // 浏览器按 retry 间隔自动重连，无需手动处理 };

读的时候注意：event字段写了什么就addEventListener什么名字；没写就落进onmessage。断线后的重连是EventSource的内置行为，这也是 SSE 对比 WebSocket 的核心红利。

断线重连：用Last-Event-ID补发事件

完整的断点续传链路分四步：

1. 服务端给每个事件编号（id字段）；
2. 浏览器自动记录收到的最后一个id；
3. 连接断开后浏览器重连，把该值放进Last-Event-ID请求头；
4. 服务端读出这个头，跳过已发过的事件，从断点继续。

以监控指标流为例（fastapi/param_functions.py 的Header会把last_event_id参数自动映射到Last-Event-ID头）：

from collections.abc import AsyncIterable from typing import Annotated from fastapi import FastAPI, Header from fastapi.sse import EventSourceResponse, ServerSentEvent app = FastAPI() def metrics_history() -> list[str]: return [f"cpu={12.5 * i:.1f}" for i in range(1, 21)] @app.get("/metrics/stream", response_class=EventSourceResponse) async def stream_metrics( last_event_id: Annotated[int | None, Header()] = None, ) -> AsyncIterable[ServerSentEvent]: history = metrics_history() start = last_event_id + 1 if last_event_id is not None else 0 for i, line in enumerate(history): if i < start: continue yield ServerSentEvent(raw_data=line, id=str(i))

这里有个容易踩的点：重连只是"从某个编号继续"，幂等由你负责。客户端收到id不大于已处理值的重复事件时应直接丢弃，服务端事件处理逻辑也应设计成可重放。

FastAPI 在底层默默做的 4 件事

框架把 SSE 部署中最常见的坑处理掉了，你不需要额外配置。以下 4 件都在 fastapi/routing.py 的 SSE 流式分支（L555-L685）中完成。

① 空闲 15 秒自动发: ping心跳

常量定义在 fastapi/sse.py L236-L241：KEEPALIVE_COMMENT = b": ping\n\n"与_PING_INTERVAL = 15.0。实现上，路由层用两条anyio内存对象流（缓冲大小 1）把"产出"和"心跳"解耦（fastapi/routing.py L615-L647）：_producer任务从生成器拉取并序列化事件，_keepalive_inserter任务用anyio.fail_after(_PING_INTERVAL)包住接收——15 秒内没等到事件就判定生成器空闲，向输出流插入一条: ping。这样拆任务是为了不让超时异常直接打穿生成器的__anext__（那会触发CancelledError并意外终止生成器）。

行为佐证：test_keepalive_ping_async 与 test_keepalive_ping_sync 把_PING_INTERVALmonkeypatch 到 0.05 秒，断言: ping出现在两条数据事件之间；test_no_keepalive_when_fast 反过来验证快速产出时不会误发心跳。

②Cache-Control: no-cache防缓存

fastapi/routing.py L682 显式写入响应头，阻止浏览器和各级代理把实时流当普通响应缓存。test_async_generator_with_model 对该头做了断言。

③X-Accel-Buffering: no关代理缓冲

L683-L684 写入该专用头，让 Nginx 这类反向代理停止对响应的攒包缓冲——否则前端会看到数据"憋半天一起到"。同一测试也断言了x-accel-buffering: no。

④ 序列化链路：先校验，再拼报文

yield出的每一项进入_serialize_sse_item（fastapi/routing.py L559-L590）：

• 产出的是ServerSentEvent→ 读raw_data原样使用；data优先走model_dump_json，其余对象走jsonable_encoder+json.dumps；连同event/id/retry/comment交给format_sse_event；
• 产出的是普通对象 → 先经返回类型派生的字段校验（L528-L553 的_serialize_data，与 JSONL 流共用），再包进data:字段。

最底层的format_sse_event（fastapi/sse.py L165-L233）负责把数据串按 SSE 规范换行切成多条data:行、给注释加:前缀，并以\n\n收尾。换行符（\r\n、\r）、空串、连续空行等边界行为被 test_format_sse_event_splitlines_behavior_in_data 和 test_format_sse_event_splitlines_behavior_in_comment 逐参数钉死。

最后澄清一点：EventSourceResponse本身（fastapi/sse.py L20-L33）只是一个轻量标记类——继承StreamingResponse、设置media_type = "text/event-stream"，真正的编码逻辑全在路由层。正因为如此，它才能安全地兼作default_response_class。

生产部署清单

上线前逐项核对：

• 代理缓冲：Nginx 靠框架自带的X-Accel-Buffering: no关闭；自研或第三方代理（HAProxy、云 LB 网关）需确认自身不缓冲，用curl -N观察数据是否逐条到达。
• 空闲超时对齐：内置心跳间隔是 15 秒，必须小于链路上每一跳代理/LB 的空闲超时（常见 30~60 秒）。若调过_PING_INTERVAL，LB 超时要同步调整。
• 重连幂等：Last-Event-ID保证"从断点续发"，但不保证"恰好一次"。客户端按id去重，消费逻辑可重放。
• POST 端点的客户端选型：EventSource不支持 POST；SSE over POST（MCP 场景）用fetch流式解析或协议 SDK，别在浏览器里硬造。
• 二进制数据不走 SSE：音视频等二进制流改用StreamingResponse（Stream Data 指南）。
• 边缘缓存旁路：no-cache已设置，但 CDN 的缓存规则可能覆盖它，确认边缘节点对该路径 no-store 或绕过。

结语

FastAPI 的 SSE 把"流式推送"压缩成一个yield和一个响应类：yield决定推什么，EventSourceResponse决定怎么推。返回类型标注给你校验和文档，ServerSentEvent给你event/id/retry的完整控制面，Last-Event-ID续传与 15 秒心跳、代理头这些细节框架已经内置。剩下的工作只剩把部署链路上的缓冲和超时对齐好。

延伸阅读（均为仓库内文件）：

• SSE 核心模型与格式化实现：fastapi/sse.py
• 路由层 SSE 检测、序列化与心跳注入：fastapi/routing.py
• 覆盖全部行为的测试：tests/test_sse.py
• 全部可运行示例：docs_src/server_sent_events/
• 逐行 JSON 流式输出：Stream JSON Lines 教程
• 流式传输任意（含二进制）数据：Stream Data 进阶指南
• 通用编码器：encoder 教程

【免费下载链接】fastapiFastAPI framework, high performance, easy to learn, fast to code, ready for production项目地址: https://gitcode.com/GitHub_Trending/fa/fastapi