← FastAPI

WebSockets & Real-time Features

← WebSockets & Real-time Features

Select a question from the sidebar

12 questions in this topic

Question 1 / 12

How do you create WebSocket endpoints in FastAPI?

FastAPI provides WebSocket support through Starlette's WebSocket class. You define WebSocket endpoints using the @app.websocket() decorator instead of the usual @app.get() or @app.post().

How it works:

Decorate an async function with @app.websocket("/path")
The function receives a WebSocket object as its first parameter
Call await websocket.accept() to complete the handshake
Use await websocket.receive_text() or receive_bytes() to read messages
Use await websocket.send_text() or send_bytes() to send messages
Close with await websocket.close()

Key differences from HTTP endpoints:

WebSocket endpoints are persistent connections, not request-response
They do not return a value -- communication happens through the websocket object
Path parameters and query parameters still work
Response models and status codes do not apply

Path parameters and query parameters:

WebSocket endpoints support the same path and query parameter extraction as regular endpoints. You can use type annotations and FastAPI will validate them.

Pseudo Code

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    while True:
        data = await websocket.receive_text()
        await websocket.send_text(f"Echo: {data}")

Code

from fastapi import FastAPI, WebSocket
import asyncio

app = FastAPI()


# --- Basic WebSocket endpoint ---
@app.websocket("/ws")
async def websocket_echo(websocket: WebSocket):
    """Simple echo server -- sends back whatever it receives."""
    await websocket.accept()
    while True:
        data = await websocket.receive_text()
        await websocket.send_text(f"Echo: {data}")


# --- With path parameters ---
@app.websocket("/ws/rooms/{room_id}")
async def room_websocket(
    websocket: WebSocket,
    room_id: int,
):
    """Path parameters work just like HTTP endpoints."""
    await websocket.accept()
    await websocket.send_text(f"Connected to room {room_id}")
    while True:
        data = await websocket.receive_text()
        await websocket.send_text(f"[Room {room_id}] {data}")


# --- With query parameters ---
@app.websocket("/ws/chat")
async def chat_websocket(
    websocket: WebSocket,
    username: str = "anonymous",
):
    """Query parameters extracted from the ws:// URL."""
    await websocket.accept()
    await websocket.send_text(f"Welcome, {username}!")
    while True:
        message = await websocket.receive_text()
        await websocket.send_text(f"{username}: {message}")


# --- Client-side JavaScript example ---
# const ws = new WebSocket("ws://localhost:8000/ws/chat?username=alice");
# ws.onmessage = (event) => console.log(event.data);
# ws.onopen = () => ws.send("Hello!");

Key Points

•Use @app.websocket("/path") to define WebSocket endpoints
•Always call await websocket.accept() before sending or receiving
•Path and query parameters work identically to HTTP endpoints
•WebSocket endpoints are long-lived -- they run until the connection closes or an error occurs

Gotchas

•WebSocket endpoints cannot use response_model or status_code -- they communicate through the WebSocket object directly
•Forgetting to call websocket.accept() will leave the client hanging with no handshake completion
•A bare while True loop will run forever if the client disconnects without sending a close frame -- always handle WebSocketDisconnect
•WebSocket URLs use ws:// or wss:// scheme, not http:// or https://

Use @app.websocket("/path") to create WebSocket endpoints. Accept the connection with await websocket.accept(), then use receive_text()/send_text() in a loop. Path and query parameters work the same as HTTP endpoints.

•@app.websocket() defines persistent bidirectional connections

from fastapi import FastAPI, WebSocket

app = FastAPI()

@app.websocket("/ws/{room_id}")
async def websocket_endpoint(websocket: WebSocket, room_id: int):
    await websocket.accept()
    while True:
        data = await websocket.receive_text()
        await websocket.send_text(f"[Room {room_id}] Echo: {data}")

You must call websocket.accept() before any send/receive operations

Question 2 / 12

How does the WebSocket connection lifecycle work (accept, receive, send)?

A WebSocket connection follows a strict lifecycle with three phases: handshake, data transfer, and close.

1. Handshake (HTTP Upgrade):

Client sends an HTTP GET request with Upgrade: websocket header
Server responds with 101 Switching Protocols
In FastAPI, await websocket.accept() completes this handshake

2. Data Transfer:

Both sides can send messages at any time (full-duplex)
Messages can be text (receive_text()/send_text()) or binary (receive_bytes()/send_bytes())
receive_json()/send_json() are convenience wrappers for JSON
The generic receive() returns a dict with a type key indicating the message type

3. Close:

Either side can initiate close with a close frame
In FastAPI, await websocket.close(code=1000) sends a clean close
When the client closes, receive_*() raises WebSocketDisconnect

WebSocket state transitions:

CONNECTING -> accept() -> CONNECTED -> close()/disconnect -> DISCONNECTED

Common close codes:

1000: Normal closure
1001: Going away (browser tab closed)
1008: Policy violation
1011: Unexpected server error

Pseudo Code

# Full lifecycle
@app.websocket("/ws")
async def ws(websocket: WebSocket):
    # Phase 1: Handshake
    await websocket.accept()
    try:
        while True:
            # Phase 2: Data transfer
            data = await websocket.receive_text()
            await websocket.send_text(f"Got: {data}")
    except WebSocketDisconnect:
        pass  # Phase 3: Close
    finally:
        cleanup()

Code

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

app = FastAPI()


@app.websocket("/ws/lifecycle")
async def lifecycle_demo(websocket: WebSocket):
    """Demonstrates the full WebSocket lifecycle."""

    # Phase 1: Accept the connection (completes HTTP upgrade)
    # You can optionally specify subprotocol and headers
    await websocket.accept(subprotocol=None)
    logger.info(
        f"Client connected: {websocket.client.host}:{websocket.client.port}"
    )

    try:
        # Phase 2: Data transfer loop
        while True:
            # receive_text() blocks until a message arrives
            message = await websocket.receive_text()
            logger.info(f"Received: {message}")

            if message == "ping":
                await websocket.send_text("pong")
            elif message == "close":
                # Server-initiated close
                await websocket.close(code=1000, reason="Requested")
                break
            else:
                await websocket.send_text(f"Echo: {message}")

    except WebSocketDisconnect as e:
        # Phase 3: Client disconnected
        logger.info(
            f"Client disconnected: code={e.code}, reason={e.reason}"
        )
    finally:
        # Cleanup runs regardless of how the connection ended
        logger.info("Connection cleanup complete")


# --- Different message types ---
@app.websocket("/ws/types")
async def message_types(websocket: WebSocket):
    await websocket.accept()
    try:
        while True:
            # Generic receive -- returns a dict with 'type' key
            raw = await websocket.receive()

            if raw["type"] == "websocket.receive":
                if "text" in raw:
                    text = raw["text"]
                    await websocket.send_text(f"Text: {text}")
                elif "bytes" in raw:
                    data = raw["bytes"]
                    await websocket.send_bytes(
                        b"Binary echo: " + data
                    )
            elif raw["type"] == "websocket.disconnect":
                break
    except WebSocketDisconnect:
        pass


# --- Rejecting a connection ---
@app.websocket("/ws/guarded")
async def guarded_endpoint(websocket: WebSocket):
    """Reject connections that lack a required token."""
    token = websocket.query_params.get("token")
    if token != "secret":
        # Close before accepting -- sends 403-like rejection
        await websocket.close(code=1008, reason="Invalid token")
        return
    await websocket.accept()
    await websocket.send_text("Authenticated!")
    while True:
        data = await websocket.receive_text()
        await websocket.send_text(data)

Key Points

•The lifecycle is: accept (handshake) -> send/receive loop (data transfer) -> close/disconnect
•receive_text(), receive_bytes(), and receive_json() handle different message formats
•WebSocketDisconnect is raised when the client closes the connection
•You can reject a connection by calling websocket.close() without first calling accept()

Gotchas

•Calling send_text() before accept() raises a RuntimeError
•The generic receive() returns a raw ASGI dict -- prefer the typed receive_text()/receive_bytes() methods
•Close codes above 4000 are reserved for application-specific use
•If the server crashes without sending a close frame, the client may not know the connection is dead until a timeout

The WebSocket lifecycle is: accept() (handshake), receive_*/send_* loop (data transfer), close()/WebSocketDisconnect (teardown). Use receive_text() and send_text() for text messages, receive_bytes()/send_bytes() for binary, and handle WebSocketDisconnect for client disconnections.

•Three phases: accept (handshake), send/receive (data), close (teardown)

from fastapi import FastAPI, WebSocket, WebSocketDisconnect

app = FastAPI()

@app.websocket("/ws")
async def ws(websocket: WebSocket):
    await websocket.accept()          # Handshake
    try:
        while True:
            data = await websocket.receive_text()  # Receive
            await websocket.send_text(f"Echo: {data}")  # Send
    except WebSocketDisconnect:
        print("Client disconnected")  # Close

