Responses and Data Streaming

To stream data to clients or manage complex JSON serialization in this project, you use specialized response classes like EventSourceResponse and utilities like jsonable_encoder.

Stream Server-Sent Events (SSE)

To stream data using the Server-Sent Events protocol, use EventSourceResponse as the response_class and yield items from your path operation.

from collections.abc import AsyncIterable
from fastapi import FastAPI
from fastapi.responses import EventSourceResponse
from pydantic import BaseModel

class Item(BaseModel):
    name: str
    description: str | None = None

app = FastAPI()

@app.get("/items/stream", response_class=EventSourceResponse)
async def sse_items() -> AsyncIterable[Item]:
    items = [
        Item(name="Plumbus", description="A multi-purpose household device."),
        Item(name="Portal Gun", description="A portal opening device."),
    ]
    for item in items:
        yield item

When you yield a plain object or a Pydantic model:

• FastAPI automatically serializes the item to JSON.
• It wraps the result in the SSE data: field.
• It sets the Content-Type to text/event-stream.
• It includes proxy-friendly headers like X-Accel-Buffering: no.

Customize SSE Events

Use the ServerSentEvent class to control metadata such as event names, IDs, and retry intervals.

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

app = FastAPI()

@app.get("/items/stream-sse-event", response_class=EventSourceResponse)
async def sse_items_event():
    # Send a named event with an ID
    yield ServerSentEvent(data={"key": "value"}, event="json-data", id="1")
    
    # Send a comment (useful for manual keep-alive)
    yield ServerSentEvent(comment="just a comment")
    
    # Tell the client to wait 5 seconds before reconnecting
    yield ServerSentEvent(data="retry-test", retry=5000)

Key Fields in ServerSentEvent

• data: The payload. Can be any JSON-serializable object. Strings are JSON-encoded (quoted).
• event: The event type name. Browser clients can listen for this specifically using addEventListener(event, ...).
• id: The event ID. The browser sends this back in the Last-Event-ID header upon reconnection.
• retry: Reconnection time in milliseconds.
• comment: Lines starting with : that are ignored by standard clients.

Stream Raw Data

If you need to send pre-formatted text, HTML fragments, or CSV lines without JSON encoding, use the raw_data field of ServerSentEvent.

@app.get("/items/stream-raw", response_class=EventSourceResponse)
async def sse_items_raw():
    # Sent as 'data: plain text without quotes'
    yield ServerSentEvent(raw_data="plain text without quotes")
    
    # Sent as 'data: <div>html fragment</div>'
    yield ServerSentEvent(raw_data="<div>html fragment</div>", event="html")

[!WARNING] data and raw_data are mutually exclusive. You cannot set both on the same ServerSentEvent instance.

Manual JSON Serialization

Use jsonable_encoder to convert complex Python objects (Pydantic models, dataclasses, Enums, etc.) into JSON-compatible types like dict, list, or str. This is useful when you need to prepare data for a custom response or for storage in a database.

from fastapi.encoders import jsonable_encoder
from pydantic import BaseModel

class Item(BaseModel):
    name: str

item = Item(name="Plumbus")
# Returns a dict: {"name": "Plumbus"}
json_compatible_item = jsonable_encoder(item)

jsonable_encoder supports standard Pydantic parameters:

• include / exclude: Set of fields to include or exclude.
• by_alias: Whether to use field aliases (defaults to True).
• exclude_none: Whether to exclude fields with None values.

Troubleshooting

Keep-Alive Pings

FastAPI automatically sends a : ping comment every 15 seconds if the generator is idle. This prevents proxies and load balancers from timing out the connection. This interval is defined by _PING_INTERVAL in fastapi.routing.

Null Characters in IDs

The SSE specification forbids null characters (\0) in event IDs. ServerSentEvent will raise a ValueError if you attempt to use one.

Deprecated Response Classes

ORJSONResponse and UJSONResponse are deprecated in this codebase. FastAPI now serializes data directly to JSON bytes via Pydantic V2, which is more efficient than using these custom response classes. Use standard return type annotations or response_model instead.

SSE with POST

EventSourceResponse is compatible with POST requests, which is required for protocols like the Model Context Protocol (MCP).

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