服务器发送事件 (SSE)¶

你可以使用服务器发送事件 (SSE) 向客户端进行流式传输数据。

这类似于流式传输 JSON 行，但使用了 text/event-stream 格式，浏览器通过 EventSource API 原生支持该格式。

信息

在 FastAPI 0.135.0 版本中添加。

什么是服务器发送事件？¶

SSE 是一种通过 HTTP 将数据从服务器流式传输到客户端的标准。

每个事件都是一个小的文本块，包含诸如 data、event、id 和 retry 等“字段”，并以空行分隔。

它看起来像这样

data: {"name": "Portal Gun", "price": 999.99}

data: {"name": "Plumbus", "price": 32.99}

SSE 常用于 AI 聊天流、实时通知、日志和可观测性，以及其他服务器需要向客户端推送更新的场景。

提示

如果你想流式传输二进制数据（例如视频或音频），请查看高级指南：流式传输数据。

使用 FastAPI 进行 SSE 流式传输¶

要在 FastAPI 中进行 SSE 流式传输，请在路径操作函数中使用 yield，并设置 response_class=EventSourceResponse。

从 fastapi.sse 导入 EventSourceResponse

Python 3.10+

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

# Code below omitted 👇

👀 完整文件预览

Python 3.10+

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


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

每个生成 (yield) 的项目都会被编码为 JSON，并发送到 SSE 事件的 data: 字段中。

如果你将返回类型声明为 AsyncIterable[Item]，FastAPI 将使用它通过 Pydantic 对数据进行验证、生成文档和序列化。

Python 3.10+

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

# Code below omitted 👇

👀 完整文件预览

Python 3.10+

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


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

提示

由于 Pydantic 将在 Rust 端进行序列化，因此你会获得比不声明返回类型高得多的性能。

非异步路径操作函数¶

你也可以使用普通的 def 函数（不带 async），并以同样的方式使用 yield。

FastAPI 会确保它被正确运行，从而不会阻塞事件循环。

由于这种情况下函数不是异步的，正确的返回类型应该是 Iterable[Item]

Python 3.10+

# Code above omitted 👆

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

# Code below omitted 👇

👀 完整文件预览

Python 3.10+

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


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

无返回类型¶

你也可以省略返回类型。FastAPI 将使用 jsonable_encoder 来转换并发送数据。

Python 3.10+

# Code above omitted 👆

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

# Code below omitted 👇

👀 完整文件预览

Python 3.10+

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


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

`ServerSentEvent`¶

如果你需要设置 SSE 字段（如 event、id、retry 或 comment），可以生成 ServerSentEvent 对象，而不是普通数据。

从 fastapi.sse 导入 ServerSentEvent

Python 3.10+

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)

data 字段始终被编码为 JSON。你可以传递任何可以序列化为 JSON 的值，包括 Pydantic 模型。

原始数据¶

如果你需要发送不经 JSON 编码的数据，请使用 raw_data 而不是 data。

这对于发送预格式化的文本、日志行或特殊的“哨兵” (sentinel) 值（如 [DONE]）非常有用。

Python 3.10+

from collections.abc import AsyncIterable

from fastapi import FastAPI
from fastapi.sse import EventSourceResponse, ServerSentEvent

app = FastAPI()


@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 和 raw_data 是互斥的。每个 ServerSentEvent 只能设置其中一个。

使用 `Last-Event-ID` 恢复¶

当浏览器在连接断开后重新连接时，它会在 Last-Event-ID 请求头中发送最后接收到的 id。

你可以将其作为请求头参数读取，并用它从客户端断开的位置恢复流式传输。

Python 3.10+

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))

POST 请求的 SSE¶

SSE 适用于任何 HTTP 方法，不仅仅是 GET。

这对于像 MCP 这样在 POST 上流式传输 SSE 的协议非常有用。

Python 3.10+

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")

技术细节¶

FastAPI 开箱即用地实现了一些 SSE 最佳实践。

当没有任何消息时，每 15 秒发送一个“保持连接”的 ping 注释，以防止某些代理关闭连接，正如 HTML 规范：服务器发送事件中建议的那样。
设置 Cache-Control: no-cache 请求头以防止缓存流数据。
设置特殊请求头 X-Accel-Buffering: no 以防止像 Nginx 这样的代理进行缓冲。

你无需为此做任何配置，它开箱即用。🤓

服务器发送事件 (SSE)¶

什么是服务器发送事件？¶

使用 FastAPI 进行 SSE 流式传输¶

非异步 路径操作函数¶

无返回类型¶

ServerSentEvent¶

原始数据¶

使用 Last-Event-ID 恢复¶

POST 请求的 SSE¶

技术细节¶

非异步路径操作函数¶

`ServerSentEvent`¶

使用 `Last-Event-ID` 恢复¶