Always handle WebSocketDisconnect -- otherwise the server logs unhandled exceptions on every client disconnect

Question 3 / 12

How do you handle WebSocketDisconnect exceptions?

WebSocketDisconnect is raised when a WebSocket client disconnects (closes the browser tab, loses network, or sends a close frame). Handling it properly is essential for resource cleanup and connection tracking.

When it is raised:

Client sends a close frame (clean disconnect)
Client's network drops (detected on next receive/send)
Client closes the browser tab or navigates away

How to handle it:

Wrap the receive loop in try/except: The most common pattern
Access the close code: e.code tells you why the client disconnected
Cleanup resources: Remove the client from connection managers, release locks, etc.
Log the disconnection: Useful for debugging and monitoring

Important close codes:

1000: Normal closure (client called ws.close())
1001: Going away (tab closed, navigation)
1006: Abnormal closure (no close frame received -- network drop)
1011: Unexpected condition on the server

What happens if you do not handle it:

Starlette catches it and logs a traceback
Connection managers will not know the client left
Resources tied to the connection will leak

Pseudo Code

@app.websocket("/ws")
async def ws(websocket: WebSocket):
    await websocket.accept()
    manager.add(websocket)
    try:
        while True:
            data = await websocket.receive_text()
            await process(data)
    except WebSocketDisconnect as e:
        manager.remove(websocket)
        log(f"Disconnected: code={e.code}")

Code

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

app = FastAPI()

# Track active connections
active_connections: set[WebSocket] = set()


@app.websocket("/ws/chat")
async def chat(websocket: WebSocket):
    await websocket.accept()
    active_connections.add(websocket)
    client_id = f"{websocket.client.host}:{websocket.client.port}"
    logger.info(f"Client {client_id} connected. Total: {len(active_connections)}")

    try:
        while True:
            message = await websocket.receive_text()
            # Broadcast to all other clients
            for conn in active_connections:
                if conn != websocket:
                    try:
                        await conn.send_text(f"[{client_id}] {message}")
                    except Exception:
                        # Other client might have disconnected
                        pass

    except WebSocketDisconnect as e:
        # Clean disconnect handling
        active_connections.discard(websocket)
        logger.info(
            f"Client {client_id} disconnected: "
            f"code={e.code}, reason={getattr(e, 'reason', 'N/A')}. "
            f"Remaining: {len(active_connections)}"
        )
        # Notify others
        for conn in active_connections:
            try:
                await conn.send_text(f"[{client_id}] left the chat")
            except Exception:
                pass

    except Exception as e:
        # Unexpected errors (not a disconnect)
        active_connections.discard(websocket)
        logger.error(f"Unexpected error for {client_id}: {e}")
        try:
            await websocket.close(code=1011, reason="Internal error")
        except Exception:
            pass  # Connection might already be closed


# --- Handling different disconnect codes ---
@app.websocket("/ws/monitored")
async def monitored_ws(websocket: WebSocket):
    await websocket.accept()
    try:
        while True:
            data = await websocket.receive_text()
            await websocket.send_text(data)
    except WebSocketDisconnect as e:
        if e.code == 1000:
            logger.info("Clean disconnect")
        elif e.code == 1001:
            logger.info("Client navigated away")
        elif e.code == 1006:
            logger.warning("Abnormal disconnect (network drop)")
        else:
            logger.warning(f"Disconnect with code {e.code}")

Key Points

•WebSocketDisconnect is raised when receive_*() or send_*() detects a closed connection
•Always remove the client from connection tracking in the except block
•Access e.code to determine the reason for disconnection
•Code 1006 means no close frame was received -- typically a network failure

Gotchas

•WebSocketDisconnect can be raised during send_text() too, not just receive_text()
•Code 1006 is never sent on the wire -- it is synthesized locally when the TCP connection drops without a close frame
•Sending a message to a disconnected client raises an exception -- always wrap broadcast sends in try/except
•If you forget to remove clients from a connection set, you leak memory and get errors on every broadcast

Wrap your receive loop in try/except WebSocketDisconnect. In the except block, clean up resources (remove from connection managers) and log the disconnect. Access e.code to learn why the client disconnected (1000 = clean, 1001 = navigated away, 1006 = network drop).

•Catch WebSocketDisconnect to clean up resources when a client leaves

from fastapi import FastAPI, WebSocket, WebSocketDisconnect

app = FastAPI()
clients: set[WebSocket] = set()

@app.websocket("/ws")
async def ws(websocket: WebSocket):
    await websocket.accept()
    clients.add(websocket)
    try:
        while True:
            data = await websocket.receive_text()
            await websocket.send_text(data)
    except WebSocketDisconnect:
        clients.discard(websocket)

Not handling WebSocketDisconnect leaks connections in your tracking data structures

Question 4 / 12

What is the ConnectionManager pattern for managing multiple clients?

The ConnectionManager is a widely used pattern for tracking and broadcasting to multiple WebSocket clients. It encapsulates connection tracking, message routing, and cleanup in a single class.

Core responsibilities:

connect(): Accept and store new WebSocket connections
disconnect(): Remove connections that have closed
send_personal(): Send a message to one specific client
broadcast(): Send a message to all connected clients

Why use a class:

Centralizes connection logic instead of scattering it across endpoints
Makes it easy to add features like rooms, user tracking, and message history
Simplifies testing -- you can mock the manager

Advanced extensions:

Room/channel support: Group connections by topic or room
User mapping: Associate connections with authenticated user IDs
Message history: Buffer recent messages for new joiners
Connection metadata: Store per-connection state (username, join time, etc.)

Concurrency note:

Since asyncio is single-threaded, a simple list/set is safe for tracking connections. You do not need locks unless you use multiple threads.

Pseudo Code

class ConnectionManager:
    def __init__(self):
        self.connections = []

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

    def disconnect(self, ws):
        self.connections.remove(ws)

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

Code

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from collections import defaultdict
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

app = FastAPI()


# --- Basic ConnectionManager ---
class ConnectionManager:
    """Manages WebSocket connections with broadcast support."""

    def __init__(self):
        self.active_connections: list[WebSocket] = []

    async def connect(self, websocket: WebSocket):
        await websocket.accept()
        self.active_connections.append(websocket)
        logger.info(f"Connected. Total: {len(self.active_connections)}")

    def disconnect(self, websocket: WebSocket):
        self.active_connections.remove(websocket)
        logger.info(f"Disconnected. Total: {len(self.active_connections)}")

    async def send_personal(self, message: str, websocket: WebSocket):
        await websocket.send_text(message)

    async def broadcast(self, message: str, exclude: WebSocket | None = None):
        """Send message to all connected clients."""
        for connection in self.active_connections:
            if connection != exclude:
                try:
                    await connection.send_text(message)
                except Exception:
                    pass  # Client may have disconnected between iterations


manager = ConnectionManager()


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


# --- Room-based ConnectionManager ---
class RoomManager:
    """Manages connections organized by rooms."""

    def __init__(self):
        self.rooms: dict[str, list[WebSocket]] = defaultdict(list)

    async def connect(self, room: str, websocket: WebSocket):
        await websocket.accept()
        self.rooms[room].append(websocket)

    def disconnect(self, room: str, websocket: WebSocket):
        self.rooms[room].remove(websocket)
        if not self.rooms[room]:
            del self.rooms[room]  # Clean up empty rooms

    async def broadcast_to_room(
        self, room: str, message: str,
        exclude: WebSocket | None = None,
    ):
        for ws in self.rooms.get(room, []):
            if ws != exclude:
                try:
                    await ws.send_text(message)
                except Exception:
                    pass

    def get_room_count(self, room: str) -> int:
        return len(self.rooms.get(room, []))


room_manager = RoomManager()


@app.websocket("/ws/rooms/{room_id}")
async def room_endpoint(websocket: WebSocket, room_id: str):
    await room_manager.connect(room_id, websocket)
    await room_manager.broadcast_to_room(
        room_id,
        f"New user joined. Users in room: {room_manager.get_room_count(room_id)}",
    )
    try:
        while True:
            data = await websocket.receive_text()
            await room_manager.broadcast_to_room(
                room_id, data, exclude=websocket
            )
    except WebSocketDisconnect:
        room_manager.disconnect(room_id, websocket)
        await room_manager.broadcast_to_room(
            room_id,
            f"User left. Users in room: {room_manager.get_room_count(room_id)}",
        )

Key Points

•ConnectionManager centralizes connect, disconnect, send, and broadcast logic
•Always wrap send_text() in try/except when broadcasting -- clients may disconnect between iterations
•Use defaultdict(list) for room-based connection grouping
•Clean up empty rooms to prevent memory leaks in long-running servers

Gotchas

