Skip to main content
FastAPI Docs

Search documentation

Type to search this documentation.

On this pageOverview

Server-Sent Events - EventSourceResponse and ServerSentEvent¶

To stream Server-Sent Events (SSE), use yield in your path operation function and set response_class=EventSourceResponse.

If you need to set SSE fields like event, id, retry, or comment, you can yield ServerSentEvent objects instead of plain data.

Read more about it in the FastAPI docs for Server-Sent Events (SSE).

You can import them directly from fastapi.sse:

from fastapi.sse import EventSourceResponse, ServerSentEvent

fastapi.sse.EventSourceResponse

Section titled “fastapi.sse.EventSourceResponse ¶”
EventSourceResponse(
    content,
    status_code=200,
    headers=None,
    media_type=None,
    background=None,
)

Bases: StreamingResponse

Streaming response with text/event-stream media type.

Use as response_class=EventSourceResponse on a path operation that uses yield to enable Server Sent Events (SSE) responses.

Works with any HTTP method (GET, POST, etc.), which makes it compatible with protocols like MCP that stream SSE over POST.

The actual encoding logic lives in the FastAPI routing layer. This class serves mainly as a marker and sets the correct Content-Type.

Source code in starlette/responses.py
def __init__(
    self,
    content: ContentStream,
    status_code: int = 200,
    headers: Mapping[str, str] | None = None,
    media_type: str | None = None,
    background: BackgroundTask | None = None,
) -> None:
    if isinstance(content, AsyncIterable):
        self.body_iterator = content
    else:
        self.body_iterator = iterate_in_threadpool(content)
    self.status_code = status_code
    self.media_type = self.media_type if media_type is None else media_type
    self.background = background
    self.init_headers(headers)

media_type class-attribute instance-attribute

Section titled “media_type class-attribute instance-attribute ¶”
media_type = 'text/event-stream'

charset class-attribute instance-attribute

Section titled “charset class-attribute instance-attribute ¶”
charset = 'utf-8'

status_code instance-attribute

Section titled “status_code instance-attribute ¶”
status_code = status_code

background instance-attribute

Section titled “background instance-attribute ¶”
background = background

body instance-attribute

Section titled “body instance-attribute ¶”
body = render(content)

headers property

Section titled “headers property ¶”
headers

body_iterator instance-attribute

Section titled “body_iterator instance-attribute ¶”
body_iterator

render

Section titled “render ¶”
render(content)
code
def render(self, content: Any) -> bytes | memoryview:
    if content is None:
        return b""
    if isinstance(content, bytes | memoryview):
        return content
    return content.encode(self.charset)  # type: ignore

init_headers

Section titled “init_headers ¶”
init_headers(headers=None)
code
def init_headers(self, headers: Mapping[str, str] | None = None) -> None:
    if headers is None:
        raw_headers: list[tuple[bytes, bytes]] = []
        populate_content_length = True
        populate_content_type = True
    else:
        raw_headers = [(k.lower().encode("latin-1"), v.encode("latin-1")) for k, v in headers.items()]
        keys = [h[0] for h in raw_headers]
        populate_content_length = b"content-length" not in keys
        populate_content_type = b"content-type" not in keys

    body = getattr(self, "body", None)
    if (
        body is not None
        and populate_content_length
        and not (self.status_code < 200 or self.status_code in (204, 304))
    ):
        content_length = str(len(body))
        raw_headers.append((b"content-length", content_length.encode("latin-1")))

    content_type = self.media_type
    if content_type is not None and populate_content_type:
        if content_type.startswith("text/") and "charset=" not in content_type.lower():
            content_type += "; charset=" + self.charset
        raw_headers.append((b"content-type", content_type.encode("latin-1")))

    self.raw_headers = raw_headers

set_cookie

