FastAPI WebSocket support with connection lifecycle management

WebSocket connections need lifecycle management (accept, receive, send, disconnect) and a connection registry for broadcasting to multiple clients.

Use FastAPI's WebSocket type. Always wrap in try/finally to handle disconnections. Use a connection manager class to track active connections.

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 connection in self.active:
            await connection.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)

WebSocket connections are long-lived. Without proper lifecycle management, disconnected clients remain in the registry causing send errors on broadcast. try/except WebSocketDisconnect ensures cleanup on client disconnect.

Building real-time features (chat, notifications, live updates) with FastAPI