Server-Sent Events (SSE)

Stream real-time updates to clients using Server-Sent Events in FastAPI

Server-Sent Events (SSE) provide a simple way to stream real-time updates from the server to the client over HTTP. Unlike WebSockets, SSE is unidirectional (server to client only) and uses the standard HTTP protocol.

Basic SSE Endpoint

Create an SSE endpoint using EventSourceResponse and yield:

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

When you yield plain objects (dicts, Pydantic models, etc.), FastAPI automatically JSON-encodes them and sends them as SSE data: fields.

ServerSentEvent Model

For fine-grained control over SSE events, use the ServerSentEvent model:

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
        )

ServerSentEvent Fields

data

The event payload. Can be any JSON-serializable value:

yield ServerSentEvent(data={"message": "Hello", "count": 42})
yield ServerSentEvent(data="hello")  # JSON-encoded as "hello" (with quotes)
yield ServerSentEvent(data=item)  # Pydantic model, auto-serialized

All data values, including plain strings, are JSON-serialized. For example, data="hello" produces data: "hello" on the wire (with quotes).

raw_data

Send raw string data without JSON encoding:

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)

Use raw_data when you need to send pre-formatted text, HTML fragments, CSV lines, or any non-JSON payload. The data and raw_data fields are mutually exclusive.

event

Optional event type name that maps to addEventListener(event, ...) on the browser:

yield ServerSentEvent(data=item, event="item_update")
yield ServerSentEvent(data=error, event="error")

When omitted, the browser dispatches on the generic message event.

id

Optional event ID for resuming streams:

for i, item in enumerate(items):
    yield ServerSentEvent(data=item, id=str(i))

The browser sends this value back as the Last-Event-ID header on automatic reconnection.

The id field must not contain null (\0) characters.

retry

Optional reconnection time in milliseconds:

yield ServerSentEvent(
    data=item,
    event="item_update",
    retry=5000  # Reconnect after 5 seconds
)

Tells the browser how long to wait before reconnecting after the connection is lost.

comment

Optional comment line(s) for keep-alive pings:

yield ServerSentEvent(comment="Keep-alive ping")

Comment lines start with : in the SSE wire format and are ignored by EventSource clients. Useful for preventing proxy/load-balancer timeouts.

FastAPI automatically sends keep-alive pings every 15 seconds to prevent connection timeouts. You don’t need to manually send comment events for this purpose.

Resuming Streams with Last-Event-ID

Handle reconnections by reading the Last-Event-ID header:

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

SSE with POST Requests

EventSourceResponse works with any HTTP method, including POST:

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

SSE over POST is useful for protocols like MCP (Model Context Protocol) that require streaming responses with request bodies.

Synchronous Generators

You can use regular (non-async) generator functions:

from collections.abc import Iterable
from fastapi import FastAPI
from fastapi.sse import EventSourceResponse
from pydantic import BaseModel

app = FastAPI()

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

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

Client-Side JavaScript

Connect to SSE endpoints using the EventSource API:

const eventSource = new EventSource("http://localhost:8000/items/stream");

// Listen to the generic "message" event
eventSource.onmessage = function(event) {
    const data = JSON.parse(event.data);
    console.log("Received:", data);
};

// Listen to custom events
eventSource.addEventListener("item_update", function(event) {
    const item = JSON.parse(event.data);
    console.log("Item update:", item);
});

// Handle errors
eventSource.onerror = function(error) {
    console.error("SSE Error:", error);
    eventSource.close();
};

// Close the connection
eventSource.close();

SSE with POST (JavaScript)

For POST requests, use fetch with a ReadableStream:

const response = await fetch("http://localhost:8000/chat/stream", {
    method: "POST",
    headers: {
        "Content-Type": "application/json",
    },
    body: JSON.stringify({ text: "Hello world" }),
});

const reader = response.body.getReader();
const decoder = new TextDecoder();

while (true) {
    const { done, value } = await reader.read();
    if (done) break;
    
    const chunk = decoder.decode(value);
    console.log("Received chunk:", chunk);
}

SSE Wire Format

SSE events follow the text/event-stream format:

event: item_update
data: {"name":"Plumbus","price":32.99}
id: 1
retry: 5000

Each event is terminated by a blank line (\n\n).

FastAPI automatically handles SSE wire format encoding. You don’t need to manually format events.

When to Use SSE vs WebSockets

Use Server-Sent Events when:

• You only need server-to-client communication
• You want automatic reconnection with event replay
• You prefer working over standard HTTP/HTTPS
• You’re building notifications, live feeds, or progress updates

Use WebSockets when:

• You need bidirectional communication
• You require lower latency
• You’re building chat applications or real-time collaboration tools

SSE has several advantages: automatic reconnection, event IDs for resuming streams, and compatibility with HTTP/2 multiplexing. It’s often the simpler choice for one-way streaming.