Server-Sent Events

FastAPI ships first-class support for Server-Sent Events through fastapi/sse.py and the routing layer. The encoding lives in the routing layer, not in the response class.

Public surface

fastapi/sse.py exposes:

Symbol	Purpose
EventSourceResponse	A StreamingResponse subclass with media_type = "text/event-stream". Use it as response_class=EventSourceResponse.
ServerSentEvent	A Pydantic BaseModel for a single event. Fields: data, raw_data, event, id, retry, comment. data and raw_data are mutually exclusive.
format_sse_event(*, data_str, event, id, retry, comment)	Build SSE wire-format bytes from already-serialized data.
KEEPALIVE_COMMENT	The default keep-alive comment frame: b": ping\n\n".
_PING_INTERVAL	Module-level float (15.0 seconds). Importable for tests.
_SSE_EVENT_SCHEMA	OpenAPI 3.2 SSE schema fragment used by the OpenAPI generator.

Wire format

Each event is one or more lines, terminated by an empty line. The serialiser mirrors the WHATWG SSE spec:

: optional comment (kept alive)
event: my-event
data: {"value": 42}
id: 17
retry: 3000

format_sse_event writes each line, splits multi-line data into multiple data: lines, ensures the terminator is \n\n, and returns UTF-8 bytes.

How a route becomes SSE

@app.get("/events", response_class=EventSourceResponse)
async def stream():
    yield {"value": 1}                # plain dict → JSON-encoded
    yield ServerSentEvent(event="m", data={"v": 2})  # explicit
    yield None                        # keep-alive tick

Inside fastapi/routing.py, request_response checks the route's response_class. When the class is EventSourceResponse (or a subclass), the inner ASGI app:

1. Calls the endpoint to get an async generator.
2. Wraps that generator in an iterator that:
   • Encodes plain dicts and Pydantic models via jsonable_encoder + json.dumps, then format_sse_event(data_str=...).
   • For ServerSentEvent instances, formats the explicit fields. data is always JSON-serialised; raw_data is sent as-is.
   • For None, sends KEEPALIVE_COMMENT.
3. Wraps the iterator in EventSourceResponse(...) so the Content-Type is text/event-stream.
4. Adds a periodic timeout-driven keep-alive every _PING_INTERVAL seconds when the generator stalls.

The stream is closed when the generator finishes or the client disconnects. Cancellation is propagated to the generator via the standard async-cancellation path; tests under tests/test_stream_cancellation.py cover this.

Why EventSourceResponse is mostly a marker

The class itself overrides nothing — its body is just media_type = "text/event-stream" and a docstring. The actual encoding lives in routing because the encoder needs access to the same machinery that handles response_model validation, jsonable_encoder, and async-generator cleanup. The class serves as a flag: "this route streams SSE; switch into the SSE branch."

OpenAPI integration

When the OpenAPI generator encounters a route with response_class=EventSourceResponse, it sets the response media type to text/event-stream and embeds _SSE_EVENT_SCHEMA (from fastapi/sse.py) under that media type. The schema follows the OpenAPI 3.2 SSE addendum:

{
  "type": "object",
  "properties": {
    "data": { "type": "string" },
    "event": { "type": "string" },
    "id": { "type": "string" },
    "retry": { "type": "integer", "minimum": 0 }
  }
}

If the route also declares a response_model, it is reflected in data (as a JSON-encoded payload). Otherwise the schema describes the SSE envelope only.

Compatibility with non-GET methods

EventSourceResponse works on any HTTP method, including POST. The class docstring calls out the use case: protocols like MCP stream SSE over POST. Most browser EventSource clients only use GET, but the framework does not enforce that — server frameworks (the MCP host, e.g.) can negotiate either method.

Tests

tests/test_sse.py exercises:

• Plain dict / Pydantic model / string yields.
• Explicit ServerSentEvent with all fields.
• Keep-alive timing (with _PING_INTERVAL monkey-patched to a small value).
• Cancellation behaviour.
• The OpenAPI fragment generation.

Integration points

• Routing — encoding is driven from fastapi/routing.py based on response_class.
• OpenAPI — _SSE_EVENT_SCHEMA is the OpenAPI 3.2 SSE schema fragment.
• Encoders — non-ServerSentEvent payloads pass through jsonable_encoder so Pydantic models, datetimes, dataclasses, etc. all serialize correctly.

Entry points for modification

• Adding new envelope fields: extend ServerSentEvent. Add the corresponding line in format_sse_event.
• Changing the keep-alive interval: monkey-patch _PING_INTERVAL (it is private but importable). The official knob is to send your own ServerSentEvent(comment="ping") and skip the default.
• Supporting binary frames: SSE is text-only; this would require a different response class. Use StreamingResponse directly instead.

See Routing for the encoding integration and OpenAPI generation for the schema flow.