When you need real‑time updates without the overhead of WebSockets, Server‑Sent Events (SSE) offer a lightweight, browser‑native solution that works perfectly with Python web frameworks. In this guide we’ll explore what SSE is, why it’s a great fit for Python web apps, and walk through building a complete SSE‑powered application using Flask (the concepts apply equally to Django, FastAPI, and other frameworks). By the end you’ll have a production‑ready codebase, SEO‑friendly content, and a solid understanding of how to scale SSE in the Python ecosystem.
What Are Server‑Sent Events?
Server‑Sent Events, standardized in the HTML5 EventSource API, allow a server to push a stream of text‑based messages to the browser over a single HTTP connection. Unlike WebSockets, SSE is unidirectional (server → client) and automatically handles reconnection, making it ideal for:
- Live dashboards and analytics
- Chat notifications and typing indicators
- Stock tickers, sports scores, or any frequently updating data
- Background task progress bars
Because SSE uses plain HTTP, it works seamlessly with proxies, load balancers, and CDNs, and it benefits from built‑in browser support for automatic reconnection and event ID tracking.
Why Choose Python for SSE?
Python’s rich ecosystem of asynchronous libraries and lightweight web frameworks makes it a natural fit for SSE. Here are a few reasons to consider Python when building an SSE web app:
- Readability: Python’s clear syntax speeds up development and maintenance.
- Async support: Modern frameworks like
FastAPIandStarletteprovide nativeasynccapabilities, crucial for handling many concurrent connections. - Extensive libraries: Packages such as
Flask‑SSE,django-sse, andaiohttpsimplify boilerplate code. - Scalability: With tools like
uvicorn,gunicorn, andRedispub/sub, you can scale SSE to thousands of clients.
Core Concepts of an SSE Endpoint
Content Type and HTTP Headers
An SSE endpoint must return the Content-Type: text/event-stream header and keep the connection open. The server streams data in the following format:
event: message
id: 42
data: {"value": 123}
Each field ends with a newline, and a double newline signals the end of a message.
Automatic Reconnection
The browser automatically attempts to reconnect if the connection drops. You can control the retry interval with the retry field:
retry: 3000
This tells the client to wait 3 seconds before reconnecting.
Building a Simple SSE App with Flask
Below is a step‑by‑step example of a minimal Flask application that broadcasts random numbers to connected browsers every second. The same logic can be adapted to Django, FastAPI, or any ASGI‑compatible framework.
1. Install Required Packages
pip install Flask Flask-SSE redis
We’ll use redis as a lightweight message broker, but for a single‑process demo you can skip it.
2. Project Structure
my_sse_app/
│
├─ app.py
├─ templates/
│ └─ index.html
└─ requirements.txt
3. Server Code (app.py)
from flask import Flask, render_template, request
from flask_sse import sse
import random
import time
import threading
app = Flask(__name__)
# Configure Redis (default localhost:6379)
app.config["REDIS_URL"] = "redis://localhost"
app.register_blueprint(sse, url_prefix='/stream')
@app.route('/')
def index():
return render_template('index.html')
def generate_numbers():
"""Background thread that publishes a new number every second."""
while True:
number = random.randint(1, 100)
sse.publish({"value": number}, type='number')
time.sleep(1)
# Start background thread only once
thread = threading.Thread(target=generate_numbers)
thread.daemon = True
thread.start()
if __name__ == '__main__':
# Use a production-ready server in real deployments
app.run(debug=True, threaded=True)
4. Client‑Side HTML (templates/index.html)
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>Python SSE Demo</title>
<style>
body {font-family: Arial, sans-serif; margin: 2rem;}
#log {border: 1px solid #ccc; padding: 1rem; height: 300px; overflow-y: auto;}
</style>
</head>
<body>
<h2>Live Random Numbers (Server‑Sent Events)</h2>
<div id="log"></div>
<script>
const log = document.getElementById('log');
const source = new EventSource('/stream');
source.addEventListener('number', function(event) {
const data = JSON.parse(event.data);
const entry = document.createElement('div');
entry.textContent = `Received: ${data.value}`;
log.appendChild(entry);
log.scrollTop = log.scrollHeight;
});
source.onerror = function(err) {
console.error('SSE error', err);
source.close();
};
</script>
</body>
</html>
5. Run the App
python app.py
Open http://localhost:5000 in a browser and watch a new random number appear every second without refreshing the page.
Scaling SSE in Production
While the example above works for development, production environments require a few extra considerations to handle high traffic and ensure reliability.
Use an ASGI Server
Flask’s built‑in server is single‑threaded and not suitable for many concurrent connections. Deploy with gunicorn + uvicorn workers or switch to an ASGI‑native framework like FastAPI:
gunicorn -k uvicorn.workers.UvicornWorker app:app --workers 4
Leverage a Message Broker
- Redis Pub/Sub: Publish events from any process; all workers receive them.
- RabbitMQ or Kafka: For enterprise‑grade durability and replay capabilities.
Load Balancer Configuration
Because SSE relies on a persistent HTTP connection, make sure your load balancer (NGINX, HAProxy, AWS ELB) is configured to:
- Disable HTTP/2 push for SSE streams.
- Set a high
proxy_read_timeout(e.g., 300s) to keep connections alive. - Enable sticky sessions if you’re not using a shared message broker.
Graceful Shutdown & Heartbeats
Long‑living connections can be terminated by network devices. Sending a comment line (:) every 15–30 seconds acts as a heartbeat:
def sse_heartbeat():
while True:
sse.publish({}, type='heartbeat')
time.sleep(20)
Clients can ignore heartbeat events or use them to detect connection health.
Advanced Features and Best Practices
Custom Event Types
Instead of the default message event, you can define meaningful names (e.g., chat, notification) to simplify client‑side handling.
Event IDs for Reliable Delivery
Including an id field lets the browser request missed events after a reconnection:
sse.publish({"value": 42}, type='number', id=str(uuid4()))
Combine this with a server‑side store (Redis list, database) to replay missed messages.
Security Considerations
- CORS: Set
Access-Control-Allow-Originonly for trusted origins. - Authentication: Use session cookies or JWT tokens; verify them before establishing the SSE stream.
- Rate Limiting: Prevent a single client from opening hundreds of SSE connections.
Testing SSE Endpoints
Automated tests can be written with pytest and httpx (async) or requests for simple cases. Example using httpx:
import httpx
import asyncio
async def test_sse():
async with httpx.AsyncClient(base_url="http://localhost:8000") as client:
async with client.stream("GET", "/stream") as response:
async for line in response.aiter_lines():
if line.startswith("data:"):