Build a WebSocket chat endpoint in FastAPI

Approach

Accept WebSocket connections per room, authenticate during the handshake (token query param or cookie), and fan out messages via Redis pub/sub so users connected to different instances receive them. Each connection runs a receive loop and a subscriber task; clean up on disconnect.

Solution

import asyncio, json
from fastapi import WebSocket, WebSocketDisconnect, status
from redis.asyncio import Redis

redis = Redis.from_url("redis://redis:6379/0")


@app.websocket("/ws/rooms/{room_id}")
async def room(websocket: WebSocket, room_id: str):
    user = await authenticate_ws(websocket)            # e.g. validate token from the cookie or a query param
    if user is None:
        await websocket.close(code=status.WS_1008_POLICY_VIOLATION)
        return
    await websocket.accept()
    pubsub = redis.pubsub()
    await pubsub.subscribe(f"room:{room_id}")

    async def forward_from_redis():
        async for message in pubsub.listen():
            if message["type"] == "message":
                await websocket.send_text(message["data"].decode())

    forwarder = asyncio.create_task(forward_from_redis())
    try:
        while True:
            text = (await websocket.receive_text())[:2000]
            payload = json.dumps({"user": user.name, "text": text})
            await redis.publish(f"room:{room_id}", payload)
    except WebSocketDisconnect:
        pass
    finally:
        forwarder.cancel()
        await pubsub.unsubscribe(f"room:{room_id}")
        await pubsub.aclose()

Pitfalls

• In-memory connection lists that only work on a single instance.
• No authentication or authorization per room.
• Unbounded message sizes and rates (add limits).

Variations

• Persist messages to a database for history; pub/sub is fire-and-forget.
• Use Redis Streams or a managed service when delivery guarantees matter.

Further reading: FastAPI: WebSockets