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 秒后砰地弹出一整块文字——这是 SSEServer-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_classEventSourceResponse) async def stream_answer() - AsyncIterable[Token]: for ch in SSE 让 AI 回答逐字送达: yield Token(textch) yield Token(text, finishTrue)三步拆解response_classEventSourceResponse从fastapi.sse导入。路由层检测到生成器端点 这个响应类的组合后会把端点标记为 SSE 流判断逻辑在 fastapi/routing.py响应头自动带上Content-Type: text/event-stream。yield每一项每个产出项都会被编码成 SSE 报文的data:字段发出去。标注AsyncIterable[Token]这是收益最大的一步。返回类型标注带来三个好处Pydantic 逐项校验不符合Token结构的产出会在校验阶段直接报ResponseValidationError而不是带病上线类型化序列化走model_dump_json快速路径比兜底编码更快OpenAPI 文档完整/docs里能看到该端点的text/event-stream媒体类型与单项 schemaitemSchema前端同学可以直接照着写消费代码。另外两种写法端点写法返回标注说明async defyieldAsyncIterable[Token]推荐写法校验 文档全开同步defyieldIterable[Token]端点被放进线程池迭代不会阻塞事件循环收益相同省略返回标注无每项走jsonable_encoder兜底编码不逐项校验也不进 OpenAPI 单项 schema还有一个偷懒选项FastAPI(default_response_classEventSourceResponse)可以把 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_classEventSourceResponse) async def stream_order_events(order_id: str) - AsyncIterable[ServerSentEvent]: for i, state in enumerate([paid, picking, shipped]): yield ServerSentEvent( dataOrderStatus(order_idorder_id, statestate), eventstatus_change, idstr(i 1), retry5000, )网络上传输的报文长这样事件之间以空行分隔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 编码——datahello线上是data: helloraw_data字符串原样写入data:字段、不做 JSON 编码与data互斥event单行事件类型名浏览器端对应addEventListener(event, ...)缺省则派发messageid单行、不含\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_dataapp.get(/deploy/logs, response_classEventSourceResponse) 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_dataline)浏览器收到的是与原文一致的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_classEventSourceResponse) 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( dataPoint(tsi * 5, cpu0.4 (i % 7) / 10), idstr(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 与 MCPSSE 不挑 HTTP 方法——EventSourceResponse对GET、POST等都生效。这一点在对接 MCP 这类协议时尤其重要它们正是通过POST承载 SSE 流的。一个接收 Prompt、按词回吐 token、末尾发结束标记的 POST 流式接口class Prompt(BaseModel): text: str app.post(/chat/stream, response_classEventSourceResponse) async def stream_chat(prompt: Prompt) - AsyncIterable[ServerSentEvent]: for word in prompt.text.split(): yield ServerSentEvent(dataword, eventtoken) yield ServerSentEvent(raw_data[DONE], eventdone)注意前端的限制浏览器原生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之间只会看到心跳、看不到数据。选型与延伸阅读一张表划清边界需求选型服务器 → 浏览器单向文本流要自动重连SSEEventSourceResponse浏览器与服务器双向实时通信聊天输入、协作编辑WebSocket服务到服务的逐行 JSON 文本流Stream JSON Lines流式传输任意/二进制数据音视频Stream Data 进阶指南延伸阅读均在当前仓库内核心实现fastapi/sse.pyServerSentEvent模型、format_sse_event线上格式拼装、心跳常量路由层序列化与心跳注入fastapi/routing.py全部可运行示例docs_src/server_sent_events/行为测试tests/test_sse.py通用编码兜底jsonable_encoderencoder 教程把最小端点 前端 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），仅供参考
