FastAPI + TypeScript 如何在 WebSocket 欄位改名時觸發編譯錯誤

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; the full story is in Part 3 of my Agent Graph Engineering series.

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. It takes about 10 minutes, and every command and output below comes from a real run (chanx 2.13.0, FastAPI 0.162, @chanx-js/client 2.3).

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 for a single process; for several workers you'd register a Redis layer from fast_channels.layers.redis instead, and nothing else changes.

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. 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.