FastAPI websockets let you push data to a client the moment it changes, without the overhead of polling. In a few lines you can expose a bidirectional channel, authenticate the user, and broadcast updates to every subscriber. Below I walk through a production-grade implementation: a simple endpoint, connection lifecycle, auth, a manager for broadcasting, and finally scaling with Uvicorn workers and Redis pub/sub.
How do I set up a basic WebSocket endpoint in FastAPI?
The first step is just a plain WebSocket route. FastAPI wraps Starlette’s WebSocket object, so you get async send/receive for free.
# app/main.py
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
app = FastAPI()
@app.websocket("/ws/echo")
async def websocket_echo(ws: WebSocket):
await ws.accept()
try:
while True:
data = await ws.receive_text()
await ws.send_text(f"ECHO: {data}")
except WebSocketDisconnect:
print("Client disconnected")
Run it with uvicorn app.main:app --reload. Open a browser console and type:
const ws = new WebSocket("ws://localhost:8000/ws/echo");
ws.onmessage = e => console.log(e.data);
ws.onopen = () => ws.send("hello");
You’ll see ECHO: hello printed. That’s the entire lifecycle in about 15 lines. In production you’ll probably want to move the loop into a helper class so you can reuse the same logic for many routes.
How can I manage the connection lifecycle (connect, receive, disconnect)?
A clean lifecycle is more than a try/except block. You need to track active sockets, clean up when a client drops, and optionally enforce a timeout. A tiny manager does the job:
# app/ws_manager.py
from typing import List
from fastapi import WebSocket, WebSocketDisconnect
class ConnectionManager:
def __init__(self):
self.active_connections: List[WebSocket] = []
async def connect(self, ws: WebSocket):
await ws.accept()
self.active_connections.append(ws)
def disconnect(self, ws: WebSocket):
self.active_connections.remove(ws)
async def send_personal_message(self, message: str, ws: WebSocket):
await ws.send_text(message)
async def broadcast(self, message: str):
for connection in self.active_connections:
await connection.send_text(message)
manager = ConnectionManager()
Now the endpoint becomes:
# app/main.py (continued)
from .ws_manager import manager
@app.websocket("/ws/chat")
async def websocket_chat(ws: WebSocket):
await manager.connect(ws)
try:
while True:
data = await ws.receive_text()
await manager.broadcast(f"User says: {data}")
except WebSocketDisconnect:
manager.disconnect(ws)
The manager isolates the state, making it easier to plug in Redis later. One thing that bites me in early prototypes is forgetting to remove the socket on disconnect – the list grows forever and you eventually hit memory limits. The disconnect method must be called even when the client aborts the TCP connection, which FastAPI surfaces as WebSocketDisconnect.
How do I add authentication and authorization for WebSocket connections?
WebSocket upgrades don’t carry the usual Authorization header the way HTTP does, but you can still validate a token that the client sends as a query param or as a sub-protocol header. I prefer a signed JWT in the query string because it works with browsers that don’t let you set custom headers on a WebSocket constructor.
# app/deps.py
from fastapi import HTTPException, status
from jose import jwt, JWTError
SECRET_KEY = "super-secret"
ALGORITHM = "HS256"
def get_current_user(token: str):
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
user_id: str = payload.get("sub")
if user_id is None:
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN)
return {"user_id": user_id}
except JWTError:
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN)
Now protect the endpoint:
# app/main.py (continued)
from fastapi import Depends, Query
from .deps import get_current_user
@app.websocket("/ws/secure")
async def websocket_secure(
ws: WebSocket,
token: str = Query(...),
user: dict = Depends(get_current_user)
):
await manager.connect(ws)
await manager.send_personal_message(f"Welcome {user['user_id']}", ws)
try:
while True:
data = await ws.receive_text()
# Example: only allow users with role "admin" to broadcast
if user.get("role") == "admin":
await manager.broadcast(f"[ADMIN] {data}")
else:
await manager.send_personal_message("You lack permission", ws)
except WebSocketDisconnect:
manager.disconnect(ws)
If the JWT is missing or invalid, FastAPI raises a 403 before the upgrade happens. This pattern works well with any async auth library, and it keeps the WebSocket code free of DB calls – you validate the token once and store the user dict in the connection’s scope.
When not to use this approach: If you need fine-grained per-message ACLs that depend on DB state, pulling the user from the DB on every message adds latency. In that case you may want to cache permissions in Redis and refresh them periodically.
How can I broadcast to multiple clients with a WebSocket manager?
The ConnectionManager.broadcast method shown earlier works for a single server process. It iterates over the in-memory list and pushes the same payload to every socket. That’s fine for a dev box, but once you run several Uvicorn workers behind a load balancer, each worker has its own list and they don’t see each other’s connections.
Enter Redis pub/sub. One worker publishes a message to a Redis channel; all workers subscribe and forward the payload to their local sockets.
# app/ws_redis_manager.py
import asyncio
import json
import aioredis
from .ws_manager import manager
REDIS_URL = "redis://localhost:6379"
CHANNEL = "fastapi_ws"
class RedisBroadcast:
def __init__(self):
self.redis = None
async def connect(self):
self.redis = await aioredis.from_url(REDIS_URL)
asyncio.create_task(self.listener())
async def listener(self):
pubsub = self.redis.pubsub()
await pubsub.subscribe(CHANNEL)
async for message in pubsub.listen():
if message["type"] == "message":
data = message["data"]
# Redis gives bytes; decode and forward
await manager.broadcast(data.decode())
async def publish(self, message: str):
await self.redis.publish(CHANNEL, message)
redis_broadcaster = RedisBroadcast()
Initialize the broadcaster on startup:
# app/main.py (continued)
@app.on_event("startup")
async def startup_event():
await redis_broadcaster.connect()
Now modify the manager’s broadcast to go through Redis:
# app/ws_manager.py (updated)
from .ws_redis_manager import redis_broadcaster
async def broadcast(self, message: str):
# Publish to Redis; every worker will receive it
await redis_broadcaster.publish(message)
With this setup, a message sent from any client reaches all connected clients, regardless of which worker they hit. The trade-off is an extra network hop and a small latency bump (usually < 30 ms). If you need sub-millisecond delivery, keep everything in a single process and use the in-memory loop.
How do I scale FastAPI websockets with Uvicorn workers and Redis pub/sub?
Running multiple workers is as simple as adding the --workers flag:
uvicorn app.main:app --host 0.0.0.0 --port 8000 --workers 4
Each worker spawns its own event loop, so you get true parallelism on multi-core machines. However, there are a few gotchas:
Sticky sessions – Browsers open a single WebSocket connection, so the load balancer must keep that connection on the same worker for the whole session. Most reverse proxies (NGINX, Traefik) support “IP hash” or “session affinity”. Without it, a client could be bounced between workers on reconnect, losing its previous state.
Redis connection limits – Each worker opens a pub/sub connection. If you run 8 workers on a small VM, you might hit the default
maxclientsof a Redis instance. Raise the limit or use a dedicated Redis for the pub/sub channel.Graceful shutdown – When you redeploy, Uvicorn sends a SIGTERM to each worker. The
WebSocketDisconnecthandler runs, but you still have pending messages in Redis. Add a shortawait asyncio.sleep(0.5)in the shutdown hook to let the listener drain.
@app.on_event("shutdown")
async def shutdown_event():
# Give the Redis listener a moment to finish pending publishes
await asyncio.sleep(0.5)
await redis_broadcaster.redis.close()
If you’re already using a message broker like RabbitMQ for other services, you could replace Redis with a topic exchange. The API is similar, but RabbitMQ gives you durability and better scaling for high-throughput streams.
When not to scale this way: If your traffic is under 100 concurrent sockets, a single worker is simpler. Adding Redis and multiple workers introduces operational overhead that may not pay off until you see contention on the event loop or memory pressure from many connections.
When should I avoid fastapi websockets altogether?
WebSockets shine for real-time dashboards, collaborative editors, or push notifications. They’re not a magic replacement for RESTful polling in every case. If the data changes rarely (once a day) or the client only needs a snapshot, a normal HTTP endpoint is cheaper - no persistent connection, no extra Redis, no connection-lifetime bugs.
Also, be careful with binary payloads. FastAPI can handle binary frames, but if you need to stream large files you might be better off with signed URLs to an object store (S3, GCS). Trying to push multi-gigabyte blobs over a WebSocket will saturate memory and cause the server to OOM.
Real-world pitfalls I’ve hit
Session leaks – I once left a
WebSocketobject in a global list after a network glitch. The memory grew until the process crashed. The fix was to always wrapdisconnectin afinallyblock and add a watchdog that prunes dead sockets every minute. (If you’re curious about session-leak detection, see my post on FastAPI SQLAlchemy Session Leak Detection.)Unexpected 403 on upgrade – I tried to read the JWT from a cookie inside the WebSocket handler. Browsers don’t send cookies on the upgrade request unless you set
withCredentialson the client, which I hadn’t. Switching to a query-string token solved the issue.Redis back-pressure – During a load test with 5 000 concurrent sockets, the Redis channel became a bottleneck. The publish command blocked, causing the event loop to stall. The remedy was to switch to a Redis Stream and let workers pull batches, or to shard the channel by topic.
FAQ
Can I use fastapi websockets with a custom protocol like GraphQL subscriptions?
Yes. FastAPI lets you read the subprotocols attribute on the WebSocket object and negotiate a protocol. After the handshake you can feed messages to a GraphQL subscription engine the same way you would to any consumer.
Do I need to run Uvicorn with --loop uvloop for websockets?
uvloop gives a modest speed boost, but it isn’t required. The default asyncio loop handles thousands of concurrent websockets just fine. Measure first; if latency is already sub-10 ms, the extra complexity isn’t worth it.
Is there a limit to how many websockets a single FastAPI worker can hold?
The limit is mainly the OS file-descriptor count and memory per socket. On Linux the default is 1024 descriptors per process; raise it with ulimit -n. Memory usage per idle socket is tiny (a few KiB), but keep an eye on the total if you expect >20 000 connections.
How do I debug a dropped connection that shows no error in the server logs?
Enable WebSocket logging: uvicorn app.main:app --log-level debug. Look for websocket.disconnect messages. Also, capture network traces with tcpdump to see if the client sent a FIN or if a middlebox timed out the idle connection.
Key Takeaways
- A basic fastapi websockets endpoint is only a few lines; the real work starts with connection management.
- Use a
ConnectionManagerto centralize connect, disconnect, and broadcast logic. - Authenticate the upgrade request with a JWT passed as a query param or sub-protocol header; validate once and store the user in the connection scope.
- For multi-worker deployments, replace in-memory broadcasting with Redis pub/sub (or another broker) to keep all sockets in sync.
- Scale with Uvicorn workers, enable sticky sessions on the load balancer, and monitor Redis client limits.
- Avoid websockets for low-frequency data or large binary streams; a simple HTTP endpoint or signed URL is cheaper and safer.
Happy coding, and may your sockets stay alive.
Top comments (0)