Type-safe, scalable WebSockets in Python: a FastAPI tutorial

This is the error this post builds up to:

src/chat.ts(8,60): error TS2339: Property 'text' does not exist on type 'ChatNotificationPayload'.

The Python server renamed a WebSocket field from text to body, and the TypeScript frontend found out at compile time, through a client generated from the server's code. Without that client the field reads as undefined, nothing throws, and the user sees an empty chat bubble. The same kind of bug (content on one side, body on the other) once cost me most of a day, and it's how I opened my talk at AsyncAPI Conf 2026, which covers the approach behind this post.

FastAPI already gives HTTP routes this kind of safety: type hints become an OpenAPI schema, a /docs page and generated clients. @app.websocket gets none of it. You receive a dict, check a string and send a dict back, and nothing describes those messages.

So in this post we build a small FastAPI room chat where every WebSocket message is a Pydantic model. From those models we get validation, an AsyncAPI docs page, and typed TypeScript and Python clients. Then we move it onto Redis so it keeps working when you run more than one server, and push messages into it from a plain HTTP endpoint. It takes about 15 minutes, and every command and output below comes from a real run (chanx 2.13.0, FastAPI 0.162, @chanx-js/client 2.3, Redis 7).

Setup

mkdir room-chat && cd room-chat
python -m venv .venv && source .venv/bin/activate
pip install "chanx[fast_channels]" uvicorn websockets

The library doing the work is chanx, which I maintain. It also runs on Django Channels and other ASGI frameworks, but this post sticks to FastAPI.

1. Messages are Pydantic models

Each message has a constant action field and a typed payload. Create main.py:

from typing import Literal

from fastapi import FastAPI, Request
from fast_channels.layers import InMemoryChannelLayer, register_channel_layer
from pydantic import BaseModel, Field

from chanx.core.decorators import channel, ws_handler
from chanx.fast_channels import asyncapi_docs, asyncapi_spec_json
from chanx.fast_channels.websocket import AsyncJsonWebsocketConsumer
from chanx.messages.base import BaseMessage
from chanx.messages.incoming import PingMessage
from chanx.messages.outgoing import PongMessage


class ChatPayload(BaseModel):
    text: str = Field(min_length=1, max_length=500)


class ChatMessage(BaseMessage):
    """A message a user posts to the room."""

    action: Literal["chat"] = "chat"
    payload: ChatPayload


class ChatNotificationPayload(BaseModel):
    room: str
    text: str


class ChatNotification(BaseMessage):
    """A message broadcast to everyone in the room."""

    action: Literal["chat_notification"] = "chat_notification"
    payload: ChatNotificationPayload

Keep the Literal on action. It becomes "const" in the JSON schema, which lets the code generator later turn "some object" into a union the TypeScript compiler can narrow on. The docstrings travel too, into the docs page and as comments in the generated clients.

2. One handler per message

Add the consumer below the models:

@channel(name="room_chat", description="Chat in a named room")
class RoomChatConsumer(AsyncJsonWebsocketConsumer):
    channel_layer_alias = "default"

    async def post_authentication(self) -> None:
        self.room = self.scope["path_params"]["room_name"]
        await self.join_group(f"room_{self.room}")

    @ws_handler
    async def handle_ping(self, _message: PingMessage) -> PongMessage:
        return PongMessage()

    @ws_handler(output_type=ChatNotification)
    async def handle_chat(self, message: ChatMessage) -> None:
        await self.broadcast_message(
            ChatNotification(
                payload=ChatNotificationPayload(room=self.room, text=message.payload.text)
            )
        )

For every incoming message, chanx picks the handler whose argument type has a matching action, so you never write a receive method or an if action == ... chain.

handle_ping returns its reply, which goes back to the sender only. handle_chat broadcasts to the room's group instead, and because its return type is None, output_type tells chanx (and the docs) what it sends. join_group adds the connection to a group named after the room in the URL, and the connection leaves that group on its own when it disconnects.

3. Wire it into FastAPI

At the bottom of main.py:

register_channel_layer("default", InMemoryChannelLayer())  # Redis in production

app = FastAPI()
app.router.add_websocket_route("/ws/room/{room_name}", RoomChatConsumer.as_asgi())


@app.get("/asyncapi")
async def docs(request: Request):
    return await asyncapi_docs(request=request, app=app)


@app.get("/asyncapi.json")
async def spec(request: Request):
    return await asyncapi_spec_json(request=request, app=app)

The channel layer is what carries broadcasts between connections. In-memory is fine while everything runs in one process; section 9 swaps it for Redis.

Run it:

uvicorn main:app

4. Try it

Here's a quick script with two users in the same room. Save it as try_it.py:

import asyncio
import json

import websockets


async def main():
    url = "ws://localhost:8000/ws/room/lobby"
    async with websockets.connect(url) as alice, websockets.connect(url) as bob:
        await alice.send(json.dumps({"action": "ping"}))
        print("alice <-", await alice.recv())

        await alice.send(json.dumps({"action": "chat", "payload": {"text": "hi bob"}}))
        print("alice <-", await alice.recv())
        print("bob   <-", await bob.recv())

        await alice.send(json.dumps({"action": "chat", "payload": {"txt": "hi"}}))
        print("alice <-", await alice.recv())