•The manager is global state -- each Uvicorn worker has its own instance with its own connections
•Broadcasting is sequential (await in a loop) -- for hundreds of clients, consider asyncio.gather()
•list.remove() raises ValueError if the connection is not found -- use discard() with sets or guard with try/except
•A slow client blocks the broadcast loop since each send is awaited sequentially

The ConnectionManager pattern tracks active WebSocket connections in a list and provides connect(), disconnect(), send_personal(), and broadcast() methods. It centralizes all connection logic and makes room-based routing straightforward.

•Centralizes connect/disconnect/broadcast in one class for clean WebSocket management

from fastapi import FastAPI, WebSocket, WebSocketDisconnect

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/{name}")
async def ws(websocket: WebSocket, name: str):
    await manager.connect(websocket)
    try:
        while True:
            data = await websocket.receive_text()
            await manager.broadcast(f"{name}: {data}")
    except WebSocketDisconnect:
        manager.disconnect(websocket)

Each Uvicorn worker has its own manager instance -- connections are not shared across workers

Question 5 / 12

How do you implement authentication for WebSocket connections?

WebSocket authentication differs from HTTP because you cannot set custom headers during the browser's WebSocket handshake. You must use alternative strategies.

Authentication strategies:

Query parameter token: Pass a JWT or API key in the WebSocket URL
- ws://host/ws?token=eyJhbG...
- Simple but tokens appear in server logs and browser history
Cookie-based: If the user is already authenticated via HTTP cookies, the browser sends cookies with the WebSocket upgrade request
- websocket.cookies["session_id"]
- Best for browser apps with existing session auth
First-message auth: Accept the connection, then require a token in the first message before allowing further communication
- More secure than URL tokens (not logged)
- Adds a round-trip before the connection is "active"
Subprotocol header: Some libraries support passing tokens via the Sec-WebSocket-Protocol header
- Requires client-side support

Why not the Authorization header?

Browser JavaScript's WebSocket API does not allow setting custom headers. Server-to-server WebSocket connections can use headers, but browser clients cannot.

Best practice:

Use short-lived tokens. The WebSocket connection can outlive the token, so implement periodic re-validation or heartbeat-based token refresh.

Pseudo Code

# Query param auth
@app.websocket("/ws")
async def ws(websocket: WebSocket, token: str):
    user = verify_jwt(token)
    if not user:
        await websocket.close(code=1008)
        return
    await websocket.accept()
    ...

# First-message auth
@app.websocket("/ws")
async def ws(websocket: WebSocket):
    await websocket.accept()
    auth_msg = await websocket.receive_json()
    user = verify_jwt(auth_msg["token"])
    if not user:
        await websocket.close(code=1008)
        return
    ...

Code

from fastapi import FastAPI, WebSocket, WebSocketDisconnect, Depends, Query
from fastapi import status
import jwt
import logging
from datetime import datetime, timezone, timedelta
from typing import Annotated

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

app = FastAPI()
SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"


def create_token(user_id: int) -> str:
    payload = {
        "sub": str(user_id),
        "exp": datetime.now(timezone.utc) + timedelta(hours=1),
    }
    return jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)


def verify_token(token: str) -> dict | None:
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        return {"user_id": int(payload["sub"])}
    except jwt.InvalidTokenError:
        return None


# --- Strategy 1: Query parameter token ---
@app.websocket("/ws/query-auth")
async def query_auth_ws(
    websocket: WebSocket,
    token: str = Query(default=""),
):
    user = verify_token(token)
    if not user:
        await websocket.close(code=status.WS_1008_POLICY_VIOLATION)
        return

    await websocket.accept()
    await websocket.send_text(f"Welcome, user {user['user_id']}!")
    try:
        while True:
            data = await websocket.receive_text()
            await websocket.send_text(f"[User {user['user_id']}] {data}")
    except WebSocketDisconnect:
        logger.info(f"User {user['user_id']} disconnected")


# --- Strategy 2: Cookie-based auth ---
@app.websocket("/ws/cookie-auth")
async def cookie_auth_ws(websocket: WebSocket):
    session_token = websocket.cookies.get("session")
    user = verify_token(session_token) if session_token else None

    if not user:
        await websocket.close(code=status.WS_1008_POLICY_VIOLATION)
        return

    await websocket.accept()
    await websocket.send_text(f"Authenticated via cookie: user {user['user_id']}")
    try:
        while True:
            data = await websocket.receive_text()
            await websocket.send_text(data)
    except WebSocketDisconnect:
        pass


# --- Strategy 3: First-message auth ---
@app.websocket("/ws/first-msg-auth")
async def first_message_auth_ws(websocket: WebSocket):
    await websocket.accept()

    # Expect the first message to be an auth token
    try:
        auth_data = await websocket.receive_json()
    except Exception:
        await websocket.close(code=status.WS_1008_POLICY_VIOLATION)
        return

    token = auth_data.get("token", "")
    user = verify_token(token)
    if not user:
        await websocket.send_json({"error": "Invalid token"})
        await websocket.close(code=status.WS_1008_POLICY_VIOLATION)
        return

    await websocket.send_json(
        {"status": "authenticated", "user_id": user["user_id"]}
    )

    try:
        while True:
            data = await websocket.receive_text()
            await websocket.send_text(f"[User {user['user_id']}] {data}")
    except WebSocketDisconnect:
        logger.info(f"User {user['user_id']} disconnected")


# --- Helper: HTTP endpoint to get a token ---
@app.get("/token")
async def get_token(user_id: int = 1):
    return {"token": create_token(user_id)}

Key Points

•Browser WebSocket API cannot set custom headers -- use query params, cookies, or first-message auth
•Query param tokens are simple but appear in server logs and browser history
•Cookie-based auth works seamlessly when users already have an HTTP session
•First-message auth is most secure for browser clients -- tokens are not in the URL

Gotchas

•Tokens in query params are logged by most web servers and proxies -- use short-lived tokens
•WebSocket connections can outlive JWT expiration -- implement periodic re-validation or heartbeat auth checks
•Closing before accept() sends an HTTP 403 but some browsers show a generic connection error
•If using first-message auth, set a timeout on the receive -- otherwise unauthenticated connections linger indefinitely

