Python Server-Sent Events for Real-Time Streaming

This article provides an overview of how Python manages Server-Sent Events (SSE) to deliver real-time, unidirectional data from a web server to clients over standard HTTP. It examines the underlying SSE protocol requirements, contrasts how synchronous (WSGI) and asynchronous (ASGI) Python frameworks handle persistent streaming connections, provides code implementations for popular frameworks like FastAPI and Flask, and outlines best practices for scaling real-time endpoints in production.

Understanding the SSE Protocol in HTTP

Server-Sent Events allow a client to establish a persistent, long-lived connection over standard HTTP/HTTPS. Unlike WebSockets, which provide full-duplex communication, SSE is strictly unidirectional: the server continuously pushes data to the client.

To handle an SSE stream, the Python web server must respond with specific HTTP headers:

• Content-Type: text/event-stream: Informs the client that the connection is an event stream.
• Cache-Control: no-cache: Prevents intermediary proxies from caching partial responses.
• Connection: keep-alive: Maintains an open TCP connection without terminating after sending data.

The data payload sent over the stream must follow a plain-text structure ending with a double newline:

data: {"message": "hello world"}

Optional fields include event: (custom event type), id: (unique message identifier for reconnection tracking), and retry: (reconnection timeout in milliseconds).

Synchronous vs. Asynchronous Python for SSE

Python handles persistent streaming connections differently depending on whether the framework runs on WSGI (Web Server Gateway Interface) or ASGI (Asynchronous Server Gateway Interface).

The WSGI Limitation

Synchronous WSGI frameworks, such as standard Flask or Django configurations, allocate one operating system thread or process per open HTTP connection. Because an SSE connection remains open indefinitely, each connected client monopolizes a server worker. In high-concurrency environments, this quickly leads to worker starvation and connection limits unless coupled with gevent or specialized event-loop patching.

The ASGI Solution

Modern Python real-time streaming relies predominantly on ASGI frameworks like FastAPI, Starlette, or Quart, executed on servers like Uvicorn or Hypercorn. Under ASGI, an open connection is suspended using Python’s native asyncio event loop. The server can maintain thousands of concurrent, idle, or intermittently streaming SSE connections using minimal memory and a single operating system thread.

Implementing SSE with FastAPI (ASGI)

FastAPI utilizes Starlette's StreamingResponse to deliver async generator streams. An asynchronous generator yields SSE-formatted strings periodically while awaiting events.

import asyncio
from datetime import datetime
from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()

async def event_generator():
    while True:
        # Check for disconnection or wait for an event source
        await asyncio.sleep(1)
        current_time = datetime.utcnow().isoformat()
        yield f"data: {{\"timestamp\": \"{current_time}\"}}\n\n"

@app.get("/stream")
async def stream_events():
    return StreamingResponse(
        event_generator(),
        media_type="text/event-stream"
    )

In this implementation, asyncio.sleep(1) yields control back to the event loop, allowing the server to handle other concurrent requests until the timer expires.

Implementing SSE with Flask (WSGI)

Flask can stream responses using standard Python generators combined with its stream_with_context decorator.

import time
from flask import Flask, Response, stream_with_context

app = Flask(__name__)

def generate_events():
    while True:
        time.sleep(1)
        yield f"data: Server heartbeat\n\n"

@app.route("/stream")
def sse_endpoint():
    return Response(
        stream_with_context(generate_events()),
        mimetype="text/event-stream"
    )

if __name__ == "__main__":
    app.run(threaded=True)

In Flask, running with threaded=True is mandatory to avoid blocking the entire server for a single client, though production deployments typically require an async worker model (such as Gunicorn running Gthread or Gevent workers).

Connection Lifecycle and Scalability

Handling production SSE endpoints requires managing edge cases around disconnections, buffering, and horizontal scaling.

Detecting Client Disconnections

When a client closes a browser tab or loses network connectivity, the server must stop the generator to prevent resource leakage. In ASGI frameworks, the underlying server detects a broken socket and raises an exception (such as asyncio.CancelledError) inside the async generator, which should be caught to clean up resources:

async def event_generator():
    try:
        while True:
            await asyncio.sleep(1)
            yield f"data: ping\n\n"
    except asyncio.CancelledError:
        # Clean up database connections or queue subscriptions
        pass

Reverse Proxies and Buffering

When deploying Python applications behind reverse proxies like Nginx, response buffering must be explicitly disabled. By default, Nginx buffers HTTP responses, preventing SSE events from arriving at the client in real time.

To disable buffering, add the following header in Python or configure it directly in Nginx:

headers = {
    "X-Accel-Buffering": "no",
    "Cache-Control": "no-cache",
}

Scaling Across Multiple Servers

Because SSE connections are stateful and persistent, distributing events across multiple Python server instances requires an external message broker. A common pattern involves using Redis Pub/Sub:

1. The client establishes an SSE connection with an arbitrary Python instance.
2. The Python worker subscribes to a specific Redis channel via an asynchronous client like redis-py.
3. When an event occurs anywhere in the infrastructure, it is published to the Redis channel.
4. The Python worker receives the message from Redis and forwards it to the client through the active SSE generator.