WebSockets

WebSockets turn an HTTP upgrade into a long-lived bidirectional channel.

Handshake essentials

Client must send valid upgrade headers, including:

• Upgrade: websocket
• Connection: Upgrade
• Sec-WebSocket-Key
• Sec-WebSocket-Version: 13

If handshake validation fails, connection is rejected.

ASGI flow

1. app receives websocket.connect
2. app sends one of:
3. websocket.accept
4. websocket.close
5. websocket.http.response.start + body (HTTP-style rejection path)
6. after accept, app handles receive/send messages
7. close handshake and disconnect complete session

Easy echo example:

from __future__ import annotations


async def app(scope, receive, send):
    """Accept websocket clients and echo text messages."""
    if scope["type"] != "websocket":
        return

    await send({"type": "websocket.accept"})
    while True:
        message = await receive()
        message_type = message["type"]

        if message_type == "websocket.disconnect":
            break

        if message_type == "websocket.receive" and "text" in message:
            await send({"type": "websocket.send", "text": message["text"]})

Auth gate example:

from __future__ import annotations

from urllib.parse import parse_qs


async def app(scope, receive, send):
    """Allow upgrades only when a known token is present."""
    if scope["type"] != "websocket":
        return

    query_string = scope.get("query_string", b"").decode("utf-8")
    params = parse_qs(query_string)
    token = params.get("token", [""])[0]

    if token != "demo-token":
        await send({"type": "websocket.close", "code": 1008, "reason": "unauthorized"})
        return

    await send({"type": "websocket.accept"})
    while True:
        message = await receive()
        if message["type"] == "websocket.disconnect":
            break
        if message["type"] == "websocket.receive" and "text" in message:
            await send({"type": "websocket.send", "text": f"secure:{message['text']}"})

Stateful room example:

from __future__ import annotations

from collections import defaultdict

ROOMS: dict[str, set] = defaultdict(set)


async def app(scope, receive, send):
    """Broadcast received messages to all clients in one room."""
    if scope["type"] != "websocket":
        return

    query = scope.get("query_string", b"").decode("ascii")
    room = "general"
    if query.startswith("room="):
        room = query.split("=", 1)[1] or "general"

    await send({"type": "websocket.accept", "subprotocol": None})
    ROOMS[room].add(send)

    try:
        while True:
            message = await receive()
            if message["type"] == "websocket.disconnect":
                break

            text = message.get("text")
            if text is None:
                continue

            for peer_send in list(ROOMS[room]):
                await peer_send({"type": "websocket.send", "text": f"[{room}] {text}"})
    finally:
        ROOMS[room].discard(send)

Runtime controls

• --ws: backend mode selection
• --ws-max-size: max frame/message size
• --ws-max-queue: receive queue sizing
• --ws-ping-interval and --ws-ping-timeout
• --ws-per-message-deflate

Failure cases you should test

• invalid handshake headers
• oversized payloads
• invalid UTF-8 text frames
• half-open disconnects
• proxy configurations that drop upgrade headers

Plain-language explanation

HTTP is a request letter. WebSocket is a live conversation. It stays open until one side closes.