Skip to main content

Streaming with Server-Sent Events

To stream real-time, one-way updates from a server to a client, FastAPI provides the EventSourceResponse class. This implementation follows the Server-Sent Events (SSE) standard, allowing you to push data to clients over a long-lived HTTP connection.

Basic Streaming​

To enable SSE, set response_class=EventSourceResponse on your path operation and yield items from an AsyncIterable or Iterable. FastAPI automatically wraps yielded objects in the SSE data: field and encodes them as JSON.

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

app = FastAPI()

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

@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

In this example, yielding an Item model results in the following wire format:

data: {"name":"Plumbus","description":"A multi-purpose household device."}

data: {"name":"Portal Gun","description":"A portal opening device."}

Customizing Events with ServerSentEvent​

For more control over the SSE protocol fields—such as event types, IDs, or reconnection times—yield ServerSentEvent objects from fastapi.sse.

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="hello", event="greeting", id="1")
    
    # Send JSON data with a different event type
    yield ServerSentEvent(data={"status": "processing"}, event="update")
    
    # Set a custom retry interval (in milliseconds) for the client
    yield ServerSentEvent(data="reconnecting...", retry=5000)
    
    # Send a comment (ignored by most clients, useful for keep-alive)
    yield ServerSentEvent(comment="heartbeat")

ServerSentEvent Fields​

The ServerSentEvent model supports the following fields:

  • data: The event payload. Any JSON-serializable value (dict, list, Pydantic model, string). Strings are JSON-encoded (e.g., data="hello" becomes data: "hello").
  • raw_data: A string sent directly in the data: field without JSON encoding. Useful for HTML fragments or CSV.
  • event: The event type. Browsers use this to trigger specific addEventListener callbacks.
  • id: The event ID. Browsers send this back in the Last-Event-ID header if they reconnect.
  • retry: Reconnection time in milliseconds.
  • comment: A comment line starting with :.

Sending Raw Data​

If you need to send pre-formatted strings, HTML, or other non-JSON content, use the raw_data field. This prevents FastAPI from adding JSON quotes around your strings.

@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.

Keep-Alive Pings​

FastAPI automatically maintains the connection by sending keep-alive pings (SSE comments) when the generator is idle. This prevents proxies and load balancers from timing out the connection.

By default, FastAPI sends : ping\n\n every 15 seconds if no data has been yielded. This behavior is handled internally by the routing layer when EventSourceResponse is used.

Type Hinting and Validation​

When you provide a type hint like AsyncIterable[Item], FastAPI validates every yielded item against the Item model before serializing it.

If you yield a mix of plain objects and ServerSentEvent objects, FastAPI skips validation for the ServerSentEvent instances to allow flexible event streams:

@app.get("/items/stream-mixed", response_class=EventSourceResponse)
async def sse_items_mixed() -> AsyncIterable[Item]:
    # Validated against Item
    yield Item(name="Plumbus") 
    
    # Validation skipped for ServerSentEvent
    yield ServerSentEvent(data="custom-event", event="special")

Usage with Different HTTP Methods​

While SSE is traditionally used with GET, EventSourceResponse works with any HTTP method, including POST. This is useful for protocols like the Model Context Protocol (MCP) that stream SSE over POST requests.

@app.post("/items/stream-post", response_class=EventSourceResponse)
async def sse_items_post():
    yield {"message": "started"}