FastAPI SSE 流式推送完整指南:从 AI 逐字输出到断线自动续传

发布时间:2026/10/31 6:50:05
FastAPI SSE 流式推送完整指南:从 AI 逐字输出到断线自动续传

FastAPI SSE 流式推送完整指南:从 AI 逐字输出到断线自动续传

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

让 AI 的回答像打字机一样逐字蹦出来,而不是让用户盯着转圈 30 秒后"砰"地弹出一整块文字——这是 SSE(Server-Sent Events)流式推送最直观的用法。FastAPI 0.135.0+ 内置了EventSourceResponse,配合yield就能把任意接口改造成text/event-stream实时流。下面按"先跑通 → 接前端 → 生产加固 → 避坑"的工程路径走一遍。

先跑通:最小 SSE 端点 3 步搭好

想象一家餐厅:SSE 的思路是"做好一道上一道",而不是等所有菜做完再一次性端上桌。数据在循环里现算现发,客户端边收边渲染。

一个把回答逐 token 吐出去的最小端点,只有三件事要做:声明response_class、在端点里yield、给返回类型加上标注。

from collections.abc import AsyncIterable from fastapi import FastAPI from fastapi.sse import EventSourceResponse from pydantic import BaseModel app = FastAPI() class Token(BaseModel): text: str finish: bool = False @app.get("/chat/stream", response_class=EventSourceResponse) async def stream_answer() -> AsyncIterable[Token]: for ch in "SSE 让 AI 回答逐字送达": yield Token(text=ch) yield Token(text="", finish=True)

三步拆解:

  1. response_class=EventSourceResponse:从fastapi.sse导入。路由层检测到"生成器端点 + 这个响应类"的组合后,会把端点标记为 SSE 流(判断逻辑在 fastapi/routing.py),响应头自动带上Content-Type: text/event-stream。
  2. yield每一项:每个产出项都会被编码成 SSE 报文的data:字段发出去。
  3. 标注AsyncIterable[Token]:这是收益最大的一步。

返回类型标注带来三个好处:

  • Pydantic 逐项校验:不符合Token结构的产出会在校验阶段直接报ResponseValidationError,而不是带病上线;
  • 类型化序列化:走model_dump_json快速路径,比兜底编码更快;
  • OpenAPI 文档完整:/docs里能看到该端点的text/event-stream媒体类型与单项 schema(itemSchema),前端同学可以直接照着写消费代码。

另外两种写法:

端点写法返回标注说明
async def+yieldAsyncIterable[Token]推荐写法,校验 + 文档全开
同步def+yieldIterable[Token]端点被放进线程池迭代,不会阻塞事件循环,收益相同
省略返回标注无每项走jsonable_encoder兜底编码,不逐项校验,也不进 OpenAPI 单项 schema

还有一个"偷懒"选项:FastAPI(default_response_class=EventSourceResponse)可以把 SSE 设为整个应用(或某个APIRouter)的默认响应类,路由下所有生成器端点默认按 SSE 输出。

完整可运行的示例在 docs_src/server_sent_events/。

前端接流:EventSource 开箱即用

服务端发完,浏览器端用原生EventSource就能接住,不需要任何第三方协议库:

const es = new EventSource("/chat/stream"); es.onmessage = (event) => { // data 字段里是 JSON,字符串会带引号,所以先 parse const token = JSON.parse(event.data); document.getElementById("answer").textContent += token.text; if (token.finish) es.close(); }; es.onerror = () => { // 连接断开后浏览器会按 retry 间隔自动重连, // 且自动携带 Last-Event-ID 请求头(见后文) };

两个要点:

  • 端点没指定event字段时,浏览器统一派发到message事件(即onmessage);指定了event字段则要改用addEventListener,下一节马上用到;
  • event.data里是 JSON 文本(哪怕是纯字符串也带引号),记得JSON.parse。

到这里"服务端发、前端收"的闭环已经成立。

精控事件:event / id / retry / comment

普通对象yield出去只会落在data:字段里。想控制 SSE 报文的其余字段,就改为yield一个ServerSentEvent对象(同样从fastapi.sse导入)。

以"订单状态流转推送"为例:每次状态变化发一个带类型、带编号的事件,并把重连等待时间设为 5 秒。

from fastapi.sse import EventSourceResponse, ServerSentEvent class OrderStatus(BaseModel): order_id: str state: str @app.get("/orders/{order_id}/events", response_class=EventSourceResponse) async def stream_order_events(order_id: str) -> AsyncIterable[ServerSentEvent]: for i, state in enumerate(["paid", "picking", "shipped"]): yield ServerSentEvent( data=OrderStatus(order_id=order_id, state=state), event="status_change", id=str(i + 1), retry=5000, )

网络上传输的报文长这样(事件之间以空行分隔):

event: status_change data: {"order_id": "SO-10086", "state": "paid"} id: 1 retry: 5000 event: status_change data: {"order_id": "SO-10086", "state": "picking"} id: 2 retry: 5000

前端对应地按类型订阅:

es.addEventListener("status_change", (event) => { const status = JSON.parse(event.data); updateOrderBadge(status.order_id, status.state); });

各字段速查(模型定义见 fastapi/sse.py):

字段约束作用
data任意可 JSON 序列化值事件负载,总是JSON 编码——data="hello"线上是data: "hello"
raw_data字符串原样写入data:字段、不做 JSON 编码;与data互斥
event单行事件类型名,浏览器端对应addEventListener(event, ...);缺省则派发message
id单行、不含\0事件编号,浏览器自动重连时以Last-Event-ID头回传
retry非负整数(毫秒)告知浏览器断线后等多久再重连
comment单行注释行,线上以:开头,客户端忽略,常用作心跳

注意这些约束是构造期校验:event里塞换行、id里带\0、retry给负数或浮点数,ServerSentEvent(...)那一行就会直接抛ValueError,问题在写代码时暴露而不是在流到一半时。相关测试见 tests/test_sse.py。

原样文本:raw_data 与 [DONE] 哨兵值

data的"永远 JSON 编码"是特性也是麻烦:当你想发日志行、HTML 片段、CSV 行这类本来就是文本的东西时,被加上 JSON 引号反而碍事。这时用raw_data:

@app.get("/deploy/logs", response_class=EventSourceResponse) async def stream_deploy_logs() -> AsyncIterable[ServerSentEvent]: for line in [ "[09:41:02] build image ok", "[09:41:05] migrate db ok", "[09:41:08] rolling update finished", ]: yield ServerSentEvent(raw_data=line)

浏览器收到的是与原文一致的data: [09:41:02] build image ok,没有多余引号。

另一个高频用法是哨兵值:流式 AI 接口常以data=[DONE]作为"本次生成结束"的标记,客户端收到该标记后停止等待。用raw_data="[DONE]"一行就能实现,前端解析时先判断哨兵、再JSON.parse,不会出错。

记住一条红线:data与raw_data互斥,同一个事件两者都设会在构造时抛ValueError,不存在"同时发两份"的可能。

断线续传:Last-Event-ID 怎么接

EventSource的杀手锏是自动重连:断线后浏览器按retry间隔重新请求,并且把收到的最后一个id放进Last-Event-ID请求头。服务端只要读这个头,就能从断点接着发,而不是从头重播。

以实时指标流为例,给每个样本编上序号,重连时跳过已发的部分:

from typing import Annotated from fastapi import Header @app.get("/metrics/stream", response_class=EventSourceResponse) async def stream_metrics( 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 in range(100): if i < start: continue yield ServerSentEvent( data=Point(ts=i * 5, cpu=0.4 + (i % 7) / 10), id=str(i), )

三个关键点:

  • FastAPI 的Header()会把参数名last_event_id自动映射到请求头Last-Event-ID(下划线转连字符),不用手写request.headers;
  • 每条事件必须带id=,浏览器才有值可回传;
  • 生产环境里"跳过已发项"通常意味着按start游标查数据库或日志系统,而不是靠内存列表——但协议层面就是上面这几行。

配合前端零代码的自动重连,你就得到一条"断线自动补发、不重不漏"的推送通道。

生产加固:心跳、缓存与代理缓冲三件套

SSE 是长连接,"连接一直开着但没数据"是最危险的状态。FastAPI 路由层已内置三个开箱即用的处理(实现见 fastapi/routing.py 的 SSE 响应构建部分):

机制具体行为为什么需要
空闲心跳生成器超过 15 秒无产出时,自动插入注释报文: ping\n\n(常量KEEPALIVE_COMMENT、间隔_PING_INTERVAL定义在 fastapi/sse.py)代理和负载均衡器会掐断"看起来死了"的长连接,心跳保活
Cache-Control: no-cache显式写入响应头防止中间缓存对实时流做缓存,导致客户端拿到旧数据
X-Accel-Buffering: no显式写入响应头关闭 Nginx 等反向代理对该流的缓冲,数据即时透传

实现上,路由层用anyio内存对象流把"生成器迭代"和"心跳插入"拆成两个任务:生产任务从生成器拉取并序列化事件,转发任务用超时包裹接收操作,一旦空闲超时就把心跳注释插进字节流——这样即使同步def端点在跑线程池,心跳也不会被阻塞的__next__卡住。生产产出足够快时不会误发心跳,这套行为在 tests/test_sse.py 里有专门的测试覆盖。

你不需要写任何代码来获得这三个保障,只需要在部署后验证一下:用curl -i看到这三个头都在,就算合格。

进阶:SSE over POST 与 MCP

SSE 不挑 HTTP 方法——EventSourceResponse对GET、POST等都生效。这一点在对接 MCP 这类协议时尤其重要:它们正是通过POST承载 SSE 流的。

一个"接收 Prompt、按词回吐 token、末尾发结束标记"的 POST 流式接口:

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")

注意前端的限制:浏览器原生EventSource只支持GET,且不能带自定义请求头。因此 POST 场景(以及需要携带鉴权头的场景)要换用基于fetch的 SSE 客户端库逐行解析text/event-stream——这也是 MCP 客户端的做法。另外,类型化 SSE 端点在 OpenAPI 文档里会完整呈现text/event-stream媒体类型与单项 schema,配合客户端生成工具可以直接产出带流式解析的调用代码。

避坑与调试清单

上线前把这几条过一遍,能省下大量排查时间:

  • 用curl -N观测流:-N(等价于--no-buffer)关闭 curl 自身的输出缓冲,否则你会看到数据"攒一堆才吐",误以为是服务端问题。加-i可以顺便核对Cache-Control与X-Accel-Buffering两个头。
    curl -i -N http://127.0.0.1:8000/metrics/stream
  • 数据"一次性吐出":几乎总是代理缓冲。先确认X-Accel-Buffering: no在响应头里;若还有 CDN/网关层,需要在其配置中显式关闭缓冲或放行text/event-stream。
  • 收到的字符串带引号:这不是 bug,是data字段的 JSON 编码行为;要原文就用raw_data。
  • 构造期报错:retry=-1、多行event、含\0的id都会在ServerSentEvent(...)处立即抛ValueError——这是设计如此,别去"修"它,修数据。
  • Swagger UI 里看不到流:/docs的测试面板不适合观测长流,curl或浏览器控制台new EventSource(...)才是标准姿势。
  • EventSource无法带鉴权头:原生 API 发不了自定义Authorization,用 Cookie、query 参数或 POST +fetch客户端绕过。
  • 同步def端点里的阻塞调用:心跳注入不依赖线程池,但阻塞会拖慢数据产出,长时间卡在一次yield之间只会看到心跳、看不到数据。

选型与延伸阅读

一张表划清边界:

需求选型
服务器 → 浏览器单向文本流,要自动重连SSE(EventSourceResponse)
浏览器与服务器双向实时通信(聊天输入、协作编辑)WebSocket
服务到服务的逐行 JSON 文本流Stream JSON Lines
流式传输任意/二进制数据(音视频)Stream Data 进阶指南

延伸阅读(均在当前仓库内):

  • 核心实现:fastapi/sse.py(ServerSentEvent模型、format_sse_event线上格式拼装、心跳常量)
  • 路由层序列化与心跳注入:fastapi/routing.py
  • 全部可运行示例:docs_src/server_sent_events/
  • 行为测试:tests/test_sse.py
  • 通用编码兜底jsonable_encoder:encoder 教程

把"最小端点 + 前端 EventSource"当作骨架,ServerSentEvent各字段当作精细调节旋钮,Last-Event-ID当作可靠性保险,再加上默认自带的心跳与防缓冲头——一个能扛生产的 FastAPI SSE 流式推送接口就是这四件事。

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

创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考