Real-Time is More Than an Echo Route

Almost every tutorial on WebSockets in FastAPI starts and ends with a simplistic echo route:

# A toy implementation: echoes text, lacks authentication, leaks memory
@app.websocket("/ws")
async def echo_endpoint(websocket: WebSocket):
    await websocket.accept()
    while True:
        data = await websocket.receive_text()
        await websocket.send_text(f"Echo: {data}")

In production, real-time messaging is vastly more nuanced:

• How do you verify client identity safely without query param leaks?
• How do you clean up ghost connections when mobile networks drop abruptly?
• How do you broadcast events across horizontally scaled container replicas without maintaining state in a single worker’s memory?

---

1. Authentication: Never Pass Tokens in Query Parameters

A common mistake is authenticating WebSockets via URLs: /ws?token=eyJhbGci....

Query strings are logged in plaintext by Nginx reverse proxies, cloud load balancers, and CDN access logs. Anyone with access to system telemetry can steal customer bearer tokens.

The Secure Alternative: First-Message Handshake

Accept the WebSocket connection and immediately await a strongly-typed authentication payload with a short timeout. If valid, proceed; otherwise, close with an RFC-compliant policy violation code (1008):

@app.websocket("/ws/notifications")
async def notifications_ws(websocket: WebSocket):
    await websocket.accept()
    try:
        # Await auth frame within 5 seconds
        auth_message = await asyncio.wait_for(websocket.receive_json(), timeout=5.0)
        token = auth_message.get("token")
        user = await verify_jwt_token(token)
        if not user:
            await websocket.close(code=1008, reason="Unauthorized")
            return
    except asyncio.TimeoutError:
        await websocket.close(code=1008, reason="Authentication timeout")
        return

    await manager.subscribe(user.id, websocket)
    try:
        while True:
            await websocket.receive_text()
    except WebSocketDisconnect:
        manager.unsubscribe(user.id, websocket)

---

2. Preventing Memory Leaks with Cleanup Blocks

Mobile devices lose Wi-Fi signal, laptops close, and browser tabs crash without cleanly completing the TCP closure handshake.

If an exception occurs while transmitting to a dead socket, your connection manager must guarantee that dead sockets are pruned:

class ConnectionManager:
    def __init__(self):
        self.active_connections: dict[str, set[WebSocket]] = defaultdict(set)

    async def broadcast_to_user(self, user_id: str, message: dict):
        dead_sockets = set()
        for ws in self.active_connections.get(user_id, []):
            try:
                await ws.send_json(message)
            except Exception:
                dead_sockets.add(ws)

        for dead_ws in dead_sockets:
            self.unsubscribe(user_id, dead_ws)

---

3. Scaling Across Multiple Workers: Redis Pub/Sub Backbone

If your FastAPI application runs across four Uvicorn workers or ten Kubernetes pods, in-memory sets cannot share events across processes.

Use a Redis Pub/Sub channel as the shared messaging backbone:

# Publishing an event from any REST endpoint or worker
async def notify_user(user_id: str, payload: dict):
    await redis.publish(f"events:user:{user_id}", json.dumps(payload))

# Background listener running on each container instance
async def redis_listener(manager: ConnectionManager):
    pubsub = redis.pubsub()
    await pubsub.psubscribe("events:user:*")
    async for message in pubsub.listen():
        if message["type"] == "pmessage":
            channel = message["channel"].decode()
            user_id = channel.split(":")[-1]
            data = json.loads(message["data"])
            await manager.broadcast_to_user(user_id, data)

Every container listens to Redis and relays events strictly to the WebSocket clients currently connected to its own local process. This architecture scales linearly with zero cross-worker locks.