SSE Starlette

https://github.com/sysid/sse-starlette
Admin
SSE Starlette provides a production-ready, standards-compliant implementation of Server-Sent Events...
Tokens:30,670
Snippets:202
Trust Score:8.6
License:BSD-3-Clause
Update:1 month ago
Show doc for...
Tokens:
Raw
### Install dependencies and run with Python

Source: https://github.com/sysid/sse-starlette/blob/main/examples/README.md

Manually install required packages and run the example script directly with Python.

```bash
pip install sse-starlette fastapi uvicorn
python examples/01_basic_sse.py
```

--------------------------------

### Install sse-starlette and optional dependencies

Source: https://github.com/sysid/sse-starlette/blob/main/README.md

Install sse-starlette with pip or uv. Use the [examples] extra to run the examples, [examples,examples-db] for example 03 which needs SQLAlchemy and aiosqlite, and [uvicorn,granian,daphne] for recommended ASGI servers.

```bash
pip install sse-starlette
uv add sse-starlette

# To run the examples (fastapi, uvicorn, pydantic)
uv add sse-starlette[examples]

# Example 03 also needs the DB extras (sqlalchemy, aiosqlite)
uv add sse-starlette[examples,examples-db]

# Recommended ASGI server
uv add sse-starlette[uvicorn,granian,daphne]
```

--------------------------------

### Run example with uv

Source: https://github.com/sysid/sse-starlette/blob/main/examples/README.md

Run an example script using uv, which reads inline dependencies declared via PEP 723.

```bash
uv run examples/01_basic_sse.py
```

--------------------------------

### Run demonstration scripts

Source: https://github.com/sysid/sse-starlette/blob/main/README.md

Commands to run example demonstrations for basic patterns, production scenarios, and advanced patterns. Each script is executed with Python.

```bash
python examples/demonstrations/basic_patterns/client_disconnect.py
python examples/demonstrations/production_scenarios/load_simulations.py
python examples/demonstrations/advanced_patterns/error_recovery.py
```

--------------------------------

### Monkey-patch installation for uvicorn

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/api-reference/AppStatus.md

At import time, this code saves the original uvicorn Server.handle_exit and replaces it with AppStatus.handle_exit. If uvicorn is not installed, it logs a debug message.

```python
try:
    from uvicorn.main import Server

    AppStatus.original_handler = Server.handle_exit
    Server.handle_exit = AppStatus.handle_exit  # type: ignore
except ImportError:
    logger.debug(
        "Uvicorn not installed. Graceful shutdown on server termination disabled."
    )
```

--------------------------------

### ensure_bytes usage example

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/api-reference/sse-module.md

Shows how to use `ensure_bytes` with different input types. The example asserts that bytes are passed through, dicts are converted to SSE frames with the given separator, ServerSentEvent objects are encoded, and other values are wrapped as data. Note that the dict input is mutated in place by adding the 'sep' key.

```python
from sse_starlette.event import ensure_bytes
from sse_starlette import ServerSentEvent

assert ensure_bytes(b"raw\x00bytes", "\r\n") == b"raw\x00bytes"
assert ensure_bytes({"data": "hi"}, "\r\n") == b"data: hi\r\n\r\n"
assert ensure_bytes(ServerSentEvent(comment="c"), "\n") == b": c\n\n"
assert ensure_bytes(42, "\r\n") == b"data: 42\r\n\r\n"
```

--------------------------------

### __call__ ASGI entrypoint

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/api-reference/EventSourceResponse.md

ASGI entrypoint that implements the application contract. It rewrites send for WebSocket scopes, runs under collapse_excgroups(), and starts tasks in an anyio task group with cancel_on_finish supervision. Returns None and propagates SendTimeoutError or content generator exceptions.

```python
async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
```

--------------------------------

### Test endpoint with curl

Source: https://github.com/sysid/sse-starlette/blob/main/examples/README.md

Use curl to test the SSE endpoint. Replace <endpoint> with the specific endpoint from the example's docstring.

```bash
curl -N http://localhost:8000/<endpoint>
```

--------------------------------

### Run test commands

Source: https://github.com/sysid/sse-starlette/blob/main/README.md

Use these make targets to run the test suite. `make test-unit` runs only unit tests, `make test` runs unit and Docker integration tests, and `make test-experimentation` runs optional multi-consumer load tests that require the `experimentation` dependency group (installed via `uv sync --group experimentation`).

```bash
make test-unit              # Unit tests only (default dev install)
make test                   # Unit + docker integration tests
make test-experimentation   # Optional experimentation tests (multi-consumer load tests)
                            # Requires the `experimentation` dependency group:
                            #   uv sync --group experimentation
```

--------------------------------

### Basic ServerSentEvent construction example

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/api-reference/ServerSentEvent.md

Constructs a ServerSentEvent with data, event, id, and retry fields, then prints the encoded output. The retry value of 5000 indicates the client will retry after 5 seconds. The output shows the encoded bytes with CRLF line separators.

```python
from sse_starlette import ServerSentEvent

event = ServerSentEvent(
    data="Custom message",
    event="notification",
    id="msg-123",
    retry=5000,          # client will retry after 5 seconds
)

print(event.encode())
# b'id: msg-123\r\nevent: notification\r\ndata: Custom message\r\nretry: 5000\r\n\r\n'
```

--------------------------------

### Quick Start: SSE endpoint with EventSourceResponse

Source: https://github.com/sysid/sse-starlette/blob/main/README.md

Creates a Starlette application with an /events route that streams server-sent events. The generate_events async generator yields events every second, and EventSourceResponse handles the SSE protocol.

