FastAPI SSE 流式推送实战指南：3 步搭建支持断线重连的 Server-Sent Events 接口-平芜编程栈

FastAPI SSE 流式推送实战指南：3 步搭建支持断线重连的 Server-Sent Events 接口

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

FastAPI 从 0.135.0 版本起原生支持 Server-Sent Events（SSE）流式推送，你只需在生成器端点上加一个response_class=EventSourceResponse，就能把 AI 聊天逐 token 输出、订单状态实时更新、服务器日志转发这类"服务器主动往客户端推数据"的场景做掉，而且断线重连、心跳保活、代理防缓冲这些生产细节框架已经内置。

先看一个典型痛点：你在做一个 AI 对话后端，模型生成 300 个字大约要 5 秒。传统"请求—响应"模式下，前端必须干等 5 秒才能看到第一个字，用户体感就是"卡死了"。如果改为流式返回，每生成一小段就立刻推给浏览器，体验完全不同。SSE 正是为此设计的标准：它建立在普通 HTTP 之上，使用text/event-stream媒体类型，浏览器内置EventSourceAPI 可以直接消费，不需要任何第三方协议库。

上图用"排队等餐"打比方：流式接口的价值就是让客户端不用干等整份结果出炉，而是边生产边取走。下面按"先跑通 → 再进阶 → 后加固"的顺序展开。

1 分钟跑通的最小 SSE 接口

最小实现只做两件事：路径操作函数里用yield产出数据，路由上声明 SSE 响应类。下面演示推送订单状态变更：

from collections.abc import AsyncIterable from fastapi import FastAPI from fastapi.sse import EventSourceResponse from pydantic import BaseModel app = FastAPI() class OrderUpdate(BaseModel): order_id: str status: str @app.get("/orders/stream", response_class=EventSourceResponse) async def stream_orders() -> AsyncIterable[OrderUpdate]: for status in ("paid", "packing", "shipped"): yield OrderUpdate(order_id="SO-1001", status=status)

浏览器收到的线上报文大致长这样（空行分隔两个事件）：

data: {"order_id": "SO-1001", "status": "paid"} data: {"order_id": "SO-1001", "status": "packing"}

为什么这样写：

声明了AsyncIterable[OrderUpdate]返回类型后，FastAPI 会用 Pydantic 对每个yield出的对象做校验、OpenAPI 文档化和序列化，这是最推荐的工程写法；
如果你不写返回类型，端点也能跑，但产出对象只能走通用的jsonable_encoder兜底编码，失去了类型化校验和 schema 文档；
路由层检测到"生成器 +EventSourceResponse"的组合后，会把每一项自动编码为data:字段下发。

同步函数同样支持。把async def换成def即可，返回类型相应声明为Iterable[OrderUpdate]：

from collections.abc import Iterable @app.get("/orders/stream-sync", response_class=EventSourceResponse) def stream_orders_sync() -> Iterable[OrderUpdate]: for status in ("paid", "packing", "shipped"): yield OrderUpdate(order_id="SO-1001", status=status)

这里有个容易踩的点：同步生成器不是async的，所以返回类型应写Iterable[...]而不是AsyncIterable[...]。FastAPI 会把它放进线程池迭代，事件循环不会被阻塞，这也是官方示例 docs_src/server_sent_events/tutorial001_py310.py 里特意区分两种写法的原因。

三种数据形态：普通对象、字段化事件、原始文本

yield出去的东西决定推给客户端的报文形态，一共三种玩法，按需取用。

普通对象：只填 data 字段

上面订单示例就是这种：对象被 JSON 编码后放进data:字段，报文里没有event、id等其他字段，浏览器统一在message事件上接收。够用、最省事。

ServerSentEvent：控制 event / id / retry / comment

需要给事件"起名字"、编号或告诉浏览器多久重连一次时，改为yield一个ServerSentEvent（同样从fastapi.sse导入）。下面演示带事件类型与重连配置的商品价格推送：

