企业官网建设流程全解析

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），前端同学可以直接照着写消费代码。

另外两种写法：

还有一个"偷懒"选项：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）：

注意这些约束是构造期校验：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 响应构建部分）：

实现上，路由层用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之间只会看到心跳、看不到数据。

选型与延伸阅读

一张表划清边界：

延伸阅读（均在当前仓库内）：

• 核心实现：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