```python
import asyncio
from starlette.applications import Starlette
from starlette.routing import Route
from sse_starlette import EventSourceResponse

async def generate_events():
    for i in range(10):
        yield {"data": f"Event {i}"}
        await asyncio.sleep(1)

async def sse_endpoint(request):
    return EventSourceResponse(generate_events())

app = Starlette(routes=[Route("/events", sse_endpoint)])
```

--------------------------------

### handle_exit

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/api-reference/AppStatus.md

Handles server exit signal by setting the shutdown flag and chaining to the original uvicorn handler. Installed via monkey-patch at import time.

```APIDOC
## handle_exit

### Description
Handles server exit signal by setting the shutdown flag and chaining to the original uvicorn handler. Installed via monkey-patch at import time.

### Method
Static method

### Signature
```python
@staticmethod
def handle_exit(*args, **kwargs):
```

### Parameters
- **args** (tuple) - Optional - Forwarded to the original uvicorn handler.
- **kwargs** (dict) - Optional - Forwarded to the original uvicorn handler.

### Returns
Whatever `original_handler` returns (or `None` if no original handler).

### Behavior
Sets `should_exit = True` (only when automatic graceful drain is enabled), then chains to the saved uvicorn handler. Installed via monkey-patch at import time:

```python
try:
    from uvicorn.main import Server

    AppStatus.original_handler = Server.handle_exit
    Server.handle_exit = AppStatus.handle_exit  # type: ignore
except ImportError:
    logger.debug(
        "Uvicorn not installed. Graceful shutdown on server termination disabled."
    )
```
```

--------------------------------

### Multi-Threaded Usage with EventSourceResponse

Source: https://github.com/sysid/sse-starlette/blob/main/README.md

Run SSE streaming in separate threads with independent asyncio event loops. This avoids 'Event bound to different loop' errors. The example starts three threads, each running an SSE response.

```python
import threading
import asyncio
from sse_starlette import EventSourceResponse

def run_sse_in_thread():
    """SSE streaming works correctly in separate threads"""
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    
    async def thread_events():
        for i in range(5):
            yield {"data": f"Thread event {i}"}
            await asyncio.sleep(1)
    
    # This works without "Event bound to different loop" errors
    response = EventSourceResponse(thread_events())
    loop.close()

# Start SSE in multiple threads
for i in range(3):
    thread = threading.Thread(target=run_sse_in_thread)
    thread.start()
```

--------------------------------

### Catch SendTimeoutError in consumer code

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/errors.md

Shows how to catch SendTimeoutError in consumer code. The exception is also catchable as plain TimeoutError. The example assumes a generator gen() and an ASGI test tool or server invocation.

```python
from sse_starlette import EventSourceResponse
from sse_starlette.sse import SendTimeoutError

try:
    response = EventSourceResponse(gen(), send_timeout=10)
    # ... invoke via your ASGI test tool / server
except SendTimeoutError:
    handle_timeout()
# Also catchable as plain TimeoutError
```

--------------------------------

### Streaming JSON Objects from a Route

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/api-reference/JSONServerSentEvent.md

Example demonstrating how to use JSONServerSentEvent with EventSourceResponse to stream JSON objects from a route.

```APIDOC
## Streaming JSON Objects from a Route

### Description
Example demonstrating how to use JSONServerSentEvent with EventSourceResponse to stream JSON objects from a route.

### Example
```python
from sse_starlette import EventSourceResponse, JSONServerSentEvent


async def status_stream():
    status = {"cpu": 42, "mem": 67}
    for percent in range(5):
        status["cpu"] = percent * 10
        yield JSONServerSentEvent(data=status, event="status")


async def endpoint(request):
    return EventSourceResponse(status_stream())
```
```

--------------------------------

### handle_exit static method

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/api-reference/AppStatus.md

Sets should_exit when automatic graceful drain is enabled, then chains to the original uvicorn handler. Installed via monkey-patch at import time.

```python
@staticmethod
def handle_exit(*args, **kwargs):
    if AppStatus.enable_automatic_graceful_drain:
        AppStatus.should_exit = True
    if AppStatus.original_handler is not None:
        AppStatus.original_handler(*args, **kwargs)
```

--------------------------------

### HAProxy timeout configuration for SSE

Source: https://github.com/sysid/sse-starlette/blob/main/README.md

Configure HAProxy timeouts to exceed the heartbeat frequency to prevent connection drops. The example sets client and server timeouts to 60 seconds for a 45-second ping interval.

```haproxy
# Ensure timeouts > ping interval
timeout client 60s    # If ping every 45s
timeout server 60s
```

--------------------------------

### EventSourceResponse Methods

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/api-reference/EventSourceResponse.md

Documentation for the EventSourceResponse methods `enable_compression` and `__call__` (ASGI entrypoint).

