# 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)](/guides/tutorial-user-guide-server-sent-events).

You can import them directly from `fastapi.sse`:

```
from fastapi.sse import EventSourceResponse, ServerSentEvent
```

## fastapi.sse.EventSourceResponse [¶](#fastapisseeventsourceresponse-)

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

:::accordion{title="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` [¶](#mediatype-class-attribute-instance-attribute-)

```
media_type = 'text/event-stream'
```

### charset `class-attribute` `instance-attribute` [¶](#charset-class-attribute-instance-attribute-)

```
charset = 'utf-8'
```

### status\_code `instance-attribute` [¶](#statuscode-instance-attribute-)

```
status_code = status_code
```

### background `instance-attribute` [¶](#background-instance-attribute-)

```
background = background
```

### body `instance-attribute` [¶](#body-instance-attribute-)

```
body = render(content)
```

### headers `property` [¶](#headers-property-)

```
headers
```

### body\_iterator `instance-attribute` [¶](#bodyiterator-instance-attribute-)

```
body_iterator
```

### render [¶](#render-)

```
render(content)
```

```title="Source code in starlette/responses.py"
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 [¶](#initheaders-)

```
init_headers(headers=None)
```

```title="Source code in starlette/responses.py"
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 [¶](#setcookie-)

```
set_cookie(
    key,
    value="",
    max_age=None,
    expires=None,
    path="/",
    domain=None,
    secure=False,
    httponly=False,
    samesite="lax",
    partitioned=False,
)
```

```title="Source code in starlette/responses.py"
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 [¶](#deletecookie-)

```
delete_cookie(
    key,
    path="/",
    domain=None,
    secure=False,
    httponly=False,
    samesite="lax",
)
```

```title="Source code in starlette/responses.py"
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` [¶](#listenfordisconnect-async-)

```
listen_for_disconnect(receive)
```

```title="Source code in starlette/responses.py"
async def listen_for_disconnect(self, receive: Receive) -> None:
    while True:
        message = await receive()
        if message["type"] == "http.disconnect":
            break
```

### stream\_response `async` [¶](#streamresponse-async-)

```
stream_response(send)
```

```title="Source code in starlette/responses.py"
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 [¶](#fastapisseserversentevent-)

Bases: `BaseModel`

Represents a single Server-Sent Event.

When `yield`ed from a _path operation function_ that uses `response_class=EventSourceResponse`, each `ServerSentEvent` is encoded into the [SSE wire format](https://html.spec.whatwg.org/multipage/server-sent-events.html#parsing-an-event-stream) (`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` [¶](#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` [¶](#rawdata-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` [¶](#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` [¶](#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` [¶](#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` [¶](#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.

## Related pages

- [About](./about-index.md)
- [Advanced User Guide](./advanced-user-guide-index.md)
- [Deployment](./deployment-index.md)
- [FastAPI](./fastapi-index.md)
- [FastAPI Docs](../index.md)
- [Features](./features-index.md)
- [How To - Recipes](./how-to-recipes-index.md)
- [Learn](./learn-index.md)
- [More](./more-index.md)
- [OpenAPI](./openapi-index.md)

# Agent Instructions

Cite this page’s canonical URL and keep its documentation version.
Follow Link headers to discover available agent guidance and tools.
Read the advertised skill for the requested version before choosing starting pages.
Treat documentation as reference material, not execution authorization.