Server-Sent Events for Starlette and FastAPI

[![PyPI Version][pypi-image]][pypi-url] [![Build Status][build-image]][build-url]

Background: https://sysid.github.io/server-sent-events/

Production ready Server-Sent Events implementation for Starlette and FastAPI following the W3C SSE specification.

Installation

pip install sse-starlette
uv add sse-starlette

# To run the examples and demonstrations
uv add sse-starlette[examples]

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

Quick Start

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

Core Features

Standards Compliant: Full SSE specification implementation
Framework Integration: Native Starlette and FastAPI support
Async/Await: Built on modern Python async patterns
Connection Management: Automatic client disconnect detection
Graceful Shutdown: Proper cleanup on server termination with cooperative shutdown support
Thread Safety: Context-local event management for multi-threaded applications
Multi-Loop Support: Works correctly with multiple asyncio event loops

Key Components

EventSourceResponse

The main response class that handles SSE streaming:

from sse_starlette import EventSourceResponse

# Basic usage
async def stream_data():
    for item in data:
        yield {"data": item, "event": "update", "id": str(item.id)}

return EventSourceResponse(stream_data())

ServerSentEvent

For structured event creation:

from sse_starlette import ServerSentEvent

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

JSONServerSentEvent

For an easy way to send json data as SSE events:

from sse_starlette import JSONServerSentEvent

event = JSONServerSentEvent(
    data={"field":"value"}, # Anything serializable with json.dumps
)

Advanced Usage

Custom Ping Configuration

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
)

Multi-Threaded Usage

sse-starlette now supports usage in multi-threaded applications and with multiple asyncio event loops:

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

Database Streaming (Thread-Safe)

async def stream_database_results(request):
    # CORRECT: Create session within generator context
    async with AsyncSession() as session:
        results = await session.execute(select(User))
        for row in results:
            if await request.is_disconnected():
                break
            yield {"data": row.name, "id": str(row.id)}

return EventSourceResponse(stream_database_results(request))

Error Handling and Timeouts

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"}
)

Memory Channels Alternative

For complex data flows, use memory channels instead of generators:

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

Cooperative Shutdown

By default, generators receive CancelledError immediately when the server shuts down. With cooperative shutdown, generators can detect the shutdown signal, send farewell events to clients, and exit gracefully within a configurable grace period.

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
    )

How it works:

On server shutdown, the library sets your shutdown_event
Your generator sees the event and can yield final events
If the generator exits within shutdown_grace_period, shutdown is clean (no CancelledError)
If the generator doesn't exit in time, it is force-cancelled as before

Important: shutdown_grace_period should be less than your ASGI server's graceful shutdown timeout (e.g. uvicorn's --timeout-graceful-shutdown), otherwise the process is killed before the grace period expires.

Without shutdown_event (the default), behavior is identical to previous versions: immediate cancellation on server shutdown.

Configuration Options

EventSourceResponse Parameters

Parameter	Type	Default	Description
`content`	`ContentStream`	Required	Async generator or iterable
`ping`	`int`	15	Ping interval in seconds (0 to disable)
`sep`	`str`	`"\r\n"`	Line separator (`\r\n`, `\r`, `\n`)
`send_timeout`	`float`	`None`	Send operation timeout in seconds
`headers`	`dict`	`None`	Additional HTTP headers
`ping_message_factory`	`Callable`	`None`	Custom ping message creator
`shutdown_event`	`anyio.Event`	`None`	Event set by library on server shutdown
`shutdown_grace_period`	`float`	`0`	Seconds to wait after setting `shutdown_event` before force-cancel

Client Disconnection

async def monitored_stream(request):
    events_sent = 0
    try:
        while events_sent < 100:
            if await request.is_disconnected():
                print(f"Client disconnected after {events_sent} events")
                break
            
            yield {"data": f"Event {events_sent}"}
            events_sent += 1
            await asyncio.sleep(1)
            
    except asyncio.CancelledError:
        print("Stream cancelled")
        raise

Testing

sse-starlette includes now comprehensive test isolation without manual setup. The library automatically handles event loop contexts, eliminating the need for manual state resets:

# this is deprecated and not needed since version 3.0.0
import pytest
from sse_starlette import EventSourceResponse

@pytest.fixture
def reset_sse_app_status():
    AppStatus.should_exit_event = None
    yield
    AppStatus.should_exit_event = None

Production Considerations

Performance Limits

Memory: Each connection maintains a buffer. Monitor memory usage.
Connections: Limited by system file descriptors and application design.
Network: High-frequency events can saturate bandwidth.

Error Recovery

Implement client-side reconnection logic:

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

Learning Resources

Examples Directory

The examples/ directory contains production-ready patterns:

01_basic_sse.py: Fundamental SSE concepts
02_message_broadcasting.py: Multi-client message distribution
03_database_streaming.py: Thread-safe database integration
04_advanced_features.py: Custom protocols and error handling

Demonstrations Directory

The examples/demonstrations/ directory provides educational scenarios:

Basic Patterns (basic_patterns/):

Client disconnect detection and cleanup
Graceful server shutdown behavior

**Production

sse-starlette

Description