```APIDOC
## `enable_compression`

### Description
This method is provided for parity with Starlette's `Response` interface but is not supported for SSE streams. It always raises `NotImplementedError`.

### Method
`enable_compression`

### Signature
```python
def enable_compression(self, force: bool = False) -> None:
```

### Parameters
- **force** (bool) - Optional - Accepted for parity with Starlette's `Response`, unused.

### Throws
- `NotImplementedError` - Always raised. Compression is not supported for SSE streams.

### Returns
`None`

## `__call__` (ASGI entrypoint)

### Description
This is the ASGI application contract. It handles the full request/response lifecycle for SSE streams, including WebSocket scope handling, task management, and background task execution.

### Method
`__call__`

### Signature
```python
async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
```

### Parameters
- **scope** (`starlette.types.Scope`) - Required - The ASGI connection scope.
- **receive** (`starlette.types.Receive`) - Required - The ASGI receive callable.
- **send** (`starlette.types.Send`) - Required - The ASGI send callable.

### Behavior
1. Rewrites `send` for WebSocket scopes so response messages become `websocket.http.response.*` (denial parity with Starlette).
2. Runs the whole body under `collapse_excgroups()` so single-exception `ExceptionGroup`s surface as the bare exception to user middleware.
3. Opens an `anyio.create_task_group()` and starts tasks, each wrapped in the `cancel_on_finish` supervisor — the first task to return cancels all siblings:
   - `_stream_response(send)` — always started
   - `_ping(send)` — started only when `ping_interval > 0`
   - `_listen_for_exit_signal_with_grace()` — always started
   - `data_sender_callable()` — started when provided
   - `_listen_for_disconnect(receive)` — always started
4. Awaits `self.background` (if set) after the task group exits.

### Returns
`None`. The response is written by calling `send(...)` with `http.response.start` and `http.response.body` ASGI messages.

### Throws
- Propagates `SendTimeoutError`; otherwise exceptions from the content generator propagate through the task group (single exceptions unwrapped by `collapse_excgroups`).
```

--------------------------------

### enable_compression method

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/api-reference/EventSourceResponse.md

Always raises NotImplementedError because compression is not supported for SSE streams. The force parameter is accepted only for parity with Starlette's Response and is unused.

```python
def enable_compression(self, force: bool = False) -> None:
    raise NotImplementedError("Compression is not supported for SSE streams.")
```

--------------------------------

### ServerSentEvent Constructor

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/api-reference/ServerSentEvent.md

Creates a ServerSentEvent instance that formats data as an SSE message. All parameters except data are keyword-only.

```APIDOC
## ServerSentEvent Constructor

### Description
Creates a ServerSentEvent instance that formats data as an SSE message. All parameters except data are keyword-only.

### Method
Constructor

### Endpoint
ServerSentEvent.__init__

### Parameters
#### Constructor Parameters
- **data** (Optional[Any]) - Optional - Payload of the event. Multi-line values are split into repeated `data:` lines when encoded.
- **event** (Optional[str]) - Optional - Event type emitted as an `event:` line. Newlines are stripped.
- **id** (Optional[str]) - Optional - Event id emitted as an `id:` line. Newlines are stripped.
- **retry** (Optional[int]) - Optional - Reconnection time in milliseconds emitted as a `retry:` line. Must be `int` — `encode()` raises `TypeError` otherwise.
- **comment** (Optional[str]) - Optional - Comment text emitted as `: comment` lines (SSE spec: lines starting with `:` are ignored by clients and can double as keep-alives). Multi-line comments become multiple `:` lines.
- **sep** (Optional[str]) - Optional - Line separator used by `encode()` (`None` falls back to `DEFAULT_SEPARATOR`).

### Request Example
```python
ServerSentEvent(
    data="hello",
    event="message",
    id="1",
    retry=3000,
    comment="keep-alive",
    sep="\r\n"
)
```

### Response
#### Success Response (200)
- **ServerSentEvent instance** - The constructed object.

#### Response Example
```python
ServerSentEvent(data='hello', event='message', id='1', retry=3000, comment='keep-alive', sep='\r\n')
```

```

--------------------------------

### Import public API symbols

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/api-reference/index.md

Import the main public classes and version from the top-level package. Use this to access EventSourceResponse, ServerSentEvent, JSONServerSentEvent, and the version string.

```python
from sse_starlette import EventSourceResponse, ServerSentEvent, JSONServerSentEvent
from sse_starlette import __version__  # "3.4.11"
```

--------------------------------

### Cooperative shutdown with farewell event

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/api-reference/EventSourceResponse.md

Creates an EventSourceResponse that yields periodic 'tick' events until a shutdown event is set, then sends a final 'shutdown' event. Uses anyio.Event and shutdown_grace_period to allow graceful termination.

```python
import anyio
from sse_starlette import EventSourceResponse


async def graceful_stream(request):
    shutdown_event = anyio.Event()

    async def generate():
        try:
            while not shutdown_event.is_set():
                yield {"data": "tick"}
                with anyio.move_on_after(1.0):
                    await shutdown_event.wait()
            yield {"event": "shutdown", "data": "Server is shutting down"}
        except anyio.get_cancelled_exc_class():
            raise

    return EventSourceResponse(
        generate(),
        shutdown_event=shutdown_event,
        shutdown_grace_period=5.0,
    )
```

--------------------------------

### _send_with_timeout implementation

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/api-reference/EventSourceResponse.md

Sends one ASGI message, applying the response's per-send timeout. Uses anyio.move_on_after with self.send_timeout and raises SendTimeoutError if the timeout expires. Source: sse_starlette/sse.py:351-357.

```python
async def _send_with_timeout(self, send: Send, message: Message) -> None:
    """Send one ASGI message, applying the response's per-send timeout."""
    with anyio.move_on_after(self.send_timeout) as cancel_scope:
        await send(message)

    if cancel_scope and cancel_scope.cancel_called:
        raise SendTimeoutError()
```

--------------------------------

### ServerSentEvent structured event creation

Source: https://github.com/sysid/sse-starlette/blob/main/README.md

Create a ServerSentEvent instance with custom data, event type, id, and retry interval.

```python
from sse_starlette import ServerSentEvent

event = ServerSentEvent(
    data="Custom message",
    event="notification", 
    id="msg-123",
    retry=5000
)
```