from collections.abc import AsyncIterable from fastapi import FastAPI from fastapi.sse import EventSourceResponse, ServerSentEvent from pydantic import BaseModel app = FastAPI() class PriceTag(BaseModel): sku: str price: float @app.get("/prices/stream", response_class=EventSourceResponse) async def stream_prices() -> AsyncIterable[ServerSentEvent]: yield ServerSentEvent(comment="price feed started") for i, (sku, price) in enumerate([("WIDGET-A", 12.5), ("WIDGET-B", 99.0)]): yield ServerSentEvent(data=PriceTag(sku=sku, price=price), event="price_update", id=str(i + 1), retry=5000)

对应线上报文（开头的:行是注释，浏览器会忽略）：

: price feed started event: price_update data: {"sku": "WIDGET-A", "price": 12.5} id: 1 retry: 5000

ServerSentEvent各字段一览（定义在 fastapi/sse.py）：

字段	类型	说明
`data`	任意 JSON 可序列化值	事件负载，总是被 JSON 序列化——连字符串都会加引号，`data="hi"`在网络上呈现为`data: "hi"`
`raw_data`	`str \\| None`	原样写入`data:`字段、不做 JSON 编码的字符串，与`data`互斥
`event`	`str \\| None`	事件类型名，对应前端`addEventListener(event, ...)`；缺省时派发通用`message`事件；必须单行
`id`	`str \\| None`	事件编号，浏览器自动重连时通过`Last-Event-ID`请求头回传；必须单行且不能含空字符`\0`
`retry`	`int \\| None`	断线后重连的等待时长，单位毫秒，必须是非负整数
`comment`	`str \\| None`	注释行，线上以`:`开头，客户端忽略，常用于心跳

这些约束不是文档建议而是硬校验：ServerSentEvent是 Pydantic 模型，内置了AfterValidator与model_validator，非法值在构造对象那一刻就抛ValueError，不会拖到响应阶段才炸。具体地：event或id里出现换行（\n、\r\n、\r）会抛错，因为会破坏 SSE 报文结构；id含\0抛错；retry为负数或浮点数抛错；同一个事件同时设置data和raw_data也抛错。这类行为在 tests/test_sse.py 中有逐项钉死的测试。

raw_data：按原文推送文本

如果data:内容本就不该是 JSON——比如日志行、CSV 行、HTML 片段，或者表示"流结束"的哨兵值[DONE]——就用raw_data：

@app.get("/logs/stream", response_class=EventSourceResponse) async def stream_logs() -> AsyncIterable[ServerSentEvent]: lines = [ "2026-10-30 06:00:01 INFO worker booted", "2026-10-30 06:00:02 WARN retrying db connection", ] for line in lines: yield ServerSentEvent(raw_data=line)

浏览器收到的将是data: 2026-10-30 06:00:01 INFO worker booted——没有 JSON 引号、空格原样保留。这是它与data的唯一区别：data永远 JSON 化，raw_data永远原样。二选一，不能都填。

生产加固：断线续传、心跳与代理缓冲

流式接口上线后最常遇到的三个生产问题是：连接断了数据丢失、长连接被中间设备掐断、代理层把流攒着不发。前两个 SSE 协议自带解法，第三个 FastAPI 已经帮你处理好了。

用 Last-Event-ID 实现断线续传

EventSource客户端断线后会自动重连，并把服务器最后下发的id值放进Last-Event-ID请求头。服务端读出这个头、算出起点、跳过已推过的事件即可：

from typing import Annotated from fastapi import Header @app.get("/orders/stream", response_class=EventSourceResponse) async def stream_orders_resumable( last_event_id: Annotated[int | None, Header()] = None, ) -> AsyncIterable[ServerSentEvent]: start = last_event_id + 1 if last_event_id is not None else 0 for i, status in enumerate(("paid", "packing", "shipped")): if i < start: continue yield ServerSentEvent( data=OrderUpdate(order_id="SO-1001", status=status), id=str(i) )

要点有二：一是靠 FastAPI 的Header依赖注入读取请求头，参数名last_event_id会自动按首字母大写映射到Last-Event-ID，无需手写request.headers；二是每个事件都必须带id，否则浏览器没有可回传的断点，续传无从谈起。客户端重连后，服务器只补发未送达的部分，避免重复消费。

空闲 15 秒自动发 keep-alive 心跳

