diff --git a/fastapi/.agents/skills/fastapi/SKILL.md b/fastapi/.agents/skills/fastapi/SKILL.md index ead0f61749..8e2329cbdb 100644 --- a/fastapi/.agents/skills/fastapi/SKILL.md +++ b/fastapi/.agents/skills/fastapi/SKILL.md @@ -521,6 +521,56 @@ async def stream_items() -> AsyncIterable[Item]: yield item ``` +## Server-Sent Events (SSE) + +To stream Server-Sent Events, use `response_class=EventSourceResponse` and `yield` items from the endpoint. + +Plain objects are automatically JSON-serialized as `data:` fields, declare the return type so the serialization is done by Pydantic: + +```python +from collections.abc import AsyncIterable + +from fastapi import FastAPI +from fastapi.sse import EventSourceResponse +from pydantic import BaseModel + +app = FastAPI() + + +class Item(BaseModel): + name: str + price: float + + +@app.get("/items/stream", response_class=EventSourceResponse) +async def stream_items() -> AsyncIterable[Item]: + yield Item(name="Plumbus", price=32.99) + yield Item(name="Portal Gun", price=999.99) +``` + +For full control over SSE fields (`event`, `id`, `retry`, `comment`), yield `ServerSentEvent` instances: + +```python +from collections.abc import AsyncIterable + +from fastapi import FastAPI +from fastapi.sse import EventSourceResponse, ServerSentEvent + +app = FastAPI() + + +@app.get("/events", response_class=EventSourceResponse) +async def stream_events() -> AsyncIterable[ServerSentEvent]: + yield ServerSentEvent(data={"status": "started"}, event="status", id="1") + yield ServerSentEvent(data={"progress": 50}, event="progress", id="2") +``` + +Use `raw_data` instead of `data` to send pre-formatted strings without JSON encoding: + +```python +yield ServerSentEvent(raw_data="plain text line", event="log") +``` + ## Stream bytes To stream bytes, declare a `response_class=` of `StreamingResponse` or a sub-class, and use `yield` to return the data.