--------------------------------

### Cooperative Shutdown with shutdown_event and shutdown_grace_period

Source: https://github.com/sysid/sse-starlette/blob/main/README.md

Implement graceful shutdown by providing a shutdown_event that the library sets on server shutdown. The generator can detect the event, send a farewell event, and exit within shutdown_grace_period. If it doesn't exit in time, it is force-cancelled. Ensure shutdown_grace_period is less than the ASGI server's graceful shutdown timeout.

```python
import anyio
from sse_starlette import EventSourceResponse

async def graceful_stream(request):
    shutdown_event = anyio.Event()

    async def generate():
        try:
            while not shutdown_event.is_set():
                yield {"data": "tick"}
                # Check for shutdown between iterations
                with anyio.move_on_after(1.0):
                    await shutdown_event.wait()
            # Shutdown detected — send farewell event
            yield {"event": "shutdown", "data": "Server is shutting down"}
        except anyio.get_cancelled_exc_class():
            # Grace period expired — generator force-cancelled
            raise

    return EventSourceResponse(
        generate(),
        shutdown_event=shutdown_event,       # Library sets this on shutdown
        shutdown_grace_period=5.0,           # Seconds to wait before force-cancel
    )
```

--------------------------------

### ServerSentEvent: comment, multi-line data, and retry validation

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/api-reference/ServerSentEvent.md

Shows how to create ServerSentEvent instances with comment, multi-line data, and retry. The comment event is ignored by clients and useful as a keep-alive ping. Multi-line data is split into multiple data: lines. Passing a non-int retry raises TypeError.

```python
from sse_starlette import ServerSentEvent

# Comment-only event — useful as a keep-alive ping (ignored by clients)
keepalive = ServerSentEvent(comment="ping - 2026-01-01 00:00:00+00:00")
print(keepalive.encode())
# b': ping - 2026-01-01 00:00:00+00:00\r\n\r\n'

# Multi-line data becomes multiple data: lines
multi = ServerSentEvent(data="line1\nline2", sep="\n")
print(multi.encode())
# b'data: line1\ndata: line2\n\n'

# retry must be an int
try:
    ServerSentEvent(data="x", retry="5000").encode()
except TypeError as exc:
    print(exc)  # retry argument must be int
```

--------------------------------

### ServerSentEvent constructor signature

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/api-reference/ServerSentEvent.md

Constructor signature for ServerSentEvent. All parameters except data are keyword-only. The retry parameter must be an int; encode() raises TypeError otherwise. The sep parameter defaults to DEFAULT_SEPARATOR when None.

```python
def __init__(
    self,
    data: Optional[Any] = None,
    *,
    event: Optional[str] = None,
    id: Optional[str] = None,
    retry: Optional[int] = None,
    comment: Optional[str] = None,
    sep: Optional[str] = None,
) -> None
```

--------------------------------

### collapse_excgroups context manager implementation

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/api-reference/sse-module.md

Internal context manager that unwraps single-exception groups recursively so middleware sees the bare exception. Requires the `exceptiongroup` backport on Python < 3.11. Used by `EventSourceResponse.__call__`.

```python
@contextmanager
def collapse_excgroups() -> Generator[None, None, None]:
    try:
        yield
    except BaseException as exc:
        if has_exceptiongroups:
            # `ty` does not narrow BaseExceptionGroup.exceptions; the runtime
            # contract is identical to starlette._utils.collapse_excgroups.
            while isinstance(exc, BaseExceptionGroup) and len(exc.exceptions) == 1:  # ty: ignore[unresolved-attribute]
                exc = exc.exceptions[0]  # ty: ignore[unresolved-attribute]

        raise exc


__all__ = ["collapse_excgroups"]
```

--------------------------------

### Construction validation errors

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/api-reference/EventSourceResponse.md

Shows invalid constructor arguments that raise ValueError or TypeError, and that enable_compression() always raises NotImplementedError. Each try/except prints the error message.

```python
from sse_starlette import EventSourceResponse

try:
    EventSourceResponse([], sep="|")            # not one of \r\n, \r, \n
except ValueError as exc:
    print("invalid sep:", exc)

try:
    EventSourceResponse([], ping="every-second")  # not int/float
except TypeError as exc:
    print("invalid ping:", exc)

try:
    EventSourceResponse([], ping=-1)              # negative interval
except ValueError as exc:
    print("negative ping:", exc)

try:
    EventSourceResponse([], shutdown_grace_period=-2)
except ValueError as exc:
    print("negative grace period:", exc)

response = EventSourceResponse([])
response.enable_compression()   # NotImplementedError always
```

--------------------------------

### Custom Ping Configuration with ping_message_factory

Source: https://github.com/sysid/sse-starlette/blob/main/README.md

Configure a custom ping message by providing a factory function that returns a ServerSentEvent with a comment. Set the ping interval in seconds via the ping parameter of EventSourceResponse.

```python
from sse_starlette import ServerSentEvent

def custom_ping():
    return ServerSentEvent(comment="Custom ping message")

return EventSourceResponse(
    generate_events(),
    ping=10,  # Ping every 10 seconds
    ping_message_factory=custom_ping
)
```

--------------------------------

### Task group architecture diagram

Source: https://github.com/sysid/sse-starlette/blob/main/ARCHITECTURE.md

ASCII diagram showing the task group structure with four concurrent tasks (stream_response, ping, exit_signal, disconnect) and the cancellation flow.

