tiangolo.fastapi/fastapi/.agents/skills/fastapi/references/streaming.md

Streaming

Stream JSON Lines

To stream JSON Lines, declare the return type and use yield to return the data.

@app.get("/items/stream")
async def stream_items() -> AsyncIterable[Item]:
    for item in items:
        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:

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:

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:

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.

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from app.utils import read_image

app = FastAPI()


class PNGStreamingResponse(StreamingResponse):
    media_type = "image/png"

@app.get("/image", response_class=PNGStreamingResponse)
def stream_image_no_async_no_annotation():
    with read_image() as image_file:
        yield from image_file

prefer this over returning a StreamingResponse directly:

# DO NOT DO THIS

import anyio
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from app.utils import read_image

app = FastAPI()


class PNGStreamingResponse(StreamingResponse):
    media_type = "image/png"


@app.get("/")
async def main():
    return PNGStreamingResponse(read_image())