SSE-Starlette

raw JSON →
3.3.3 verified Tue May 12 auth: no python install: verified quickstart: verified

sse-starlette is a production-ready Python library that provides a Server-Sent Events (SSE) plugin for Starlette and FastAPI frameworks. It offers a standards-compliant implementation of the W3C SSE specification, including features like automatic client disconnect detection, graceful shutdown, and thread-safe event management. The current version is 3.3.3, and the library maintains an active release cadence with regular updates and dependency management.

pip install sse-starlette
error ModuleNotFoundError: No module named 'sse_starlette'
cause The `sse-starlette` library has not been installed in the active Python environment.
fix
pip install sse-starlette
error TypeError: 'list' object is not async iterable
cause `EventSourceResponse` requires an asynchronous generator (an `async def` function that uses `yield`), but a synchronous iterable like a list was provided.
fix
Define an async def function that uses yield to produce events, then pass this generator function to EventSourceResponse.
error TypeError: expected string or bytes-like object, got dict
cause The `async` generator is yielding Python objects (like dictionaries) directly, but `EventSourceResponse` expects event data to be a string or bytes.
fix
Serialize the data to a JSON string (e.g., using json.dumps()) before yielding it from the async generator.
error RuntimeError: generator didn't stop itself
cause The `async` generator function, typically within a `while True` loop, does not properly handle `GeneratorExit` when the client disconnects, leading to the ASGI server reporting an unstopped generator.
fix
Modify the async generator to catch GeneratorExit and break its loop gracefully, for example, by wrapping the loop in try...except GeneratorExit: break.
gotcha SSE streaming does not work in combination with Starlette's GZipMiddleware. Applying GZipMiddleware to an SSE endpoint will buffer the response, preventing real-time delivery.
fix Avoid using GZipMiddleware on routes that serve `EventSourceResponse`. Consider configuring your reverse proxy for compression if needed for other endpoints, but bypass it for SSE streams.
gotcha Reverse proxies (e.g., Nginx, Apache, Cloudflare, Akamai, HAProxy) often buffer responses by default, which can break the real-time nature of SSE. This can cause events to be delayed or batched.
fix Configure your reverse proxy to disable buffering for SSE endpoints. For Nginx, add `proxy_buffering off;` and `X-Accel-Buffering: no` header. For other proxies, consult their documentation for equivalent settings.
breaking Prior to v3.2.0, handling graceful shutdown of SSE streams could be problematic, often leading to 'Waiting for background tasks to complete' warnings or immediate force-cancellation of generators upon server termination. This prevented proper cleanup or sending final events.
fix Upgrade to v3.2.0 or newer. Use the `shutdown_event` exposed in generators (v3.3.0+) for cooperative shutdown, or configure `AppStatus.enable_automatic_graceful_drain_mode` to manage how streams terminate during server shutdown. Ensure your graceful shutdown period in `sse-starlette` is less than your ASGI server's graceful shutdown timeout.
gotcha Versions prior to v3.1.2 had a 'watcher task leak' due to incorrect `threading.local` usage, potentially leading to resource exhaustion or unexpected behavior in long-running applications.
fix Upgrade to v3.1.2 or newer to benefit from the fix preventing this task leak.
gotcha When using database sessions (e.g., SQLAlchemy's `AsyncSession`) or other non-thread-safe objects within SSE generators, care must be taken to avoid passing them directly. The streaming internally uses `anyio.TaskGroups`, which can lead to thread-safety issues.
fix Ensure that any objects passed to SSE generators are thread-safe or are instantiated within the generator's context to prevent concurrent access issues. For database sessions, open and close them within the generator's async scope.
gotcha The test environment generated warnings related to pip usage, specifically concerning running pip as the 'root' user and suggesting a pip version upgrade. These are not direct failures of the tested library but indicate sub-optimal or potentially problematic environment setup for package management.
fix It is recommended to use a virtual environment for Python projects to avoid permission issues and conflicts with the system package manager. Avoid running pip as the 'root' user. Regularly update pip to its latest version (e.g., `pip install --upgrade pip`).
python os / libc status wheel install import disk
3.10 alpine (musl) wheel - 0.21s 20.8M
3.10 alpine (musl) - - 0.22s 20.8M
3.10 slim (glibc) wheel 2.0s 0.16s 21M
3.10 slim (glibc) - - 0.15s 21M
3.11 alpine (musl) wheel - 0.36s 22.9M
3.11 alpine (musl) - - 0.37s 22.9M
3.11 slim (glibc) wheel 2.2s 0.29s 23M
3.11 slim (glibc) - - 0.28s 23M
3.12 alpine (musl) wheel - 0.50s 14.6M
3.12 alpine (musl) - - 0.54s 14.6M
3.12 slim (glibc) wheel 1.9s 0.47s 15M
3.12 slim (glibc) - - 0.48s 15M
3.13 alpine (musl) wheel - 0.52s 14.0M
3.13 alpine (musl) - - 0.52s 13.9M
3.13 slim (glibc) wheel 1.8s 0.46s 15M
3.13 slim (glibc) - - 0.47s 14M
3.9 alpine (musl) wheel - 0.22s 20.1M
3.9 alpine (musl) - - 0.22s 20.2M
3.9 slim (glibc) wheel 2.3s 0.18s 21M
3.9 slim (glibc) - - 0.18s 21M

This quickstart sets up a basic Starlette application with an `/events` endpoint that streams ten Server-Sent Events, each separated by a one-second delay. The `EventSourceResponse` handles the SSE protocol details, including event formatting and connection management.

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)])

# To run this example, you would typically use an ASGI server like uvicorn:
# uvicorn your_module_name:app --reload