```text
EventSourceResponse.__call__(scope, receive, send)
   |
   v
anyio.create_task_group()
   |
   +-- cancel_on_finish(_stream_response)      # pushes SSE data to client
   +-- cancel_on_finish(_ping)                 # keepalive pings every ~15s
   |                                           # (skipped entirely when ping=0)
   +-- cancel_on_finish(_listen_for_exit_signal_with_grace)  # server shutdown
   +-- cancel_on_finish(_listen_for_disconnect) # client disconnect
   |   (+ optional: data_sender_callable)
   |
   v
All tasks cancelled --> background task (if any) --> return
```

--------------------------------

### EventSourceResponse constructor signature

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/api-reference/EventSourceResponse.md

The constructor for EventSourceResponse. It accepts an async iterable or sync iterator of SSE event data, along with optional parameters for status code, headers, media type, background task, ping interval, separator, ping message factory, data sender callable, send timeout, client close handler, shutdown event, and shutdown grace period. The constructor does not call super().__init__() and sets its own attributes.

```python
def __init__(
    self,
    content: ContentStream,          # ContentStream = Union[
                                     #     AsyncIterable[Union[str, bytes, dict, ServerSentEvent, Any]],
                                     #     Iterator[Union[str, bytes, dict, ServerSentEvent, Any]],
                                     # ]
    status_code: int = 200,
    headers: Optional[Mapping[str, str]] = None,
    media_type: str = "text/event-stream",
    background: Optional[BackgroundTask] = None,
    ping: Optional[Union[int, float]] = None,
    sep: Optional[str] = None,
    ping_message_factory: Optional[Callable[[], ServerSentEvent]] = None,
    data_sender_callable: Optional[
        Callable[[], Coroutine[None, None, None]]
    ] = None,
    send_timeout: Optional[float] = None,
    client_close_handler_callable: Optional[
        Callable[[Message], Awaitable[None]]
    ] = None,
    shutdown_event: Optional[anyio.Event] = None,
    shutdown_grace_period: float = 0,
) -> None
```

--------------------------------

### Error Handling and Timeouts with send_timeout

Source: https://github.com/sysid/sse-starlette/blob/main/README.md

Handle client disconnections and timeouts in an SSE stream. Uses try/except for asyncio.CancelledError to perform cleanup, and sets send_timeout to 30 seconds and custom headers.

```python
async def robust_stream(request):
    try:
        for i in range(100):
            if await request.is_disconnected():
                break
            yield {"data": f"Item {i}"}
            await asyncio.sleep(0.5)
    except asyncio.CancelledError:
        # Client disconnected - perform cleanup
        raise

return EventSourceResponse(
    robust_stream(request),
    send_timeout=30,  # Timeout hanging sends
    headers={"Cache-Control": "no-cache"}
)
```

--------------------------------

### EventSourceResponse Constructor

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/api-reference/EventSourceResponse.md

Creates an EventSourceResponse instance for streaming Server-Sent Events. The constructor accepts an iterable of event data and various configuration options for pings, timeouts, and shutdown behavior.

```APIDOC
## EventSourceResponse Constructor

### Description
Creates an EventSourceResponse instance for streaming Server-Sent Events. The constructor accepts an iterable of event data and various configuration options for pings, timeouts, and shutdown behavior.

### Method
Constructor

### Endpoint
EventSourceResponse.__init__

### Parameters
#### Constructor Parameters
- **content** (ContentStream) - Required - Async iterable or sync iterator yielding SSE event data. Sync iterators are wrapped with `starlette.concurrency.iterate_in_threadpool`.
- **status_code** (int) - Optional - Default: `200` - HTTP status code of the response.
- **headers** (Mapping[str, str]) - Optional - Default: `None` - Additional HTTP headers. `Cache-Control` defaults to `no-store` (via setdefault, so user-provided cache headers win, e.g. for fan-out proxies), `Connection: keep-alive` and `X-Accel-Buffering: no` are always set.
- **media_type** (str) - Optional - Default: `"text/event-stream"` - Response media type, passed to `self.init_headers(_headers)` for the `Content-Type` header.
- **background** (Optional[BackgroundTask]) - Optional - Default: `None` - Starlette background task awaited after the response completes.
- **ping** (Optional[Union[int, float]]) - Optional - Default: `None` - Ping (heartbeat) interval in seconds; `None` falls back to `DEFAULT_PING_INTERVAL` (15). `0` disables keep-alive pings entirely — the ping task is not started.
- **sep** (Optional[str]) - Optional - Default: `None` - SSE line separator. Must be one of `"\r\n"`, `"\r"`, `"\n"`; `None` falls back to `DEFAULT_SEPARATOR`. Any other value raises `ValueError`.
- **ping_message_factory** (Optional[Callable[[], ServerSentEvent]]) - Optional - Default: `None` - Zero-argument callable returning a custom ping `ServerSentEvent`. When `None`, the default ping is a comment line `` `ping - {datetime.now(timezone.utc)}` ``.
- **data_sender_callable** (Optional[Callable[[], Coroutine[None, None, None]]]) - Optional - Default: `None` - Async callable started as an extra task-group task for push-based data sending (e.g. a producer writing to an anyio memory channel).
- **send_timeout** (Optional[float]) - Optional - Default: `None` - Timeout in seconds for an individual data or heartbeat send. On expiry, `SendTimeoutError` is raised.
- **client_close_handler_callable** (Optional[Callable[[Message], Awaitable[None]]]) - Optional - Default: `None` - Async callback invoked with the `http.disconnect` ASGI message when the client disconnects.
- **shutdown_event** (Optional[anyio.Event]) - Optional - Default: `None` - An `anyio.Event` the library sets when server shutdown is detected, allowing generators to send farewell events and exit cooperatively.
- **shutdown_grace_period** (float) - Optional - Default: `0` - Seconds to wait after setting `shutdown_event` before force-cancelling the generator. Must be `>= 0` (else `ValueError`); should be less than your ASGI server's graceful shutdown timeout. `0` = immediate cancellation.

### Request Example
```python
from sse_starlette.sse import EventSourceResponse

async def event_generator():
    yield {"data": "Hello"}

response = EventSourceResponse(
    content=event_generator(),
    ping=10,
    sep="\r\n",
)
```

### Response
#### Success Response (200)
- **body_iterator** (AsyncIterable) - The async iterable that yields SSE formatted bytes.
- **status_code** (int) - The HTTP status code (default 200).
- **media_type** (str) - The media type (default "text/event-stream").
- **headers** (Headers) - The response headers, including `Cache-Control`, `Connection`, and `X-Accel-Buffering`.

#### Response Example
```python
# The response object is returned by the constructor.
# It can be returned directly from a Starlette route.
```

### Error Handling
- **ValueError** - Raised if `sep` is not one of `None`, `"\r\n"`, `"\r"`, `"\n"`, or if `shutdown_grace_period` is negative.
- **TypeError** - Raised if `ping` is not an `int` or `float`.
- **ValueError** - Raised if `ping` is negative.
- **SendTimeoutError** - Raised if a data or heartbeat send exceeds `send_timeout`.
```

--------------------------------

### Client-side reconnection logic with EventSource

Source: https://github.com/sysid/sse-starlette/blob/main/README.md

Implements a reconnection mechanism for SSE clients. On error, it schedules a reconnect after a 5-second delay. Use this to handle network interruptions and server restarts.

```javascript
function createEventSource(url) {
    const eventSource = new EventSource(url);
    
    eventSource.onerror = function() {
        setTimeout(() => {
            createEventSource(url);  // Reconnect after delay
        }, 5000);
    };
    
    return eventSource;
}
```

--------------------------------

### Sequence diagram: heartbeat send timeout (5b)

Source: https://github.com/sysid/sse-starlette/blob/main/ARCHITECTURE.md

Shows the flow when the heartbeat send times out on an idle stream: _ping holds _send_lock, times out, raises SendTimeoutError, and the task group cancels siblings, throwing CancelledError into the suspended __anext__ and running the generator's finally block.

```text
_stream_response         _ping        _exit_signal    _disconnect
      |                    |               |               |
  suspended in       await sleep(interval) |               |
  __anext__()             |                |               |
  (generator idle)   async with _send_lock:|               |
      |              move_on_after(timeout):               |
      |                send(ping)          |               |
      |                  [send hangs!]     |               |
      |              [timeout expires]     |               |
      |              timeout detected      |               |
      |              raise SendTimeoutError|               |
      |                    |               |               |
  EXCEPTION propagates through cancel_on_finish into task group
  Task group cancels all siblings
      |                    |               |               |
  CancelledError thrown into the suspended __anext__
  Generator's finally block runs
```

--------------------------------

### ensure_bytes

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/api-reference/sse-module.md

Documentation for the ensure_bytes function from the sse_starlette.event module.

```APIDOC
## ensure_bytes

### Description
Converts a stream item (bytes, dict, ServerSentEvent, or any other value) into a ready-to-send SSE frame.

### Function Signature
```python
def ensure_bytes(data: Union[bytes, dict, ServerSentEvent, Any], sep: str) -> bytes:
```

### Parameters
- **data** (`Union[bytes, dict, ServerSentEvent, Any]`) - Required - Stream item to convert to an SSE frame.
- **sep** (`str`) - Required - Line separator injected into the generated frame.

### Return Type
`bytes` - the ready-to-send SSE frame.

### Behavior
- `bytes`/`memoryview` are passed through verbatim.
- `ServerSentEvent` instances are encoded via `.encode()`.
- `dict` items are injected with `"sep"` and passed to `ServerSentEvent(**data)`.
- Anything else is wrapped as `ServerSentEvent(str(data), sep=sep)`.

### Side Effects
- For `dict` inputs, the caller's dict is mutated in place: `data["sep"] = sep`.

### Usage
This is the conversion routine `EventSourceResponse._stream_response` applies to every item of `content` (`sse.py:365`), and `_ping` applies to heartbeat messages (`sse.py:454`).

### Example
```python
from sse_starlette.event import ensure_bytes
from sse_starlette import ServerSentEvent

assert ensure_bytes(b"raw\x00bytes", "\r\n") == b"raw\x00bytes"
assert ensure_bytes({"data": "hi"}, "\r\n") == b"data: hi\r\n\r\n"
assert ensure_bytes(ServerSentEvent(comment="c"), "\n") == b": c\n\n"
assert ensure_bytes(42, "\r\n") == b"data: 42\r\n\r\n"
```
```

--------------------------------

### Sequence diagram: data send timeout (5a)

Source: https://github.com/sysid/sse-starlette/blob/main/ARCHITECTURE.md

Shows the flow when a data send times out: _stream_response detects the timeout, closes the iterator, raises SendTimeoutError, and the exception propagates through cancel_on_finish into the task group, which cancels all siblings.

