WebSockets

This is the v1.0 documentation, for a release that is still in development. The API it describes is not on PyPI and can still change.

Reading about the version you have installed? Switch to v0.x.

WebSockets

A WebSocket is a long-lived, bidirectional connection between a client and your server. Unlike an HTTP request, which ends when the response is sent, a WebSocket stays open and either side can send at any time. That makes it the right tool for chat, live dashboards, collaborative editing, multiplayer state, and push notifications.

The core ships the connection itself, and nothing about talking to more than one of them at a time:

Layer	What it gives you	Where
WebSocketContext	The raw connection: accept, send, receive, close	This page
Peer, Hub, RoomConsumer	Rooms, presence, broadcast, replay	sillo-wire

Rooms and fan-out moved out of the core in v1. The reason is a dependency direction rather than size: the room layer needs a socket, a socket needs nothing from the room layer, and fan-out is the part that grows a backend once groups have to span more than one worker process.

pip install sillo-wire

Start here regardless. Most applications end up using a hub, but understanding the raw connection first makes the rest obvious rather than magical.

When to reach for a WebSocket

Use one when the server needs to initiate messages, and often: a chat room, a price ticker, a progress feed, a presence indicator.

Do not use one when the client drives everything. A form submission, a search, a file upload, a CRUD API: all of these are request/response, and HTTP does request/response better, with caching, retries, status codes, and every piece of debugging tooling ever built.

Consider Server-Sent Events instead when the traffic is one-directional server-to-client. SSE runs over plain HTTP, reconnects automatically in the browser, survives proxies that mangle upgrades, and needs no special infrastructure. See Streaming Responses. Reach for a WebSocket when you genuinely need the client to push too.

Routing

Register a WebSocket endpoint with the ws_route decorator:

a minimal echo endpoint

from sillo import SilloApp
from sillo.websockets import WebSocketContext, WebSocketDisconnect

app = SilloApp()


@app.ws_route("/ws/echo")
async def echo(websocket: WebSocketContext):
    await websocket.accept()
    try:
        while True:
            message = await websocket.receive_text()
            await websocket.send_text(f"echo: {message}")
    except WebSocketDisconnect:
        pass

The handler takes the context first, exactly as an HTTP handler does — a WebSocketContext rather than an HttpContext. It does not return anything; the connection ends when the handler returns.

There is no @app.websocket(...) decorator. The name is ws_route.

Three equivalent registration forms:

the three ways to register

from sillo import WebSocketContext

@app.ws_route("/ws/chat")
async def chat(websocket: WebSocketContext): ...

app.add_ws_route(path="/ws/chat", handler=chat)

from sillo.core.routing import WebsocketRoute
app.add_ws_route(WebsocketRoute("/ws/chat", chat))

Path parameters

Declared the same way as HTTP routes, and delivered the same way: as keyword arguments after the context.

path parameters

from sillo import WebSocketContext

@app.ws_route("/ws/room/{room_id}")
async def room(websocket: WebSocketContext, room_id: str):
    await websocket.accept()
    await websocket.send_json({"room": room_id})

They are also on websocket.path_params, which is what you read when the handler is generic over the route:

the same value, off the connection

from sillo import WebSocketContext

@app.ws_route("/ws/room/{room_id}")
async def room(websocket: WebSocketContext):
    await websocket.accept()
    room_id = websocket.path_params["room_id"]      # '42' — a str
    await websocket.send_json({"room": room_id})

Index individual parameters; do not serialize the mapping

websocket.path_params["room_id"] gives you a plain str. The mapping itself is not directly JSON-serializable, passing it whole to send_json raises TypeError: Object of type RouteParam is not JSON serializable and, because that happens inside the connection handler, closes the socket.

await websocket.send_json(websocket.path_params)          # breaks
await websocket.send_json(dict(websocket.path_params))    # also risky
await websocket.send_json({"room": websocket.path_params["room_id"]})   # fine

Dependency injection

A WebSocket handler’s signature is analysed the same way an HTTP handler’s is, so it can declare Depend(...) parameters. The whole tree is resolved once, when the socket connects — nothing is re-solved per message. Like every dependency, a WebSocket dependency is called with the context as its first positional argument.

a dependency on a socket

from sillo import Depend, WebSocketContext


async def get_room_service(_):
    return RoomService()


@app.ws_route("/ws/room/{room_id}")
async def room(
    websocket: WebSocketContext,
    room_id: str,
    rooms=Depend(get_room_service),
):
    await websocket.accept()
    await rooms.join(room_id, websocket)

Generator dependencies work too: the part after yield runs when the connection handler returns, so a resource opened for the socket is released when the socket closes.

a resource scoped to the connection

