The official FastAPI WebSocket tutorial fits in twenty lines. That is sufficient to understand the API, not to build something reliable. Here is what the examples do not show: authentication, distributed state management, broadcasting, and dead connection cleanup.
The Problem with the Basic Example
# What every tutorial shows
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
await websocket.accept()
while True:
data = await websocket.receive_text()
await websocket.send_text(f"Message: {data}")
This code has several production problems: no authentication, no handling of unexpected disconnections, no broadcasting to other clients, and state that does not survive a pod restart.
ConnectionManager: Handling Multiple Clients
The first step is a manager that maintains the list of active connections:
import asyncio
from fastapi import WebSocket
from typing import Any
import logging
logger = logging.getLogger(__name__)
class ConnectionManager:
def __init__(self):
# user_id -> list of websockets (a user may have multiple tabs open)
self._connections: dict[str, list[WebSocket]] = {}
self._lock = asyncio.Lock()
async def connect(self, websocket: WebSocket, user_id: str) -> None:
await websocket.accept()
async with self._lock:
if user_id not in self._connections:
self._connections[user_id] = []
self._connections[user_id].append(websocket)
logger.info(f"WebSocket connected: user={user_id}, total={self.total_connections}")
async def disconnect(self, websocket: WebSocket, user_id: str) -> None:
async with self._lock:
if user_id in self._connections:
self._connections[user_id] = [
ws for ws in self._connections[user_id] if ws != websocket
]
if not self._connections[user_id]:
del self._connections[user_id]
logger.info(f"WebSocket disconnected: user={user_id}")
async def send_to_user(self, user_id: str, message: dict) -> None:
"""Sends a message to all connections belonging to a given user."""
connections = self._connections.get(user_id, [])
dead_connections = []
for websocket in connections:
try:
await websocket.send_json(message)
except Exception:
dead_connections.append(websocket)
for ws in dead_connections:
await self.disconnect(ws, user_id)
async def broadcast(self, message: dict, exclude_user: str | None = None) -> None:
"""Broadcasts a message to all connected users."""
tasks = []
for user_id in list(self._connections.keys()):
if user_id != exclude_user:
tasks.append(self.send_to_user(user_id, message))
await asyncio.gather(*tasks, return_exceptions=True)
@property
def total_connections(self) -> int:
return sum(len(ws_list) for ws_list in self._connections.values())
manager = ConnectionManager()
The asyncio.Lock() protects dictionary modifications — in Python, dict operations are not safe across concurrent coroutines. A user may have several simultaneous connections (multiple open tabs), which the dict[str, list[WebSocket]] structure handles natively.
Authenticating a WebSocket Connection
This is where most implementations stumble. WebSockets do not support custom HTTP headers from the browser — it is not possible to send Authorization: Bearer ... in the initial handshake via the standard browser WebSocket API.
Two viable approaches:
Approach 1: Token in the Query Parameter
from fastapi import WebSocket, WebSocketException, status, Depends, Query
from app.core.security import verify_token
async def get_websocket_user(
websocket: WebSocket,
token: str = Query(...),
) -> str:
"""Extracts and validates the user from the query parameter token."""
payload = verify_token(token)
if payload is None:
await websocket.close(code=status.WS_1008_POLICY_VIOLATION)
raise WebSocketException(code=status.WS_1008_POLICY_VIOLATION)
return payload["sub"]
@app.websocket("/ws/notifications")
async def notifications_ws(
websocket: WebSocket,
user_id: str = Depends(get_websocket_user),
):
await manager.connect(websocket, user_id)
try:
while True:
await websocket.receive_text() # Keep the connection alive
except Exception:
await manager.disconnect(websocket, user_id)
On the Vue.js client side:
// composables/useWebSocket.ts
export function useWebSocket() {
const token = useCookie("access_token")
const ws = ref<WebSocket | null>(null)
const connect = () => {
ws.value = new WebSocket(
`wss://api.myapp.com/ws/notifications?token=${token.value}`,
)
ws.value.onmessage = (event) => {
const message = JSON.parse(event.data)
handleMessage(message)
}
ws.value.onclose = () => {
// Automatic reconnection after 3 seconds
setTimeout(connect, 3000)
}
}
onMounted(connect)
onUnmounted(() => ws.value?.close())
return { ws }
}
Approach 2: Session Cookie (More Secure)
If you are using the BFF pattern with HttpOnly cookies, the WebSocket connection automatically sends the domain cookies — this is the browser’s native behaviour:
@app.websocket("/ws/notifications")
async def notifications_ws(
websocket: WebSocket,
session: dict = Depends(get_websocket_session),
):
user_id = session["user_id"]
await manager.connect(websocket, user_id)
# ...
async def get_websocket_session(websocket: WebSocket) -> dict:
session_id = websocket.cookies.get("session_id")
if not session_id:
await websocket.close(code=status.WS_1008_POLICY_VIOLATION)
raise WebSocketException(code=status.WS_1008_POLICY_VIOLATION)
session = await session_manager.get_session_by_id(session_id)
if not session:
await websocket.close(code=status.WS_1008_POLICY_VIOLATION)
raise WebSocketException(code=status.WS_1008_POLICY_VIOLATION)
return session
The cookie approach is preferable on a BFF — the token never appears in plaintext in the URL, which would otherwise be visible in server logs.
Handling Unexpected Disconnections Cleanly
A WebSocket connection can die in several ways: the user closes the tab, the network drops, the client pod restarts. These cases must be detected and cleaned up:
@app.websocket("/ws/notifications")
async def notifications_ws(
websocket: WebSocket,
user_id: str = Depends(get_websocket_user),
):
await manager.connect(websocket, user_id)
try:
while True:
try:
# Timeout on receive — detects dead connections
data = await asyncio.wait_for(
websocket.receive_text(),
timeout=30.0
)
await handle_client_message(user_id, data)
except asyncio.TimeoutError:
# Send a ping to verify the client is still alive
try:
await websocket.send_json({"type": "ping"})
except Exception:
break # Dead connection — exit the loop
except Exception as e:
logger.info(f"WebSocket closed for user={user_id}: {type(e).__name__}")
finally:
# Cleanup guaranteed regardless of how the connection ended
await manager.disconnect(websocket, user_id)
The try/finally pattern around the main loop guarantees that disconnect is always called, regardless of the reason for closure.
Broadcasting Events From Anywhere in the Application
The real use case: a backend process completes and needs to notify connected clients in real time.
class CertificateService:
def __init__(self, repo: CertificateRepository, ws_manager: ConnectionManager):
self.repo = repo
self.ws_manager = ws_manager
async def process_certificate(self, cert_id: str, user_id: str) -> Certificate:
certificate = await self.repo.process(cert_id)
# Notify the user in real time
await self.ws_manager.send_to_user(user_id, {
"type": "certificate_processed",
"data": {
"id": certificate.id,
"status": certificate.status,
"volume": certificate.volume,
}
})
return certificate
The ConnectionManager is injected as a FastAPI dependency — a singleton shared across the entire process:
# app/api/dependencies.py
from app.api.websockets import manager
def get_ws_manager() -> ConnectionManager:
return manager
The Multi-Pod Problem: Distributed State with Redis Pub/Sub
The ConnectionManager as described above has a critical limitation: it is in-memory. On a multi-pod OpenShift deployment with three replicas, each pod has its own manager. An event processed on pod A will not be broadcast to clients connected to pod B or C.
The solution: Redis Pub/Sub as an inter-pod event bus.
import redis.asyncio as aioredis
import json
import asyncio
class DistributedConnectionManager(ConnectionManager):
def __init__(self, redis: aioredis.Redis):
super().__init__()
self.redis = redis
self.channel = "ws:broadcast"
async def publish_to_user(self, user_id: str, message: dict) -> None:
"""Publishes an event to Redis — all pods receive it."""
await self.redis.publish(
f"ws:user:{user_id}",
json.dumps(message)
)
async def publish_broadcast(self, message: dict) -> None:
"""Publishes a broadcast to Redis."""
await self.redis.publish(self.channel, json.dumps(message))
async def start_subscriber(self) -> None:
"""To be started at pod startup — listens for Redis events."""
pubsub = self.redis.pubsub()
await pubsub.psubscribe("ws:user:*", self.channel)
async for message in pubsub.listen():
if message["type"] != "pmessage" and message["type"] != "message":
continue
channel = message["channel"].decode()
data = json.loads(message["data"])
if channel == self.channel:
# Local broadcast to clients on THIS pod
await super().broadcast(data)
elif channel.startswith("ws:user:"):
user_id = channel.split(":")[-1]
# Send to clients on THIS pod for this user
await super().send_to_user(user_id, data)
Starting the subscriber in the FastAPI lifespan:
@asynccontextmanager
async def lifespan(app: FastAPI):
task = asyncio.create_task(distributed_manager.start_subscriber())
background_tasks.add(task)
task.add_done_callback(background_tasks.discard)
yield
task.cancel()
await asyncio.gather(task, return_exceptions=True)
With this architecture, a service on pod A calls publish_to_user() — Redis propagates the event to all pods, and each pod delivers it locally to the relevant clients.
Key Takeaways
Production-grade WebSockets require solving four distinct problems: authentication (cookie or query parameter depending on the architecture), dead connection handling (ping/timeout with try/finally), broadcasting to multiple connections per user (list[WebSocket] per user_id), and multi-pod distribution (Redis Pub/Sub). Each problem is straightforward in isolation — it is their combination that determines whether a WebSocket implementation holds up under real-world conditions.