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/fastapiFastAPI 从 0.135.0 版本起原生支持 Server-Sent EventsSSE流式推送你只需在生成器端点上加一个response_classEventSourceResponse就能把 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_classEventSourceResponse) async def stream_orders() - AsyncIterable[OrderUpdate]: for status in (paid, packing, shipped): yield OrderUpdate(order_idSO-1001, statusstatus)浏览器收到的线上报文大致长这样空行分隔两个事件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_classEventSourceResponse) def stream_orders_sync() - Iterable[OrderUpdate]: for status in (paid, packing, shipped): yield OrderUpdate(order_idSO-1001, statusstatus)这里有个容易踩的点同步生成器不是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_classEventSourceResponse) async def stream_prices() - AsyncIterable[ServerSentEvent]: yield ServerSentEvent(commentprice feed started) for i, (sku, price) in enumerate([(WIDGET-A, 12.5), (WIDGET-B, 99.0)]): yield ServerSentEvent(dataPriceTag(skusku, priceprice), eventprice_update, idstr(i 1), retry5000)对应线上报文开头的:行是注释浏览器会忽略: price feed started event: price_update data: {sku: WIDGET-A, price: 12.5} id: 1 retry: 5000ServerSentEvent各字段一览定义在 fastapi/sse.py字段类型说明data任意 JSON 可序列化值事件负载总是被 JSON 序列化——连字符串都会加引号datahi在网络上呈现为data: hiraw_datastr \| None原样写入data:字段、不做 JSON 编码的字符串与data互斥eventstr \| None事件类型名对应前端addEventListener(event, ...)缺省时派发通用message事件必须单行idstr \| None事件编号浏览器自动重连时通过Last-Event-ID请求头回传必须单行且不能含空字符\0retryint \| None断线后重连的等待时长单位毫秒必须是非负整数commentstr \| 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_dataapp.get(/logs/stream, response_classEventSourceResponse) 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_dataline)浏览器收到的将是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_classEventSourceResponse) 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( dataOrderUpdate(order_idSO-1001, statusstatus), idstr(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: noNginx 专用头关闭其对响应的缓冲让每个事件即时透传给客户端。这两行对部署在 Nginx 之后的 SSE 接口至关重要且同样无需你手动添加。另外EventSourceResponse本身只是继承自 StarletteStreamingResponse的轻量标记类真正的工作是设置text/event-stream内容类型并让路由层识别 SSE 端点因此它也可以作为FastAPI(default_response_classEventSourceResponse)或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_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)两个细节值得留意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_classEventSourceResponse让数据过 Pydantic 校验并生成 OpenAPI schema返回类型声明AsyncIterable[T]/Iterable[T]设置事件名 / 编号 / 重连间隔 / 心跳注释yield ServerSentEvent(event..., id..., retry..., comment...)推送非 JSON 原文日志、[DONE]哨兵ServerSentEvent(raw_data...)断线续传事件带id服务端读Last-Event-ID头跳过已推事件POST 承载 SSEMCP 场景直接换app.post响应类不变心跳 / 防缓存 / 关 Nginx 缓冲框架内置: ping15s、no-cache、X-Accel-Buffering: no延伸阅读均在仓库内逐行 JSON 流式输出SSE 的近亲媒体类型为application/x-ndjsonStream JSON Lines 教程流式传输任意数据含二进制SSE 不适用时Stream Data 进阶指南SSE 核心模型、字段校验与format_sse_event格式化实现fastapi/sse.py覆盖字段校验、心跳、raw_data、POST 流与 OpenAPI 的全量测试tests/test_sse.py【免费下载链接】fastapiFastAPI framework, high performance, easy to learn, fast to code, ready for production项目地址: https://gitcode.com/GitHub_Trending/fa/fastapi创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考