async def get_conn(_):
    conn = await pool.acquire()
    try:
        yield conn
    finally:
        await pool.release(conn)

To reach the connection itself from inside a dependency, take it as the first parameter — it is the live WebSocketContext, the same object the handler receives.

the connection, inside a dependency

from sillo import Depend, WebSocketContext


def client_ip(ctx: WebSocketContext) -> str:
    return ctx.client.host if ctx.client else "unknown"


@app.ws_route("/ws/feed")
async def feed(websocket: WebSocketContext, ip: str = Depend(client_ip)):
    await websocket.accept()
    await websocket.send_json({"from": ip})

Grouping routes on a router

a websocket router

from sillo.core.routing import Router

ws_router = Router(prefix="/ws")
ws_router.add_ws_route(path="/chat", handler=chat)
ws_router.add_ws_route(path="/presence", handler=presence)

app.mount_router(ws_router)

mount_router does not take a prefix. Set it on the Router. And add_ws_route wants either a WebsocketRoute instance or the path=/handler= keyword pair; a bare positional function will not do.

The connection lifecycle

Every WebSocket goes through the same four phases.

Handshake. The client sends an HTTP request with Upgrade: websocket. Your handler is invoked, and nothing can be sent yet.

Accept. You call await websocket.accept(). Until this happens the client is waiting; if you return without accepting or closing, the connection is rejected.

Exchange. Both sides send and receive freely until one stops.

Close. Either side closes. A client disconnect raises WebSocketDisconnect from whatever receive call is in flight.

the full shape

from sillo import WebSocketContext

@app.ws_route("/ws/session")
async def session(websocket: WebSocketContext):
    await websocket.accept()
    try:
        async for message in websocket.iter_json():
            await handle(message)
    except WebSocketDisconnect:
        pass
    finally:
        await cleanup()

iter_json(), iter_text(), and iter_bytes() are async generators that loop until disconnect and then stop cleanly. They swallow WebSocketDisconnect internally, so the except above is belt and braces for anything raised by handle.

Do work before accepting when you need to reject

Authentication belongs before accept(). Once accepted, a rejection costs the client a round trip and looks like an error rather than a refusal.

authenticating before accept

from sillo import WebSocketContext

@app.ws_route("/ws/private")
async def private(websocket: WebSocketContext):
    token = websocket.query_params.get("token")
    user = await authenticate(token)
    if user is None:
        await websocket.close(code=1008)   # policy violation
        return
    await websocket.accept()
    ...

Browsers cannot set custom headers on a WebSocket handshake, which is why tokens usually arrive as a query parameter or in the first message after accept. A query parameter ends up in access logs and proxy logs. Prefer a short-lived, single-use ticket over a long-lived session token, and treat it as compromised the moment it is used.

The state machine

WebSocketContext tracks two states (client_state and application_state) and enforces legal transitions. This is why the error messages are specific rather than mysterious.

Attempt	Result
receive_text() before accept()	RuntimeError: WebSocket is not connected. Need to call "accept" first.
send_text() after close()	RuntimeError: Cannot call "send" once a close message has been sent.
receive() after a disconnect message	RuntimeError: Cannot call "receive" once a disconnect message has been received.
send() when the peer is gone (OSError)	WebSocketDisconnect(code=1006)

Two of these matter in practice. Accepting exactly once is easy to get wrong when a base class also accepts. See the consumer guide. And is_connected() is a method, not a property:

if websocket.is_connected():      # correct
    await websocket.send_text("still here")

if websocket.is_connected:        # always truthy — this is a bound method
    await websocket.send_text("sent even after the peer is gone")

The second form is always true and silently defeats whatever guard you wrote it for.

Sending and receiving

the message API

await websocket.send_text("hello")
await websocket.send_bytes(b"\x00\x01")
await websocket.send_json({"type": "ping"})
await websocket.send_json(payload, mode="binary")   # JSON over a binary frame

text = await websocket.receive_text()
data = await websocket.receive_bytes()
obj = await websocket.receive_json()
obj = await websocket.receive_json(mode="binary")

send_json serializes with separators=(",", ":") and ensure_ascii=False, compact output, and non-ASCII characters sent as real UTF-8 rather than escapes.

Frame type must match on both ends. Calling receive_text() when the client sent a binary frame raises KeyError: 'text', because the ASGI message has a bytes key instead. Pick one encoding per endpoint and document it, or use receive() and branch on what arrived:

handling either frame type

message = await websocket.receive()
if message["type"] == "websocket.disconnect":
    return
payload = message.get("text") or message["bytes"].decode()

Closing

await websocket.close(code=1000, reason="session ended")

The close codes worth knowing, all available as constants in sillo.websockets.status:

Code	Constant	Meaning
1000	WS_1000_NORMAL_CLOSURE	Done, no error
1001	WS_1001_GOING_AWAY	Server shutting down
1003	WS_1003_UNSUPPORTED_DATA	Wrong frame type or unparseable payload
1008	WS_1008_POLICY_VIOLATION	Auth failure, rate limit
1011	WS_1011_INTERNAL_ERROR	Unhandled exception

Codes in the 4000 to 4999 range are yours to define. Use them for application-level reasons the client should act on differently: “token expired, refresh and reconnect” versus “you were banned, do not reconnect”.

Note that browsers do not expose the close reason reliably and, for abnormal closures, report code 1006 regardless of what you sent. Anything the client must act on should be a normal message sent before closing.

Scaling

Groups live in one process's memory

ChannelBox.CHANNEL_GROUPS is a plain class-level dictionary. It is not shared between processes, workers, or machines.

Run uvicorn --workers 4 and you have four independent group registries. A user connected to worker 1 will never receive a broadcast issued on worker 3. In development with one worker everything appears to work; the bug appears only under the deployment configuration you actually ship.

Three ways out, in increasing order of effort:

One worker per process, sticky by room. Route each room to a fixed worker at the load balancer. Simple, and it caps a room’s size at one process.

A Redis pub/sub fan-out layer. Every worker subscribes to the rooms its local connections care about, and broadcast publishes to Redis instead of iterating a local dict. This is what Django Channels’ channel layer does, and it is roughly a hundred lines here.

A dedicated realtime service. Move WebSocket termination out of the application entirely.

Whichever you choose, decide before you ship, retrofitting the fan-out means touching every broadcast call site.

The other scaling limit is per-connection memory. Each open WebSocket holds a task, two ASGI callables, and whatever your handler keeps in scope. Ten thousand connections holding a loaded user object each is a memory profile worth measuring rather than assuming.

Testing

TestClient speaks WebSocket, so endpoints are testable without a running server. The connection is a context manager and the send/receive calls are synchronous.

testing a websocket endpoint

from sillo.testclient import TestClient


def test_echo():
    with TestClient(app) as client:
        with client.websocket_connect("/ws/echo") as ws:
            ws.send_text("hello")
            assert ws.receive_text() == "echo: hello"


def test_room_path_param():
    with TestClient(app) as client:
        with client.websocket_connect("/ws/room/42") as ws:
            assert ws.receive_json() == {"room": "42"}

Leaving the inner block closes the connection, which is how you exercise disconnect handling, assert on whatever your finally was supposed to clean up. A rejected connection surfaces as WebSocketDisconnect raised by websocket_connect, so an auth test wraps it in pytest.raises.

What not to do

Do not use @app.websocket(...). It does not exist; use ws_route.

Do not treat is_connected as a property. It is a method, and the un-called form is always truthy.

Do not accept before authenticating when the endpoint is private.

Do not send a path_params mapping to send_json. Index the values.

Do not mix frame types on one endpoint without branching on receive().

Do not rely on the close reason reaching the browser. Send a normal message first.

Do not run multiple workers with in-memory groups. Broadcasts will silently reach a fraction of your users.

Do not do slow work inside the receive loop. A blocking call stalls that connection entirely; hand it to a background task.

Performance notes

Each connection is one asyncio task. Idle connections cost almost nothing; the cost is in what you keep alive alongside them.

send_json serializes on every call. Broadcasting one payload to a thousand channels currently serializes it a thousand times, because Channel._send calls websocket.send_json(payload) per channel. For large payloads to large groups, pre-serialize once and use text channels.

Backpressure is the failure mode that surprises people. If a client stops reading and you keep sending, the frames queue in the server’s buffers until memory runs out. Cap what you queue per connection and drop or disconnect a client that falls too far behind. A slow consumer must not be allowed to degrade the process.

API reference

Member	Signature	Notes
accept	(subprotocol=None, headers=None)	Waits for websocket.connect first
receive_text / receive_bytes	() -> str / bytes	Raise if the frame type differs
receive_json	(mode="text") -> Any	mode is "text" or "binary"
iter_text / iter_bytes / iter_json	async generators	Stop cleanly on disconnect
send_text / send_bytes	(data)
send_json	(data, mode="text")	Compact separators, ensure_ascii=False
close	(code=1000, reason=None)
is_connected	() -> bool	Method, not a property
path_params / query_params / headers	mappings	Inherited from BaseContext

Related

• Wire: rooms, presence, broadcast and replay, in a package that installs alongside the core
• Events: routing messages to handlers by type
• Streaming Responses: SSE, for one-directional push
• Routing: how paths and parameters are matched
• Dependency injection: Depend(...) on a handler, HTTP or WebSocket