Skip to main content

Advanced SSE Event Control

FastAPI provides the ServerSentEvent class in fastapi.sse to give you granular control over the Server-Sent Events (SSE) protocol. While you can yield plain dictionaries or Pydantic models from an async generator to stream JSON data, using the ServerSentEvent model allows you to set specific SSE fields like event types, IDs, and reconnection intervals.

Data Serialization and Raw Payloads​

A key design choice in FastAPI's SSE implementation is how it handles the data: field. The ServerSentEvent model provides two mutually exclusive fields for the payload: data and raw_data.

JSON-Serialized Data​

The data field is designed for JSON-serializable content. FastAPI automatically serializes any value provided to this field to JSON. This includes strings, which are wrapped in quotes. For example, ServerSentEvent(data="hello") results in data: "hello" on the wire.

Raw Data for Non-JSON Payloads​

If you need to send pre-formatted text, HTML fragments, or other non-JSON content, use the raw_data field. This field accepts a string and places it directly into the data: field without any additional encoding or quoting.

FastAPI enforces this distinction using a Pydantic model validator:

@model_validator(mode="after")
def _check_data_exclusive(self) -> "ServerSentEvent":
    if self.data is not None and self.raw_data is not None:
        raise ValueError(
            "Cannot set both 'data' and 'raw_data' on the same "
            "ServerSentEvent. Use 'data' for JSON-serialized payloads "
            "or 'raw_data' for pre-formatted strings."
        )
    return self

This ensures that the wire format remains predictable and compliant with the SSE specification.

Controlling Client Behavior​

The ServerSentEvent model includes fields that map directly to the SSE protocol's control mechanisms, allowing you to manage how the client (usually a browser's EventSource) processes the stream.

Event Types​

The event field allows you to name the event. In the browser, this maps to specific event listeners. If omitted, the browser dispatches the event to the generic message listener.

# Client-side: eventSource.addEventListener("item_update", ...)
yield ServerSentEvent(data=item, event="item_update")

Reconnection and IDs​

The id and retry fields manage connection resilience:

  • id: An optional identifier for the event. If the connection drops, the browser will send this ID back in the Last-Event-ID HTTP header when it attempts to reconnect. FastAPI validates that this string does not contain null (\0) characters via the _check_id_no_null validator.
  • retry: A non-negative integer representing the reconnection time in milliseconds. It tells the client how long to wait before attempting to reconnect after a failure.
yield ServerSentEvent(
    data={"status": "processing"},
    id="step-10",
    retry=5000  # Reconnect after 5 seconds if lost
)

Keep-Alives and Comments​

SSE connections are long-lived, which can lead to timeouts by intermediate proxies or load balancers if no data is sent for a period. FastAPI addresses this in two ways:

  1. The comment field: You can manually send comments (lines starting with :) which are ignored by standard SSE clients but keep the connection active.
  2. Automatic Pings: FastAPI includes an internal mechanism that sends a : ping comment if the generator is idle. This is controlled by _PING_INTERVAL, which defaults to 15.0 seconds in fastapi/sse.py.

Wire Format Encoding​

The conversion from a ServerSentEvent object to the actual bytes sent over the network is handled by the format_sse_event utility function. This function ensures that multi-line data or comments are correctly formatted according to the spec:

def format_sse_event(
    *,
    data_str: str | None = None,
    event: str | None = None,
    id: str | None = None,
    retry: int | None = None,
    comment: str | None = None,
) -> bytes:
    lines: list[str] = []
    # ... formatting logic ...
    if data_str is not None:
        for line in data_str.splitlines():
            lines.append(f"data: {line}")
    # ...
    lines.append("")
    lines.append("")
    return "\n".join(lines).encode("utf-8")

By using EventSourceResponse as your response_class, FastAPI automatically utilizes these structures to ensure your stream is a valid text/event-stream.