Browser WebSocket APIs cannot set custom headers, so use query parameter tokens (ws://host/ws?token=...), cookies, or first-message authentication. Verify the token before processing any messages. Close with code 1008 (Policy Violation) on auth failure.

•Query params, cookies, or first-message are the three browser-compatible auth strategies

from fastapi import FastAPI, WebSocket, WebSocketDisconnect, Query, status

app = FastAPI()

@app.websocket("/ws")
async def ws(websocket: WebSocket, token: str = Query(default="")):
    user = verify_token(token)
    if not user:
        await websocket.close(code=status.WS_1008_POLICY_VIOLATION)
        return
    await websocket.accept()
    try:
        while True:
            data = await websocket.receive_text()
            await websocket.send_text(data)
    except WebSocketDisconnect:
        pass

Query parameter tokens appear in server logs -- use short-lived tokens and rotate frequently

Question 6 / 12

How do you send and receive JSON over WebSockets?

FastAPI's WebSocket object provides built-in JSON support through receive_json() and send_json(). You can also combine this with Pydantic models for validation.

Built-in methods:

await websocket.receive_json() -- deserializes incoming text as JSON
await websocket.send_json(data) -- serializes a dict/list to JSON and sends as text

The `mode` parameter:

Both methods accept a mode parameter:

mode="text" (default): JSON is sent/received as text frames
mode="binary": JSON is sent/received as binary frames

Pydantic validation:

FastAPI does not automatically validate WebSocket messages with Pydantic (unlike HTTP request bodies). You must manually parse and validate incoming JSON using Pydantic models.

Structured message protocol:

For real apps, define a message protocol with a type field to distinguish between different message kinds (chat, typing indicator, system messages, etc.).

Pseudo Code

from pydantic import BaseModel

class Message(BaseModel):
    type: str
    content: str

@app.websocket("/ws")
async def ws(websocket: WebSocket):
    await websocket.accept()
    while True:
        data = await websocket.receive_json()
        msg = Message(**data)  # Validate with Pydantic
        await websocket.send_json({"echo": msg.content})

Code

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from pydantic import BaseModel, ValidationError
from typing import Literal
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

app = FastAPI()


# --- Pydantic models for message protocol ---
class ChatMessage(BaseModel):
    type: Literal["chat"]
    content: str
    username: str

class TypingIndicator(BaseModel):
    type: Literal["typing"]
    username: str
    is_typing: bool

class SystemCommand(BaseModel):
    type: Literal["command"]
    action: str  # "clear", "list_users", etc.


# --- Basic JSON echo ---
@app.websocket("/ws/json")
async def json_echo(websocket: WebSocket):
    """Send and receive JSON messages."""
    await websocket.accept()
    try:
        while True:
            # receive_json() parses the text frame as JSON
            data = await websocket.receive_json()
            logger.info(f"Received JSON: {data}")

            # send_json() serializes dict to JSON text
            await websocket.send_json({
                "echo": data,
                "status": "received",
            })
    except WebSocketDisconnect:
        pass


# --- Validated JSON with Pydantic ---
@app.websocket("/ws/chat")
async def validated_chat(websocket: WebSocket):
    """Validates incoming JSON with Pydantic models."""
    await websocket.accept()
    try:
        while True:
            raw = await websocket.receive_json()
            msg_type = raw.get("type")

            try:
                if msg_type == "chat":
                    msg = ChatMessage(**raw)
                    await websocket.send_json({
                        "type": "chat",
                        "content": msg.content,
                        "from": msg.username,
                    })

                elif msg_type == "typing":
                    indicator = TypingIndicator(**raw)
                    await websocket.send_json({
                        "type": "typing",
                        "username": indicator.username,
                        "is_typing": indicator.is_typing,
                    })

                elif msg_type == "command":
                    cmd = SystemCommand(**raw)
                    await websocket.send_json({
                        "type": "system",
                        "message": f"Executed: {cmd.action}",
                    })

                else:
                    await websocket.send_json({
                        "type": "error",
                        "message": f"Unknown message type: {msg_type}",
                    })

            except ValidationError as e:
                await websocket.send_json({
                    "type": "error",
                    "message": "Validation failed",
                    "details": e.errors(),
                })

    except WebSocketDisconnect:
        logger.info("Client disconnected")


# Client-side JavaScript:
# ws.send(JSON.stringify({type: "chat", content: "Hello!", username: "alice"}));
# ws.send(JSON.stringify({type: "typing", username: "alice", is_typing: true}));
# ws.send(JSON.stringify({type: "command", action: "list_users"}));

Key Points

•receive_json() and send_json() handle JSON serialization/deserialization automatically
•FastAPI does not auto-validate WebSocket JSON -- use Pydantic models manually
•Define a message protocol with a 'type' field to distinguish message kinds
•Send validation errors back to the client as JSON error messages

Gotchas

•receive_json() raises a JSONDecodeError if the client sends invalid JSON -- wrap in try/except
•send_json() uses Python's json.dumps internally -- datetime and other non-serializable types need custom encoders
•mode='binary' sends JSON as binary frames which some client libraries do not expect
•Unlike HTTP endpoints, WebSocket messages are not documented in OpenAPI/Swagger

Use await websocket.receive_json() and await websocket.send_json(data) for automatic JSON serialization. FastAPI does not auto-validate WebSocket messages -- manually validate with Pydantic using Model(**data) and catch ValidationError.

•receive_json()/send_json() handle JSON automatically; validate manually with Pydantic

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from pydantic import BaseModel, ValidationError

class ChatMsg(BaseModel):
    type: str
    content: str

app = FastAPI()

@app.websocket("/ws")
async def ws(websocket: WebSocket):
    await websocket.accept()
    try:
        while True:
            raw = await websocket.receive_json()
            try:
                msg = ChatMsg(**raw)
                await websocket.send_json({"echo": msg.content})
            except ValidationError as e:
                await websocket.send_json({"error": e.errors()})
    except WebSocketDisconnect:
        pass

Invalid JSON from the client raises JSONDecodeError, not ValidationError -- handle both

Question 7 / 12

How do you implement a chat room with broadcasting?

A chat room requires a ConnectionManager that supports multiple rooms, user tracking, and message broadcasting. Each room maintains its own list of connected clients.

Architecture:

RoomConnectionManager: Tracks connections per room
Join/leave notifications: Broadcast when users enter or exit
Message routing: Messages go only to clients in the same room
User metadata: Track usernames alongside connections

Message protocol:

Use a structured JSON format with a type field:

{"type": "chat", "content": "..."} -- regular message
{"type": "join"} -- system join notification
{"type": "leave"} -- system leave notification
{"type": "users"} -- list of users in the room

Scalability considerations:

In-memory rooms work for a single server process
For multiple workers/servers, use Redis Pub/Sub to relay messages across processes
Consider message history for new joiners (ring buffer or database-backed)

Pseudo Code

class ChatRoom:
    rooms: dict[str, dict[WebSocket, str]] = {}

    async def join(self, room, ws, username):
        rooms[room][ws] = username
        await broadcast(room, f"{username} joined")

    async def leave(self, room, ws):
        username = rooms[room].pop(ws)
        await broadcast(room, f"{username} left")

    async def broadcast(self, room, message):
        for ws in rooms[room]:
            await ws.send_json(message)

Code

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from collections import defaultdict
from datetime import datetime, timezone
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

app = FastAPI()


class ChatRoomManager:
    """Manages multiple chat rooms with user tracking."""

    def __init__(self):
        # room_id -> {websocket: username}
        self.rooms: dict[str, dict[WebSocket, str]] = defaultdict(dict)

    async def join(
        self, room_id: str, websocket: WebSocket, username: str,
    ):
        await websocket.accept()
        self.rooms[room_id][websocket] = username
        # Notify everyone in the room
        await self.broadcast(room_id, {
            "type": "system",
            "content": f"{username} joined the room",
            "users": self.get_users(room_id),
        })

    async def leave(self, room_id: str, websocket: WebSocket):
        username = self.rooms[room_id].pop(websocket, "unknown")
        if not self.rooms[room_id]:
            del self.rooms[room_id]  # Clean up empty rooms
        else:
            await self.broadcast(room_id, {
                "type": "system",
                "content": f"{username} left the room",
                "users": self.get_users(room_id),
            })
        return username

    async def broadcast(
        self,
        room_id: str,
        message: dict,
        exclude: WebSocket | None = None,
    ):
        """Send a message to all clients in a room."""
        disconnected = []
        for ws in self.rooms.get(room_id, {}):
            if ws == exclude:
                continue
            try:
                await ws.send_json(message)
            except Exception:
                disconnected.append(ws)
        # Clean up any broken connections found during broadcast
        for ws in disconnected:
            self.rooms[room_id].pop(ws, None)

    def get_users(self, room_id: str) -> list[str]:
        return list(self.rooms.get(room_id, {}).values())

    def get_room_count(self, room_id: str) -> int:
        return len(self.rooms.get(room_id, {}))


chat_manager = ChatRoomManager()


@app.websocket("/ws/chat/{room_id}")
async def chat_endpoint(
    websocket: WebSocket,
    room_id: str,
    username: str = "anonymous",
):
    await chat_manager.join(room_id, websocket, username)
    try:
        while True:
            data = await websocket.receive_json()
            msg_type = data.get("type", "chat")

            if msg_type == "chat":
                await chat_manager.broadcast(
                    room_id,
                    {
                        "type": "chat",
                        "username": username,
                        "content": data.get("content", ""),
                        "timestamp": datetime.now(timezone.utc).isoformat(),
                    },
                )
            elif msg_type == "list_users":
                await websocket.send_json({
                    "type": "users",
                    "users": chat_manager.get_users(room_id),
                })

    except WebSocketDisconnect:
        left_user = await chat_manager.leave(room_id, websocket)
        logger.info(f"{left_user} left room {room_id}")


# --- REST endpoint to list active rooms ---
@app.get("/rooms")
async def list_rooms():
    return {
        room_id: {
            "users": chat_manager.get_users(room_id),
            "count": chat_manager.get_room_count(room_id),
        }
        for room_id in chat_manager.rooms
    }


# Client usage:
# ws = new WebSocket("ws://localhost:8000/ws/chat/general?username=alice");
# ws.send(JSON.stringify({type: "chat", content: "Hello everyone!"}));
# ws.send(JSON.stringify({type: "list_users"}));

Key Points

•Use a dict[str, dict[WebSocket, str]] to map rooms to their connected users
•Broadcast join/leave notifications so all users see who is online
•Clean up empty rooms after the last user leaves to prevent memory leaks
•Remove broken connections discovered during broadcast to keep the room state clean

Gotchas

•In-memory rooms only work within a single process -- multiple Uvicorn workers each have their own rooms
•Broadcasting to many clients sequentially can be slow -- consider asyncio.gather() for large rooms
•A user can open multiple tabs and appear as multiple connections in the same room
•No message persistence -- users who join late miss previous messages unless you implement history

Use a ChatRoomManager that maps room IDs to {WebSocket: username} dicts. On join, broadcast a system message. Route chat messages only to clients in the same room. Clean up on disconnect.

•Map room IDs to connection dicts for room-scoped broadcasting

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from collections import defaultdict

app = FastAPI()

class ChatRoomManager:
    def __init__(self):
        self.rooms: dict[str, dict[WebSocket, str]] = defaultdict(dict)

    async def join(self, room, ws, username):
        await ws.accept()
        self.rooms[room][ws] = username

    async def leave(self, room, ws):
        self.rooms[room].pop(ws, None)

    async def broadcast(self, room, message):
        for ws in self.rooms.get(room, {}):
            await ws.send_json(message)

manager = ChatRoomManager()

@app.websocket("/ws/chat/{room}")
async def chat(websocket: WebSocket, room: str, username: str = "anon"):
    await manager.join(room, websocket, username)
    try:
        while True:
            data = await websocket.receive_json()
            await manager.broadcast(room, {"user": username, "msg": data["content"]})
    except WebSocketDisconnect:
        await manager.leave(room, websocket)

In-memory rooms are per-process -- use Redis Pub/Sub for multi-worker deployments

Question 8 / 12

How do you use WebSocket dependencies?

FastAPI supports dependency injection for WebSocket endpoints, similar to HTTP endpoints. However, there are important differences and limitations.

What works:

Regular function dependencies (Depends())
Class-based dependencies (Depends(MyClass))
Query parameters and path parameters
Nested/sub-dependencies
BackgroundTasks injection

What does NOT work:

Body(), Form(), File() -- WebSocket has no HTTP body
Header() for custom headers -- only standard WebSocket headers are available
Security() with OAuth2 -- browser WebSockets cannot set Authorization headers
Response injection -- WebSockets do not have HTTP responses

Dependency lifecycle:

Regular dependencies run once per WebSocket connection (not per message)
Yield dependencies clean up when the WebSocket handler function exits
This is different from HTTP where dependencies run per request

Accessing WebSocket in dependencies:

Dependencies can declare a WebSocket parameter to access the connection directly. This is useful for auth dependencies that need to read cookies or query params.

Pseudo Code

async def verify_ws_token(websocket: WebSocket) -> dict:
    token = websocket.query_params.get("token")
    user = decode_jwt(token)
    if not user:
        await websocket.close(code=1008)
        raise WebSocketException(code=1008)
    return user

@app.websocket("/ws")
async def ws(
    websocket: WebSocket,
    user: Annotated[dict, Depends(verify_ws_token)],
):
    await websocket.accept()
    ...

Code

from fastapi import (
    FastAPI, WebSocket, WebSocketDisconnect, WebSocketException,
    Depends, Query,
    status,
)
from typing import Annotated
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

app = FastAPI()


# --- Auth dependency for WebSocket ---
async def get_ws_user(
    websocket: WebSocket,
    token: str = Query(default=""),
) -> dict:
    """Validates token and returns user info.
    Raises WebSocketException to reject the connection."""
    # Simulated token validation
    valid_tokens = {
        "token-alice": {"user_id": 1, "name": "Alice"},
        "token-bob": {"user_id": 2, "name": "Bob"},
    }
    user = valid_tokens.get(token)
    if not user:
        raise WebSocketException(
            code=status.WS_1008_POLICY_VIOLATION,
            reason="Invalid token",
        )
    return user


@app.websocket("/ws/protected")
async def protected_ws(
    websocket: WebSocket,
    user: Annotated[dict, Depends(get_ws_user)],
):
    """WebSocket endpoint with auth dependency."""
    await websocket.accept()
    await websocket.send_json({"welcome": user["name"]})
    try:
        while True:
            data = await websocket.receive_text()
            await websocket.send_text(
                f"[{user['name']}] {data}"
            )
    except WebSocketDisconnect:
        logger.info(f"{user['name']} disconnected")


# --- Rate limiter dependency ---
from collections import defaultdict
import time

class WebSocketRateLimiter:
    def __init__(self, max_messages: int, window_seconds: float):
        self.max_messages = max_messages
        self.window = window_seconds
        self.message_counts: dict[str, list[float]] = defaultdict(list)

    def check(self, client_id: str) -> bool:
        now = time.time()
        timestamps = self.message_counts[client_id]
        # Remove old timestamps
        self.message_counts[client_id] = [
            t for t in timestamps if now - t < self.window
        ]
        if len(self.message_counts[client_id]) >= self.max_messages:
            return False
        self.message_counts[client_id].append(now)
        return True

rate_limiter = WebSocketRateLimiter(max_messages=10, window_seconds=60)


def get_rate_limiter() -> WebSocketRateLimiter:
    return rate_limiter


@app.websocket("/ws/limited")
async def rate_limited_ws(
    websocket: WebSocket,
    limiter: Annotated[WebSocketRateLimiter, Depends(get_rate_limiter)],
):
    await websocket.accept()
    client_id = f"{websocket.client.host}:{websocket.client.port}"
    try:
        while True:
            data = await websocket.receive_text()
            if not limiter.check(client_id):
                await websocket.send_json({
                    "error": "Rate limit exceeded. Try again later."
                })
                continue
            await websocket.send_text(f"Echo: {data}")
    except WebSocketDisconnect:
        pass


# --- Yield dependency with cleanup ---
async def ws_session_tracker(websocket: WebSocket):
    """Tracks session lifetime with cleanup."""
    client = f"{websocket.client.host}:{websocket.client.port}"
    logger.info(f"Session started: {client}")
    yield client
    # Cleanup runs when the WebSocket handler exits
    logger.info(f"Session ended: {client}")


@app.websocket("/ws/tracked")
async def tracked_ws(
    websocket: WebSocket,
    client_id: Annotated[str, Depends(ws_session_tracker)],
):
    await websocket.accept()
    try:
        while True:
            data = await websocket.receive_text()
            await websocket.send_text(f"[{client_id}] {data}")
    except WebSocketDisconnect:
        pass  # Yield dependency cleanup runs after this

Key Points

•WebSocket dependencies run once per connection, not per message
•Raise WebSocketException to reject a connection from within a dependency
•Yield dependencies clean up when the WebSocket handler function exits (on disconnect)
•Dependencies can access the WebSocket object to read cookies, query params, and headers

Gotchas

•Body(), Form(), File() do not work with WebSocket endpoints -- there is no HTTP body
•Security() with OAuth2 does not work because browsers cannot set the Authorization header on WebSocket upgrades
•WebSocketException can be raised both before and after accept() -- before accept it rejects the HTTP upgrade, after accept it sends a WebSocket close frame with the specified code
•Dependencies are cached per-connection, so the same dependency used in multiple sub-dependencies returns the same value

WebSocket endpoints support Depends() for dependency injection. Dependencies run once per connection (not per message). Use WebSocketException to reject connections from within dependencies. Body(), Form(), and Security() with OAuth2 do not work for WebSockets.

•Dependencies run once per connection and support yield for cleanup

from fastapi import FastAPI, WebSocket, WebSocketException, Depends, Query, status
from typing import Annotated

app = FastAPI()

async def verify_token(token: str = Query(default="")) -> dict:
    if token != "valid":
        raise WebSocketException(code=status.WS_1008_POLICY_VIOLATION)
    return {"user": "alice"}

@app.websocket("/ws")
async def ws(
    websocket: WebSocket,
    user: Annotated[dict, Depends(verify_token)],
):
    await websocket.accept()
    await websocket.send_json({"hello": user["user"]})

Body(), Form(), and OAuth2 Security() are not compatible with WebSocket endpoints

Question 9 / 12

What is Server-Sent Events (SSE) and how does it compare to WebSockets?

Server-Sent Events (SSE) is a standard for server-to-client streaming over HTTP. Unlike WebSockets, SSE is unidirectional -- only the server pushes data to the client.

SSE vs WebSockets:

Feature	SSE	WebSockets
Direction	Server -> Client only	Bidirectional
Protocol	HTTP/1.1 or HTTP/2	WebSocket (ws://)
Reconnection	Automatic (built into EventSource API)	Manual (you must implement it)
Data format	Text only (UTF-8)	Text or binary
Authentication	Standard HTTP headers (cookies, Bearer)	Query params or cookies only
Proxy support	Excellent (standard HTTP)	Can be problematic
Browser API	EventSource	WebSocket
Max connections	~6 per domain (HTTP/1.1)	No hard limit

When to use SSE:

Live feeds (stock prices, news, notifications)
Progress updates for long-running operations
Real-time dashboards where the client only reads
Situations where you need standard HTTP auth headers

When to use WebSockets:

Chat applications (bidirectional)
Online games (low latency, bidirectional)
Collaborative editing (bidirectional)
Binary data streaming

SSE message format:

Each event is a text block with optional event:, data:, id:, and retry: fields, separated by double newlines.

Pseudo Code

from fastapi.responses import StreamingResponse

async def event_stream():
    while True:
        data = await get_next_event()
        yield f"event: update\ndata: {json.dumps(data)}\n\n"

@app.get("/events")
async def sse():
    return StreamingResponse(
        event_stream(),
        media_type="text/event-stream",
    )

Code

from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
import asyncio
import json
import time

app = FastAPI()


# --- Basic SSE endpoint ---
@app.get("/events")
async def server_sent_events(request: Request):
    """Pushes events to the client every second."""

    async def event_generator():
        counter = 0
        while True:
            # Check if client disconnected
            if await request.is_disconnected():
                break

            data = {
                "counter": counter,
                "timestamp": time.time(),
            }
            # SSE format: "data: ...\n\n"
            yield f"data: {json.dumps(data)}\n\n"
            counter += 1
            await asyncio.sleep(1)

    return StreamingResponse(
        event_generator(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive",
            "X-Accel-Buffering": "no",  # Disable Nginx buffering
        },
    )


# --- SSE with named events and IDs ---
@app.get("/events/named")
async def named_events(request: Request):
    """SSE with event types and IDs for reconnection."""

    async def event_generator():
        event_id = 0
        while True:
            if await request.is_disconnected():
                break

            # Named event with ID (client can resume from last ID)
            event_id += 1
            yield (
                f"id: {event_id}\n"
                f"event: price_update\n"
                f"data: {{\"symbol\": \"AAPL\", \"price\": 150.{event_id}}}\n\n"
            )

            await asyncio.sleep(2)

            event_id += 1
            yield (
                f"id: {event_id}\n"
                f"event: notification\n"
                f"data: {{\"message\": \"Market update #{event_id}\"}}\n\n"
            )

            await asyncio.sleep(2)

    return StreamingResponse(
        event_generator(),
        media_type="text/event-stream",
        headers={"Cache-Control": "no-cache"},
    )


# --- Progress updates for a long-running task ---
@app.get("/tasks/{task_id}/progress")
async def task_progress(task_id: str, request: Request):
    async def progress_stream():
        for percent in range(0, 101, 10):
            if await request.is_disconnected():
                break
            yield f"data: {{\"task_id\": \"{task_id}\", \"progress\": {percent}}}\n\n"
            await asyncio.sleep(0.5)
        yield f"event: complete\ndata: {{\"task_id\": \"{task_id}\"}}\n\n"

    return StreamingResponse(
        progress_stream(),
        media_type="text/event-stream",
        headers={"Cache-Control": "no-cache"},
    )


# Client-side JavaScript:
# const es = new EventSource("/events/named");
# es.addEventListener("price_update", (e) => console.log(JSON.parse(e.data)));
# es.addEventListener("notification", (e) => console.log(JSON.parse(e.data)));
# es.onerror = () => console.log("Reconnecting...");  // Automatic!

Key Points

•SSE is server-to-client only, using standard HTTP -- simpler than WebSockets for push-only use cases
•The EventSource browser API has built-in automatic reconnection (WebSockets do not)
•SSE supports named events (event:) and IDs (id:) for resuming after reconnection
•Standard HTTP auth works with SSE -- unlike WebSockets, you can use Authorization headers

Gotchas

•HTTP/1.1 limits browsers to ~6 SSE connections per domain -- use HTTP/2 for more
•Nginx and some proxies buffer SSE by default -- set X-Accel-Buffering: no and proxy_buffering off
•SSE only supports UTF-8 text -- no binary data (use WebSockets for binary)
•Check request.is_disconnected() in the loop -- otherwise the generator runs forever even after the client leaves

SSE is server-to-client push over standard HTTP using StreamingResponse with media_type="text/event-stream". Unlike WebSockets, it is unidirectional, has automatic reconnection via the EventSource API, and supports standard HTTP auth headers. Use SSE for live feeds and notifications; use WebSockets for bidirectional communication.

•SSE = server-to-client push over HTTP with automatic reconnection

from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
import asyncio, json

app = FastAPI()

@app.get("/events")
async def sse(request: Request):
    async def stream():
        while not await request.is_disconnected():
            yield f"data: {json.dumps({'time': 'now'})}\n\n"
            await asyncio.sleep(1)

    return StreamingResponse(
        stream(),
        media_type="text/event-stream",
        headers={"Cache-Control": "no-cache"},
    )

HTTP/1.1 limits ~6 SSE connections per domain -- use HTTP/2 to avoid this

Question 10 / 12

How do you implement heartbeat/ping-pong for connection health?

Heartbeat (ping/pong) mechanisms detect dead connections that the TCP stack has not yet noticed. Without heartbeats, a server may keep sending to a client whose network dropped minutes ago.

Two levels of ping/pong:

WebSocket protocol-level (control frames):
- Built into the WebSocket standard (RFC 6455)
- Uvicorn/Starlette handle these automatically at the ASGI level
- Configure --ws-ping-interval and --ws-ping-timeout in Uvicorn
- Transparent to application code
Application-level heartbeat:
- Send custom {"type": "ping"} messages from the server
- Client responds with {"type": "pong"}
- Gives you full control over timeouts and reconnection logic
- Useful when protocol-level pings are not sufficient (e.g., behind proxies that intercept control frames)

Uvicorn configuration:

uvicorn app:app --ws-ping-interval 20 --ws-ping-timeout 20

--ws-ping-interval: Seconds between pings (default: 20)
--ws-ping-timeout: Seconds to wait for pong before closing (default: 20)
Set to None to disable

Application-level heartbeat pattern:

Use asyncio.wait_for() with a timeout on receive(), or run a concurrent ping task alongside the receive loop.

Pseudo Code

async def heartbeat_ws(websocket):
    await websocket.accept()
    async def ping_loop():
        while True:
            await asyncio.sleep(30)
            await websocket.send_json({"type": "ping"})
    ping_task = asyncio.create_task(ping_loop())
    try:
        while True:
            data = await asyncio.wait_for(
                websocket.receive_text(), timeout=60
            )
            process(data)
    except asyncio.TimeoutError:
        await websocket.close()  # Dead connection
    finally:
        ping_task.cancel()

Code

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
import asyncio
import time
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

app = FastAPI()


# --- Application-level heartbeat with concurrent ping task ---
@app.websocket("/ws/heartbeat")
async def heartbeat_endpoint(websocket: WebSocket):
    await websocket.accept()
    last_pong = time.time()
    PING_INTERVAL = 15  # Send ping every 15 seconds
    PONG_TIMEOUT = 30   # Close if no pong for 30 seconds

    async def ping_loop():
        """Periodically sends pings and checks for timeouts."""
        nonlocal last_pong
        while True:
            await asyncio.sleep(PING_INTERVAL)
            try:
                await websocket.send_json({"type": "ping"})
                # Check if we have received a pong recently
                if time.time() - last_pong > PONG_TIMEOUT:
                    logger.warning("Pong timeout -- closing connection")
                    await websocket.close(code=1001, reason="Pong timeout")
                    return
            except Exception:
                return  # Connection already closed

    ping_task = asyncio.create_task(ping_loop())

    try:
        while True:
            data = await websocket.receive_json()
            if data.get("type") == "pong":
                last_pong = time.time()
                logger.debug("Pong received")
            else:
                # Handle normal messages
                await websocket.send_json({
                    "type": "echo",
                    "data": data,
                })
    except WebSocketDisconnect:
        logger.info("Client disconnected")
    finally:
        ping_task.cancel()
        try:
            await ping_task
        except asyncio.CancelledError:
            pass


# --- Timeout-based approach (simpler) ---
@app.websocket("/ws/timeout")
async def timeout_endpoint(websocket: WebSocket):
    """Closes connection if no message received within timeout."""
    await websocket.accept()
    IDLE_TIMEOUT = 60  # Close after 60 seconds of silence

    try:
        while True:
            try:
                data = await asyncio.wait_for(
                    websocket.receive_text(),
                    timeout=IDLE_TIMEOUT,
                )
                await websocket.send_text(f"Echo: {data}")
            except asyncio.TimeoutError:
                logger.info("Idle timeout -- closing connection")
                await websocket.close(
                    code=1001, reason="Idle timeout"
                )
                break
    except WebSocketDisconnect:
        logger.info("Client disconnected")


# --- Uvicorn protocol-level pings (run with these flags) ---
# uvicorn app:app --ws-ping-interval 20 --ws-ping-timeout 20
#
# These are WebSocket control frames handled automatically:
# - Server sends a Ping frame every 20 seconds
# - If no Pong frame is received within 20 seconds, connection is closed
# - Transparent to application code (you do not see these in receive())


# Client-side JavaScript with heartbeat:
# const ws = new WebSocket("ws://localhost:8000/ws/heartbeat");
# ws.onmessage = (e) => {
#     const msg = JSON.parse(e.data);
#     if (msg.type === "ping") {
#         ws.send(JSON.stringify({type: "pong"}));
#     }
# };

Key Points

•Uvicorn provides protocol-level ping/pong via --ws-ping-interval and --ws-ping-timeout flags
•Application-level heartbeats give full control -- useful when proxies intercept protocol-level pings
•Use asyncio.create_task() for a concurrent ping loop alongside the receive loop
•asyncio.wait_for() with a timeout is the simplest way to detect idle connections

Gotchas

•Protocol-level pings are invisible to your application code -- you cannot detect them in receive()
•Some proxies (AWS ALB, Cloudflare) have their own idle timeouts that close connections even with heartbeats active
•asyncio.wait_for() cancels the underlying receive() on timeout -- the WebSocket may be in an inconsistent state
•Always cancel the ping task in a finally block to prevent it from running after the connection closes

Use Uvicorn's --ws-ping-interval and --ws-ping-timeout for protocol-level pings (transparent, no code changes). For application-level heartbeats, run a concurrent asyncio.create_task() that sends {"type": "ping"} periodically and tracks when the last pong was received.

•Uvicorn's --ws-ping-interval handles protocol-level pings automatically

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
import asyncio, time

app = FastAPI()

@app.websocket("/ws")
async def ws(websocket: WebSocket):
    await websocket.accept()
    last_pong = time.time()

    async def ping_loop():
        while True:
            await asyncio.sleep(15)
            await websocket.send_json({"type": "ping"})

    task = asyncio.create_task(ping_loop())
    try:
        while True:
            data = await websocket.receive_json()
            if data.get("type") == "pong":
                last_pong = time.time()
    except WebSocketDisconnect:
        pass
    finally:
        task.cancel()

Proxies like AWS ALB have their own idle timeouts that override your heartbeat settings

Question 11 / 12

How do you handle reconnection logic on the client side?

WebSocket reconnection is a client-side concern -- the browser's WebSocket API does not reconnect automatically (unlike SSE's EventSource). You must implement reconnection logic yourself.

Key reconnection strategies:

Exponential backoff: Wait 1s, 2s, 4s, 8s... between attempts to avoid overwhelming the server
Max retries: Give up after N attempts or a maximum elapsed time
Jitter: Add randomness to backoff intervals to prevent thundering herd when many clients reconnect simultaneously
Last event ID: Track the last received message ID so the server can replay missed events after reconnection

Server-side support:

Accept a last_event_id query parameter on reconnection
Buffer recent messages in memory or a database
Replay missed messages when a client reconnects

Libraries:

reconnecting-websocket (npm): Drop-in replacement for WebSocket with automatic reconnection
Custom wrapper: Full control over backoff, jitter, and state management

What the server should do:

Support idempotent reconnection (same room, same user)
Send a "caught up" or "replay complete" message after sending buffered events
Handle duplicate connections gracefully (same user, new socket)

Pseudo Code

// Client-side reconnection with backoff
class ReconnectingWS {
    constructor(url) {
        this.url = url;
        this.retries = 0;
        this.connect();
    }
    connect() {
        this.ws = new WebSocket(this.url);
        this.ws.onclose = () => {
            delay = Math.min(1000 * 2**this.retries, 30000);
            setTimeout(() => this.connect(), delay);
            this.retries++;
        };
        this.ws.onopen = () => this.retries = 0;
    }
}

Code

from fastapi import FastAPI, WebSocket, WebSocketDisconnect, Query
from collections import defaultdict, deque
from datetime import datetime, timezone
import json
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

app = FastAPI()


# --- Server-side: message buffer for replay on reconnect ---
class MessageBuffer:
    """Stores recent messages per room for reconnection replay."""

    def __init__(self, max_size: int = 100):
        self.buffers: dict[str, deque] = defaultdict(
            lambda: deque(maxlen=max_size)
        )
        self.counter = 0

    def add(self, room: str, message: dict) -> int:
        self.counter += 1
        entry = {"id": self.counter, **message}
        self.buffers[room].append(entry)
        return self.counter

    def get_since(self, room: str, last_id: int) -> list[dict]:
        """Returns all messages after the given ID."""
        return [
            msg for msg in self.buffers[room]
            if msg["id"] > last_id
        ]


buffer = MessageBuffer(max_size=200)
rooms: dict[str, set[WebSocket]] = defaultdict(set)


@app.websocket("/ws/chat/{room_id}")
async def chat_with_replay(
    websocket: WebSocket,
    room_id: str,
    last_event_id: int = Query(default=0),
):
    """WebSocket endpoint that supports message replay on reconnect."""
    await websocket.accept()
    rooms[room_id].add(websocket)

    # Replay missed messages
    if last_event_id > 0:
        missed = buffer.get_since(room_id, last_event_id)
        for msg in missed:
            await websocket.send_json(msg)
        await websocket.send_json({
            "type": "system",
            "content": f"Replayed {len(missed)} missed messages",
        })

    try:
        while True:
            data = await websocket.receive_json()
            message = {
                "type": "chat",
                "content": data.get("content", ""),
                "timestamp": datetime.now(timezone.utc).isoformat(),
            }
            msg_id = buffer.add(room_id, message)
            message["id"] = msg_id

            # Broadcast to all in room
            dead = set()
            for ws in rooms[room_id]:
                try:
                    await ws.send_json(message)
                except Exception:
                    dead.add(ws)
            rooms[room_id] -= dead

    except WebSocketDisconnect:
        rooms[room_id].discard(websocket)
        logger.info(f"Client disconnected from room {room_id}")


# --- Client-side JavaScript with reconnection ---
# Include this in your HTML served by FastAPI
CLIENT_JS = """
class ReconnectingWebSocket {
    constructor(url, options = {}) {
        this.url = url;
        this.maxRetries = options.maxRetries || 10;
        this.retries = 0;
        this.lastEventId = 0;
        this.onmessage = null;
        this.onopen = null;
        this.onclose = null;
        this.connect();
    }

    connect() {
        // Append last_event_id for replay
        const sep = this.url.includes('?') ? '&' : '?';
        const wsUrl = `${this.url}${sep}last_event_id=${this.lastEventId}`;
        this.ws = new WebSocket(wsUrl);

        this.ws.onopen = () => {
            console.log('Connected');
            this.retries = 0;  // Reset on successful connection
            if (this.onopen) this.onopen();
        };

        this.ws.onmessage = (event) => {
            const data = JSON.parse(event.data);
            // Track last event ID for reconnection
            if (data.id) this.lastEventId = data.id;
            if (this.onmessage) this.onmessage(data);
        };

        this.ws.onclose = (event) => {
            if (this.retries >= this.maxRetries) {
                console.error('Max retries reached');
                if (this.onclose) this.onclose(event);
                return;
            }
            // Exponential backoff with jitter
            const base = Math.min(1000 * Math.pow(2, this.retries), 30000);
            const jitter = base * 0.5 * Math.random();
            const delay = base + jitter;
            console.log(`Reconnecting in ${Math.round(delay)}ms...`);
            this.retries++;
            setTimeout(() => this.connect(), delay);
        };
    }

    send(data) {
        if (this.ws.readyState === WebSocket.OPEN) {
            this.ws.send(JSON.stringify(data));
        }
    }
}

// Usage:
// const ws = new ReconnectingWebSocket('ws://localhost:8000/ws/chat/general');
// ws.onmessage = (data) => console.log(data);
// ws.send({content: 'Hello!'});
"""

@app.get("/client.js")
async def serve_client_js():
    from fastapi.responses import Response
    return Response(content=CLIENT_JS, media_type="application/javascript")

Key Points

•The browser WebSocket API has no automatic reconnection -- you must implement it yourself
•Use exponential backoff with jitter to avoid thundering herd on server restarts
•Track last_event_id on the client and pass it on reconnection for message replay
•The server should buffer recent messages and replay missed ones when a client reconnects

Gotchas

•Without jitter, all clients reconnect at the same time after a server restart, causing a spike
•Message buffers consume memory -- use a bounded deque or persist to Redis for large-scale apps
•A reconnecting client gets a new WebSocket instance -- the server sees it as a new connection, not a resume
•If the server was restarted too, the message buffer is lost -- use Redis or a database for durability

Browser WebSockets have no automatic reconnection. Implement a wrapper class with exponential backoff and jitter. Track the last received message ID and pass it as a query parameter on reconnection so the server can replay missed messages.

•Exponential backoff with jitter prevents thundering herd on reconnection

// Client-side reconnecting WebSocket
class ReconnectWS {
    constructor(url) {
        this.url = url;
        this.retries = 0;
        this.lastId = 0;
        this.connect();
    }
    connect() {
        this.ws = new WebSocket(`${this.url}?last_event_id=${this.lastId}`);
        this.ws.onopen = () => { this.retries = 0; };
        this.ws.onclose = () => {
            const delay = Math.min(1000 * 2 ** this.retries, 30000);
            setTimeout(() => this.connect(), delay + Math.random() * delay * 0.5);
            this.retries++;
        };
        this.ws.onmessage = (e) => {
            const data = JSON.parse(e.data);
            if (data.id) this.lastId = data.id;
        };
    }
}

Without a server-side message buffer, clients miss all messages sent while they were disconnected

Question 12 / 12

What are the limitations of in-memory WebSocket connection management?

In-memory WebSocket management (storing connections in a Python dict/list) has fundamental limitations that become critical as your application scales.

Limitations:

Per-process isolation: Each Uvicorn worker has its own set of connections. A message sent in worker A cannot reach clients connected to worker B.
No persistence: If the server restarts, all connections and state are lost. There is no replay or recovery.
Memory growth: Each WebSocket connection consumes RAM. Thousands of connections can exhaust server memory.
No horizontal scaling: Adding more servers does not help -- clients on different servers cannot communicate.
Single point of failure: If the process dies, all connected clients are dropped simultaneously.

Solutions:

Redis Pub/Sub: Each worker subscribes to a Redis channel. When a message arrives, it is published to Redis, and all workers relay it to their local clients.
Dedicated WebSocket service: Use a specialized service (Socket.IO with Redis adapter, Centrifugo, Pusher, Ably) that handles connection management at scale.
Sticky sessions: Configure your load balancer to route a client to the same worker. Solves cross-worker issues but not horizontal scaling.
Event streaming: Use Kafka or NATS for cross-service message routing with persistence.

Redis Pub/Sub pattern:

This is the most common solution. Each FastAPI worker subscribes to a Redis channel and forwards messages to its local connections.

Pseudo Code

# Each worker subscribes to Redis
async def lifespan(app):
    redis = await aioredis.from_url("redis://localhost")
    pubsub = redis.pubsub()
    await pubsub.subscribe("chat")
    # Background task reads from Redis
    asyncio.create_task(relay_from_redis(pubsub))
    yield

# Publishing: worker puts message in Redis
async def broadcast(room, message):
    await redis.publish(room, json.dumps(message))

# Relay: reads from Redis, sends to local clients
async def relay_from_redis(pubsub):
    async for msg in pubsub.listen():
        for ws in local_connections:
            await ws.send_text(msg["data"])

Code

from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from contextlib import asynccontextmanager
from collections import defaultdict
import asyncio
import json
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


# --- Redis Pub/Sub solution for cross-worker broadcasting ---

class RedisBroadcaster:
    """Bridges local WebSocket connections with Redis Pub/Sub."""

    def __init__(self):
        self.local_connections: dict[str, set[WebSocket]] = defaultdict(set)
        self.redis = None
        self.pubsub = None

    async def connect(self, redis_url: str = "redis://localhost:6379"):
        import redis.asyncio as aioredis
        self.redis = aioredis.from_url(redis_url)
        self.pubsub = self.redis.pubsub()
        logger.info("Redis broadcaster connected")

    async def disconnect(self):
        if self.pubsub:
            await self.pubsub.unsubscribe()
            await self.pubsub.close()
        if self.redis:
            await self.redis.close()

    async def subscribe(self, channel: str):
        """Subscribe to a Redis channel and start relaying."""
        await self.pubsub.subscribe(channel)

    async def relay_loop(self):
        """Reads from Redis and forwards to local connections."""
        try:
            async for message in self.pubsub.listen():
                if message["type"] != "message":
                    continue
                data = json.loads(message["data"])
                channel = message["channel"]
                if isinstance(channel, bytes):
                    channel = channel.decode()

                # Forward to local connections in this channel
                dead = set()
                for ws in self.local_connections.get(channel, set()):
                    try:
                        await ws.send_json(data)
                    except Exception:
                        dead.add(ws)
                self.local_connections[channel] -= dead
        except asyncio.CancelledError:
            pass

    async def publish(self, channel: str, message: dict):
        """Publish to Redis -- all workers will receive it."""
        await self.redis.publish(channel, json.dumps(message))

    def add_local(self, channel: str, websocket: WebSocket):
        self.local_connections[channel].add(websocket)

    def remove_local(self, channel: str, websocket: WebSocket):
        self.local_connections[channel].discard(websocket)
        if not self.local_connections[channel]:
            del self.local_connections[channel]


broadcaster = RedisBroadcaster()


@asynccontextmanager
async def lifespan(app: FastAPI):
    # Connect to Redis and start relay loop
    await broadcaster.connect()
    await broadcaster.subscribe("chat:general")
    relay_task = asyncio.create_task(broadcaster.relay_loop())
    yield
    relay_task.cancel()
    await broadcaster.disconnect()


app = FastAPI(lifespan=lifespan)


@app.websocket("/ws/chat/{room_id}")
async def scalable_chat(
    websocket: WebSocket,
    room_id: str,
):
    await websocket.accept()
    channel = f"chat:{room_id}"
    broadcaster.add_local(channel, websocket)

    # Ensure we are subscribed to this room's channel
    await broadcaster.subscribe(channel)

    try:
        while True:
            data = await websocket.receive_json()
            # Publish to Redis -- ALL workers receive this
            await broadcaster.publish(channel, {
                "type": "chat",
                "content": data.get("content", ""),
                "room": room_id,
            })
    except WebSocketDisconnect:
        broadcaster.remove_local(channel, websocket)
        logger.info(f"Client left room {room_id}")


# --- In-memory limitations summary ---
# Problem: Worker A has clients [Alice, Bob], Worker B has [Charlie]
# Alice sends a message -> only Bob sees it (same worker)
# Charlie never gets the message (different worker)
#
# Solution: All workers publish to Redis, all workers relay from Redis
# Alice sends -> Worker A publishes to Redis -> Worker B relays to Charlie

Key Points

•In-memory connections are per-process -- clients on different workers cannot communicate
•Redis Pub/Sub is the standard solution for cross-worker WebSocket broadcasting
•Each worker subscribes to Redis channels and relays messages to its local connections
•The publish-subscribe pattern decouples message production from delivery

Gotchas

•Redis Pub/Sub is fire-and-forget -- if a worker is down during publish, it misses messages (use Redis Streams for durability)
•Subscribing to many channels per worker increases Redis connection overhead -- consider pattern subscriptions
•The relay loop must handle CancelledError for clean shutdown
•Running multiple Uvicorn workers with --workers flag creates separate processes, each with their own memory -- this is where in-memory management breaks down

In-memory WebSocket management is per-process. Clients on different Uvicorn workers cannot communicate. The standard solution is Redis Pub/Sub: each worker publishes messages to Redis and subscribes to relay them to local connections. For durability, use Redis Streams or Kafka instead of Pub/Sub.

•In-memory = per-process; use Redis Pub/Sub for cross-worker communication

# Each worker publishes to Redis instead of broadcasting locally
import redis.asyncio as aioredis

async def publish(channel: str, message: dict):
    redis = aioredis.from_url("redis://localhost")
    await redis.publish(channel, json.dumps(message))

# Each worker runs a relay loop reading from Redis
async def relay_loop(pubsub, local_connections):
    async for msg in pubsub.listen():
        if msg["type"] == "message":
            data = json.loads(msg["data"])
            for ws in local_connections:
                await ws.send_json(data)

Redis Pub/Sub is fire-and-forget -- use Streams or Kafka if messages must not be lost

Built by onllm.dev

Home GitHub Support awesomePrep

How do you create WebSocket endpoints in FastAPI?

How it works:

Key differences from HTTP endpoints:

Path parameters and query parameters:

How does the WebSocket connection lifecycle work (accept, receive, send)?

1. Handshake (HTTP Upgrade):

2. Data Transfer:

3. Close:

WebSocket state transitions:

Common close codes:

How do you handle WebSocketDisconnect exceptions?

When it is raised:

How to handle it:

Important close codes:

What happens if you do not handle it:

What is the ConnectionManager pattern for managing multiple clients?

Core responsibilities:

Why use a class:

Advanced extensions:

Concurrency note:

How do you implement authentication for WebSocket connections?

Authentication strategies:

Why not the Authorization header?

Best practice:

How do you send and receive JSON over WebSockets?

Built-in methods:

The mode parameter:

Pydantic validation:

Structured message protocol:

How do you implement a chat room with broadcasting?

Architecture:

Message protocol:

Scalability considerations:

How do you use WebSocket dependencies?

What works:

What does NOT work:

Dependency lifecycle:

Accessing WebSocket in dependencies:

What is Server-Sent Events (SSE) and how does it compare to WebSockets?

SSE vs WebSockets:

When to use SSE:

When to use WebSockets:

SSE message format:

How do you implement heartbeat/ping-pong for connection health?

Two levels of ping/pong:

Uvicorn configuration:

Application-level heartbeat pattern:

How do you handle reconnection logic on the client side?

Key reconnection strategies:

Server-side support:

Libraries:

What the server should do:

What are the limitations of in-memory WebSocket connection management?

Limitations:

Solutions:

Redis Pub/Sub pattern:

WebSockets & Real-time Features

The `mode` parameter: