TopolearnChat, live dashboards and collaborative editing need the server to push messages without the client polling.
A WebSocket starts as an HTTP request and is then upgraded to a persistent two-way channel. Declare an endpoint with @app.websocket("/ws/{room}") and an async def that takes a WebSocket:
await websocket.accept()completes the handshake. Until you accept, you can still reject the connection.receive_text(),receive_json()andreceive_bytes()wait for the next message;send_text(),send_json()andsend_bytes()push one.- When the client goes away, the next receive raises
WebSocketDisconnect. Catch it to clean up, such as removing the connection from a room.
Path and query parameters, Depends and Annotated work as usual. Authentication needs care: browsers can't set custom headers like Authorization on a WebSocket, so pass a token as a query parameter or rely on a cookie. To refuse, raise WebSocketException(code=status.WS_1008_POLICY_VIOLATION); HTTPException doesn't apply once you're talking WebSocket.
Each open connection is a coroutine parked on await, so thousands are cheap, provided you never block the loop. Use async libraries throughout.
The classic "connection manager" that keeps a list of sockets and loops over them to broadcast only works within one process. With several workers or replicas, a user connected to worker A never sees messages sent from worker B. For real fan-out, publish through Redis pub/sub, NATS, Postgres LISTEN/NOTIFY or a managed service, with each worker forwarding to its own sockets.
In production, make sure your proxy or load balancer passes the Upgrade headers and that idle timeouts are longer than your ping interval. Test WebSockets with TestClient.websocket_connect.
Example
from typing import Annotated
from fastapi import FastAPI, Query, WebSocket, WebSocketDisconnect, WebSocketException, status
from fastapi.testclient import TestClient
app = FastAPI()
rooms: dict[str, list[WebSocket]] = {}
@app.websocket("/ws/{room}")
async def chat(websocket: WebSocket, room: str, token: Annotated[str, Query()]):
if token != "valid":
raise WebSocketException(code=status.WS_1008_POLICY_VIOLATION)
await websocket.accept()
rooms.setdefault(room, []).append(websocket)
try:
while True:
text = await websocket.receive_text()
for peer in rooms[room]: # in-process broadcast only
await peer.send_json({"room": room, "text": text})
except WebSocketDisconnect:
rooms[room].remove(websocket)
with TestClient(app).websocket_connect("/ws/lobby?token=valid") as ws:
ws.send_text("hi")
print(ws.receive_json()) # {'room': 'lobby', 'text': 'hi'}Remember
- @app.websocket + await websocket.accept() opens the channel
- WebSocketDisconnect signals the client left: clean up there
- Authenticate with a query token or cookie; reject with WebSocketException
- In-memory broadcasting only reaches clients on the same process
Watch out for
- Shipping an in-memory connection manager to a multi-worker deployment and losing messages between workers.