FastAPI SSE 参考指南：`EventSourceResponse` 与 `ServerSentEvent` 全解

本文以 FastAPI 参考文档 fastapi.sse 模块为核心，系统讲解 Server-Sent Events（SSE）流式响应体系：如何用 EventSourceResponse 以 text/event-stream 媒体类型输出事件流，如何用 ServerSentEvent 模型精确控制 data、event、id、retry、comment 五个 SSE 线协议字段，并深入源码剖析线格式编码函数 format_sse_event、字段校验规则与内置最佳实践（心跳 ping、禁用缓存、禁用代理缓冲）。读完后你能完整掌握 SSE 端点的编写、校验约束、断线重连（Last-Event-ID）实现方式，以及其背后的路由层编码逻辑与 OpenAPI 文档生成机制。

一、模块概览：fastapi.sse 提供什么

要流式输出 Server-Sent Events（SSE），在 path operation function（路径操作函数）中使用 yield，并设置 response_class=EventSourceResponse。如果还需要设置 event、``id、retry、comment等 SSE 字段，则yield ServerSentEvent对象而不是普通数据。两者都可以直接从fastapi.sse` 导入：

from fastapi.sse import EventSourceResponse, ServerSentEvent

从源码结构看，fastapi/sse.py 模块共暴露四组关键构件：

构件	类型	职责
EventSourceResponse	类（StreamingResponse 子类）	标记 SSE 响应，设置 Content-Type: text/event-stream
ServerSentEvent	类（Pydantic BaseModel）	描述单条 SSE 事件的全部字段并做协议级校验
format_sse_event	函数	将预序列化数据拼装为 SSE 线格式字节流
KEEPALIVE_COMMENT / _PING_INTERVAL	常量	心跳注释 : ping\n\n 与空闲 ping 间隔（默认 15 秒）

SSE 能力自 FastAPI 0.135.0 起提供（见 SSE 教程文档 的版本标注）。

二、EventSourceResponse：SSE 的响应载体

EventSourceResponse 在 fastapi/sse.py#L20-L33 中定义，源码非常精简：

class EventSourceResponse(StreamingResponse):
    media_type = "text/event-stream"

其设计意图在 docstring 中写得很明确：

• 它作为 response_class=EventSourceResponse 用在带 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; charset=utf-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 出事件流

参考文档给出的核心用法就是"在路径操作中 yield + response_class=EventSourceResponse"。以下是仓库示例代码 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(name="Plumbus", description="A multi-purpose household device."),
    Item(name="Portal Gun", description="A portal opening device."),
    Item(name="Meeseeks Box", description="A box that summons a Meeseeks."),
]


@app.get("/items/stream", response_class=EventSourceResponse)
async def sse_items() -> AsyncIterable[Item]:
    for item in items:
        yield item


@app.get("/items/stream-no-async", response_class=EventSourceResponse)
def sse_items_no_async() -> Iterable[Item]:
    for item in items:
        yield item


@app.get("/items/stream-no-annotation", response_class=EventSourceResponse)
async def sse_items_no_annotation():
    for item in items:
        yield item

三种写法的差异与要点：

1. async def + AsyncIterable[Item]：推荐写法。声明 Pydantic 模型返回类型后，FastAPI 会用它对每条 yield 的数据做校验、序列化并生成 OpenAPI 文档；且由 Pydantic 在 Rust 侧执行序列化，性能显著更高。
2. 普通 def + Iterable[Item]：同步生成器同样可用，FastAPI 会确保它在后台正确运行，不阻塞事件循环。注意此时正确的类型注解是 Iterable[Item] 而非 AsyncIterable[Item]。
3. 省略返回类型：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 字段时，yield ServerSentEvent 对象即可。它是 fastapi/sse.py#L52-L156 中定义的 Pydantic 模型，六个字段及约束如下：

字段	类型	默认值	约束与行为
data	Any	None	事件载荷，可为任意可 JSON 序列化值（Pydantic 模型、dict、list、字符串、数字等）。始终序列化为 JSON——即使是纯字符串：data="hello" 在网络上输出为 data: "hello"（带引号）。与 raw_data 互斥
raw_data	str | None	None	不做 JSON 编码，原样放入 data: 字段。适合发送预格式化文本、日志行、HTML 片段、CSV 行或 [DONE] 这类哨兵值。与 data 互斥
event	str | None	None	事件类型名，浏览器端对应 addEventListener(event, ...)。省略时浏览器按通用 message 事件分发。必须单行（不允许 \r/\n，见 _check_event_single_line 校验器，fastapi/sse.py#L42-L43）
id	str | None	None	事件 ID。浏览器自动重连时会将其作为 Last-Event-ID 请求头回传。必须单行，且不得包含空字符 \0（见 _check_id_valid，fastapi/sse.py#L46-L49）
retry	int | None	None	重连等待时间（毫秒），告知浏览器断线后多久重连。必须为非负整数（Field(ge=0)，浮点数会被拒绝）
comment	str | None	None	注释行。线格式中以 : 前缀发送，EventSource 客户端会忽略。常用于 keep-alive ping，防止代理/负载均衡器超时断开连接

其中 data 与 raw_data 的互斥性由模型级校验器强制（fastapi/sse.py#L148-L156）：

@model_validator(mode="after")
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 line（test_server_sent_event_single_line_fields_reject_newlines）、retry=-1 与 retry=1.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(name="Plumbus", price=32.99),
    Item(name="Portal Gun", price=999.99),
    Item(name="Meeseeks Box", price=49.99),
]


@app.get("/items/stream", response_class=EventSourceResponse)
async def stream_items() -> AsyncIterable[ServerSentEvent]:
    yield ServerSentEvent(comment="stream of item updates")
    for i, item in enumerate(items):
        yield ServerSentEvent(data=item, event="item_update", id=str(i + 1), retry=5000)

普通对象与 ServerSentEvent 也可以混用——测试文件中的 /items/stream-mixed 端点先 yield Pydantic 模型，再 yield 一条 ServerSentEvent(data="custom-event", event="special")，最后再 yield 模型，全部正常工作。

原始字符串：raw_data

需要发送不做 JSON 编码的数据时，使用 raw_data。例如流式发送日志行（示例来自 docs_src/server_sent_events/tutorial003_py310.py）：

@app.get("/logs/stream", response_class=EventSourceResponse)
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_data=log_line)

此时线上输出为 data: 2025-01-01 INFO Application started（无 JSON 引号），与 data="2025-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(f"event: {event}")
    if data_str is not None:
        for line in _split_sse_lines(data_str):
            lines.append(f"data: {line}")
    if id is not None:
        lines.append(f"id: {id}")
    if retry is not None:
        lines.append(f"retry: {retry}")
    lines.append("")
    lines.append("")
    return "\n".join(lines).encode("utf-8")

两个实现细节值得理解：

1. 多行数据如何拆分：_split_sse_lines（fastapi/sse.py#L159-L162）只按 SSE 规范的行终止符（\n、\r\n、\r）拆分并保留尾部空串。因此包含换行的数据会被拆成多条 data: 行——这正是 SSE 规范表达多行载荷的标准方式。tests/test_sse.py 的参数化用例精确覆盖了这些边界：

   输入 data_str	输出（bytes）
   "Hello\n"	b"data: Hello\ndata: \n\n"
   "Hello\n\n"	b"data: Hello\ndata: \ndata: \n\n"
   "\n"	b"data: \ndata: \n\n"
   "Hello\r\nWorld"	b"data: Hello\ndata: World\n\n"
   ""	b"data: \n\n"

   注意 "A\u2028B"（Unicode 行分隔符）与 "A\vB"（垂直制表符）不会被拆行——\u2028、\v 不是 SSE 规范的行终止符，会原样进入单条 data: 行。

2. 心跳注释：模块末尾定义了 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 Details）

FastAPI 默认实现了若干 SSE 最佳实践，无需任何额外配置（同样来自 SSE 教程文档 并在源码与测试中可验证）：

• 每 15 秒发送 "keep alive" ping 注释（当期间没有任何消息时），防止部分代理关闭连接——这是 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.py：

from 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(name="Plumbus", price=32.99),
    Item(name="Portal Gun", price=999.99),
    Item(name="Meeseeks Box", price=49.99),
]


@app.get("/items/stream", response_class=EventSourceResponse)
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(data=item, id=str(i))

注意 ServerSentEvent.id 的校验约束（单行、无空字符）正是为了保障该值能安全地走 HTTP 头往返。

SSE over POST

SSE 不仅限于 GET，兼容任意 HTTP 方法。这对 MCP 这类通过 POST 流式返回 SSE 的协议尤为重要，示例来自 docs_src/server_sent_events/tutorial005_py310.py：

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_class=EventSourceResponse)
async def stream_chat(prompt: Prompt) -> AsyncIterable[ServerSentEvent]:
    words = prompt.text.split()
    for word in words:
        yield ServerSentEvent(data=word, event="token")
    yield ServerSentEvent(raw_data="[DONE]", event="done")

POST 场景的端到端验证见 tests/test_sse.py 的 test_post_method_sse：client.post("/items/stream-post") 返回 200 且 content-type 为 text/event-stream; charset=utf-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 文档会把每个流式项包装进 itemSchema，data 字段携带 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_class=EventSourceResponse 设在 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_event）	fastapi/sse.py
官方参考页（本页对应的仓库文档）	docs/en/docs/reference/sse.md
SSE 教程（完整用法与最佳实践）	docs/en/docs/tutorial/server-sent-events.md
示例代码（基础流 / 字段组合 / raw_data / Last-Event-ID / POST）	docs_src/server_sent_events/tutorial001_py310.py、tutorial002、tutorial003、tutorial004、tutorial005
端到端测试（线格式、校验、心跳、POST、OpenAPI schema）	tests/test_sse.py

适用前提与限制小结：fastapi.sse 的 SSE 支持自 FastAPI 0.135.0 起提供；data 恒为 JSON 序列化（字符串带引号），需要原样输出时必须显式使用 raw_data；event/id 必须单行、id 禁止空字符、retry 必须为非负整数，这些约束由 Pydantic 校验在构造 ServerSentEvent 时即抛出 ValueError；心跳 ping 间隔在源码中默认为 15 秒（_PING_INTERVAL）。