```text
_stream_response         _ping        _exit_signal    _disconnect
      |                    |               |               |
  async for data:          |               |               |
    move_on_after(timeout):|               |               |
      send(chunk)          |               |               |
        [send hangs!]      |               |               |
           ...             |               |               |
    [timeout expires]      |               |               |
    timeout detected       |               |               |
    aclose() iterator      |               |               |
    raise SendTimeoutError |               |               |
      |                    |               |               |
  EXCEPTION propagates through cancel_on_finish into task group
  Task group cancels all siblings
```

--------------------------------

### enable_automatic_graceful_drain_mode static method

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/api-reference/AppStatus.md

Re-enables automatic SSE stream termination on server shutdown, restoring the default behavior where SIGTERM triggers immediate stream draining. Use to undo a previous call to disable_automatic_graceful_drain().

```python
@staticmethod
def enable_automatic_graceful_drain_mode():
    """
    Re-enable automatic SSE stream termination on server shutdown.

    This restores the default behavior where SIGTERM triggers immediate
    stream draining. Call this to undo a previous call to
    disable_automatic_graceful_drain().
    """
    AppStatus.enable_automatic_graceful_drain = True
```

--------------------------------

### EventSourceResponse with heartbeat, headers, background task, and send timeout

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/api-reference/EventSourceResponse.md

Configures EventSourceResponse with a custom ping message factory for heartbeats, a send timeout for hang detection, custom headers, and a background task that runs after the stream finishes. The stream checks for client disconnection and handles cancellation for cleanup.

```python
from datetime import datetime, timezone

from starlette.background import BackgroundTask
from sse_starlette import EventSourceResponse, ServerSentEvent


def custom_ping():
    return ServerSentEvent(comment="keep-alive")


async def stream(request):
    try:
        for i in range(100):
            if await request.is_disconnected():
                break
            yield {"data": f"Item {i}"}
            await asyncio.sleep(0.5)
    except asyncio.CancelledError:  # client disconnected / server shutdown
        print("cleanup on cancellation")
        raise


def mark_done():
    print("stream finished")


response = EventSourceResponse(
    stream(request),
    ping=10,                                  # heartbeat every 10s
    ping_message_factory=custom_ping,         # custom comment ping
    send_timeout=30,                          # hang detection
    headers={"Cache-Control": "no-cache"},    # overrides default no-store
    background=BackgroundTask(mark_done),
)
```

--------------------------------

### Push-based streaming via memory channel

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/api-reference/EventSourceResponse.md

Uses an anyio memory object stream to push data from a producer coroutine to the EventSourceResponse. The data_sender_callable is invoked with the send channel, and the response consumes the receive channel.

```python
import anyio
from functools import partial
from sse_starlette import EventSourceResponse


async def data_producer(send_channel):
    async with send_channel:
        for i in range(10):
            await send_channel.send({"data": f"Item {i}"})
            await anyio.sleep(1)


async def channel_endpoint(request):
    send_channel, receive_channel = anyio.create_memory_object_stream(10)
    return EventSourceResponse(
        receive_channel,
        data_sender_callable=partial(data_producer, send_channel),
    )
```

--------------------------------

### JSONServerSentEvent Constructor

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/api-reference/JSONServerSentEvent.md

Creates a new JSONServerSentEvent instance, automatically JSON-serializing the data argument. The data parameter accepts any JSON-serializable value and is serialized with ensure_ascii=False, allow_nan=False, indent=None, and separators=(",", ":"). Additional arguments are forwarded to the base ServerSentEvent constructor.

```APIDOC
## JSONServerSentEvent Constructor

### Description
Creates a new JSONServerSentEvent instance, automatically JSON-serializing the data argument. The data parameter accepts any JSON-serializable value and is serialized with ensure_ascii=False, allow_nan=False, indent=None, and separators=(",", ":"). Additional arguments are forwarded to the base ServerSentEvent constructor.

### Method
Constructor

### Signature
```python
def __init__(
    self,
    data: Optional[Any] = None,
    *args,
    **kwargs,
) -> None:
```

### Parameters
- **data** (Optional[Any]) - Optional - Any value serializable with `json.dumps`. Passed through `json.dumps(data, ensure_ascii=False, allow_nan=False, indent=None, separators=(",", ":"))`. When `None`, the encoded event omits the `data:` field.
- **\*args** (positional) - Optional - Forwarded to `ServerSentEvent.__init__`.
- **\*\*kwargs** (keyword) - Optional - Forwarded to `ServerSentEvent.__init__` — i.e. any of `event`, `id`, `retry`, `comment`, `sep`.

### JSON Serialization Details
- `ensure_ascii=False` — non-ASCII characters are emitted as-is (UTF-8), not `\uXXXX` escapes.
- `allow_nan=False` — `NaN`, `Infinity`, `-Infinity` raise `ValueError` ("Out of range float values are not JSON compliant") instead of being emitted as non-standard JSON.
- `indent=None` — compact output, no pretty-printing.
- `separators=(",", ":")` — minimal separators: `{"field":"value"}` with no spaces.

### Example
```python
from sse_starlette import JSONServerSentEvent

event = JSONServerSentEvent(
    data={"field": "value"},   # anything json.dumps can handle
    event="update",
    id="msg-123",
)

print(event.encode())
# b'id: msg-123\r\nevent: update\r\ndata: {"field":"value"}\r\n\r\n'
```
```

--------------------------------

### Flow 1: Normal Generator Exhaustion

Source: https://github.com/sysid/sse-starlette/blob/main/ARCHITECTURE.md

Sequence diagram showing the normal flow when the generator exhausts: stream_response sends chunks, then sets active=False, sends more_body=False, and returns, causing cancel_on_finish to cancel all sibling tasks.

