SSE with FastAPI — Complete Guide
In this tutorial, you will learn about SSE with FastAPI. We cover key concepts, practical examples, and best practices to help you master this topic.
Implement Server-Sent Events in FastAPI using StreamingResponse, async generators, Dependency Injection, and background tasks for real-time streaming with Python async support.
What You Learn
You will learn how to implement SSE endpoints in FastAPI using async generators and StreamingResponse, handle client disconnection with asyncio, use dependency injection for SSE, and broadcast events with background tasks.
Why It Matters
FastAPI's async support makes it ideal for SSE. Unlike Django's synchronous workers, FastAPI handles thousands of concurrent SSE connections without blocking. Its dependency injection system integrates seamlessly with SSE authentication and validation.
Real-World Use
DodaTech's FastAPI-based API Gateway uses SSE for real-time API metrics. Each connected admin sees request rates, error rates, and latency distributions. FastAPI handles 5000+ concurrent SSE connections on a single instance.
Basic SSE Endpoint
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import asyncio
import json
app = FastAPI()
@app.get("/events")
async def sse_events():
async def event_generator():
counter = 0
try:
while True:
counter += 1
data = json.dumps({
"count": counter,
"timestamp": __import__("time").time(),
})
yield f"id: {counter}\ndata: {data}\n\n"
await asyncio.sleep(1)
if counter >= 10:
break
except asyncio.CancelledError:
print("Client disconnected")
return StreamingResponse(
event_generator(),
media_type="text/event-stream",
headers={
"Cache-Control": "no-cache",
"X-Accel-Buffering": "no",
},
)
Client Disconnection
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
import asyncio
import json
app = FastAPI()
@app.get("/stream")
async def stream_events(request: Request):
async def event_generator():
try:
yield f"event: connected\ndata: {json.dumps({'status': 'streaming'})}\n\n"
while True:
# Check if client is still connected
if await request.is_disconnected():
print("Client disconnected, stopping stream")
break
data = json.dumps({
"message": "update",
"time": __import__("time").time(),
})
yield f"event: update\ndata: {data}\n\n"
await asyncio.sleep(1)
except asyncio.CancelledError:
print("Stream cancelled")
finally:
print("Cleaning up stream resources")
return StreamingResponse(
event_generator(),
media_type="text/event-stream",
headers={"Cache-Control": "no-cache"},
)
Dependency Injection for SSE
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.responses import StreamingResponse
import asyncio
import json
app = FastAPI()
# Authentication dependency
async def verify_sse_token(token: str = None):
if token is None:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Missing token",
)
# Verify token (simplified)
if token != "valid-token":
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Invalid token",
)
return {"user": "authenticated", "token": token}
# Channel authorization
async def verify_channel_access(channel: str, user: dict = Depends(verify_sse_token)):
allowed_channels = ["public", "metrics", "logs"]
if channel not in allowed_channels:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Channel not found",
)
return {"channel": channel, "user": user}
@app.get("/sse/{channel}")
async def sse_channel(
channel_info: dict = Depends(verify_channel_access),
):
channel = channel_info["channel"]
user = channel_info["user"]
async def event_generator():
yield f"event: connected\ndata: {json.dumps({'channel': channel, 'user': user['user']})}\n\n"
for i in range(20):
data = json.dumps({
"channel": channel,
"event_id": i,
"timestamp": __import__("time").time(),
})
yield f"event: {channel}_update\ndata: {data}\n\n"
await asyncio.sleep(1)
return StreamingResponse(
event_generator(),
media_type="text/event-stream",
headers={"Cache-Control": "no-cache"},
)
Broadcasting with Background Tasks
from fastapi import FastAPI, BackgroundTasks
from fastapi.responses import StreamingResponse
import asyncio
import json
app = FastAPI()
class SSEBroadcaster:
def __init__(self):
self.subscribers = []
self._lock = asyncio.Lock()
async def subscribe(self):
queue = asyncio.Queue()
async with self._lock:
self.subscribers.append(queue)
return queue
async def unsubscribe(self, queue):
async with self._lock:
self.subscribers.remove(queue)
async def broadcast(self, event, data):
message = f"event: {event}\ndata: {json.dumps(data)}\n\n"
async with self._lock:
for subscriber in self.subscribers:
await subscriber.put(message)
async def publisher(self):
counter = 0
while True:
counter += 1
await self.broadcast("update", {
"count": counter,
"time": __import__("time").time(),
})
await asyncio.sleep(2)
broadcaster = SSEBroadcaster()
@app.on_event("startup")
async def start_broadcaster():
asyncio.create_task(broadcaster.publisher())
@app.get("/subscribe")
async def subscribe():
async def event_generator():
queue = await broadcaster.subscribe()
try:
yield f"event: connected\ndata: {json.dumps({'status': 'subscribed'})}\n\n"
while True:
message = await queue.get()
yield message
except asyncio.CancelledError:
pass
finally:
await broadcaster.unsubscribe(queue)
return StreamingResponse(
event_generator(),
media_type="text/event-stream",
headers={"Cache-Control": "no-cache"},
)
@app.post("/publish")
async def publish(event: str = "message", data: dict = {}):
await broadcaster.broadcast(event, data)
return {"published": True, "subscribers": len(broadcaster.subscribers)}
Event Types and Filtering
from fastapi import FastAPI, Query
from fastapi.responses import StreamingResponse
import asyncio
import json
app = FastAPI()
@app.get("/sse")
async def sse_with_filters(
events: str = Query(None, description="Comma-separated event types"),
):
allowed_events = set(events.split(",")) if events else None
async def event_generator():
event_types = ["notification", "alert", "metrics", "heartbeat"]
counter = 0
try:
yield f"event: connected\ndata: {json.dumps({'filter': events or 'all'})}\n\n"
while True:
counter += 1
event_type = event_types[counter % len(event_types)]
if allowed_events and event_type not in allowed_events:
await asyncio.sleep(1)
continue
data = json.dumps({
"type": event_type,
"id": counter,
"timestamp": __import__("time").time(),
"value": counter * 10,
})
yield f"event: {event_type}\ndata: {data}\n\n"
await asyncio.sleep(1)
except asyncio.CancelledError:
pass
return StreamingResponse(
event_generator(),
media_type="text/event-stream",
headers={"Cache-Control": "no-cache"},
)
Common Mistakes
1. Not Using async Generator
Using a regular (sync) generator blocks the event loop. SSE generators must be async to avoid blocking other requests.
2. Ignoring CancelledError
When the client disconnects, FastAPI cancels the task. Catch asyncio.CancelledError to clean up resources.
3. No Connection Check
For long-running streams, periodically check request.is_disconnected(). This detects client-side disconnection faster.
4. Using json.dumps for Every Event
Pre-compute JSON when possible. For high-frequency events, string formatting is faster than repeated json.dumps calls.
5. Blocking Operations Inside Generator
Avoid blocking calls (time.sleep, sync I/O) in the async generator. Use asyncio.sleep and async I/O.
Practice Questions
1. What FastAPI class is used for SSE responses?
StreamingResponse with media_type="text/event-stream". It streams content from an async generator.
2. How do you detect client disconnection in FastAPI SSE?
Use await request.is_disconnected() or catch asyncio.CancelledError when the generator is cancelled.
3. How do you broadcast events to multiple SSE clients?
Use a broadcaster pattern with asyncio.Queue. Each client subscribes with a queue. A publisher task broadcasts to all queues.
4. Can FastAPI handle 10K concurrent SSE connections?
Yes. FastAPI with ASGI (uvicorn) handles thousands of concurrent connections because SSE endpoints are async and non-blocking.
Challenge
Build a FastAPI SSE system for a live analytics dashboard: SSE endpoint with token authentication, three event types (pageviews, errors, latency), Redis pub/sub for broadcasting across multiple server instances, client-side event filtering by type, and connection count tracking.
FAQ
Mini Project: FastAPI SSE Dashboard
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
import asyncio
import json
import random
app = FastAPI()
@app.get("/dashboard/stream")
async def dashboard_stream(request: Request):
async def event_generator():
try:
yield f"event: connected\ndata: {json.dumps({'status': 'dashboard live'})}\n\n"
while True:
if await request.is_disconnected():
break
metrics = {
"cpu": round(random.uniform(10, 90), 1),
"memory": round(random.uniform(40, 80), 1),
"requests": random.randint(100, 5000),
"errors": random.randint(0, 10),
"latency_ms": round(random.uniform(20, 200), 1),
}
yield f"event: metrics\ndata: {json.dumps(metrics)}\n\n"
yield f": heartbeat\n\n"
await asyncio.sleep(2)
except asyncio.CancelledError:
pass
return StreamingResponse(
event_generator(),
media_type="text/event-stream",
headers={
"Cache-Control": "no-cache",
"X-Accel-Buffering": "no",
},
)
What's Next
Now that you understand SSE with FastAPI, learn SSE with plain Node.js HTTP, then explore event types and custom events.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro