Imagine a web application that pushes live data to users the instant something changes—no page reloads, no extra JavaScript libraries, just a simple, reliable stream. That’s the power of Server‑Sent Events (SSE) in Python. In this guide we’ll explore why SSE is a perfect fit for real‑time web apps, walk through a complete implementation using Flask (and a quick look at Django and FastAPI), and share best practices to keep your code clean, scalable, and SEO‑friendly.
What Are Server‑Sent Events?
Server‑Sent Events are a standard HTML5 technology that lets a server push text‑based updates to a browser over a single, long‑lived HTTP connection. Unlike WebSockets, SSE is unidirectional (server → client) and uses the simple text/event-stream MIME type, making it easy to implement and friendly to firewalls and proxies.
Key Benefits of SSE
- Simple API: Only the
EventSourceobject is needed on the client side. - Automatic Reconnect: Browsers automatically retry lost connections.
- Built‑in Heartbeat: Keep‑alive messages prevent idle timeouts.
- SEO‑Friendly: Since the initial page is rendered server‑side, search engines can index content normally.
- Low Overhead: No binary framing or handshake—just plain UTF‑8 text.
When to Choose SSE Over WebSockets
SSE shines in scenarios where you need real‑time updates but don’t require bi‑directional communication. Typical use‑cases include:
- Live sports scores or stock tickers.
- Chat notifications (not the chat messages themselves).
- Server health dashboards.
- Progress bars for long‑running background jobs.
If your app needs client‑to‑server messaging (e.g., multiplayer games), WebSockets may be a better fit. For most notification‑style updates, SSE is the lighter, more maintainable choice.
Setting Up a Basic SSE Endpoint with Flask
Flask’s minimalistic design makes it a great playground for SSE. Below is a step‑by‑step example that streams random numbers to the browser every second.
1. Install Required Packages
pip install Flask
2. Create the Flask Application
from flask import Flask, Response, stream_with_context
import time
import random
app = Flask(__name__)
def generate_numbers():
"""Yield a new random number every second as an SSE message."""
while True:
# SSE format: "data: \n\n"
yield f"data: {random.randint(1, 100)}\n\n"
time.sleep(1)
@app.route('/stream')
def stream():
# Use stream_with_context to keep the generator alive across requests
return Response(stream_with_context(generate_numbers()),
mimetype='text/event-stream')
@app.route('/')
def index():
return '''
SSE Demo
Random Numbers (SSE)
'''
if __name__ == '__main__':
app.run(debug=True, threaded=True)
Key points to note:
- The
generate_numbersfunction yields strings in the SSE format (data: …\n\n). Responsewithmimetype='text/event-stream'tells the browser to treat the output as an event stream.- Setting
threaded=True(or using a production WSGI server) ensures each client gets its own connection.
3. Run and Test
Start the app with python app.py and navigate to http://localhost:5000. You should see a number updating every second without any page refresh.
Scaling SSE with Production Servers
While Flask’s built‑in server works for development, production environments need a robust WSGI server that supports streaming responses. Here are three popular choices:
- Gunicorn: Use the
geventworker class for asynchronous I/O.gunicorn -k gevent -w 4 app:app - Uvicorn (ASGI): Ideal for FastAPI or Starlette apps, also works with Flask via
asgiref.wsgi.WsgiToAsgi. - mod_wsgi (Apache): Configure
WSGIApplicationGroup %{GLOBAL}to allow streaming.
Regardless of the server, remember to enable HTTP/2 or HTTPS for better performance and security. Browsers still support SSE over HTTP/1.1, but TLS eliminates middle‑box interference.
Implementing SSE in Django
Django’s request/response cycle is synchronous by default, but with Django Channels you can serve SSE without leaving the Django ecosystem.
Minimal Django SSE View
# views.py
from django.http import StreamingHttpResponse
import time
import random
def sse_stream(request):
def event_generator():
while True:
yield f"data: {random.randint(1, 100)}\n\n"
time.sleep(1)
response = StreamingHttpResponse(event_generator(),
content_type='text/event-stream')
response['Cache-Control'] = 'no-cache'
return response
Add the URL pattern:
# urls.py
from django.urls import path
from . import views
urlpatterns = [
path('stream/', views.sse_stream, name='sse_stream'),
]
Combine this with a standard Django template that includes the same EventSource JavaScript shown earlier, and you have a fully functional SSE endpoint inside a full‑featured web framework.
FastAPI: Asynchronous SSE Made Easy
FastAPI is built on Starlette, which natively supports asynchronous streaming. The following example demonstrates a non‑blocking SSE endpoint that pushes the server’s current timestamp every second.
# main.py
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import asyncio
import datetime
app = FastAPI()
async def time_stream():
while True:
now = datetime.datetime.utcnow().isoformat()
yield f"data: {now}\\n\\n"
await asyncio.sleep(1)
@app.get("/time")
async def stream_time():
return StreamingResponse(time_stream(),
media_type="text/event-stream")
Run with uvicorn main:app --reload and open http://localhost:8000/time in the browser. The timestamps will appear instantly, proving that asynchronous SSE scales efficiently with many concurrent connections.
Advanced SSE Patterns
Sending Custom Event Types
Beyond the default message event, you can define custom event names using the event: field. This enables you to handle different payloads with distinct callbacks.
# Python generator example
def custom_events():
while True:
yield "event: price\\n"
yield "data: 123.45\\n\\n"
time.sleep(2)
yield "event: alert\\n"
yield "data: Threshold exceeded!\\n\\n"
time.sleep(3)
Client‑side handling:
const source = new EventSource('/custom');
source.addEventListener('price', e => console.log('Price:', e.data));
source.addEventListener('alert', e => alert(e.data));
Heartbeat / Keep‑Alive Messages
Some proxies close idle connections after 30 seconds. Sending a comment line (:) every 15 seconds prevents this:
def heartbeat():
while True:
yield ": keep-alive\\n\\n"
time.sleep(15)
Combine heartbeat with your data generator using itertools.chain or an async asyncio.gather loop.
Authentication & Security
- Token‑based auth: Verify a JWT or API key before returning the
StreamingResponse. Reject unauthorized requests with401. - CORS: Set
Access-Control-Allow-Originheader if your frontend lives on a different domain. - Rate limiting: Use Redis or a middleware to cap the number of concurrent streams per user.
Testing and Debugging SSE
Because SSE is a streaming protocol, traditional request‑response testing tools (like curl) need the -N flag to disable buffering:
curl -N http://localhost:5000/stream
For automated tests, consider Requests with stream=True or use httpx.AsyncClient in an async test
