Skip to main content

WebSocket Basics

WebSockets enable bidirectional, persistent connections between client and server — essential for real-time features like notifications, live updates, and chat. FastAPI has first-class WebSocket support built on Starlette.

Learning Focus

By the end of this lesson you can: accept WebSocket connections, send and receive text and JSON messages, handle disconnection gracefully, and authenticate WebSocket clients via query parameters.

Basic WebSocket Endpoint​

app/routers/ws.py
from fastapi import APIRouter, WebSocket, WebSocketDisconnect

router = APIRouter()

@router.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket) -> None:
    await websocket.accept()
    try:
        while True:
            data = await websocket.receive_text()
            await websocket.send_text(f"Echo: {data}")
    except WebSocketDisconnect:
        pass  # Client disconnected — clean up here if needed

Sending and Receiving Data Types​

MethodSendsReceives
send_text(str)Text frame—
send_bytes(bytes)Binary frame—
send_json(dict)JSON text frame—
receive_text()—Text frame
receive_bytes()—Binary frame
receive_json()—JSON text frame
app/routers/ws.py
import json
from fastapi import APIRouter, WebSocket, WebSocketDisconnect

router = APIRouter()

@router.websocket("/ws/json")
async def json_websocket(websocket: WebSocket) -> None:
    await websocket.accept()
    try:
        while True:
            data = await websocket.receive_json()
            response = {"echo": data, "status": "ok"}
            await websocket.send_json(response)
    except WebSocketDisconnect:
        pass

WebSocket Authentication​

WebSocket handshakes don't support custom headers in browser clients. Use a query parameter token instead:

app/routers/ws.py
from fastapi import APIRouter, WebSocket, WebSocketDisconnect, Query, HTTPException
from app.core.security import decode_access_token
from jose import JWTError

router = APIRouter()

@router.websocket("/ws/protected")
async def protected_ws(
    websocket: WebSocket,
    token: str = Query(...),
) -> None:
    # Validate JWT before accepting
    try:
        payload = decode_access_token(token)
        user_id = payload.get("sub")
    except JWTError:
        await websocket.close(code=4001, reason="Invalid token")
        return

    await websocket.accept()
    try:
        while True:
            msg = await websocket.receive_text()
            await websocket.send_text(f"User {user_id}: {msg}")
    except WebSocketDisconnect:
        pass
client.js
const token = localStorage.getItem("access_token");
const ws = new WebSocket(`ws://localhost:8000/ws/protected?token=${token}`);
ws.onmessage = (event) => console.log(event.data);

WebSocket State and Path Parameters​

app/routers/ws.py
@router.websocket("/ws/rooms/{room_id}")
async def room_ws(room_id: str, websocket: WebSocket) -> None:
    await websocket.accept()
    await websocket.send_json({"joined": room_id})
    try:
        while True:
            message = await websocket.receive_json()
            # Broadcast to room (see WebSocket Chat Example lesson)
            await websocket.send_json({"room": room_id, "message": message})
    except WebSocketDisconnect:
        pass

Graceful Close​

app/routers/ws.py
from fastapi import APIRouter, WebSocket, WebSocketDisconnect
import asyncio

@router.websocket("/ws/timed")
async def timed_ws(websocket: WebSocket) -> None:
    await websocket.accept()
    try:
        for i in range(5):
            await websocket.send_json({"countdown": 5 - i})
            await asyncio.sleep(1)
        await websocket.send_json({"done": True})
        await websocket.close()
    except WebSocketDisconnect:
        pass

Common Pitfalls​

PitfallCause / SymptomFix
Connection closes immediatelyException before accept()Handle auth errors with close(code=...) not exceptions
WebSocketDisconnect not caughtUnhandled exception on client closeAlways wrap receive loop in try/except WebSocketDisconnect
Blocking in WebSocket handlerSync DB call inside async handlerUse async DB calls or asyncio.to_thread
Token in URL logged by proxyJWT in query param visible in Nginx logsAccept token, then immediately authenticate and remove from logs
Cannot send after closeSending after WebSocketDisconnectGuard sends with connection state check

Hands-On Practice​

test-websocket.sh
uvicorn app.routers.ws:router --reload --port 8011

# Install wscat for CLI WebSocket testing
npm install -g wscat

# Connect and test
wscat -c ws://localhost:8011/ws
> Hello FastAPI
< Echo: Hello FastAPI
> {"type": "ping"}
< (echo)

# Test protected endpoint
TOKEN="<your-jwt-token>"
wscat -c "ws://localhost:8011/ws/protected?token=$TOKEN"

What's Next​