代理和负载均衡器普遍会掐断长时间无数据的连接。FastAPI 在路由层内置了解法：生成器每空闲 15 秒（常量_PING_INTERVAL = 15.0，定义在 fastapi/sse.py）就自动插入一条注释心跳KEEPALIVE_COMMENT = b": ping\n\n"。对客户端来说注释被EventSource直接忽略，对中间设备来说"连接上有流量"，于是保活生效。

实现上并不简单：路由层用anyio内存对象流把"从生成器拉取并序列化事件"的任务与"按超时注入心跳"的任务解耦，接收端用anyio.fail_after(_PING_INTERVAL)包裹——超时即说明生成器空闲，随即下发: ping。这个设计对同步生成器（线程池执行）同样有效，快速产出时则绝不误发心跳。你什么都不用配，它开箱即用。

缓存与 Nginx 缓冲：两个响应头

实时流不能被缓存、不能被代理攒批。FastAPI 构建 SSE 响应时会显式设置两个头：

Cache-Control: no-cache：阻止浏览器与共享代理缓存这条实时流；
X-Accel-Buffering: no：Nginx 专用头，关闭其对响应的缓冲，让每个事件即时透传给客户端。

这两行对部署在 Nginx 之后的 SSE 接口至关重要，且同样无需你手动添加。另外EventSourceResponse本身只是继承自 StarletteStreamingResponse的轻量标记类，真正的工作是设置text/event-stream内容类型并让路由层识别 SSE 端点，因此它也可以作为FastAPI(default_response_class=EventSourceResponse)或APIRouter(default_response_class=...)的全局默认，让整组路由下的生成器端点默认按 SSE 输出。

SSE 不限于 GET：对接 MCP 等 POST 流式协议

SSE 端点可以挂在任意 HTTP 方法上。对 MCP 这类通过POST承载 SSE 流的协议来说，这条能力是刚需。下面演示接收 JSON 请求体、按词逐 token 推送、最后以哨兵值收尾的聊天流：

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

两个细节值得留意：

raw_data="[DONE]"这里展示了哨兵值用法——客户端识别到这个原文即知道流已结束，不需要 JSON 解析；
浏览器原生EventSourceAPI 只支持GET，所以 POST 型 SSE 需要配合基于fetch的第三方 SSE 客户端，这正是 MCP 协议的选择。

类型化 SSE 端点在 OpenAPI 文档中会以text/event-stream媒体类型加itemSchema的形式完整呈现，客户端代码生成和手工调试都更方便。可运行示例见 docs_src/server_sent_events/tutorial005_py310.py。

浏览器端接入：EventSource 三行起步

服务端就绪后，前端接入简单到几乎可以忽略。消费前面商品流的最小脚本：

const es = new EventSource("/prices/stream"); es.addEventListener("price_update", (event) => { // event 字段为 price_update 的事件在这里收到 console.log(JSON.parse(event.data)); }); es.onmessage = (event) => { // 未设置 event 字段时，浏览器派发 message 事件 console.log(event.data); }; es.onerror = () => { // 浏览器会按 retry 间隔自动重连， // 并自动携带 Last-Event-ID 请求头 };

配合服务端的事件编号（id:）与Last-Event-ID续传逻辑，你就拥有了一条自带自动重连、断点续传能力的实时通道——重连等待时长由retry字段控制，心跳由框架注入，前端一行保活代码都不用写。

能力速查与延伸阅读

需求	做法
最小 SSE 接口	生成器端点 +`response_class=EventSourceResponse`
让数据过 Pydantic 校验并生成 OpenAPI schema	返回类型声明`AsyncIterable[T]`/`Iterable[T]`
设置事件名 / 编号 / 重连间隔 / 心跳注释	`yield ServerSentEvent(event=..., id=..., retry=..., comment=...)`
推送非 JSON 原文（日志、`[DONE]`哨兵）	`ServerSentEvent(raw_data=...)`
断线续传	事件带`id`，服务端读`Last-Event-ID`头跳过已推事件
POST 承载 SSE（MCP 场景）	直接换`@app.post`，响应类不变
心跳 / 防缓存 / 关 Nginx 缓冲	框架内置：`: ping`（15s）、`no-cache`、`X-Accel-Buffering: no`