FastAPI SSE 参考指南EventSourceResponse与ServerSentEvent全解【免费下载链接】fastapiFastAPI framework, high performance, easy to learn, fast to code, ready for production项目地址: https://gitcode.com/GitHub_Trending/fa/fastapi本文以 FastAPI 参考文档fastapi.sse模块为核心系统讲解 Server-Sent EventsSSE流式响应体系如何用EventSourceResponse以text/event-stream媒体类型输出事件流如何用ServerSentEvent模型精确控制data、event、id、retry、comment五个 SSE 线协议字段并深入源码剖析线格式编码函数format_sse_event、字段校验规则与内置最佳实践心跳 ping、禁用缓存、禁用代理缓冲。读完后你能完整掌握 SSE 端点的编写、校验约束、断线重连Last-Event-ID实现方式以及其背后的路由层编码逻辑与 OpenAPI 文档生成机制。一、模块概览fastapi.sse提供什么要流式输出 Server-Sent EventsSSE在path operation function路径操作函数中使用yield并设置response_classEventSourceResponse。如果还需要设置event、id、retry、comment等 SSE 字段则yieldServerSentEvent对象而不是普通数据。两者都可以直接从fastapi.sse 导入from fastapi.sse import EventSourceResponse, ServerSentEvent从源码结构看fastapi/sse.py 模块共暴露四组关键构件构件类型职责EventSourceResponse类StreamingResponse子类标记 SSE 响应设置Content-Type: text/event-streamServerSentEvent类PydanticBaseModel描述单条 SSE 事件的全部字段并做协议级校验format_sse_event函数将预序列化数据拼装为 SSE 线格式字节流KEEPALIVE_COMMENT/_PING_INTERVAL常量心跳注释: ping\n\n与空闲 ping 间隔默认 15 秒SSE 能力自 FastAPI 0.135.0 起提供见 SSE 教程文档 的版本标注。二、EventSourceResponseSSE 的响应载体EventSourceResponse在 fastapi/sse.py#L20-L33 中定义源码非常精简class EventSourceResponse(StreamingResponse): media_type text/event-stream其设计意图在 docstring 中写得很明确它作为response_classEventSourceResponse用在带yield的路径操作上用于启用 SSE 响应兼容任意 HTTP 方法GET、POST等因此适用于像 MCP 这类通过POST流式返回 SSE 的协议实际的编码逻辑位于 FastAPI 的路由层routing layer这个类本身主要是一个标记负责设置正确的Content-Type。最后一点值得注意EventSourceResponse并不自己做逐事件编码。它继承自 Starlette 的StreamingResponse逐条yield出来的对象Pydantic 模型、dict、ServerSentEvent由 FastAPI 路由层在流式序列化阶段统一处理。这一点可以从测试用例得到印证tests/test_sse.py 中同一个带类型注解的端点无论是async def、同步def还是无注解版本返回的响应头都一致assert response.headers[content-type] text/event-stream; charsetutf-8 assert response.headers[cache-control] no-cache assert response.headers[x-accel-buffering] no即text/event-stream媒体类型、Cache-Control: no-cache与X-Accel-Buffering: no三个响应头均由框架统一保证。三、最小可运行示例yield 出事件流参考文档给出的核心用法就是在路径操作中yieldresponse_classEventSourceResponse。以下是仓库示例代码 docs_src/server_sent_events/tutorial001_py310.py 的完整形式覆盖三种常见写法from collections.abc import AsyncIterable, Iterable from fastapi import FastAPI from fastapi.sse import EventSourceResponse from pydantic import BaseModel app FastAPI() class Item(BaseModel): name: str description: str | None items [ Item(namePlumbus, descriptionA multi-purpose household device.), Item(namePortal Gun, descriptionA portal opening device.), Item(nameMeeseeks Box, descriptionA box that summons a Meeseeks.), ] app.get(/items/stream, response_classEventSourceResponse) async def sse_items() - AsyncIterable[Item]: for item in items: yield item app.get(/items/stream-no-async, response_classEventSourceResponse) def sse_items_no_async() - Iterable[Item]: for item in items: yield item app.get(/items/stream-no-annotation, response_classEventSourceResponse) async def sse_items_no_annotation(): for item in items: yield item三种写法的差异与要点async defAsyncIterable[Item]推荐写法。声明 Pydantic 模型返回类型后FastAPI 会用它对每条 yield 的数据做校验、序列化并生成 OpenAPI 文档且由 Pydantic 在 Rust 侧执行序列化性能显著更高。普通defIterable[Item]同步生成器同样可用FastAPI 会确保它在后台正确运行不阻塞事件循环。注意此时正确的类型注解是Iterable[Item]而非AsyncIterable[Item]。省略返回类型FastAPI 会退回使用jsonable_encoder对数据做转换后发送但放弃逐条 Pydantic 校验OpenAPI 文档中的事件 schema 也会缺失。yield出的每个普通对象Pydantic 模型、dict 等都会被编码为 JSON 并放入 SSE 事件的data:字段。线上输出形如data: {name:Plumbus,description:A multi-purpose household device.} data: {name:Portal Gun,description:A portal opening device.}四、ServerSentEvent字段参考含校验规则当需要设置event、id、retry、comment等 SSE 字段时yieldServerSentEvent对象即可。它是 fastapi/sse.py#L52-L156 中定义的 Pydantic 模型六个字段及约束如下字段类型默认值约束与行为dataAnyNone事件载荷可为任意可 JSON 序列化值Pydantic 模型、dict、list、字符串、数字等。始终序列化为 JSON——即使是纯字符串datahello在网络上输出为data: hello带引号。与raw_data互斥raw_datastr \| NoneNone不做 JSON 编码原样放入data:字段。适合发送预格式化文本、日志行、HTML 片段、CSV 行或[DONE]这类哨兵值。与data互斥eventstr \| NoneNone事件类型名浏览器端对应addEventListener(event, ...)。省略时浏览器按通用message事件分发。必须单行不允许\r/\n见_check_event_single_line校验器fastapi/sse.py#L42-L43idstr \| NoneNone事件 ID。浏览器自动重连时会将其作为Last-Event-ID请求头回传。必须单行且不得包含空字符\0见_check_id_validfastapi/sse.py#L46-L49retryint \| NoneNone重连等待时间毫秒告知浏览器断线后多久重连。必须为非负整数Field(ge0)浮点数会被拒绝commentstr \| NoneNone注释行。线格式中以:前缀发送EventSource客户端会忽略。常用于 keep-alive ping防止代理/负载均衡器超时断开连接其中data与raw_data的互斥性由模型级校验器强制fastapi/sse.py#L148-L156model_validator(modeafter) def _check_data_exclusive(self) - ServerSentEvent: if self.data is not None and self.raw_data is not None: raise ValueError( Cannot set both data and raw_data on the same ServerSentEvent. Use data for JSON-serialized payloads or raw_data for pre-formatted strings. ) return self这些约束在 tests/test_sse.py 中有对应的逐项验证id含空字符抛错test_server_sent_event_null_id_rejected、event/id含换行抛SSE event must be a single linetest_server_sent_event_single_line_fields_reject_newlines、retry-1与retry1.5均被拒绝、data与raw_data同时设置抛Cannot set both错误。组合示例带完整字段的流参考文档对应的示例 docs_src/server_sent_events/tutorial002_py310.py 展示了注释 带event/id/retry的数据事件组合from collections.abc import AsyncIterable from fastapi import FastAPI from fastapi.sse import EventSourceResponse, ServerSentEvent from pydantic import BaseModel app FastAPI() class Item(BaseModel): name: str price: float items [ Item(namePlumbus, price32.99), Item(namePortal Gun, price999.99), Item(nameMeeseeks Box, price49.99), ] app.get(/items/stream, response_classEventSourceResponse) async def stream_items() - AsyncIterable[ServerSentEvent]: yield ServerSentEvent(commentstream of item updates) for i, item in enumerate(items): yield ServerSentEvent(dataitem, eventitem_update, idstr(i 1), retry5000)普通对象与ServerSentEvent也可以混用——测试文件中的/items/stream-mixed端点先 yield Pydantic 模型再 yield 一条ServerSentEvent(datacustom-event, eventspecial)最后再 yield 模型全部正常工作。原始字符串raw_data需要发送不做 JSON 编码的数据时使用raw_data。例如流式发送日志行示例来自 docs_src/server_sent_events/tutorial003_py310.pyapp.get(/logs/stream, response_classEventSourceResponse) async def stream_logs() - AsyncIterable[ServerSentEvent]: logs [ 2025-01-01 INFO Application started, 2025-01-01 DEBUG Connected to database, 2025-01-01 WARN High memory usage detected, ] for log_line in logs: yield ServerSentEvent(raw_datalog_line)此时线上输出为data: 2025-01-01 INFO Application started无 JSON 引号与data2025-01-01 INFO Application started产生的data: 2025-01-01 INFO Application started形成鲜明对比。raw_data也常用于发送[DONE]哨兵值等结束标记。五、线格式实现format_sse_event如何编码SSE 线格式wire format的编码集中在format_sse_event函数fastapi/sse.py#L159-L237。它接收已序列化的数据字符串按固定顺序拼装各字段结果始终以\n\n事件终止符结尾def format_sse_event( *, data_str: str | None None, # 预序列化后的 data 字段 event: str | None None, # event: 字段 id: str | None None, # id: 字段 retry: int | None None, # retry: 字段毫秒 comment: str | None None, # 注释行: 前缀 ) - bytes: lines: list[str] [] if comment is not None: for line in _split_sse_lines(comment): lines.append(f: {line}) if event is not None: lines.append(fevent: {event}) if data_str is not None: for line in _split_sse_lines(data_str): lines.append(fdata: {line}) if id is not None: lines.append(fid: {id}) if retry is not None: lines.append(fretry: {retry}) lines.append() lines.append() return \n.join(lines).encode(utf-8)两个实现细节值得理解多行数据如何拆分_split_sse_linesfastapi/sse.py#L159-L162只按 SSE 规范的行终止符\n、\r\n、\r拆分并保留尾部空串。因此包含换行的数据会被拆成多条data:行——这正是 SSE 规范表达多行载荷的标准方式。tests/test_sse.py 的参数化用例精确覆盖了这些边界输入data_str输出bytesHello\nbdata: Hello\ndata: \n\nHello\n\nbdata: Hello\ndata: \ndata: \n\n\nbdata: \ndata: \n\nHello\r\nWorldbdata: Hello\ndata: World\n\nbdata: \n\n注意A\u2028BUnicode 行分隔符与A\vB垂直制表符不会被拆行——\u2028、\v不是 SSE 规范的行终止符会原样进入单条data:行。心跳注释模块末尾定义了 keep-alive 常量fastapi/sse.py#L236-L241# Keep-alive comment, per the SSE spec recommendation KEEPALIVE_COMMENT b: ping\n\n # Seconds between keep-alive pings when a generator is idle. _PING_INTERVAL: float 15.0当生成器空闲超过_PING_INTERVAL默认 15 秒没有产出任何消息时路由层会自动插入: ping注释行。测试 tests/test_sse.py 的test_keepalive_ping_async/test_keepalive_ping_sync将间隔 monkeypatch 到 0.05 秒验证两个数据事件之间确实出现: ping\n而快速产出数据的流test_no_keepalive_when_fast则不会出现 ping。六、内置最佳实践Technical DetailsFastAPI 默认实现了若干 SSE 最佳实践无需任何额外配置同样来自 SSE 教程文档 并在源码与测试中可验证每 15 秒发送 keep aliveping注释当期间没有任何消息时防止部分代理关闭连接——这是 HTML 规范Server-Sent Events章节Authoring notes的建议对应源码常量KEEPALIVE_COMMENT b: ping\n\n与_PING_INTERVAL 15.0设置Cache-Control: no-cache响应头防止流被缓存设置X-Accel-Buffering: no响应头防止 Nginx 等代理对响应做缓冲确保事件实时下发。三条响应头的断言见 tests/test_sse.py#L118-L120。七、进阶用法断线重连利用Last-Event-ID浏览器在连接断开后自动重连时会把最后收到的id作为Last-Event-ID请求头发回。将其声明为 Header 参数即可实现从断点续流示例来自 docs_src/server_sent_events/tutorial004_py310.pyfrom collections.abc import AsyncIterable from typing import Annotated from fastapi import FastAPI, Header from fastapi.sse import EventSourceResponse, ServerSentEvent from pydantic import BaseModel app FastAPI() class Item(BaseModel): name: str price: float items [ Item(namePlumbus, price32.99), Item(namePortal Gun, price999.99), Item(nameMeeseeks Box, price49.99), ] app.get(/items/stream, response_classEventSourceResponse) async def stream_items( 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, item in enumerate(items): if i start: continue yield ServerSentEvent(dataitem, idstr(i))注意ServerSentEvent.id的校验约束单行、无空字符正是为了保障该值能安全地走 HTTP 头往返。SSE over POSTSSE 不仅限于GET兼容任意 HTTP 方法。这对 MCP 这类通过POST流式返回 SSE 的协议尤为重要示例来自 docs_src/server_sent_events/tutorial005_py310.pyfrom 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]: words prompt.text.split() for word in words: yield ServerSentEvent(dataword, eventtoken) yield ServerSentEvent(raw_data[DONE], eventdone)POST 场景的端到端验证见 tests/test_sse.py 的test_post_method_sseclient.post(/items/stream-post)返回 200 且content-type为text/event-stream; charsetutf-8。八、OpenAPI 文档中的 SSE 表达从源码结构看fastapi/sse.py#L7-L17 定义了与 OpenAPI 3.2 规范对齐的 SSE 事件 schema规范 4.14.4 节 Special Considerations for Server-Sent Events_SSE_EVENT_SCHEMA: dict[str, Any] { type: object, properties: { data: {type: string}, event: {type: string}, id: {type: string}, retry: {type: integer, minimum: 0}, }, }当端点声明了具体类型如AsyncIterable[Item]时OpenAPI 文档会把每个流式项包装进itemSchemadata字段携带contentMediaType: application/json与指向该模型 schema 的$ref。tests/test_sse.py 的test_sse_router_typed_openapi_schema与test_default_response_class_on_app_openapi_schema断言了完整结构content: { text/event-stream: { itemSchema: { type: object, properties: { data: { type: string, contentMediaType: application/json, contentSchema: {$ref: #/components/schemas/Item} }, event: {type: string}, id: {type: string}, retry: {type: integer, minimum: 0} }, required: [data] } } }该测试还覆盖了default_response_classEventSourceResponse设在FastAPI()或父级APIRouter上的场景test_default_response_class_on_app_stream、test_default_response_class_on_parent_router_openapi_schema——即在应用/路由级别一次性指定 SSE 默认响应类后子路由中的yield端点同样获得正确的媒体类型与 OpenAPI schema。九、参考路径汇总资源相对路径SSE 模块源码EventSourceResponse、ServerSentEvent、format_sse_eventfastapi/sse.py官方参考页本页对应的仓库文档docs/en/docs/reference/sse.mdSSE 教程完整用法与最佳实践docs/en/docs/tutorial/server-sent-events.md示例代码基础流 / 字段组合 / raw_data / Last-Event-ID / POSTdocs_src/server_sent_events/tutorial001_py310.py、tutorial002、tutorial003、tutorial004、tutorial005端到端测试线格式、校验、心跳、POST、OpenAPI schematests/test_sse.py适用前提与限制小结fastapi.sse的 SSE 支持自 FastAPI 0.135.0 起提供data恒为 JSON 序列化字符串带引号需要原样输出时必须显式使用raw_dataevent/id必须单行、id禁止空字符、retry必须为非负整数这些约束由 Pydantic 校验在构造ServerSentEvent时即抛出ValueError心跳 ping 间隔在源码中默认为 15 秒_PING_INTERVAL。【免费下载链接】fastapiFastAPI framework, high performance, easy to learn, fast to code, ready for production项目地址: https://gitcode.com/GitHub_Trending/fa/fastapi创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考