FastAPI WebSockets

HTTP is a one-way street — the client asks, the server answers, and the connection closes. WebSockets open a two-way channel that stays open. The server can push data to the client any time without the client asking. This makes WebSockets ideal for chat apps, live notifications, dashboards with live data, and multiplayer games.

HTTP vs WebSocket — Side by Side

HTTP (request-response):
  Client → "Give me the latest score"
  Server → "3-2"    [connection closes]
  Client → "Give me the latest score" (asks again)
  Server → "4-2"    [connection closes]
  ... (polling every few seconds, wasteful)

WebSocket (persistent connection):
  Client connects once → connection stays open
  Server → "3-2"
  Server → "4-2"    (pushed instantly when score changes)
  Server → "5-2"
  ... (no repeated requests needed)

A Basic WebSocket Endpoint

from fastapi import FastAPI, WebSocket

app = FastAPI()

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()          ← open the connection

    while True:
        data = await websocket.receive_text()   ← wait for message
        await websocket.send_text(f"Echo: {data}")  ← reply

WebSocket lifecycle:
  1. Client connects to ws://localhost:8000/ws
  2. Server calls websocket.accept()
  3. Both sides can now send and receive freely
  4. Connection stays open until one side closes it

Testing a WebSocket with JavaScript

// Run this in your browser console
const ws = new WebSocket("ws://localhost:8000/ws");

ws.onopen = () => {
    ws.send("Hello from browser");
};

ws.onmessage = (event) => {
    console.log("Server says:", event.data);
    // prints: "Echo: Hello from browser"
};

Sending and Receiving Different Data Types

# Text
data = await websocket.receive_text()
await websocket.send_text("hello")

# JSON
data = await websocket.receive_json()   ← parses JSON automatically
await websocket.send_json({"status": "ok", "count": 42})

# Raw bytes (for files or binary data)
data = await websocket.receive_bytes()
await websocket.send_bytes(b"\x00\x01\x02")

Managing Multiple Connected Clients

In a real chat app, many users connect at the same time. A connection manager tracks all active connections and broadcasts messages to everyone:

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("/chat")
async def chat(websocket: WebSocket):
    await manager.connect(websocket)
    try:
        while True:
            message = await websocket.receive_text()
            await manager.broadcast(f"User: {message}")
    except:
        manager.disconnect(websocket)

3 users connected:
  User A sends "Hi"
  manager.broadcast("User: Hi")
     → User A receives it
     → User B receives it
     → User C receives it

WebSocket Connection Lifecycle

Connect   → websocket.accept()
Send      → websocket.send_text() / send_json()
Receive   → websocket.receive_text() / receive_json()
Disconnect → WebSocketDisconnect exception raised
             handle in except block: manager.disconnect(ws)

Path and Query Parameters in WebSocket Routes

@app.websocket("/chat/{room_id}")
async def room_chat(room_id: str, websocket: WebSocket, token: str = None):
    # room_id from path, token from query: ws://host/chat/room1?token=abc
    await websocket.accept()
    ...

Key Points

• WebSockets keep a persistent two-way connection open between client and server.
• Use @app.websocket("/path") to define a WebSocket endpoint.
• Always call await websocket.accept() to open the connection.
• Handle WebSocketDisconnect to clean up when a client leaves.
• Use a connection manager class to track multiple clients and broadcast messages.