asyncio.run(main())

$ python try_it.py
alice <- {"action": "pong", "payload": null}
alice <- {"action": "chat_notification", "payload": {"room": "lobby", "text": "hi bob"}}
bob   <- {"action": "chat_notification", "payload": {"room": "lobby", "text": "hi bob"}}
alice <- {"action": "error", "payload": [{"type": "missing", "loc": ["chat", "payload", "text"], "msg": "Field required"}]}

Look at the last line: txt instead of text never reached handle_chat. Pydantic rejected it first, and the sender got an error naming the missing field.

5. The docs page

Open http://localhost:8000/asyncapi.

This is the official AsyncAPI viewer reading /asyncapi.json, which chanx builds from the handlers: the channel and its room_name parameter, each operation, and each message's schema and docstring. Since it comes from the code on every request, it can't drift out of date the way a wiki page does.

6. A typed TypeScript client

The same spec feeds a client generator. In a frontend project (or an empty folder):

npm install @chanx-js/client
npm install -D @chanx-js/codegen typescript
npx chanx-codegen http://localhost:8000/asyncapi.json -o src/generated

The generator reads the spec and writes plain TypeScript. Part of src/generated/schemas.ts:

/** A message a user posts to the room. */
export interface ChatMessage {
  action: "chat";
  payload: ChatPayload;
}

/** A message broadcast to everyone in the room. */
export interface ChatNotification {
  action: "chat_notification";
  payload: ChatNotificationPayload;
}

And src/generated/channels.ts knows which messages go in which direction, and which reply each request gets:

export type RoomChatToServer = ChatMessage | PingMessage;
export type RoomChatToClient = ChatNotification | PongMessage;
export type RoomChatReplies = {
  chat: ChatNotification;
  ping: PongMessage;
};

/** Chat in a named room */
export const roomChat = defineChannel<RoomChatToServer, RoomChatToClient, RoomChatReplies>()({
  name: "room_chat",
  address: "/ws/room/{room_name}",
  heartbeat: true,
});

Using it (src/chat.ts):

import { createClient } from '@chanx-js/client';
import { roomChat } from './generated/index.js';

const client = createClient({ baseUrl: 'ws://localhost:8000' });
const room = client.connect(roomChat, { params: { room_name: 'lobby' } });

room.on('chat_notification', (message) => {
  console.log(`[${message.payload.room}] ${message.payload.text}`);
});

const pong = await room.request({ action: 'ping' });
console.log('server replied with', pong.action);

room.send({ action: 'chat', payload: { text: 'hello from TypeScript' } });

$ npx tsx src/chat.ts
server replied with pong
[lobby] hello from TypeScript

message in the on callback is already a ChatNotification, and request({ action: 'ping' }) resolves to a PongMessage, both inferred from the spec. In React, Vue, Svelte or Solid you'd use the matching hook from @chanx-js/client/react (and so on) instead of createClient.

7. Mistakes become compile errors

Now some common mistakes:

client.connect(roomChat, { params: { room: 'lobby' } });
room.send({ action: 'chatt', payload: { text: 'hi' } });
room.send({ action: 'chat', payload: { txt: 'hi' } });
room.send({ action: 'chat_notification', payload: { room: 'lobby', text: 'hi' } });

$ npx tsc --noEmit
error TS2353: Object literal may only specify known properties, and 'room' does not exist in type '{ room_name: string | number; }'.
error TS2820: Type '"chatt"' is not assignable to type '"chat" | "ping"'. Did you mean '"chat"'?
error TS2561: Object literal may only specify known properties, but 'txt' does not exist in type 'ChatPayload'. Did you mean to write 'text'?
error TS2322: Type '"chat_notification"' is not assignable to type '"chat" | "ping"'.

The last one is easy to miss: chat_notification is a message the server sends, and the spec says so, so the client isn't allowed to send it.

8. The rename

Back to the error from the top of the post. On the server, rename text to body in ChatNotificationPayload (and in handle_chat), restart, and regenerate:

npx chanx-codegen http://localhost:8000/asyncapi.json -o src/generated
npx tsc --noEmit

src/chat.ts(8,60): error TS2339: Property 'text' does not exist on type 'ChatNotificationPayload'.

The build fails on the line that reads the old field. Without the generated client, that same line would compile and quietly render nothing.

In my projects I commit the generated folder and run the generator in CI against the backend's spec. A backend PR that renames a field then shows the client diff in the same PR, and the frontend type check fails before anything merges.

9. Scale it out with Redis

So far everything runs in one process. In production you'd run several servers behind a load balancer, and two people in the same room can easily land on different ones. Before testing that, one more feature real apps need.

Messages don't only come from other WebSocket clients. A notification usually starts in an HTTP view or a background job, and for that chanx sends events through the channel layer. Add one line to the consumer so it forwards ChatNotification events straight to its client:

class RoomChatConsumer(AsyncJsonWebsocketConsumer):
    channel_layer_alias = "default"
    passthrough_events = [ChatNotification]
    ...