Section titled “set_cookie ¶”
set_cookie(
    key,
    value="",
    max_age=None,
    expires=None,
    path="/",
    domain=None,
    secure=False,
    httponly=False,
    samesite="lax",
    partitioned=False,
)
code
def set_cookie(
    self,
    key: str,
    value: str = "",
    max_age: int | None = None,
    expires: datetime | str | int | None = None,
    path: str | None = "/",
    domain: str | None = None,
    secure: bool = False,
    httponly: bool = False,
    samesite: Literal["lax", "strict", "none"] | None = "lax",
    partitioned: bool = False,
) -> None:
    cookie: http.cookies.BaseCookie[str] = http.cookies.SimpleCookie()
    cookie[key] = value
    if max_age is not None:
        cookie[key]["max-age"] = max_age
    if expires is not None:
        if isinstance(expires, datetime):
            cookie[key]["expires"] = format_datetime(expires, usegmt=True)
        else:
            cookie[key]["expires"] = expires
    if path is not None:
        cookie[key]["path"] = path
    if domain is not None:
        cookie[key]["domain"] = domain
    if secure:
        cookie[key]["secure"] = True
    if httponly:
        cookie[key]["httponly"] = True
    if samesite is not None:
        assert samesite.lower() in [
            "strict",
            "lax",
            "none",
        ], "samesite must be either 'strict', 'lax' or 'none'"
        cookie[key]["samesite"] = samesite
    if partitioned:
        if sys.version_info < (3, 14):
            raise ValueError("Partitioned cookies are only supported in Python 3.14 and above.")  # pragma: no cover
        cookie[key]["partitioned"] = True  # pragma: no cover

    cookie_val = cookie.output(header="").strip()
    self.raw_headers.append((b"set-cookie", cookie_val.encode("latin-1")))

delete_cookie

Section titled “delete_cookie ¶”
delete_cookie(
    key,
    path="/",
    domain=None,
    secure=False,
    httponly=False,
    samesite="lax",
)
code
def delete_cookie(
    self,
    key: str,
    path: str = "/",
    domain: str | None = None,
    secure: bool = False,
    httponly: bool = False,
    samesite: Literal["lax", "strict", "none"] | None = "lax",
) -> None:
    self.set_cookie(
        key,
        max_age=0,
        expires=0,
        path=path,
        domain=domain,
        secure=secure,
        httponly=httponly,
        samesite=samesite,
    )

listen_for_disconnect async

Section titled “listen_for_disconnect async ¶”
listen_for_disconnect(receive)
code
async def listen_for_disconnect(self, receive: Receive) -> None:
    while True:
        message = await receive()
        if message["type"] == "http.disconnect":
            break

stream_response async

Section titled “stream_response async ¶”
stream_response(send)
code
async def stream_response(self, send: Send) -> None:
    await send({"type": "http.response.start", "status": self.status_code, "headers": self.raw_headers})
    async for chunk in self.body_iterator:
        if not isinstance(chunk, bytes | memoryview):
            chunk = chunk.encode(self.charset)
        await send({"type": "http.response.body", "body": chunk, "more_body": True})

    await send({"type": "http.response.body", "body": b"", "more_body": False})

fastapi.sse.ServerSentEvent

Section titled “fastapi.sse.ServerSentEvent ¶”

Bases: BaseModel

Represents a single Server-Sent Event.

When yielded from a path operation function that uses response_class=EventSourceResponse, each ServerSentEvent is encoded into the SSE wire format (text/event-stream).

If you yield a plain object (dict, Pydantic model, etc.) instead, it is automatically JSON-encoded and sent as the data: field.

All data values including plain strings are JSON-serialized.

For example, data="hello" produces data: "hello" on the wire (with quotes).

data class-attribute instance-attribute

Section titled “data class-attribute instance-attribute ¶”
data = None

The event payload.

Can be any JSON-serializable value: a Pydantic model, dict, list, string, number, etc. It is always serialized to JSON: strings are quoted ("hello" becomes data: "hello" on the wire).

Mutually exclusive with raw_data.

raw_data class-attribute instance-attribute

Section titled “raw_data class-attribute instance-attribute ¶”
raw_data = None

Raw string to send as the data: field without JSON encoding.

Use this when you need to send pre-formatted text, HTML fragments, CSV lines, or any non-JSON payload. The string is placed directly into the data: field as-is.

Mutually exclusive with data.

event class-attribute instance-attribute

Section titled “event class-attribute instance-attribute ¶”
event = None

Optional event type name.

Maps to addEventListener(event, ...) on the browser. When omitted, the browser dispatches on the generic message event. Must be a single line.

id class-attribute instance-attribute

Section titled “id class-attribute instance-attribute ¶”
id = None

Optional event ID.

The browser sends this value back as the Last-Event-ID header on automatic reconnection. Must be a single line and must not contain null (\0) characters.

retry class-attribute instance-attribute

Section titled “retry class-attribute instance-attribute ¶”
retry = None

Optional reconnection time in milliseconds.

Tells the browser how long to wait before reconnecting after the connection is lost. Must be a non-negative integer.

comment class-attribute instance-attribute

Section titled “comment class-attribute instance-attribute ¶”
comment = None

Optional comment line(s).

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

Suggest an edit

Propose a replacement for this page. The site team reviews it before applying any changes.

Export
Documentation menu