WebSockets in FastAPI

Difficulty: Advanced

Question

How does FastAPI support WebSockets? Describe the basic pattern.

Answer

FastAPI supports WebSockets natively via Starlette. Define a WebSocket endpoint using `@app.websocket('/ws')` and accept a `WebSocket` parameter. The connection lifecycle:

1. `await websocket.accept()` - handshake 2. Loop: `await websocket.receive_text()` / `receive_json()` / `receive_bytes()` 3. `await websocket.send_text()` / `send_json()` / `send_bytes()` 4. Handle `WebSocketDisconnect` exception for clean disconnection

For multi-client broadcasting, maintain a connection manager that tracks active connections.

Code examples

WebSocket chat example

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from typing import List

app = FastAPI()

class ConnectionManager:
    def __init__(self):
        self.active: List[WebSocket] = []

    async def connect(self, ws: WebSocket):
        await ws.accept()
        self.active.append(ws)

    def disconnect(self, ws: WebSocket):
        self.active.remove(ws)

    async def broadcast(self, message: str):
        for ws in self.active:
            await ws.send_text(message)

manager = ConnectionManager()

@app.websocket("/ws/{client_id}")
async def websocket_endpoint(websocket: WebSocket, client_id: str):
    await manager.connect(websocket)
    try:
        while True:
            data = await websocket.receive_text()
            await manager.broadcast(f"{client_id}: {data}")
    except WebSocketDisconnect:
        manager.disconnect(websocket)
        await manager.broadcast(f"{client_id} left the chat")

Key points

• WebSocket connections are long-lived - handle disconnects with try/except
• The connection manager pattern is essential for multi-client scenarios
• For production, consider Redis Pub/Sub for multi-instance broadcasting

Concepts covered

WebSocket, real-time, bidirectional, connection manager