and an ordinary FastAPI endpoint that broadcasts one into a room:

class Announcement(BaseModel):
    text: str


@app.post("/rooms/{room_name}/announce")
async def announce(room_name: str, announcement: Announcement):
    await RoomChatConsumer.broadcast_event(
        ChatNotification(payload=ChatNotificationPayload(room=room_name, text=announcement.text)),
        groups=[f"room_{room_name}"],
    )
    return {"sent": True}

The event is the same Pydantic model the clients already know, so it's typed end to end, and the spec picks it up as a send operation.

Now start two copies of the app, still on the in-memory layer:

pip install httpx
uvicorn main:app --port 8001   # server A
uvicorn main:app --port 8002   # server B

Connect Alice to A and Bob to B, both in lobby, then post an announcement to B (save as two_servers.py):

import asyncio
import json

import httpx
import websockets


async def recv(ws):
    try:
        return await asyncio.wait_for(ws.recv(), timeout=2)
    except TimeoutError:
        return "(nothing)"


async def main():
    alice_url = "ws://localhost:8001/ws/room/lobby"  # server A
    bob_url = "ws://localhost:8002/ws/room/lobby"  # server B
    async with websockets.connect(alice_url) as alice, websockets.connect(bob_url) as bob:
        await alice.send(json.dumps({"action": "chat", "payload": {"text": "hi bob"}}))
        print("alice <-", await recv(alice))
        print("bob   <-", await recv(bob))

        async with httpx.AsyncClient() as http:
            await http.post("http://localhost:8002/rooms/lobby/announce", json={"text": "deploy at 5pm"})
        print("alice <-", await recv(alice))
        print("bob   <-", await recv(bob))


asyncio.run(main())

$ python two_servers.py
alice <- {"action": "chat_notification", "payload": {"room": "lobby", "text": "hi bob"}}
bob   <- (nothing)
alice <- (nothing)
bob   <- {"action": "chat_notification", "payload": {"room": "lobby", "text": "deploy at 5pm"}}

Each server only knows about its own connections, so Bob never sees Alice's message and Alice never sees the announcement. The fix is a channel layer every server shares. Start Redis and install the Redis extra:

docker run -d -p 6379:6379 redis:7-alpine
pip install "fast-channels[redis]"

Then change the layer setup in main.py. The consumer, the handlers and the endpoint stay as they are:

import os

from fast_channels.layers import register_channel_layer
from fast_channels.layers.redis import RedisChannelLayer

redis_url = os.environ.get("REDIS_URL", "redis://localhost:6379")
register_channel_layer("default", RedisChannelLayer(hosts=[redis_url]))

Restart both servers and run the script again:

$ python two_servers.py
alice <- {"action": "chat_notification", "payload": {"room": "lobby", "text": "hi bob"}}
bob   <- {"action": "chat_notification", "payload": {"room": "lobby", "text": "hi bob"}}
alice <- {"action": "chat_notification", "payload": {"room": "lobby", "text": "deploy at 5pm"}}
bob   <- {"action": "chat_notification", "payload": {"room": "lobby", "text": "deploy at 5pm"}}

Both servers now deliver both messages, including the announcement that was posted to server B only. A third or tenth server works the same way. The same broadcast_event call also works from an async background worker, and sync code such as a Celery task can use broadcast_event_sync. If you regenerate the TypeScript client from section 6, it still compiles.

10. Bonus: a Python client for another service

The same spec works for a Python service talking to this one (in my case it's usually an AI agent service talking to the backend):

pip install "chanx[cli,client]"
chanx generate-client --schema http://localhost:8000/asyncapi.json --output ./chat_client

import asyncio

from chat_client.room_chat import RoomChatClient
from chat_client.room_chat.messages import ChatMessage, ChatNotification, ChatPayload, PongMessage


class Bot(RoomChatClient):
    async def send_init_message(self) -> None:
        await self.send_message(ChatMessage(payload=ChatPayload(text="hello from a Python service")))

    async def handle_message(self, message: ChatNotification | PongMessage) -> None:
        match message:
            case ChatNotification():
                print(f"[{message.payload.room}] {message.payload.text}")
                await self.disconnect()
            case PongMessage():
                pass


if __name__ == "__main__":
    asyncio.run(Bot("ws://localhost:8000", path_params={"room_name": "lobby"}).handle())

$ python bot.py
[lobby] hello from a Python service

The generated messages are Pydantic models, so mypy and pyright check this side too.

Where to go from here

• The same consumer runs on Django Channels: import it from chanx.channels.websocket instead and route it with Channels. Django quick start
• Auth, testing helpers, events pushed from HTTP views or background jobs: chanx docs
• TypeScript client, hooks for each framework, and a Node server that speaks the same protocol: chanx-js
• How Python handlers map to AsyncAPI 3 channels, operations and replies, in detail: my AsyncAPI Conf 2026 talk

Code: github.com/chanx-dev/chanx (BSD).

Hope it saves you the day it cost me. If you keep WebSocket messages in sync some other way, share your tips in the comments; I'm always happy to pick up a better trick.