```text
Generator yields all items and finishes naturally.

_stream_response         _ping        _exit_signal    _disconnect
      |                    |               |               |
  async for data:          |               |               |
    send(chunk)       sleep(15)       wait(event)     receive()
    send(chunk)            |               |               |
      ...                  |               |               |
  [generator ends]         |               |               |
  self.active = False      |               |               |
  send(more_body=False)    |               |               |
  return                   |               |               |
      |                    |               |               |
  cancel_on_finish ----> CANCEL          CANCEL          CANCEL
      |
      v
  Task group exits cleanly
```

--------------------------------

### encode

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/api-reference/ServerSentEvent.md

Serializes the ServerSentEvent object into a UTF-8 encoded SSE message. The output includes comment lines, id, event, data, and retry fields in a fixed order, omitting None values, and terminates with a blank line.

```APIDOC
## encode

### Description
Serializes the ServerSentEvent object into a UTF-8 encoded SSE message. The output includes comment lines, id, event, data, and retry fields in a fixed order, omitting None values, and terminates with a blank line.

### Method
encode

### Signature
```python
def encode(self) -> bytes:
```

### Parameters
None (method on ServerSentEvent instance)

### Return Type
- **bytes** - The UTF-8 encoded SSE message.

### Behavior
- Field order in the output is fixed: comment lines, `id:`, `event:`, `data:` lines, `retry:`, terminated by one blank line (`self._sep`).
- Fields with `None` values are omitted entirely.
- Multi-line data is split into multiple `data:` lines.
- Newlines in `id` and `event` are cleaned.

### Throws
- **TypeError** - If `retry` is not an `int` (only raised when `retry is not None`).

### Example
```python
from sse_starlette import ServerSentEvent

event = ServerSentEvent(
    data="Custom message",
    event="notification",
    id="msg-123",
    retry=5000,          # client will retry after 5 seconds
)

print(event.encode())
# b'id: msg-123\r\nevent: notification\r\ndata: Custom message\r\nretry: 5000\r\n\r\n'
```
```

--------------------------------

### Using ServerSentEvent with EventSourceResponse and custom ping

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/api-reference/ServerSentEvent.md

Demonstrates using ServerSentEvent inside an async generator consumed by EventSourceResponse. EventSourceResponse encodes each ServerSentEvent via .encode(). A custom ping factory can be provided via ping_message_factory along with the ping interval.

```python
from sse_starlette import EventSourceResponse, ServerSentEvent


async def stream_events():
    yield ServerSentEvent(data="progress 10 %", event="progress")
    yield ServerSentEvent(comment="still alive")

# Any item that is a ServerSentEvent is encoded via .encode() by EventSourceResponse.
response = EventSourceResponse(stream_events())

# Custom ping factory:
def custom_ping() -> ServerSentEvent:
    return ServerSentEvent(comment="Custom ping message")


custom_response = EventSourceResponse(stream_events(), ping=10, ping_message_factory=custom_ping)
```

--------------------------------

### Memory Channels Alternative with data_sender_callable

Source: https://github.com/sysid/sse-starlette/blob/main/README.md

Use anyio memory channels for complex data flows instead of generators. The data_producer sends items to a channel, and EventSourceResponse consumes from the channel via data_sender_callable.

```python
import anyio
from functools import partial

async def data_producer(send_channel):
    async with send_channel:
        for i in range(10):
            await send_channel.send({"data": f"Item {i}"})
            await anyio.sleep(1)

async def channel_endpoint(request):
    send_channel, receive_channel = anyio.create_memory_object_stream(10)
    
    return EventSourceResponse(
        receive_channel,
        data_sender_callable=partial(data_producer, send_channel)
    )
```

--------------------------------

### JSONServerSentEvent constructor

Source: https://github.com/sysid/sse-starlette/blob/main/_autodocs/api-reference/JSONServerSentEvent.md

Constructor for JSONServerSentEvent. It serializes the data argument with json.dumps using ensure_ascii=False, allow_nan=False, indent=None, and separators=(",", ":"). If data is None, the encoded event omits the data: field. Additional arguments are forwarded to ServerSentEvent.__init__.

```python
def __init__(
    self,
    data: Optional[Any] = None,
    *args,
    **kwargs,
) -> None:
    super().__init__(
        json.dumps(
            data,
            ensure_ascii=False,
            allow_nan=False,
            indent=None,
            separators=(",", ":"),
        )
        if data is not None
        else None,
        *args,
        **kwargs,
    )
```

--------------------------------

### ConnectionLimiter for performance optimization

Source: https://github.com/sysid/sse-starlette/blob/main/README.md

Use this Python class to limit concurrent connections using an asyncio.Semaphore. The limited_endpoint method wraps an EventSourceResponse with the semaphore to control concurrency.

```python
# Connection limits
class ConnectionLimiter:
    def __init__(self, max_connections=100):
        self.semaphore = asyncio.Semaphore(max_connections)
    
    async def limited_endpoint(self, request):
        async with self.semaphore:
            return EventSourceResponse(generate_events())
```

--------------------------------

### Nginx configuration to disable buffering for SSE

Source: https://github.com/sysid/sse-starlette/blob/main/README.md

Use this Nginx configuration to disable proxy buffering for the /events location, which prevents SSE events from being delayed until ~16KB accumulates. This is needed when you cannot modify application headers.

```nginx
location /events {
    proxy_pass http://localhost:8000;
    proxy_http_version 1.1;
    proxy_set_header Connection '';
    proxy_buffering off;           # Disable for this location
    chunked_transfer_encoding off;
}
```
Tokens:
Raw