DEV Community: Huy Nguyen

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

Huy Nguyen — Sat, 31 Oct 2026 06:51:02 +0000

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.

Agent Graph Engineering, Part 3: Contracts, Not Conventions

Huy Nguyen — Sat, 31 Oct 2026 02:30:56 +0000

TL;DR Three services in two languages means three seams, and a hand-maintained seam is a slow leak. Declare the shapes once on the server, generate everything else: OpenAPI becomes a typed REST client, AsyncAPI becomes a typed WebSocket client in TypeScript and another in Python. The compiler then tells you what broke instead of a user telling you. The same declarations build the approval form, so a reviewer cannot approve a call whose fields they were never shown.

Part 2 split the agent off from the business and drew the line at state ownership. That split creates a boundary, and a boundary between two codebases is where systems quietly go wrong. This post is about making it impossible for that boundary to drift.

The failure mode, which is not a crash

I have debugged a WebSocket message shape that disagreed between a backend and a frontend, and it cost me most of a day. Not because it is hard to find. Because both sides are right.

The backend sends {ticket_id, content}. The frontend reads {ticketId, body}. Each codebase is internally consistent, each has tests, and each team is certain. Nothing throws. The field is simply undefined, which in JavaScript renders as nothing at all, so the symptom is an empty space in the UI where a message should be. You go looking for a bug in rendering, then in the socket, then in the serializer, and the actual defect is that two people wrote down the same idea twice and one of them was six weeks ago.

The REST world mostly solved this a decade ago and we stopped noticing. You write a serializer, a spec falls out, the frontend generates a client, and a renamed field becomes a compile error. Nobody considers that remarkable any more.

The agent layer has no such habit. WebSocket message shapes are still usually a hand-written union on one side and a hand-written type on the other, with a code review as the only thing holding them together. That is a convention, not a contract. Conventions survive exactly as long as the people who remember them.

Types are the stance, generation is the machinery

Part 1 claimed that the structure of the system should be a declared artifact rather than something you reconstruct by reading code. Contracts are where that claim stops being philosophy.

The order matters, and it is worth being explicit because it is easy to get backwards. Typing is a design stance you adopt first. Declare a message as a model. Declare a tool's arguments as a schema. Declare an agent's output as a type. None of that needs a generator, and all of it pays off immediately in one codebase.

Generation is the machinery you add once a second codebase exists. Before that it is ceremony. The moment two processes have to agree, it stops being optional, because you have introduced a place where two people's understanding can differ silently.

So: declare on the server, generate everywhere else, and never write the same shape twice.

Three seams, three generators

The demo application has three services and two languages, which produces three places a shape can cross a boundary.

Seam	Source of truth	Spec	Generated artifact
Browser to backend, REST	DRF serializers	OpenAPI	Zodios client with Zod schemas
Browser to backend, WebSocket	chanx consumers and topics	AsyncAPI	TypeScript client via `chanx-codegen`
Backend to agent, WebSocket	the agent's chanx topics	AsyncAPI	a Python client the backend imports

Drawn out, with the tool that does each step on the arrow:

graph LR
    A[DRF serializers] -->|drf-spectacular| B[OpenAPI]
    B -->|openapi-zod-client| C[Zodios client]
    B -->|script| D[TypeScript types]

    E[Backend consumers] -->|chanx| F[AsyncAPI]
    F -->|chanx-js codegen| G[Typed WS client]

    H[Agent consumers] -->|chanx generate-client| I[Python WS client]

Every arrow is a generator, and nothing on the right of one is hand-written. One command runs all of it:

gen:all = gen:schema && gen:type && gen:ws

Which is deliberately boring. The interesting part is what comes out.

What generation actually preserves

Here is the Python definition of the team's half of a ticket feed. It is the whole thing, and it is mostly a list of what may travel on that channel:

class TicketTeamTopic(Topic[TeamFeedEvent]):
    """Outbound only: the actions a reviewer takes are sent on the ticket's own
    topic, which they are also subscribed to."""

    pattern = "ticket:{ticket_id}:team"
    passthrough_events = [
        NewEventMessage,
        AnswerStreamingMessage,
        AgentProgressMessage,
        ApprovalRequiredMessage,
        ToolProposalMessage,
        ReasoningStreamingMessage,
    ]

And here is what the frontend generator produces from it, after the AsyncAPI document in between:

export type HubTicketTeamTopicToServer = never;
export type HubTicketTeamTopicToClient =
  | AgentProgressMessage
  | AnswerStreamingMessage
  | ApprovalRequiredMessage
  | NewEventMessage
  | ReasoningStreamingMessage
  | ToolProposalMessage;

/** Outbound only: the actions a reviewer takes are sent on the ticket's own
 * topic, which they are also subscribed to. */
ticketTeamTopic: defineTopic<
  HubTicketTeamTopicToServer,
  HubTicketTeamTopicToClient
>()({
  name: 'ticket_team_topic',
  pattern: 'ticket:{ticket_id}:team',
}),

Ignore the spelling. The same name appears three ways across those two blocks, and that is a case transform you configure once and forget, not anything the contract is doing. What matters is which fields crossed.

The list did, exactly. Six message types on the Python side, the same six in the TypeScript union, and each one carries its own generated type behind it. Add a seventh to passthrough_events and it appears in the union on the next generation; remove one and every frontend branch still handling it stops compiling. That is the property worth having, and it is about fields rather than names.

Two more details are worth stopping on.

The first is ToServer = never. The Python class says outbound only, by listing no inbound handlers, and that fact survives all the way into the type system of a different language. A frontend developer who tries to send on that topic gets a compile error rather than a message that vanishes into a channel nobody reads. Nobody wrote that rule down twice. It was declared once, as the absence of handlers, and the absence travelled.

The second is the docstring. The sentence explaining why the topic is outbound only is now a JSDoc comment hovering in the editor of whoever is about to use it. I did not expect to care about that and now would not give it up. The reasoning lives next to the code that depends on it, in both languages, without anyone maintaining a second copy of the explanation.

This is also why I keep saying your WebSocket layer deserves what your REST layer already has. It is the same trick. There is no technical reason the channel your agent streams on should be less well specified than the endpoint that lists your tickets.

The schema can also build the interface

Here is where it stops being about typing and starts being about safety.

The agent can propose a tool call that a human has to approve before it runs. A refund, for instance. The reviewer needs to see the arguments, check them, possibly correct them, and then approve. That is a form.

The naive version is a hand-written form per tool. It works on the day you write it, and then the tool grows a field. The comment on the component that replaced all of them says it better than I will paraphrase it:

Hand-written forms are what rot: a field is added server-side, the form is
not updated, and nobody finds out until a request fails or - worse, for the
tool gate - until a reviewer approves a call whose arguments they were never
shown.

Read that last clause again, because it is the whole argument. A stale form is not a cosmetic bug in an approval flow. It is a human being asked to authorize something while part of it is hidden from them. The approval is still recorded, the tool still runs with the field the reviewer never saw, and the audit trail says a person checked it.

So the form is generated too. One engine reads a Zod schema and renders fields, validation, and defaults from it. It has two sources feeding it, which is the part I like:

REST forms get their schema from the OpenAPI generator.
A tool's arguments arrive over the WebSocket as JSON Schema, at runtime, and are converted to Zod on the way in.

Either way the form, the validation, and the defaults are read off the same object, so a form cannot offer a field the server will reject, and it cannot hide one the server will accept. A tool nobody has built yet already has a working approval UI.

Here is the result, for a refund the agent proposed and a reviewer has not yet approved:

Nobody wrote that form. "The customer being refunded", "Amount in GBP" and "Why the refund is owed, for the audit trail" are the field descriptions off the tool's schema, rendered straight through. Add a field to the tool and it appears here, described, validated, and in front of the person approving it.

That property is only available because the tool's arguments were a declared schema in the first place. It is the compound interest on a decision made two posts ago.

What it costs

A generated contract is not free, and the costs are the kind that annoy people on a Tuesday.

It is a build step, and build steps get forgotten. Change a serializer, forget to regenerate, and you get a type error that points at the generated file rather than at what you actually did. You learn to read it quickly, but the first few are confusing. A pre-commit hook or a CI check is the real answer, and the repo validates the schema separately with --fail-on-warn so an ambiguous spec fails before it reaches a generator.

Generation needs the server running. The frontend generator fetches the live schema endpoint, so just gen starts whatever service it needs first. That is convenient and it is also a dependency, and it means you cannot regenerate on a plane.

Generated code is ugly and you must not edit it. Both generated files open with a line telling you not to. Someone will anyway, once.

You inherit the generator's opinions. Field naming, optionality, and how a discriminated union maps into the target language are all decisions the tool makes for you. Mostly fine, occasionally not what you would have picked, and essentially never worth fighting.

None of those compare to the afternoon I described at the top.

When you do not need any of this

One service and one language: skip it. If your types are already shared because everything imports from the same module, a generator adds a step and buys nothing. The whole value is in crossing a process or a language boundary.

A prototype you intend to throw away: also skip it. The cost here is front-loaded and the payoff accrues over months.

A public API with external consumers you cannot recompile: generation still helps you, but it stops being a safety net, because the compiler that catches drift is yours and theirs is not. Versioning becomes the actual problem and that is a different post by a different person.

The sweet spot is exactly the shape in this series: a handful of services, one team, two languages, changing weekly. That is where a generated seam turns a class of silent production bug into a red squiggle.

Next up is "the agent and the graph together", which is where the series finally gets into the part everyone came for: a typed model call, and the flow that decides what happens next. Hope this one is useful, particularly the form argument, which I think is underrated.

I write these deep-dives and maintain a handful of Python libraries, so if that is your kind of thing, follow along on GitHub and LinkedIn.

Agent Graph Engineering, Part 2: Split the Services, and Know Where the Line Goes

Huy Nguyen — Fri, 23 Oct 2026 04:34:44 +0000

TL;DR Put your product's domain in one service and your agent graph in another, and draw the line at state ownership rather than at anything that sounds like a layer. The agent owns execution state, which is a LangGraph checkpoint nobody can rebuild by hand. The backend owns the record, which is what a page reload needs. Updates travel one way, agent to backend, over a generated WebSocket client. The payoff is not tidiness, it is that a node is a function you can call with a state you invented, while the same decision inside a socket consumer needs a connection and a user before it will run at all.

Part 1 listed the stack and said, in one line, to keep business logic out of the agent. That line does a lot of work and deserves its own post, because it is the decision that is nearly free this month and expensive in a year, and because "split it into two services" is useless advice until somebody tells you where the cut goes.

What the merged version looks like

I have shipped the single-service version. It does not fail dramatically. It degrades, in a way that is hard to argue with in a code review because every individual step was reasonable.

It goes something like this. Your agent needs to know whether a customer is on the paid plan, so the prompt grows a line about entitlements. Then billing adds an annual tier, and now the prompt is wrong in a way no test catches, because the prompt is a string and nothing validates strings against your billing model. Meanwhile a graph node needs the ticket's history, so it reads the ORM directly, and now your graph cannot run in a test without a database and fixtures. Six months later somebody asks whether the agent can be deployed separately, because the model calls are slow and the web workers are blocked waiting on them, and the answer is no, not without unpicking all of it.

The two failure directions are worth naming separately, because they feel different while you are in them. Business rules leak into prompts, where they cannot be type-checked or unit-tested. Graph concerns leak into models, where a tickets table starts carrying columns about node execution. Neither side can be tested alone by then, which is the point at which the split stops being a refactor and becomes a rewrite.

The test you cannot write

That last sentence is the one that matters, so here is what it looks like from inside.

In the merged version the agent's decision happens inside the socket consumer, in a method that has already authenticated the user, loaded the conversation and written a row. To assert that a particular message routes to a particular capability, you have to arrive at that line the way a user does: open a connection, authenticate it, seed a session and a conversation, and let the whole thing run until the part you care about executes. The assertion is one line. Everything above it is travel.

The mocks are worse than the setup. Because the model is reached through that object rather than across a boundary, there is no single place to stub it. You stub each collaborator where it happens to be used, so the fakes multiply with the call graph instead of with the number of seams. A suite like that gains patches faster than it gains assertions, and past a certain density nobody can tell you which parts are still being exercised and which are being asserted against a mock of themselves.

A graph node does not have that problem, and not because graphs are virtuous. It is because a node is a function from state to state. You build the state you want, call it, and look at what came back.

async def test_knowledge_base_branch_grounds_the_answer() -> None:
    thread = "kb-branch"
    with mock_openai(CLASSIFY, SEARCH_KB, ANSWER_FROM_KB) as route:
        state = (
            await SupportGraph()
            .compile(checkpointer())
            .ainvoke({"context": ticket()}, config=config(thread))
        )

    assert route.call_count == 3
    assert isinstance(state["decision"], SearchKnowledgeBase)
    # The retrieval step ran and fed the answer agent.
    assert state["kb_snippets"]
    assert any("kb-002" in snippet for snippet in state["kb_snippets"])

That is the whole test, not an excerpt. The three capitalized names are canned model replies declared at the top of the file. There is no socket to open and nobody to log in, because the node never needed either. The only thing faked is the model's HTTP response, which is the one real boundary in the picture, and everything underneath it runs: three model calls, a retrieval step, a routing decision, and an answer that had to cite what retrieval handed back.

So the argument for splitting is narrower and more useful than "separation of concerns". It was never about tidiness. It was about being able to call the thing.

The cut is about state, not layers

The instinct is to split by technology or by layer: an API tier, a service tier, maybe a cache in front. That produces a diagram your architect likes and a boundary that leaks within a month, because layers are not what is actually different about these two things.

What is actually different is the kind of state each one holds.

The agent owns execution state: which node runs next, the current channel values, the payload of a pending interrupt. In LangGraph that is a checkpoint, and it lives in Postgres. The important property is that you cannot rebuild it by hand. If you lose a checkpoint, you have not lost a row you could recompute from other rows. You have lost the ability to resume, because reconstructing it means reimplementing LangGraph's execution model, and you are not going to do that. Resumption is the whole mechanism behind human approval, so this state is load-bearing in a way ordinary application state is not.

The backend owns the record: messages, ticket events, and the two things a page reload needs to find, which are a drafted reply waiting for approval and a proposed tool call. None of those are copies of graph state. They exist because a browser cannot ask the agent what is currently waiting, and because a stored approval row is what lets the approval card come back after a refresh.

Once the line is drawn there, a lot of smaller questions answer themselves. Permissions, billing, and user records are obviously the backend's, because they are the record. Prompts, tools, and routing are obviously the agent's, because they are execution. The service the agent exposes has one job, and the service your product lives in never imports LangGraph.

In the reference implementation it comes out as three services:

Service	Owns	Port
`backend/`	users, tickets, the thread as a person reads it	8000
`agent/`	graphs, tools, checkpoints, the model's memory	8001
`web/`	nothing; it renders what the other two generate	5173

Three services and four processes, because a run is driven by a worker rather than by whichever web process accepted the request. That worker is the backend's own code pulling from a queue, so it owns nothing the table does not already list.

The agent is internal. Only the backend connects to it, never a browser, and that matters more than it looks. The approval gate lives inside the graph, so anything that can open a socket on the agent can both propose a tool call and approve it. Network isolation is the primary control there and a shared token is the second one, which is why the token is optional in a fresh clone and not optional anywhere real.

Two threads, not one

I got this wrong the first time, and the symptom was confusing enough to be worth the detail.

There are two conversation histories in a system like this, and they are not the same object.

One is the thread as a person reads it: an ordered list of events you can render, paginate, and show to a customer. That is rows, and it belongs to the backend.

The other is the thread as the model remembers it, which in Pydantic AI is its own message history. That is not a list of rows. It carries tool calls and their results, the shapes of previous structured outputs, and the sequencing the model needs to behave consistently on the next turn. Flattening it into "a list of messages" throws away exactly the parts that make the next call work.

I first gave that model memory its own table, filled by a background job once the message itself had been saved. Two copies of one thread, written by two processes at two different times, which is the shape of a problem whether or not you have hit it yet. The gap between those writes is not a rare crash window either, which is what I had told myself. It is ordinary operation: the job queues, it retries, occasionally it does not arrive, and the write that already succeeded neither knows nor cares. When the two disagreed, neither had been derived from the other, so neither could be rebuilt. There was no correct recovery, only a choice about which copy to believe.

The fix was to stop treating it as a second source of truth. The model's memory now rides on the graph's state, so the checkpointer that already persists a parked run persists the memory too. One write, one thing to lose. And when a lane has nothing remembered yet, it starts from the record the backend sends, which quietly turns the memory into a cache over the record rather than a peer of it. Lose a checkpoint now and what you lose is the shape of past tool calls, not the conversation.

That is the test I would apply to any state you are tempted to put on the agent side. If losing it means losing data, it belongs in the record. If losing it only means losing speed or fidelity, the agent can keep it.

Updates travel one way

Neither service mirrors the other, and only one direction carries updates.

A graph node broadcasts an event on its topic. The backend is subscribed, and its relay does two things with each event: writes down whatever belongs in the record, and fans the event out to the browser group. So the backend learns what happened by being told. It never reads the agent's tables, and the agent never writes to the backend's.

The useful consequence is that the record is derived, which means it can lag but it cannot diverge. A terminal event persists the turn. A parked run persists the approval card. A tool proposal is deliberately not persisted as a turn, because it only becomes one if it actually runs.

The place this shows up as real code is a small protocol on the backend describing everything the agent can announce:

class Sink:
    """Where a run's events are written down. Every hook is optional."""

    async def classified(self, category: str, priority: str, why: str) -> None: ...
    async def decided(self, decision: str, why: str) -> None: ...
    async def drafted(self, content: str, model: str) -> None: ...
    async def approval_required(self, draft: str, findings: list[str]) -> None: ...
    async def reply_sent(self, content: str) -> None: ...
    async def reply_blocked(self, findings: list[str]) -> None: ...
    async def token(self, delta: str) -> None: ...
    async def answered(self, content: str, model: str) -> None: ...
    async def tool_proposed(self, payload: Any) -> None: ...
    async def tool_ran(self, payload: Any) -> None: ...
    async def reasoning_delta(self, step: str, delta: str) -> None: ...
    async def reasoned(
        self, step: str, content: str, decision: str, model: str
    ) -> None: ...
    async def failed(self, message: str) -> None: ...

That interface is the boundary, written down. Three different callers open connections to the agent in this app: a ticket's customer-facing run, the same ticket's internal team run, and a standalone conversation. They turned out to differ only in what they wrote down, never in how they talked to the agent, so the talking lives in one place and the writing-down is whichever Sink you hand it. Every hook being optional is the bit that makes that work: a caller that does not care about token deltas simply does not implement token.

A boundary you can read in twenty lines is also a boundary you can test. Pass a recording Sink, drive the relay, and assert on what it would have written, with no database and no model.

The boundary is generated, not hand-written

Two services talking means a contract, and a hand-maintained contract between two codebases is a slow leak. I have watched a WebSocket message shape drift between a backend and a frontend that each believed they were right, and there is no satisfying way to debug it, because both sides are internally consistent.

So nothing here is hand-written. The agent's consumers produce an AsyncAPI document, and a generated Python client is what the backend imports:

from helpdesk.agent_client.agent_hub_support_topic.client import (
    AgentHubSupportTopicClient,
)
from helpdesk.agent_client.agent_hub_support_topic.messages import (
    ApprovalRequiredMessage,
    ReplySentMessage,
    RunRequestMessage,
    ToolApprovalMessage,
)

Change a message on the agent, regenerate, and the backend stops type-checking. That is the entire mechanism, and it is the same trick your REST API has had through OpenAPI for a decade. The next post covers how the generation works and what drift actually costs. For now the only claim is that the seam between two services is the last place you want a shape that exists only in two people's heads.

What it costs

Two services is more than one service, and I would rather say the costs out loud than discover them in your comments.

You now run two services locally, each with its own environment file and a shared token that has to match in both, plus a worker and the web app. The repo hides that behind a single setup command, but somebody wrote that command.

You have a network hop on the critical path. For this workload it disappears into model latency, which is measured in seconds, so a few milliseconds of local socket is noise. If your agent were doing something fast and chatty, the calculus would change.

You have a contract to regenerate. It is one command, but it is a step, and forgetting it produces a confusing failure until you learn the shape of it.

And you have two deploys, two sets of logs, and two things that can be down independently. Tracing across the boundary stops being free, which is a large part of why observability gets a post of its own rather than a paragraph.

When not to do this

I do not think this is unconditional. If you are building a prototype to find out whether an idea works, one service is correct, and splitting early is the kind of discipline that feels responsible while slowing you down for no return. The same goes for a tool with one user, which is you.

The signals that it is time: the model calls are slow enough that they are blocking your web workers; you want to scale the agent independently, or on different hardware; more than one person is working on the system and the prompts and the business rules are colliding in review; or you have started writing graph concepts into your application tables.

The reason to do it before you are forced to is simply that the cost curve is steep. Drawing the line while there is nothing across it is an afternoon. Drawing it after a year of leaks is the rewrite that gets postponed every quarter, and that one never gets cheaper.

Next up is types and contracts, which is the other idea the rest of the series leans on and the thing that makes this boundary enforceable rather than aspirational. Hope this one is useful if you are staring at a single service and wondering whether to cut it.

I write these deep-dives and maintain a handful of Python libraries, so if that is your kind of thing, follow along on GitHub and LinkedIn.

Agent Graph Engineering, Part 1: The Stack, and Why Each Piece Is There

Huy Nguyen — Fri, 23 Oct 2026 04:26:37 +0000

TL;DR The stack is Python, Pydantic AI for agents, LangGraph for the flow, chanx for typed streaming WebSockets and a generated frontend contract, OpenTelemetry into a trace store you own, pytest with a mocked LLM for fast tests, and real-model evals for regressions. Business logic stays in a separate service from the agent graph. The single idea holding it together: every boundary in the system is a declared, typed artifact rather than something you hold in your head. This post makes the case for each piece so you can bring it to your team, then lays out what we build in the rest of the series.

Part 0 was the story: years of shipping LLM features as intent detection plus nested conditionals, and why observability tooling never fixed the underlying problem. This post is the pragmatic half.

One clarification first, because the word is overloaded. This series is about the execution graph of an agent: state, nodes, edges, routing, interrupts, resumption. It is not graph RAG, and not knowledge graphs. Those are retrieval techniques, where a graph is the data you query. Graph RAG could sit behind one node here as one tool among several. Same word, different concept, and that collision is a large part of why there is so little material on this subject.

The one idea

Everything below follows from a single principle: the structure of your AI system should be a declared artifact, not something reconstructed by reading code.

Your flow should be a graph you can render. Your tool inputs and outputs should be schemas. Your WebSocket messages should be a contract your frontend can generate a client from. Your agent's output should have a type.

The reason is not elegance. It is that declared structure can be rendered, validated, traced, and diffed automatically, while structure that only exists as control flow can only be understood by a human reading it. That difference is what decides whether your system survives its third feature and its second engineer.

Why this is worth your time

The case for each piece is below. Before that, the reasons you might care beyond whatever you are shipping this week, because I think there are three different readers here.

If your team already runs an agent in production, you do not need convincing that it is hard. The useful part is the set of practices that remove whole categories of bug rather than fixing them one at a time. Typed tool arguments turn a wrong type into a validation error at the edge instead of a TypeError three frames into your own code. Structured output means you route on isinstance rather than on whether the response happened to contain the word "escalate". Generated contracts mean a message shape cannot drift between two codebases without the compiler noticing. Checkpointing means a crashed run resumes instead of vanishing. Guardrails at both edges mean the text you screened is the text that ships. Evals mean a model upgrade is a measurement rather than a gamble. None of those are features you gain. They are bugs you stop having, which is a different and better thing.

If you are about to build your first one, the useful part is the order. These decisions are sequenced the way they actually bite, and the one that is cheap this month and expensive next year is the service split, which is why it turns up early rather than in a footnote. Reading ahead costs an afternoon and might save you a rewrite.

If you are building for yourself or for a client, the repository is the point. It is distilled from a production system of five services, out of roughly five years on realtime infrastructure and agent systems, so the parts that look over-careful are mostly the parts that broke somewhere else first. It is production-shaped rather than notebook-shaped, and it runs on one API key, or none at all. Moving it to another domain is mostly swapping tools, prompts, and edges. I want to be careful with the word production, though. The shape is right and the last mile is real, so the final post is a plain list of what you still owe before paying users touch it. Nobody is helped by pretending a checkout is a deployment.

Then there is the market, if you are weighing where to put your learning time. Stanford's AI Index 2026 found that mentions of the "agentic AI" skill cluster in job postings grew more than 280% in a year, from 0.06% of postings in 2024 to 0.23% in 2025, which is roughly 90,000 US listings. Fast growth from a small base, and both halves of that matter: it is early, and it is moving. The detail I find more telling sits in the same report. Mentions of "ChatGPT", "conversational AI" and "chatbots" went down over the same period. Demand moved off demos. Korn Ferry's 2026 survey of 1,674 talent leaders has 52% planning to deploy autonomous agents by the end of this year, which is the same shift seen from the other side of the table.

The job ads are the more useful signal, though, and they say something slightly different from the headline. Reading through current senior backend openings, the ones that touch AI are not asking for prompt engineering. They ask for distributed systems, availability and latency targets, production Python, schema and API design, architecture review across teams, and operational practice: service levels, incident response, containers and orchestration. AI turns up as a capability expected inside all that, rather than instead of it. That is a more demanding ask than the hype implies, and it maps almost exactly onto what this series builds, which is a system with contracts, traces, tests and a deployment story that happens to have a model in it.

The sentence that keeps turning up in hiring write-ups is that a demo agent is easy and a reliable one is not, and that eval literacy is the clearest sign somebody has actually shipped. That is more or less this entire series, arrived at independently and the slow way.

The stack

Layer	Choice	What it buys you
Language	Python	The only ecosystem where every provider, every serving runtime, and every eval tool shows up first
Agent	Pydantic AI	Typed tools, typed outputs, typed deps, provider independence
Flow	LangGraph	Declared graph, renderable diagram, inspectable state, checkpointing
Transport	chanx	Typed WebSocket messages, broadcast from anywhere, generated AsyncAPI contract
Tracing	OpenTelemetry	One connected trace per user message, readable with no account and exportable to any OTLP backend
Tests	pytest + mocked LLM	Fast, deterministic, no API spend
Regressions	Evals with real models	Catch what mocks cannot when you change a model or a prompt
Business logic	A separate service	Keeps your product's domain out of your agent's graph
Runs	A worker, on Celery	A run outlives the process that asked for it, so a deploy does not take the work in flight with it

Python

Not a controversial pick, but worth stating the actual reason. It is not that Python has the best SDK for calling an API, because every language has that now. It is everything around the call: local model serving, embeddings, evaluation harnesses, vector stores, document parsing. When you eventually need to run a model yourself or plug in a tool nobody has wrapped yet, Python is where it already exists.

Pydantic AI, for the agent

An agent is a thing with a declared output type and a declared set of tools. Pydantic AI is built on exactly that, and being Pydantic-native means your tool schemas and output schemas are the same objects you already validate with.

What that gets you in practice:

Structured output. The agent returns a typed object, not a string you regex. Routing on isinstance beats routing on substring matching.
Validated tool calls. The model's arguments are checked before your code runs. A wrong type is a caught validation error, not a TypeError three frames deep.
Provider independence. Switching providers is a config change. This matters more than it sounds like it does, because you will want a cheap model for routing decisions and a strong one for synthesis.
Typed dependencies. Your tools receive a typed context object instead of reading globals.

A word on determinism, because it gets oversold. You cannot make the model deterministic, and anyone promising that is selling something. What you can do is make everything around it deterministic: the model's only job is to return a value of a declared type, and every other step on the path is ordinary code you can test. That shift is most of what separates a system from a prompt.

You could use a provider's own SDK or an ADK instead. I would not, because the moment you want a second provider or a self-hosted model, the abstraction you skipped has to be written anyway, and it will be worse.

If you want to see this layer on its own before the graph arrives, I wrote it up separately in Streaming Agents with Pydantic AI over Typed WebSockets. That post is a good prerequisite for this series.

LangGraph, for the flow

This is the piece that changes how you work, and the piece your team will push back on. So be specific about the trade.

The cost is real. You stop writing handler functions and start declaring state, nodes, and conditional edges. A branch becomes an edge. A handler becomes a subgraph. It takes a few weeks to stop fighting it, and the first graph you write will be wrong.

What you get back:

A diagram from the code. LangGraph exports Mermaid. Your architecture picture is generated from the thing that actually runs, so it cannot go stale. No more hand-drawn draw.io files that lie.
Inspectable state at every step. You can see what the state was going into a node and coming out. When something breaks, you know which transition did it rather than guessing.
Checkpointing. Graph state persists. A run can be paused, resumed, and recovered after a crash, which is also the mechanism that makes human-in-the-loop approvals work rather than being a bolt-on.
Structure that survives change. Adding a capability means adding a node and an edge, not threading another condition through an existing chain.

The honest downsides: it is a LangChain-ecosystem library, so some of the ergonomics assume LangChain; and streaming from non-LangChain code goes through a generic custom channel, which is the gap chanx fills.

chanx, for transport and contract

Disclosure: I wrote chanx, because I kept rebuilding the same thing. The intro post has the full background.

Three things it does for an agent system:

Broadcast from anywhere. A node buried in a subgraph can emit a progress event to the right user without the graph passing a stream handle down through every call. You can stream the model's reasoning while it works, instead of going silent and then dumping everything.
Typed messages instead of a receive_json if/else chain. Every message is a model, routing is automatic, and bad payloads fail at the edge.
Generated AsyncAPI docs. This is the one nobody else does. Your WebSocket contract becomes a spec, and your frontend generates a TypeScript client from it. Change the graph, regenerate, and the compiler tells the frontend what broke. Your REST API has had this via OpenAPI for a decade; there is no reason your agent layer should not.

It works with Django, FastAPI, or Litestar, which matters because we use two of them.

OpenTelemetry, into a trace store you own

The usual advice here is to name a vendor. I would rather name the standard, because the vendor is the part you should be able to change.

Both LangGraph and Pydantic AI instrument themselves with OpenTelemetry, so once a graph node opens a span, every model call inside it is already a child. You get one connected trace covering graph transitions and model calls together, rather than the flat list of completions a dashboard usually shows you.

What this stack does with those spans is the part I would argue for. They go into a small trace store inside the agent service, which means /traces renders a run as the chain of steps it was with no account, no signup, and no configuration. That matters more than it sounds. A reference implementation that needs a vendor account before it shows you anything has lost most of its readers at step one, and an on-call engineer at 2am should not need a third party to answer "what did this run do".

Exporting onward is then one environment variable: off, files on disk, an OTLP collector, or both. Any OTLP-speaking backend should work, since that is the point of a standard, and the repo covers that path with a fake collector that checks spans really leave the process with the right endpoint and auth header. Honest limit: that proves the export path, not any particular vendor's ingestion. Langfuse, Jaeger and Grafana are all OTLP endpoints plus a header, but I have verified the path rather than each product.

When a user reports that the agent did something strange in production, you want evidence, not a reproduction attempt. This is the piece that gives you evidence, and it is yours.

pytest with a mocked LLM

Mock at the HTTP layer, not at the framework layer. If you stub Pydantic AI's client you test your stubs. If you intercept the HTTP call and return a recorded provider response, the real pipeline runs: SSE parsing, tool call assembly, streaming, validation. Your tests then cover the parts that actually break.

These tests are fast, deterministic, and free, so you can run them on every commit and cover full WebSocket flows end to end.

Evals with real models

Mocked tests prove your code works. They cannot tell you whether the model still picks the right tool after you reworded a prompt, or whether the new model release routes the same way.

That is what evals are for: scenarios run against real providers, on demand and never in CI, because they cost money. Pydantic AI has an eval module, or a plain script works. Without this you cannot upgrade a model without fear, which in practice means you stop upgrading.

Keep business logic out of the agent

Put your product's domain (users, permissions, billing, persistence) in one service and your agent graph in another. I have shipped the merged version and it becomes a mess: business rules leak into prompts, graph concerns leak into models, and neither side can be tested alone.

Separated, the agent service has one job, and the boundary between them is a contract you can generate a client from.

What we build

A support ticket triage assistant. A ticket arrives, and the agent decides what to do with it: answer it directly, search the knowledge base, escalate it to a human, or draft a reply. Posting that reply to the customer requires human approval; searching the knowledge base does not.

Two views of one ticket. The customer sees it being worked, step by step:

The team sees the same ticket with their own lane on it, internal notes beside the agent's answers and the step it chose:

I picked this deliberately. It needs real routing, so the graph earns its place instead of being a two-node demo. It has a naturally irreversible action, so the approval machinery is solving an actual problem rather than an invented one. And it runs on an LLM key alone, with no OAuth flows or third-party signups standing between you and a working checkout.

The shape:

flowchart LR
    FE["Frontend<br/>React"]
    BE["Backend<br/>Django"]
    AG["Agent<br/>FastAPI"]

    FE -->|"REST + WS<br/>OpenAPI"| BE
    BE -->|"typed WS<br/>AsyncAPI"| AG

    BE --- BEOWNS[["users, tickets,<br/>the thread a person reads"]]
    AG --- AGOWNS[["graphs, tools, checkpoints,<br/>the thread a model remembers"]]

    classDef owns fill:none,stroke-dasharray:4 3;
    class BEOWNS,AGOWNS owns;

Three services, one repository. The frontend generates its clients from OpenAPI and AsyncAPI. The backend owns users, tickets, and history. The agent service owns the graphs, the tools, and the checkpoints, and talks to nobody's database but its own.

It is small enough to read in an afternoon and complete enough to adopt: generated contracts on both boundaries, real WebSocket worker processes, approval interrupts on irreversible tools, guardrails on the input and output edges, streamed progress, a span per graph node with per-run cost, three test suites that spend nothing, evals against real providers, and container images serving ASGI. The architecture is production-grade and the deployment is production-shaped. What it does not do yet, context budgeting and spend caps among others, is written down in the repo rather than glossed over, and the last post walks that gap. Every post is pinned to a tag so you can check out the exact state being described.

What it costs

I do not want to undersell the ramp. Declaring flows as graphs is a genuine shift in how you think, and the first few weeks are slower than writing the conditionals would have been. Expect to throw away your first graph design once you understand state properly.

What you get on the other side is an AI system you can point at, trace, test, and change. The agent stops being magic and starts being software.

Next up is the service split, then types and contracts, and then the posts get into code. Hope this one is useful in the meantime, particularly if you are the person who has to justify the stack to everyone else.

I write these deep-dives and maintain a handful of Python libraries, so if that is your kind of thing, follow along on GitHub and LinkedIn.

Agent Graph Engineering, Part 0: I Needed This Before It Had a Name

Huy Nguyen — Sat, 10 Oct 2026 02:01:12 +0000

TL;DR Building a chatbot is easy. Building an AI system you can trace, test, and change without fear is not. I spent years shipping LLM features as intent detection plus nested if/else, and no amount of observability tooling fixed it, because the problem was that the flow only existed in code, in my head, and in a stale diagram. The fix was to stop writing the flow as control flow and start declaring it as a graph. This post is the story of how I got there, years before the practice had a name, and why I think graphs are a durable answer rather than a trend. If you only want the stack and the arguments to bring to your team, skip to Part 1.

Building an AI assistant today is easy. A few lines of code, an API key, and you have something that chats. Add a couple of tool definitions and you have an agent. Done, right?

Yes, for a hobby project. Production is a different animal, and the gap is wider than most people expect.

The usual first answer is observability. Drop in Langfuse, Langsmith, or Logfire and you can finally see what the model said. That helps. It also does not touch the questions that actually keep an AI system from scaling:

How do you split work across dedicated agents, so that document handling and web research are not the same tangle of prompt instructions?
How do you pause a tool call for human approval before something irreversible like sending an email, while letting safe calls like reading a public profile run untouched?
How does anyone see the full picture of your agent flow? Intent routing lives in conditionals. There is no artifact to point at in a design review.
How do you trace one user message from arrival to final answer? Observability platforms give you a list of LLM spans. Reconstructing the flow means reading code backwards or writing a script to reshape the traces into something legible.
How does your frontend know which messages the backend can send it? When the flow changes, what stops a schema from silently going stale on one side?
How do you stream intermediate progress in a predictable way, instead of going quiet for forty seconds and then dumping everything at once?
How do you test any of this? Mocked responses for fast feedback, real model runs to catch regressions when you swap models or edit a prompt.
When the agent calls the wrong tool, or two tools, or none, what do you actually look at?

If none of those have bitten you yet, you probably have a chatbot rather than a system. I know because that was me.

One thing before the story, because I would want to know. None of what follows is a composite or a hypothetical. It was a real product in the financial domain, with real users, a real team, and mistakes that were mine to make. There is a lot of generated agent writing around at the moment and I understand the reflex to skim a post hunting for the tell. Skim away. The details here are specific because they happened, and that is not something you can fake at length.

The code that goes with this series is the same story. It is distilled from a production system of five services, out of about five years spent on realtime infrastructure and agent systems, three of those on one agent product and two on another. What carries over is not the domain, which is deliberately gone, but the failure modes. The parts that look over-careful are the parts that broke somewhere else first.

Where I was

Years ago, right around when ChatGPT landed, my team and I were building an AI product in the financial domain. I was not starting from zero: I had trained GPT-2 models, built realtime systems with Django Channels, shipped chatbots before.

The product worked. Users used it. But it was built the only way anyone knew how at the time. The term "agent" did not exist yet. We had assistants with function calls, which later became tool calls, plus prompting techniques like chain of thought and tree of thought. Our domain required us to classify what the user wanted and route to the right handler, so the core of the system was an intent classifier feeding a tree of conditionals.

We were not careless about it. We refactored, we applied every best practice we knew, and the code was reasonable by normal backend standards. But something felt wrong, and it took me a while to name it.

Here is what was actually missing:

No system view. Nothing showed the shape of the flow. Understanding it meant reading from entry point to exit, holding every branch in your head, and drawing it by hand in draw.io. Then someone shipped a change and the diagram was a lie.

No end-to-end trace. We could see individual model calls. We could not follow one user message through routing, tool calls, intermediate results, and the final answer as a single connected thing.

No generated contract. Our REST API had OpenAPI. Change a serializer, regenerate, the frontend knows. Our AI layer had nothing equivalent. The WebSocket message shapes lived in two hand-maintained places and drifted.

Flaky streaming. It worked, mostly, until it lagged or dropped in ways nobody could reproduce.

Onboarding a teammate onto that system took weeks. Debugging a production report meant a code-reading expedition. That is not a scaling problem you fix with a better logger.

The state machine itch

At some point the shape of it clicked. Each step has a state. The model processes that state. Based on the result you transition to another state. That is a state machine, the thing I learned in university and had not thought about in years.

Once you see it that way, the conditionals look like an implementation detail of something that wants to be declared instead of written. And a declared state machine can be drawn, inspected, and traced automatically, because the structure exists as data rather than as control flow.

The framework landscape then was basically LangChain, and early LangChain was rough. If you were there, you know. Debugging through those abstractions when something went wrong was worse than debugging our own code.

So I wanted a graph library for agents: something that routes on state, that can render itself, that lets a teammate understand the system without reading a thousand lines. I started building a small package for it and gave up, because doing it properly is a lot of work and I had a product to ship.

Finding the pieces

Time passed, the field moved, and the libraries got much better. CrewAI appeared. Pydantic AI appeared. Even LangChain grew into something genuinely useful, quite different from the version that burned me.

Pydantic AI won me over immediately. Tool calls validated by Pydantic models, typed outputs, typed dependencies, provider independence, and an API that does not fight you. It treats an agent as a thing with a declared output type and a declared set of tools, which is exactly right.

I still wanted the graph half. Then I ran into LangGraph, I think from a Reddit thread. It was the thing I had been sketching: state, nodes, conditional edges, and a flow you can export as a diagram. I went through Andrew Ng's course on it and the concepts landed fast, because I had already arrived at the same model from the other direction.

One thing bothered me. LangGraph is a LangChain-ecosystem library, and I wanted Pydantic AI for the agent layer. So I went looking for people who had combined them, and found plenty. The combination came up again and again, with strong opinions in its favor, and some teams already running it in production.

I tried it on a side project and it went better than expected. Two rough edges remained. Interrupts between Pydantic AI and LangGraph needed care. And streaming was awkward, because LangGraph only offers a generic custom-stream channel for anything outside LangChain.

The streaming problem I had already solved elsewhere. chanx is a WebSocket library I built for typed messages and generated AsyncAPI docs, and it can broadcast from anywhere in your process without threading a stream object through the call graph. Dropping it in meant a graph node deep in a subgraph could emit a progress event to the user without the graph flow knowing or caring.

LangGraph, Pydantic AI, and chanx. That is the combination I had wanted for years, and it is still the one I reach for.

Why this outlives the frameworks

I have watched a lot of AI trends arrive and leave. Most did not feel solid. Graphs do, and I think that is because the idea is older than the hype around it. It is the same structure that shows up in build systems, dataflow engines, and state machines, applied to a new execution substrate.

The signals are lining up too. Google's ADK material now teaches graph engineering directly. The discussion in my corner of the internet has shifted from prompt tricks toward flow structure.

The vocabulary has caught up as well. Graph engineering has guides now, a growing pile of papers, and a definition that has more or less settled: designing the graph an agentic system executes in, the nodes that do the work, the edges that route between them, and the state that travels along them. There is a distinction in that writing I wish I had had years ago. Loop engineering designs the cycle a single agent repeats until it finishes. Graph engineering designs the coordination between those cycles. Most of my pain was from trying to solve the second problem with the tools for the first.

What is still missing is production material. Nearly everything written is a tutorial: here is a state graph, here are nodes and edges, here is a two-node example that answers a question. Very little covers what happens when the graph has to live inside a real system, with a typed contract to a frontend, a person approving an action that cannot be taken back, traces you can hand someone after an incident, evals that let you change a model without crossing your fingers, and a deployment at the end of it. That gap is what this series is for.

One more thing worth clearing up, because it is the reason this is hard to search for. Put "graph" and "AI" in a search box and you mostly get knowledge graphs and graph RAG, which are retrieval techniques where a graph is the data you query. This is the execution graph of the agent itself. Graph RAG could sit behind one node here as one tool among several. Same word, unrelated concept.

What this series is

A dozen posts, working from the stack and the reasoning down to a running system. We will build a reference implementation that is deliberately production-shaped rather than notebook-shaped: a React frontend, a Django backend that owns business data, and a separate FastAPI agent service running the graphs, talking over a typed WebSocket contract, with human-in-the-loop approvals, streaming progress, end-to-end tracing, mocked tests, and real-model evals.

Part 1 covers the stack and the case for each piece, which is the part you can forward to your tech lead. Everything after that is code.

The frameworks in this series will age. Some of them will be replaced, and you may end up building your own. The structural idea underneath, that your agent flow should be declared as a graph rather than written as control flow, is the part I expect to outlive the tooling.

Hope the story is useful, especially if you recognize yourself in the first half of it.

I write these deep-dives and maintain a handful of Python libraries, so if that is your kind of thing, follow along on GitHub and LinkedIn.

Benchmarking Python API Frameworks with Real Workloads: FastAPI vs Litestar vs DRF vs Django Ninja vs Django Bolt

Huy Nguyen — Tue, 11 Aug 2026 13:42:01 +0000

Most Python framework benchmarks measure "hello world" JSON. Real APIs do authentication,
database access and complex queries - so that's what I benchmarked, across FastAPI,
Litestar, Django REST Framework, Django Ninja and Django Bolt, each with multiple
production servers, under strict Docker resource limits.

Everything is reproducible: https://github.com/huynguyengl99/python-api-frameworks-benchmark

The core design decisions

Latest versions of everything - Django Bolt 0.10, Django 6.0, FastAPI 0.141, Litestar 2.24, DRF 3.18, Django Ninja 1.6.2.
Each framework uses its own native serializer. Feeding Litestar Pydantic models makes it convert twice (Pydantic then msgspec), so you'd be benchmarking Pydantic instead of Litestar - a point the Litestar author raised on an earlier version of this benchmark. Litestar and Bolt use msgspec structs, FastAPI and Ninja use Pydantic, DRF uses its serializers. The response models are defined twice, field for field, and the output is byte-identical: 4,796 bytes for /articles/1 from both FastAPI and Litestar.
JWT in an httpOnly cookie for the auth endpoints - the setup I'd actually use in production. Each framework uses its own ecosystem's library instead of something hand-rolled: drf-auth-kit (DRF), django-ninja-jwt (Ninja), AuthX (FastAPI), and the built-in cookie JWT support in Litestar and Bolt.
Median across container restarts, not repeated runs. Some servers (notably Bolt's Rust worker pool under a 1-CPU cgroup) pick a throughput regime at container startup and hold it, so best-of-N inside one container flatters the lucky starts. Details in the repo's Methodology section.

Setup

MacBook M2 Pro, 32GB RAM; PostgreSQL 16; 500 articles, 2000 comments, 100 tags, 50 authors
Each framework alone in a Docker container: --cpus=1, --memory=750m, one at a time
bombardier, 100 connections, 10s per measurement
5 independent container starts per framework, median reported with min-max spread

Endpoints:

Endpoint	What it measures
`/json-1k`, `/json-10k`	Pure serialization
`/db`	10 simple rows
`/articles?page=1&page_size=20`	Pagination + nested author/tags
`/articles/1`	Nested author + tags + comments
`/auth/me`	Cookie JWT verify + load user
`/auth/articles`	Cookie JWT + the paginated query

One fairness detail that matters: all five frameworks load the user row from the DB on
both auth endpoints. Litestar, DRF and Ninja do it as part of authenticating; AuthX and
Bolt's guards only verify the signature, so I made FastAPI and Bolt load the user
explicitly. Otherwise they'd do strictly less work and the comparison would be junk.

Results

RPS, median over 5 container starts, zero errors across all 70 measurements:

Config	json-1k	json-10k	/db	/articles	/articles/1	/auth/me	/auth/articles
bolt	38,576	19,089	1,986	208	432	3,024	196
litestar-uvicorn	31,284	24,547	1,039	246	443	976	193
litestar-granian	19,006	15,166	1,180	250	488	1,104	210
fastapi-uvicorn	13,845	2,641	984	224	428	820	193
fastapi-granian	8,484	2,280	952	201	410	749	199
drf-gunicorn	3,925	3,132	282	140	193	261	133
drf-granian	2,703	2,200	830	198	321	726	179
ninja-granian	1,566	1,422	680	130	295	610	117
ninja-uvicorn	1,533	1,424	699	126	236	584	114
drf-uvicorn	1,035	973	495	153	234	447	137

The gap collapses once you hit the database

37x between fastest and slowest on /json-1k. 1.9x on /articles/1. If your endpoint
touches PostgreSQL, your framework is not the bottleneck - your queries are.

What authentication actually costs

Comparing /auth/articles against the identical unauthenticated /articles:

Config	public	authed	cost
fastapi-granian	201	199	-1%
drf-gunicorn	140	133	-5%
bolt	208	196	-6%
drf-granian	198	179	-10%
ninja-uvicorn	126	114	-10%
fastapi-uvicorn	224	193	-14%
litestar-granian	250	210	-16%
litestar-uvicorn	246	193	-22%

Roughly 5-20%, cheaper than I expected for "verify a token and load a user". Two things
stand out:

Bolt's /auth/me is 2.7x ahead of everything (3,024 vs 1,104) because it validates the JWT in Rust before Python runs at all.
Litestar pays the most for a concrete reason: its JWTCookieAuth runs in middleware, before dependency injection, so retrieve_user_handler opens its own DB session - that request pays for two connection acquisitions instead of one.

Memory and CPU

Most configs peak at 195-260MB RAM
drf-granian is the outlier at 456MB - untuned --blocking-threads, per the Granian maintainer (see "unaddressed feedback" below)
Nearly everything saturates ~85% of the 1-CPU budget under load - except Bolt at 67%, while leading most endpoints

A word on Django Bolt

If you're open to a young framework, this is the one I'd watch. It won or tied the top
spot on 4 of 7 endpoints while leaving ~18% more CPU headroom than every other config -
room to grow under load, not just a good number on a chart. And unlike moving to Litestar
or FastAPI, you keep the Django ORM, admin and package ecosystem.

Honest trade-offs: it's young and moving fast, the Rust internals (Actix Web + PyO3 +
msgspec) mean you can't monkey-patch your way out of a corner, and under a hard 1-CPU cap
its throughput varies between container starts under a hard 1-CPU cap. For a side project or a new
internal service, I'd have no problem reaching for it today.

Feedback I have NOT addressed yet

Being upfront, because it affects one result. The Granian maintainer (u/gi0baro) explained
that drf-granian's memory number comes from me not setting --blocking-threads or
backpressure, so it spawns a lot of threads and burns time on GIL contention - drf-granian's
456MB is that same unfixed issue. He also noted that --cpus=1 in Docker is a time-slice
scheduler limit rather than a real core pin, and that it penalizes Granian more than other
servers since Granian runs I/O in a separate runtime with extra threads.

I'm keeping the CPU cap because it makes runs reproducible and comparable, but he's right
on both counts. Tuning --blocking-threads is top of my list for the next iteration, along with cold
start time and image size (both suggested by the Litestar author).

Takeaways

Framework choice barely matters once you touch I/O. Optimize your queries - the N+1 you didn't notice costs more than any framework migration will ever save.
For raw serialization, Bolt and Litestar lead, then FastAPI, then a big gap.
Cookie JWT auth costs 5-20% on a DB-heavy endpoint; where the token is validated (Rust vs Python, middleware vs dependency) decides where in that range you land.
uvicorn vs granian is workload-dependent: uvicorn wins CPU-bound JSON on ASGI frameworks, granian wins DB-bound endpoints, and granian clearly beats uvicorn for WSGI DRF.
Measure across container restarts and publish the spread - it changed these results more than any framework upgrade did.

Everything - code, Docker setup, runner, raw per-framework JSON results, and dated
snapshots - is in the repo:

https://github.com/huynguyengl99/python-api-frameworks-benchmark

Issues and PRs are very welcome, especially if you know these servers better than I do.
And if you find it useful, a star would be appreciated 😄

Streaming Agents with Pydantic AI over Typed WebSockets

Huy Nguyen — Sat, 08 Aug 2026 14:34:43 +0000

I've been building realtime products for years: group chat, voice pipelines, streaming AI assistants. Two kinds of pain kept following me around.

The first was WebSockets. Every project grew the same untyped receive_json handler with an if/else chain, hand-written validation, and no contract the frontend could rely on. Messages were stringly-typed, bugs surfaced at runtime, and onboarding a teammate meant "read the consumer, end to end". That pain is why I built chanx: typed messages, automatic routing, AsyncAPI docs (the full story is in my intro post).

The second was agent frameworks. I shipped my first LLM features on LangChain and spent more time debugging its abstractions than my own code. Pydantic AI won me over the first time I tried it: typed tools, typed outputs, typed dependency injection. It applies to agents the same philosophy chanx applies to WebSockets, and since both are built on Pydantic they snap together with almost no glue.

There's just one gap. At the UI layer, pydantic-ai's official story is SSE-shaped: adapters for AG-UI and the Vercel AI protocol, and a built-in Agent.to_web() chat page that the docs explicitly label as a dev tool. Real agent products want more than a one-way stream. A conversation is bidirectional: the user sends prompts, the agent streams back text and tool activity, and (the interesting part) the agent sometimes has to stop and ask permission before running a destructive tool. That round trip (pause, ask, resume) fits a WebSocket perfectly, and there is currently no WebSocket example anywhere in the pydantic-ai docs.

So I built one, the same way I build REST APIs: contract-first. The server defines every message as a Pydantic model, chanx turns those models into routed handlers plus an AsyncAPI 3.0 schema, and the React client generates its TypeScript types from that schema. Backend and frontend can't drift, teammates integrate from the contract instead of reading your Python, and the whole thing (streaming, tool calls, approvals) is testable without ever calling an LLM.

Companion repo with everything runnable: huynguyengl99/pydantic-ai-ws-agent. Snippets below are trimmed for reading and current as of pydantic-ai 2.9 and chanx 2.8; the repo is the source of truth.

The big picture

The demo agent ("Tasklet") manages a task list in SQLite. Safe tools run freely; destructive ones pause for approval.

pydantic-ai Agent ──run_stream events──▶ AgentHarness (background task)
      │                                        │ broadcast_event
  typed tools                                  ▼
  requires_approval=True          group: conversation.{id}   ◀── HTTP POST /notify
                                               │ channel layer (in-memory / Redis)
                              ┌────────────────┼────────────────┐
                              ▼                ▼                ▼
                      chanx consumer    chanx consumer   chanx consumer      AsyncAPI 3.0 schema
                        (tab 1)           (tab 2)        (other device)  ──▶ generated TS types

Four pieces: a plain pydantic-ai agent (typed tools, typed deps, requires_approval on the dangerous ones, structured output); a harness that runs it and translates each event into a typed message; a channel layer that carries those events to a per-conversation group rather than a single socket; and a chanx consumer that validates, routes, and documents the protocol the React client generates its types from.

The agent: tools that know they're dangerous

Nothing here is WebSocket-aware. It's a standard pydantic-ai agent with two deliberate choices: a structured output type, and typed deps carrying the conversation id so every tool operates on that conversation's tasks.

class TextAnswer(BaseModel):
    """Final answer to the user, with suggested follow-up prompts."""

    content: str
    follow_ups: list[str] = Field(default_factory=list)


@dataclass
class AgentDeps:
    conversation_id: str


# Every run ends as either a final answer or a request for tool approval.
AgentOutput = TextAnswer | DeferredToolRequests


def build_agent(model: Model | str | None = None) -> Agent[AgentDeps, AgentOutput]:
    agent: Agent[AgentDeps, AgentOutput] = Agent(
        model or settings.resolved_model,
        deps_type=AgentDeps,
        output_type=[TextAnswer, DeferredToolRequests],
        instructions=INSTRUCTIONS,
    )

    @agent.tool
    async def add_task(ctx: RunContext[AgentDeps], title: str) -> db.TaskItem:
        """Add a new task with the given title."""
        return await db.add_task(ctx.deps.conversation_id, title)

    @agent.tool(requires_approval=True)
    async def delete_task(ctx: RunContext[AgentDeps], task_id: int) -> str:
        """Permanently delete the task with the given id. Requires user approval."""
        await db.delete_task(ctx.deps.conversation_id, task_id)
        return f"Task {task_id} deleted"

    return agent

(The repo has six tools; clear_all_tasks is the other one behind approval.)

requires_approval=True is pydantic-ai's deferred tools feature. When the model calls delete_task, the tool does not execute. Instead the run ends early and result.output is a DeferredToolRequests carrying the pending calls, which is exactly why output_type is a union: every run ends as either a final answer or a request for permission, and the type system makes you handle both.

TextAnswer instead of plain str is the structured-output half of the story. Alongside the answer text, the model fills follow_ups with three suggested next prompts that the UI renders as clickable chips. The final answer isn't prose with conventions bolted on; it's a validated object, and the chips ride along in the same schema.

build_agent() being a factory instead of a module-level singleton matters later: tests build the same agent with a fake model.

The message contract

Every WebSocket message is a Pydantic model with a Literal action field. chanx uses that field as a discriminator for validation, routing, and docs:

class ChatMessage(BaseMessage):
    """User sends a prompt to the agent."""
    action: Literal["chat"] = "chat"
    payload: ChatPayload

The full protocol is small. Client → server: chat, tool_decision. Server → client: user_message, stream_start, text_delta, tool_call, tool_result, approval_request, stream_end, suggestions, history, tasks_updated, notification, agent_error.

The consumer binds it together: handlers declare what they accept and what they may reply with, and that declaration is the documentation:

@channel(name="agent", description="Pydantic AI task assistant over a typed WebSocket")
class AgentConsumer(BaseConsumer):
    passthrough_events: ClassVar[list[type[BaseMessage]]] = BROADCAST_MESSAGES

    @ws_handler(
        summary="Send a prompt to the agent",
        output_type=AgentServerMessage,  # the union of everything we may send back
    )
    async def handle_chat(self, message: ChatMessage) -> None:
        self._spawn_run(self.harness.run(message.payload.text))

    @ws_handler(summary="Approve or deny pending tool calls", output_type=AgentServerMessage)
    async def handle_tool_decision(self, message: ToolDecisionMessage) -> None:
        self._spawn_run(self.harness.resolve(message.payload.decisions))

No if/else routing, no manual validation, and /asyncapi.json now serves an AsyncAPI 3.0 spec describing every message above: the WebSocket equivalent of what drf-spectacular gives you for REST (I wrote about that workflow in the schema-first post).

Streaming structured output: snapshots into deltas

Streaming plain text is easy. Streaming a structured answer is the interesting part: the final output is a TextAnswer object, so there is no raw text stream to forward. pydantic-ai's answer is run_stream() plus two hooks.

Tool events arrive through an event_stream_handler, which the harness maps straight onto the message contract:

async def forward_tool_events(
    _ctx: RunContext[AgentDeps], events: AsyncIterable[AgentStreamEvent]
) -> None:
    async for event in events:
        match event:
            case FunctionToolCallEvent(part=part):
                await self.send(ToolCallMessage(payload=ToolCallPayload(
                    tool_call_id=part.tool_call_id,
                    tool_name=part.tool_name,
                    args=part.args_as_dict(),
                )))
            case FunctionToolResultEvent(part=part):
                await self.send(ToolResultMessage(payload=ToolResultPayload(
                    tool_call_id=part.tool_call_id,
                    content=to_jsonable_python(part.content),
                )))

The answer itself streams as partially validated snapshots: stream_output() yields a growing TextAnswer each time more of the model's JSON arrives. Diffing successive content values recovers the text deltas to send:

sent_text = ""
async with self.agent.run_stream(
    user_prompt,
    message_history=history,
    deferred_tool_results=deferred_tool_results,
    deps=AgentDeps(conversation_id=self.conversation_id),
    event_stream_handler=forward_tool_events,
) as stream:
    async for partial in stream.stream_output(debounce_by=None):
        match partial:
            case TextAnswer(content=content) if content and content != sent_text:
                delta = (
                    content.removeprefix(sent_text)
                    if content.startswith(sent_text)
                    else content
                )
                await self.send(TextDeltaMessage(payload=TextDeltaPayload(delta=delta)))
                sent_text = content
        output = partial
    await db.save_history(self.conversation_id, stream.all_messages_json().decode())

The client sees text_delta for prose and tool_call/tool_result pairs for tool activity (rendered as live cards in the UI), and the loop stays completely ignorant of which tools exist.

After the loop, the union output type earns its keep:

match output:
    case DeferredToolRequests() as requests:
        # run_stream stops forwarding events once the final result (the paused
        # calls) is found, so announce them here: same ids, same wire order.
        for part in requests.approvals:
            await self.send(ToolCallMessage(payload=ToolCallPayload(...)))
        PENDING_APPROVALS[self.conversation_id] = requests
        await self.send(approval_request_message(requests))
    case TextAnswer(content=content, follow_ups=follow_ups):
        await self.send(StreamEndMessage(payload=StreamEndPayload(text=content, usage=...)))
        if follow_ups:
            await db.save_suggestions(self.conversation_id, follow_ups)
            await self.send(SuggestionsMessage(payload=SuggestionsPayload(follow_ups=follow_ups)))

One subtlety worth calling out: a deferred tool call never reaches the event_stream_handler (deferred means "not executed", and only executing tools emit events), so the harness announces the paused calls itself before the approval_request. The client renders the same tool cards either way; it can't tell which path produced them. And on the happy path, the suggestions arrive as their own message after stream_end, so the chips pop in once the answer settles.

Broadcast, don't send: background runs and groups

My first version awaited the agent run inside the chat handler and wrote messages straight to the socket. It worked, but with three quiet problems: the handler blocks for the whole run, a page refresh mid-run kills the stream, and a second tab sees nothing.

The fix is the pattern chanx (and Django Channels before it) is built around: broadcast to a group, not to a connection. Every consumer joins a per-conversation group on connect, and the run executes as a background task that broadcasts events to that group:

async def post_authentication(self) -> None:
    group = f"conversation.{self.conversation_id}"
    await self.channel_layer.group_add(group, self.channel_name)
    self.harness = AgentHarness(agent, self.conversation_id, self._broadcast)

async def _broadcast(self, message: BaseMessage) -> None:
    await self.broadcast_event(message, groups=f"conversation.{self.conversation_id}")

def _spawn_run(self, coro) -> None:
    # The task outlives this connection: a client that refreshes mid-run
    # keeps receiving events through the conversation group.
    RUNNING[self.conversation_id] = asyncio.create_task(guarded(coro))

On the receiving side there is nothing to write. chanx's passthrough_events generates the event handlers: any message type in the list that arrives via the group is forwarded to the WebSocket client verbatim, and documented in AsyncAPI like everything else.

What this buys, concretely (all covered by tests):

The chat handler returns immediately; a per-conversation RUNNING dict rejects concurrent runs with a typed agent_error
Refresh-proof streams: the run keeps going without any socket attached; reconnect replays history and picks up live events
Multi-tab for free: every tab renders the same deltas, tool cards, and approval requests. Even the user's own prompt travels this way, echoed as a user_message before stream_start, so the sending tab renders the bubble from the same broadcast as everyone else instead of appending locally
Approve from anywhere: pending approvals are keyed by conversation, not connection; request approval on one device, approve on another

Beyond the socket: HTTP, workers, other frameworks

broadcast_event is a classmethod, so anything can push typed messages into a conversation. The demo exposes POST /conversations/{id}/notify, and a curl fires a toast on every connected tab. The agent uses the same door from the inside: a schedule_reminder tool reads the conversation id from ctx.deps, schedules a job, and returns immediately; when the job fires long after stream_end, it broadcasts a notification into the group.

That escape hatch scales in two directions. Group broadcast isn't the only delivery mode: chanx can address a single connection, so a handler can hand heavy work to a real task queue (taskiq, Celery, ARQ) using self.channel_name as a typed reply address:

# consumer: queue the job with a reply address
await run_agent_job.kiq(prompt, reply_to=self.channel_name)

# worker process: send typed events back to that exact connection
await AgentConsumer.send_event(JobProgressEvent(payload=...), reply_to)

# consumer: typed, routed, and documented like everything else
@event_handler
async def handle_job_progress(self, event: JobProgressEvent) -> ProgressMessage:
    return ProgressMessage(payload=event.payload)

The consumer stays a thin protocol adapter while the LLM work happens in a process built for it. And since the channel layer is just Redis, and chanx speaks the same API on Django Channels and FastAPI, the socket and the agent don't even have to share a service. A split I run in production: Django owns the user-facing WebSocket (sessions, ORM auth, a consumer that's little more than passthrough_events) while a separate FastAPI service runs the models and broadcasts into the same conversation groups. Django never awaits an LLM; the agent service never touches a session. The contract is the only thing they share, and that's the part AsyncAPI documents.

The demo runs on an in-memory layer with zero dependencies; set REDIS_URL and the same code spans processes and instances.

Human-in-the-loop: the approval round trip

This is the part WebSockets were made for. The full sequence:

Model calls delete_task → the run ends with DeferredToolRequests, and the harness announces the paused calls as tool_call messages, so the UI shows the tool card immediately (status: awaiting approval)
The server sends approval_request, persists the message history, and simply stops. No polling, no hanging request
The user clicks Approve or Deny → client sends tool_decision
The harness builds DeferredToolResults and resumes the run with the saved history:

async def resolve(self, decisions: list[ToolDecision]) -> None:
    approvals = {
        d.tool_call_id: (
            ToolApproved(override_args=d.override_args)
            if d.approved
            else ToolDenied(message=d.reason or "Denied by the user")
        )
        for d in decisions
    }
    del PENDING_APPROVALS[self.conversation_id]
    await self.run(None, DeferredToolResults(approvals=approvals))

The resumed run is just another run_stream() call: approved tools execute (their events stream through the same handler), denied tools return the denial message to the model, and the model streams a final response either way. ToolApproved(override_args=...) is a lovely detail: the user can edit the arguments as part of approving, which is how real review UIs should work.

Approvals also survive reconnects, in both directions. Pending requests are keyed by conversation rather than connection, and the consumer re-sends the open approval_request to every new connection, so a refresh mid-approval gets a working approval card instead of a dead end. The E2E test literally approves from a second browser tab. One UI subtlety falls out of multi-tab: when a run resumes, other tabs still showing an open approval card mark it "resolved from another session" on the next stream_end.

The gotcha: tell the model the harness has it covered

My first instructions said "destructive operations require user approval". The model took that personally. Instead of calling delete_task, it asked in chat: "This is permanent, do I have your approval? (yes/no)". Technically obedient, completely bypassing the typed approval flow.

The fix is to be explicit about the division of labor:

When the user asks for a destructive operation, call the tool directly — do NOT
ask for confirmation in chat. The system automatically pauses destructive tool
calls and asks the user for approval in the UI.

If you build approval flows, this prompt bug will find you. The model needs to know that safety is infrastructure, not its job.

AsyncAPI → TypeScript: the client writes itself

chanx serves the contract at /asyncapi.json; every payload above appears in components.schemas as plain JSON Schema. A ~100-line script fetches the spec, collects the client-bound and server-bound messages from the AsyncAPI operations (inputs vs replies), and feeds them to json-schema-to-typescript:

// generated - src/generated/messages.ts
export interface ToolCallMessage {
  action: "tool_call";
  payload: ToolCallPayload;
}
export type ClientMessage = ChatMessage | ToolDecisionMessage;
export type ServerMessage = /* union of all twelve server messages */;

One codegen detail worth knowing: Pydantic marks action as optional in JSON Schema (it has a default), but it's the discriminator, so the script forces it into required for TypeScript to narrow on. With that, the entire client state layer is a single exhaustive switch:

function applyServer(state: ChatState, msg: ServerMessage): ChatState {
  switch (msg.action) {
    case "text_delta":       // append to the streaming bubble
    case "tool_call":        // add a tool card
    case "approval_request": // render Approve/Deny
    case "suggestions":      // show follow-up chips
    case "stream_end":       // finalize + count tokens
    // TypeScript errors if any server message is unhandled
  }
}

Rename a field in a Pydantic payload, run pnpm generate, and tsc lists every client line that just became wrong. Same loop as schema-first REST: the transport changed, the discipline didn't.

Persistence: history as a typed artifact

pydantic-ai ships ModelMessagesTypeAdapter, a TypeAdapter for full message histories, so conversation state is just another validated blob: stream.all_messages_json() on the way out, ModelMessagesTypeAdapter.validate_json(raw) on the way back in. Two functions, no bespoke serialization.

On connect, the consumer replays the conversation as a history message plus the current task list (tasks_updated) and the latest follow-up chips. The transcript builder walks the typed history and reconstructs the interleaving: user text, assistant text, and tool cards in order, each card carrying a done, denied, or awaiting status, so a denied deletion still renders as denied after a refresh. Since everything is stored per conversation, a sidebar with /conversations (titles derived from the first prompt) and conversation delete falls out almost for free. Multi-turn context, reconnects, and resume-after-approval all sit on these same two functions.

Testing the whole flow without an LLM

The part I care most about: the entire WebSocket protocol (streaming, tool execution, approval, denial, history replay) runs in pytest with no API key. pydantic-ai's FunctionModel plays the model; a stream function scripts exactly what "the LLM" does. One wrinkle of structured output: the final answer is itself a call to the output tool, so the script emits that too:

async def delete_task_flow(messages, info):
    tool_return = next(
        (p for p in messages[-1].parts if isinstance(p, ToolReturnPart)), None
    )
    if tool_return is not None:
        # After the tool ran, answer via the structured-output tool
        yield {0: DeltaToolCall(
            name=info.output_tools[0].name,
            json_args=json.dumps({"content": f"Result: {tool_return.content}"}),
        )}
    else:
        yield {0: DeltaToolCall(name="delete_task", json_args='{"task_id": 1}')}

chanx's WebsocketCommunicator drives the consumer, and the assertions read like the protocol spec:

await comm.send_message(ChatMessage(payload=ChatPayload(text="Delete task 1")))
replies = await comm.receive_all_messages(stop_action="approval_request")

# the prompt echoes to the group, the paused call is announced, the run pauses
assert [m.action for m in replies] == [
    "user_message", "stream_start", "tool_call", "approval_request",
]
assert (await db.list_tasks("conv-delete")) != []     # nothing deleted yet

await comm.send_message(ToolDecisionMessage(...approved=True...))
replies = await comm.receive_all_messages(stop_action="stream_end")
assert await db.list_tasks("conv-delete") == []       # now it's gone

The rest of the suite covers the deny path, two tabs receiving identical broadcast feeds, approving from a fresh connection after a disconnect, the concurrent-run guard, transcript replay with denied and awaiting tool cards, and a contract test that pulls /asyncapi.json and asserts every message type is present, so the schema the frontend generates from can't silently lose a message. Twenty tests, a couple of seconds, no LLM.

One trap worth sharing: Agent.override(model=...) uses contextvars, which don't propagate into the consumer's task under the test communicator. The factory pattern sidesteps it: tests call build_agent(FunctionModel(...)) and patch the consumer's agent directly.

Production notes

The demo makes three simplifications you'd revisit for production:

Scaling out: the repo ships a docker-compose.yml for Redis, and the README walks through the two-instance demo (chat on one port, watch the same stream on another).
Auth: the consumer trusts the conversation query param. chanx has an authentication hook (post_authentication) where you'd validate a session or token before joining the group and replaying history.
Durable pending approvals: pending DeferredToolRequests live in an in-process dict; store them in the database if approvals must survive restarts. And if you ever accept message history from the client, run it through pydantic-ai's sanitize_messages first.

Takeaways

WebSockets are the natural transport for agents with approvals: pause and resume are just messages, in both directions, on one connection. pydantic-ai's run_stream() + DeferredToolRequests/DeferredToolResults hand you streaming and human-in-the-loop as library features; the harness adapting them to a transport is ~150 lines
Structured output streams too: partial-validation snapshots diff into text deltas, and extra fields like follow-up suggestions ride along in the same schema
Treat the WebSocket protocol as a contract: Pydantic models with discriminated actions on the server, AsyncAPI in the middle, generated TypeScript on the client
Run in the background, broadcast to a group: refresh-proof streams, multi-tab sync, approve-from-anywhere, and server-initiated notifications all come from one decision; events go to the conversation, not the socket. The same layer lets you offload runs to a worker or put a Django socket in front of a FastAPI agent
Tell the model that approval is handled by infrastructure, or it will improvise its own confirmation dialog in prose
Script the model with FunctionModel and test the entire flow offline; deterministic agent tests are entirely achievable

I hope this saves you some of the pain it took me to learn. If you try the pattern (or have your own tricks for agent WebSockets), the repo issues are open.

🚀 Supercharge Your FastAPI WebSockets with Channel Layers & Group Messaging

Huy Nguyen — Thu, 18 Sep 2025 02:38:11 +0000

Building real-time features in FastAPI? Hit the wall trying to send messages from HTTP endpoints to WebSocket clients? Struggling with group messaging and scalability?

Fast Channels brings Django Channels' battle-tested architecture to FastAPI and the entire ASGI ecosystem.

The Pain Points We've All Hit

FastAPI's native WebSocket support gets you started, but quickly becomes limiting:

❌ No group messaging - can't broadcast to multiple connections

❌ No cross-process communication - HTTP endpoints can't reach WebSocket clients

❌ Manual connection management - tracking users, rooms, cleanup

❌ No scalability - everything breaks with multiple server instances

❌ Testing nightmare - WebSocket unit tests are painful to write

What Fast Channels Delivers

✅ Group messaging - broadcast to chat rooms, user segments, notification groups

✅ Cross-process communication - HTTP → WebSocket, background workers → clients

✅ Auto connection management - join/leave groups automatically

✅ Multi-instance scaling - Redis-backed message routing

✅ Built-in testing framework - proper WebSocket test utilities

✅ Consumer patterns - WebSocket handlers that feel like FastAPI views

Battle-Tested in Production

This isn't experimental tech. Django Channels has powered real-time features for millions of users across thousands of production applications for years. Fast Channels brings that same proven architecture to FastAPI.

# Simple but powerful
class ChatConsumer(AsyncWebsocketConsumer):
    groups = ["chat_room"]  # Auto-join on connect

    async def receive(self, text_data=None, **kwargs):
        # Broadcast to all users in the room
        await self.channel_layer.group_send("chat_room", {
            "type": "chat_message",
            "message": text_data
        })

Production-Ready Features

Redis Integration: Full Redis support with Sentinel for high availability

Multiple Backends: In-memory for testing, Redis Queue for reliability, Redis Pub/Sub for speed

Type Safety: Complete mypy and pyright support

ASGI Universal: Works with FastAPI, Starlette, Quart, any ASGI framework

Testing Built-in: Comprehensive WebSocket testing utilities

FastAPI vs Fast Channels vs Alternatives

Feature	Native FastAPI	Broadcaster	Fast Channels
Basic WebSocket	✅	✅	✅
Group messaging	❌	✅	✅
Consumer patterns	❌	❌	✅
Message persistence	❌	❌	✅
Testing framework	❌	❌	✅
Auto connection mgmt	❌	❌	✅
Cross-process msgs	❌	❌	✅
Production battle-tested	❌	❌	✅

Real-World Impact

Teams are using Fast Channels for:

🏪 E-commerce: Live inventory, order tracking, flash sale updates

💬 Chat platforms: Real-time messaging, typing indicators, presence

📊 Dashboards: Live metrics, system monitoring, alerts

🎮 Gaming: Multiplayer sync, leaderboards, real-time events

🔔 Notifications: Push alerts, system updates, user notifications

Quick Start

pip install fast-channels[redis]

Register a channel layer, create a consumer, add to FastAPI - done. Your WebSockets now scale across multiple processes and servers.

Why It Matters

Real-time features shouldn't require reinventing messaging infrastructure. Fast Channels gives you Django's proven patterns with FastAPI's developer experience - the best of both worlds.

Production-ready: Redis Sentinel support, proper error handling, connection cleanup

Developer-friendly: Type safety, testing utilities, clear documentation

Scalable: Multi-process, multi-server, battle-tested architecture

Stop building WebSocket boilerplate. Start building features.

🔗 Links:

Building real-time features? What's your biggest WebSocket challenge? 👇

Building a Modern, Type-Safe Authentication System for Django REST Framework

Huy Nguyen — Wed, 09 Jul 2025 13:59:32 +0000

Have you ever spent hours fighting with authentication setup in your Django REST API? Or struggled with type errors that could have been caught at development time? Or wished your API documentation would just... work automatically?

I've been there. After years of wrestling with existing Django authentication packages, I decided to build something better: DRF Auth Kit – a modern, type-safe authentication toolkit that just works.

The Problem: Authentication Shouldn't Be This Hard

Let me paint a picture. You're building a modern web app with a Django REST API backend. You need:

✅ JWT authentication with cookie support

✅ Multi-factor authentication

✅ Social login (Google, GitHub, etc.)

✅ Automatic API documentation

✅ Type safety throughout

With existing solutions, you'd typically:

Install 3-4 different packages
Write custom serializers and views
Manually configure complex URL patterns
Struggle with type hints and documentation
Debug cryptic authentication errors

There had to be a better way.

The Solution: DRF Auth Kit

DRF Auth Kit is what I wish existed when I started building Django APIs. Here's what makes it different:

🎯 One Package, Everything Included

# Get everything you need
pip install drf-auth-kit[all]

# Or pick what you need
pip install drf-auth-kit[mfa]     # + Multi-factor auth
pip install drf-auth-kit[social]  # + Social authentication

🔒 Universal Authentication Backend

Instead of juggling multiple authentication classes:

# settings.py
REST_FRAMEWORK = {
    'DEFAULT_AUTHENTICATION_CLASSES': [
        'auth_kit.authentication.AuthKitAuthentication',
    ],
}

AUTH_KIT = {
    'AUTH_TYPE': 'jwt',  # or 'token', 'custom'
    'USE_AUTH_COOKIE': True,
    'USE_MFA': True,
}

This single configuration handles:

JWT tokens with automatic refresh
DRF token authentication
Custom authentication backends
HTTP-only cookies with security headers
Multi-factor authentication

🛠️ Intelligent URL Management

Conditional URL patterns - URLs automatically included/excluded based on settings:

# When USE_MFA=True, automatically creates:
# POST /api/auth/login/         # First step (password)
# POST /api/auth/login/verify/  # Second step (MFA code)
# POST /api/auth/login/change-method/
# POST /api/auth/login/resend/
# GET /api/auth/mfa/           # MFA management endpoints

# When USE_MFA=False, creates:
# POST /api/auth/login/         # Direct login

Social authentication - automatic endpoint generation:

# Just add a provider to INSTALLED_APPS:
INSTALLED_APPS = [
    'allauth.socialaccount.providers.google',
    'auth_kit.social',
]

# Automatically creates:
# POST /api/auth/social/google/
# POST /api/auth/social/google/connect/
# GET /api/auth/social/accounts/

📝 Type Safety That Actually Works

Every component includes comprehensive type hints:

from auth_kit.authentication import AuthKitAuthentication
from auth_kit.serializers import LoginRequestSerializer
from auth_kit.views import LoginView

Full mypy and pyright compatibility with comprehensive type hints throughout the codebase. Your IDE will thank you. No more guessing what parameters a function expects or what a setting does.

📚 API Documentation That Updates Itself

Integration with DRF Spectacular means your API docs are always current:

# Zero configuration needed
INSTALLED_APPS = [
    'drf_spectacular',
    'auth_kit',
]

Visit /api/docs/ and you'll see:

Interactive Swagger UI
Complete request/response schemas
Authentication flow examples
Working test interface

🛡️ MFA Made Simple

Multi-factor authentication that doesn't require a PhD:

# Enable MFA
INSTALLED_APPS = [
    'auth_kit',
    'auth_kit.mfa',  # Just add this!
]

AUTH_KIT = {
    'USE_MFA': True,  # URLs automatically reconfigure
}

Automatic URL reconfiguration when MFA is enabled:

Login endpoint becomes two-step flow
MFA management endpoints automatically created
All endpoints documented in your OpenAPI schema

You get:

Email-based MFA with HTML/text templates
Authenticator app support (Google Authenticator, Authy)
Backup codes for account recovery
Extensible system for custom MFA methods

The login flow becomes a clean two-step process:

Username/password → Get ephemeral token
MFA code → Get final auth tokens

🌐 Social Authentication That Just Works

Social login with 50+ providers and automatic URL generation:

INSTALLED_APPS = [
    'auth_kit',
    'auth_kit.social',  # Automatic URL generation
    'allauth.socialaccount.providers.google',
    'allauth.socialaccount.providers.github',
    # Add any provider, URLs created automatically!
]

# URLs are automatically created for each provider:
# POST /api/auth/social/google/
# POST /api/auth/social/github/
# POST /api/auth/social/google/connect/
# POST /api/auth/social/github/connect/
# GET /api/auth/social/accounts/

No manual URL configuration needed! The system scans your installed providers and creates endpoints automatically.

🍪 Security-First Cookie Authentication

HTTP-only cookies with all the security features:

AUTH_KIT = {
    'USE_AUTH_COOKIE': True,
    'AUTH_COOKIE_SECURE': True,      # HTTPS only
    'AUTH_COOKIE_HTTPONLY': True,    # No JavaScript access
    'AUTH_COOKIE_SAMESITE': 'Lax',   # CSRF protection
}

Perfect for modern web apps where you want security without complexity.

Real-World Example: Complete Auth Setup

Here's how you'd set up a complete authentication system:

# settings.py
INSTALLED_APPS = [
    'rest_framework',
    'auth_kit',
    'auth_kit.mfa',
    'auth_kit.social',
    'allauth',
    'allauth.account',
    'allauth.socialaccount',
    'allauth.socialaccount.providers.google',
]

REST_FRAMEWORK = {
    'DEFAULT_AUTHENTICATION_CLASSES': [
        'auth_kit.authentication.AuthKitAuthentication',
    ],
}

AUTH_KIT = {
    'AUTH_TYPE': 'jwt',
    'USE_AUTH_COOKIE': True,
    'USE_MFA': True,
}

# Social auth providers
SOCIALACCOUNT_PROVIDERS = {
    'google': {
        'SCOPE': ['profile', 'email'],
        'AUTH_PARAMS': {'access_type': 'online'},
        'OAUTH_PKCE_ENABLED': True,
        'APP': {
            'client_id': 'your-google-client-id',
            'secret': 'your-google-client-secret',
        }
    }
}

# urls.py
from django.urls import path, include

urlpatterns = [
    path('api/auth/', include('auth_kit.urls')),
    path('api/auth/social/', include('auth_kit.social.urls')),
]

That's it! You now have:

JWT authentication with cookies
User registration and email verification
Password reset functionality
Multi-factor authentication
Social login with Google
Complete API documentation
Type safety throughout

Customization: Make It Yours

Need custom behavior? Override any component:

# Custom login serializer
class CustomLoginSerializer(LoginRequestSerializer):
    remember_me = serializers.BooleanField(default=False)

    def validate(self, attrs):
        attrs = super().validate(attrs)
        # Your custom logic here
        return attrs

# Apply it
AUTH_KIT = {
    'LOGIN_REQUEST_SERIALIZER': 'myapp.serializers.CustomLoginSerializer',
}

Want a custom MFA method? Easy:

class SMSMFAHandler(BaseMFAHandler):
    name = "sms"

    def send_code(self, user: User, code: str) -> None:
        # Your SMS implementation
        send_sms(user.phone, f"Your verification code: {code}")

# Register it
AUTH_KIT = {
    'MFA_HANDLERS': [
        'auth_kit.mfa.handlers.app.MFAAppHandler',
        'auth_kit.mfa.handlers.email.MFAEmailHandler',
        'myapp.handlers.SMSMFAHandler',
    ],
}

Production-Ready Features

DRF Auth Kit isn't just for prototypes:

Comprehensive test coverage (>95%) across multiple Python/Django versions
Internationalization support for 57 languages
Security best practices built-in
Performance optimized with efficient database queries
Rate limiting and audit logging (coming soon)

The Numbers Don't Lie

Here's how DRF Auth Kit compares to existing solutions:

Feature	dj-rest-auth	django-trench	DRF Auth Kit
Type Safety	❌	❌	✅ Full mypy/pyright
Auto OpenAPI	⚠️ Partial	❌	✅ Complete
Cookie Auth	✅	❌	✅ Enhanced
MFA Support	❌	✅	✅ Simplified
Social Auth	✅	❌	✅ Better integration
Custom Auth	⚠️ Complex	❌	✅ Easy
I18n Support	❌	❌	✅ 57 languages
Test Coverage	⚠️	⚠️	✅ >95%

Try It Today

pip install drf-auth-kit[all]

Check out the complete documentation for detailed guides, examples, and API reference.

What's Next?

The roadmap includes exciting features:

WebAuthn support for passwordless authentication
Hardware security keys (YubiKey, FIDO2)
Rate limiting and audit logging
Advanced security features

Join the Community

DRF Auth Kit is open source and built by developers, for developers:

GitHub: github.com/forthecraft/drf-auth-kit
Documentation: drf-auth-kit.readthedocs.io
PyPI: pypi.org/project/drf-auth-kit

Final Thoughts

Building authentication shouldn't be a painful experience. With DRF Auth Kit, you get a modern, type-safe solution that grows with your project – from simple JWT authentication to complex multi-factor flows.

The combination of type safety, automatic documentation, and developer-friendly APIs makes it perfect for both new projects and existing ones looking to modernize their authentication.

What are you waiting for? Give it a try in your next project and let me know what you think! 🚀

Chanx: The Missing Toolkit for Django Channels - Production-Ready WebSockets Made Simple

Huy Nguyen — Wed, 11 Jun 2025 08:13:18 +0000

Building WebSocket applications with Django Channels just got a whole lot easier

The Problem with Django Channels

Don't get me wrong - Django Channels is fantastic for adding WebSocket support to Django applications. But after building several real-time apps, I kept running into the same challenges:

Setting up authentication for WebSocket connections
Validating message structures and preventing runtime errors
Testing multi-user WebSocket interactions
Managing group messaging patterns
Debugging WebSocket connections during development

I found myself writing the same boilerplate code over and over. So I built Chanx to solve these problems once and for all.

What is Chanx?

Chanx is a comprehensive toolkit that extends Django Channels with production-ready features. Think of it as what Django REST Framework is to Django views - but for WebSocket consumers.

Key Features That'll Save You Hours

🔐 DRF-Style Authentication

Use your existing authentication classes with WebSockets:

from chanx.generic.websocket import AsyncJsonWebsocketConsumer
from rest_framework.authentication import SessionAuthentication
from rest_framework.permissions import IsAuthenticated

class ChatConsumer(AsyncJsonWebsocketConsumer[ChatMessage]):
    authentication_classes = [SessionAuthentication]
    permission_classes = [IsAuthenticated]

    async def receive_message(self, message: ChatMessage, **kwargs):
        # Only authenticated users reach here
        # self.user is automatically populated
        pass

📝 Type-Safe Messaging with Pydantic

No more runtime errors from malformed messages:

from typing import Literal
from chanx.messages.base import BaseMessage

class ChatMessage(BaseMessage):
    action: Literal["chat"] = "chat"
    payload: str

class NotificationMessage(BaseMessage):
    action: Literal["notify"] = "notify"
    payload: dict

# Union type for type safety
AllMessages = ChatMessage | NotificationMessage

class MyConsumer(AsyncJsonWebsocketConsumer[AllMessages]):
    async def receive_message(self, message: AllMessages, **kwargs):
        # Pattern matching with full type safety
        match message:
            case ChatMessage(payload=text):
                await self.handle_chat(text)
            case NotificationMessage(payload=data):
                await self.handle_notification(data)

🎮 WebSocket Playground

Like Swagger, but for WebSockets! Auto-discovers your endpoints and lets you test them interactively:

# settings.py
INSTALLED_APPS = [
    # ...
    'chanx.playground',
]

# urls.py
urlpatterns = [
    path('playground/', include('chanx.playground.urls')),
]

Visit /playground/websocket/ and test your WebSocket endpoints without writing JavaScript!

👥 Simplified Group Messaging

Pub/sub messaging made easy:

class ChatConsumer(AsyncJsonWebsocketConsumer[ChatMessage]):
    async def build_groups(self) -> list[str]:
        room_id = self.scope["url_route"]["kwargs"]["room_id"]
        return [f"chat_room_{room_id}"]

    async def receive_message(self, message: ChatMessage, **kwargs):
        # Broadcast to all users in the room
        await self.send_group_message(message)

Messages automatically include metadata:

{
  "action": "chat",
  "payload": "Hello everyone!",
  "is_mine": false,
  "is_current": false
}

🧪 Testing That Actually Works

Multi-user WebSocket testing without the headaches:

from chanx.testing import WebsocketTestCase

class ChatTest(WebsocketTestCase):
    ws_path = "/ws/chat/room1/"

    async def test_multi_user_chat(self):
        # First user (automatic setup)
        await self.auth_communicator.connect()
        await self.auth_communicator.assert_authenticated_status_ok()

        # Second user with different credentials
        user2_comm = self.create_communicator(headers=user2_headers)
        await user2_comm.connect()

        # Test group messaging
        await self.auth_communicator.send_message(ChatMessage(payload="Hello!"))
        responses = await user2_comm.receive_all_json(wait_group=True)

        assert responses[0]["payload"] == "Hello!"
        assert responses[0]["is_mine"] == False

Real-World Example: Building a Chat App

Here's how simple it is to build a production-ready chat application:

1. Define Your Messages

from typing import Literal
from chanx.messages.base import BaseMessage, BaseGroupMessage

class ChatMessage(BaseMessage):
    action: Literal["chat"] = "chat"
    payload: str

class UserJoinedMessage(BaseGroupMessage):
    action: Literal["user_joined"] = "user_joined"
    payload: dict

ChatMessages = ChatMessage | PingMessage  # Include ping for health checks

2. Create Your Consumer

from chanx.generic.websocket import AsyncJsonWebsocketConsumer
from chanx.messages.incoming import PingMessage
from chanx.messages.outgoing import PongMessage

class ChatConsumer(AsyncJsonWebsocketConsumer[ChatMessages]):
    authentication_classes = [SessionAuthentication]
    permission_classes = [IsAuthenticated]

    async def build_groups(self) -> list[str]:
        room_id = self.scope["url_route"]["kwargs"]["room_id"]
        return [f"chat_room_{room_id}"]

    async def post_authentication(self):
        """Called after successful authentication"""
        await self.send_group_message(
            UserJoinedMessage(payload={"username": self.user.username})
        )

    async def receive_message(self, message: ChatMessages, **kwargs):
        match message:
            case PingMessage():
                await self.send_message(PongMessage())
            case ChatMessage(payload=text):
                # Save to database and broadcast
                await self.save_message(text)
                await self.send_group_message(message)

3. Set Up Routing

# chat/routing.py
from chanx.routing import path
from channels.routing import URLRouter

router = URLRouter([
    path("<int:room_id>/", ChatConsumer.as_asgi()),
])

# project/routing.py
from chanx.routing import include, path
from channels.routing import URLRouter

router = URLRouter([
    path("ws/chat/", include("chat.routing")),
])

4. Test It

class ChatRoomTest(WebsocketTestCase):
    ws_path = "/ws/chat/123/"

    async def test_user_can_join_and_chat(self):
        await self.auth_communicator.connect()
        await self.auth_communicator.assert_authenticated_status_ok()

        # Should receive user_joined message
        messages = await self.auth_communicator.receive_all_json()
        assert any(msg["action"] == "user_joined" for msg in messages)

        # Send a chat message
        await self.auth_communicator.send_message(ChatMessage(payload="Hello!"))
        responses = await self.auth_communicator.receive_all_json(wait_group=True)

        assert responses[0]["action"] == "chat"
        assert responses[0]["payload"] == "Hello!"

Why Developers Love Chanx

Complete Type Safety

# Generic type parameters for compile-time checking
class MyConsumer(AsyncJsonWebsocketConsumer[
    IncomingMessages,  # Required: Your message types
    ChannelEvents,     # Optional: Channel layer events
    Room              # Optional: Model for object permissions
]):
    queryset = Room.objects.all()  # For object-level permissions

Enhanced Routing

Django-style routing specifically designed for WebSockets:

from chanx.routing import path, re_path, include

# Use familiar Django patterns
router = URLRouter([
    path("room/<str:room_name>/", ChatConsumer.as_asgi()),
    re_path(r"^admin/(?P<id>\d+)/$", AdminConsumer.as_asgi()),
    path("api/", include("api.routing")),
])

Channel Events

Type-safe communication between consumers:

# Send events from views, tasks, or other consumers
ChatConsumer.send_channel_event(
    "chat_room_123",
    NotificationEvent(payload={"message": "System update"})
)

# Handle in consumer
async def receive_event(self, event: NotificationEvent):
    match event:
        case NotificationEvent():
            await self.send_message(SystemMessage(payload=event.payload))

Getting Started

pip install chanx

Essential links:

📚 Documentation - Comprehensive guides and examples
💻 GitHub - Source code and issues
🚀 Complete Example - Production-ready chat app

Quick setup:

# settings.py
INSTALLED_APPS = [
    'channels',
    'rest_framework', 
    'chanx.playground',  # For WebSocket testing UI
]

CHANX = {
    'SEND_COMPLETION': True,  # Important for testing
    'SEND_AUTHENTICATION_MESSAGE': True,
}

Perfect for AI Applications

Chanx is especially powerful for AI chatbots and streaming applications:

async def handle_ai_chat(self, user_message: str):
    """Stream AI response token by token"""
    async for token in self.get_ai_stream(user_message):
        await self.send_message(AIStreamingMessage(payload=token))

Community and Feedback

Chanx is actively maintained and used in production applications. The community is growing, and I'm always looking for feedback:

What challenges are you facing with Django Channels?
Which features would be most valuable to you?
How can we make WebSocket development even easier?

Conclusion

Django Channels is powerful, but building production WebSocket applications shouldn't require reinventing the wheel every time. Chanx provides the missing pieces:

✅ Authentication - DRF integration out of the box

✅ Type Safety - Catch errors at development time

✅ Testing - Multi-user scenarios made simple

✅ Developer Tools - WebSocket playground for debugging

✅ Group Messaging - Pub/sub patterns simplified

✅ Production Ready - Battle-tested in real applications

Whether you're building a chat application, real-time collaboration tool, or AI-powered interface, Chanx helps you ship faster with fewer bugs.

Try it out and let me know what you think! 🚀

What's your experience with Django Channels? Have you run into similar challenges? Share your thoughts in the comments!

🚀 Streamline Your Python & Django Development with Production-Ready Cookiecutter Templates

Huy Nguyen — Mon, 09 Jun 2025 15:18:53 +0000

TL;DR: I've created a comprehensive cookiecutter template collection that generates production-ready Python apps, PyPI packages, Django REST APIs, and DRF packages with modern tooling, best practices, and zero configuration hassle.

The Problem: Starting Projects Shouldn't Be This Hard

How many times have you started a new Python project and spent hours setting up the same boilerplate code? Configuring pyproject.toml, setting up pre-commit hooks, writing GitHub Actions, configuring Docker, setting up Django with all the modern best practices...

The list goes on. And then you realize you forgot to configure type checking, or your test coverage setup is broken, or you're missing that one crucial piece of tooling that makes development smooth.

I've been there. We've all been there.

The Solution: One Template to Rule Them All (Actually, Four)

That's why I created cookiecutter-python — a curated collection of four specialized cookiecutter templates that cover the most common Python development scenarios:

🐍 Python App - For General-Purpose Applications

Perfect for CLI tools, data processing scripts, or any standalone Python application.

📦 Python Package - For PyPI Publishing

Ready-to-publish package structure with documentation, CI/CD, and multiple license options.

🌐 Django REST App - For Full-Stack Web APIs

Complete Django REST Framework setup with authentication, database, optional Celery/WebSockets, and cloud deployment support.

🧩 DRF Package - For Reusable Django Components

Lightweight structure for building and distributing Django REST Framework extensions.

What Makes These Templates Special?

✨ Modern Tooling Out of the Box

Each template is carefully configured with specialized tooling for its use case:

🔧 Universal Tools (All Templates):

UV for lightning-fast dependency management with intelligent dependency groups
Ruff for blazing-fast linting and formatting
Black for consistent code formatting
Dual type checking with both mypy and pyright for maximum type safety
Pre-commit hooks for automated quality checks
pytest with parallel execution (pytest-xdist) and coverage reporting
GitHub Actions CI/CD pipelines

📦 Package Templates (PyPackage & DRF Package) Add:

Commitizen for conventional commits and automated versioning
Interrogate for docstring coverage monitoring with badge generation
Sphinx ecosystem for professional documentation (myst-parser, ReadTheDocs theme)
Automated publishing workflows to PyPI

🌐 Django Templates (DRF & DRF Package) Add:

Factory Boy for realistic test data generation
Faker for generating fake data
django-stubs + djangorestframework-stubs for comprehensive Django type checking
django-debug-toolbar for development debugging
IPython for enhanced Django shell experience
Freezegun for time-based testing

🏗️ Production-Ready Architecture

The Django templates include:

PostgreSQL integration with connection pooling
Redis support for Celery and WebSockets
JWT authentication with django-rest-auth
API documentation with Swagger/ReDoc
Docker containerization
Cloud storage support (AWS S3, GCP, Azure)
Email providers integration (SendGrid, Mailgun, etc.)

🎯 Zero Configuration Philosophy

Each generated project is immediately runnable. No additional setup required. Just generate, install dependencies, and start developing.

Quick Start Guide

Prerequisites

First, you'll need to install cookiecutter. I recommend using Homebrew (macOS) or pipx for a clean installation:

macOS (Recommended):

brew install cookiecutter

Linux/Windows with pipx:

pipx install cookiecutter

Alternative (with pip):

python3 -m pip install --user cookiecutter

Why not just pip install cookiecutter? Since we'll be using UV for Python dependency management in the generated projects, keeping cookiecutter isolated prevents potential conflicts and keeps your global Python environment clean.

Generate Your Project

cookiecutter gh:huynguyengl99/cookiecutter-python

You'll be prompted to select your template type:

Select a template
1 - Python app (A cookiecutter template for a normal python app)
2 - Python package (A cookiecutter template for a python package)  
3 - DRF (A cookiecutter template for a django rest framework app)
4 - DRF package (A cookiecutter template for a drf package)
Choose from [1/2/3/4] (1):

Start Developing Immediately

For a Django REST App:

cd your-project
uv sync                    # Install dependencies
docker compose up -d       # Start PostgreSQL/Redis
scripts/manage.sh migrate  # Run migrations
scripts/manage.sh runserver # Start development server

Your API documentation is immediately available at:

Swagger UI: http://localhost:8000/api/schema/swg/
ReDoc: http://localhost:8000/api/schema/redoc/

Deep Dive: What Each Template Provides

🐍 Python App Template

Perfect for: CLI tools, data processing, automation scripts

Includes:

Modern pyproject.toml configuration
Docker setup for easy deployment
Comprehensive testing framework
Code quality tools (Ruff, Black, mypy, pyright)
GitHub Actions for CI/CD

Generated Structure:

my-app/
├── pyproject.toml
├── docker-compose.yml
├── Dockerfile
├── scripts/
│   └── lint.sh
└── my_app/
    └── __init__.py

📦 Python Package Template

Perfect for: Open-source libraries, reusable components

Includes:

PyPI-ready project structure
Sphinx documentation with ReadTheDocs integration
Multiple license options (MIT, BSD, Apache, GPL)
Automated versioning with Commitizen
Publishing workflow to PyPI

Generated Structure:

my-package/
├── pyproject.toml
├── LICENSE
├── docs/
├── tests/
├── .github/workflows/
└── my_package/
    └── __init__.py

🌐 Django REST App Template

Perfect for: Web APIs, SaaS backends, microservices

Configuration Options:

Database: PostgreSQL with multiple version options
Authentication: JWT with optional Google OAuth
Background Tasks: Optional Celery integration
Real-time: Optional WebSocket support with Channels
Cloud Storage: AWS S3, Google Cloud, Azure support
Email: 10+ provider integrations (SendGrid, Mailgun, etc.)
API Style: Optional camelCase conversion

Generated Structure:

my-api/
├── pyproject.toml
├── docker-compose.yml
├── Dockerfile
├── scripts/
│   └── manage.sh
└── my_api/
    ├── config/
    │   ├── settings/
    │   ├── urls.py
    │   └── wsgi.py
    ├── accounts/        # Custom user model
    ├── core/           # Utilities
    └── status/         # Health check

Key Features:

Custom User Model with email authentication
JWT Authentication with refresh token rotation
API Versioning ready structure
Comprehensive Testing with factories and fixtures
Admin Interface customization
Structured Logging with structured logs
Health Checks and monitoring endpoints

🧩 DRF Package Template

Perfect for: Django REST Framework extensions, reusable API components

Includes:

Sandbox Django project for testing
Documentation with Django integration
Package publishing to PyPI
Testing across multiple Django/DRF versions

Unique Feature: The sandbox directory serves dual purposes:

Testing Environment: Run tests against your package
Development Playground: Manual testing with a real Django app

Advanced Features That Set These Apart

🔧 Intelligent Dependency Management with UV

Each template uses UV's dependency groups for optimal development experience:

[dependency-groups]
dev = ["commitizen", "factory-boy", "pytest-django"]  # Core development
lint = ["black", "ruff", "mypy", "pyright"]           # Code quality  
test = ["pytest-cov", "pytest-xdist", "coverage"]    # Testing tools
docs = ["sphinx", "myst-parser", "interrogate"]      # Documentation

Smart defaults with uv sync:

Application templates: Auto-installs dev + lint groups
Package templates: Auto-installs all groups for comprehensive development
Selective installation: uv sync --group test for specific workflows

🛡️ Type Safety First

Every template includes comprehensive type checking with dual tools for maximum coverage:

mypy with specialized Django and DRF plugins for framework-specific type checking
pyright for advanced static analysis and fast feedback
django-stubs for comprehensive Django typing (Django templates)

This dual approach catches different categories of type issues and provides the most robust type safety available in the Python ecosystem.

📊 Quality Metrics & Automation

Package templates include sophisticated quality monitoring:

# Docstring coverage with automatic badge generation
interrogate --generate-badge docs/_static/ my_package

# Conventional commits with automated versioning
cz commit  # Interactive commit creation
cz bump    # Automatic version bumping and changelog

# Parallel testing for faster feedback
pytest -n5  # Runs tests across 5 parallel processes

Quality gates enforce standards:

80% docstring coverage requirement
100% type hint coverage with mypy strict mode
Zero linting violations with Ruff and Black
Conventional commit format for automated changelog generation

⚡ Modern Python Practices

PEP 621 compliant pyproject.toml
UV for dependency management (10x faster than pip)
Ruff for linting (100x faster than flake8)
Python 3.10+ with modern syntax support

🚀 Production Deployment Ready

Django templates include:

Multi-stage Docker builds for optimized images
Gunicorn configuration for production WSGI
Environment-based settings (dev/test/production)
Static file handling with WhiteNoise
Database connection pooling

Real-World Usage Examples

Scenario 1: Building a SaaS API

cookiecutter gh:huynguyengl99/cookiecutter-python
# Select: 3 - DRF
# Configure: PostgreSQL, JWT auth, Celery, SendGrid emails, AWS S3

cd my-saas-api
uv sync
docker compose up -d
scripts/manage.sh migrate
scripts/manage.sh createsuperuser
scripts/manage.sh runserver

Result: Full-featured API with user authentication, background tasks, email integration, and cloud storage. Ready for production deployment.

Scenario 2: Publishing a Python Library

cookiecutter gh:huynguyengl99/cookiecutter-python
# Select: 2 - Python package
# Configure: MIT license, GitHub username, PyPI publishing

cd my-library
uv sync
pytest  # Tests pass immediately
scripts/lint.sh  # Code quality checks pass
cz commit  # Conventional commits

Result: Professional package structure ready for PyPI publishing with documentation, CI/CD, and automated releases.

Scenario 3: Building a DRF Extension

cookiecutter gh:huynguyengl99/cookiecutter-python
# Select: 4 - DRF package

cd my-drf-extension
uv sync
docker compose up -d  # Test database
pytest sandbox  # Test against real Django app
python sandbox/manage.py runserver  # Manual testing

Result: Professional DRF package with testing sandbox and publishing workflow.

Best Practices Baked In

🧪 Testing Excellence

Each template includes sophisticated testing infrastructure:

🏃‍♂️ Performance-Optimized Testing:

Parallel execution with pytest-xdist (5 parallel processes by default)
Smart test discovery with proper Django integration
Coverage reporting with branch coverage and XML output for CI integration

🏭 Test Data Generation (Django Templates):

# Factory Boy integration for realistic test data
from accounts.factories import UserFactory

def test_user_creation():
    user = UserFactory.create(email="test@example.com")
    assert user.is_active

⏰ Time-Based Testing:

# Freezegun for testing time-dependent code
from freezegun import freeze_time

@freeze_time("2023-01-01")
def test_timestamp_behavior():
    # Test behavior at specific time
    pass

🔧 Advanced Testing Features:

pytest-mock for sophisticated mocking
pytest-django for Django-specific fixtures and database handling
Factory Boy with custom factory metaclass for type-safe model factories
Coverage exclusions for migrations, settings, and other non-testable code

📝 Documentation Standards

Package templates include professional documentation tooling:

📖 Sphinx Documentation:

Automatic API documentation generation
ReadTheDocs integration with automated builds
MyST parser for Markdown support in reStructuredText projects
Cross-references and intersphinx linking

📊 Documentation Quality Monitoring:

Interrogate for docstring coverage monitoring
Automatic badge generation showing documentation coverage percentage
80% coverage requirement enforced in CI/CD
SVG badges integrated into documentation and README

# Check docstring coverage
interrogate -vv my_package

# Generate coverage badge
interrogate --generate-badge docs/_static/

📋 API Documentation (Django Templates):

Swagger UI and ReDoc auto-generation from code
OpenAPI 3.0 specification export
Interactive API testing directly in documentation

🔄 CI/CD Pipeline

GitHub Actions workflows include:

🧪 Comprehensive Testing:

Multi-version testing across Python 3.10-3.13
Parallel test execution with pytest-xdist for faster feedback
Coverage reporting with Codecov integration and XML output
Matrix testing across multiple dependency versions (Django templates)

✅ Code Quality Gates:

Linting pipeline with Ruff for import sorting and style checks
Dual type checking with both mypy and pyright
Code formatting verification with Black
TOML file validation and sorting with toml-sort

📦 Automated Publishing (Package Templates):

Semantic versioning with Commitizen integration
Automated PyPI publishing on tagged releases
GitHub releases with automated changelog generation
Build artifact caching for faster CI runs

🔒 Security & Quality:

Dependency vulnerability scanning
Pre-commit hook validation in CI
Build reproducibility with UV lock files

🎨 Code Quality

Pre-commit hooks automatically run:

Ruff for linting and import sorting
Black for code formatting
toml-sort for TOML file organization
Type checking with mypy/pyright

Common Questions

Q: Why four separate templates instead of one configurable template?

A: Each template is optimized for its specific use case. A Python package has different needs than a Django API. Separate templates mean:

Cleaner code without excessive conditionals
Faster generation with only relevant files
Better maintenance with focused concerns
Easier customization for specific needs

Q: Why UV instead of pip/poetry?

A: UV is incredibly fast (10-100x faster than pip) and handles both Python version management and dependency resolution. It's the future of Python package management.

Q: Can I customize the generated templates?

A: Absolutely! The templates are designed to be starting points. All configuration is in standard files (pyproject.toml, settings.py, etc.) that you can modify as needed.

Q: What about database migrations in Django templates?

A: The Django templates include a custom user model and initial migrations. After generation, run scripts/manage.sh migrate and you're ready to go.

Future Roadmap

I'm continuously improving these templates based on community feedback:

FastAPI template for modern async APIs
Next.js + Django full-stack template
Kubernetes deployment configurations
More cloud provider integrations
Additional authentication providers

Getting Started Today

Ready to streamline your Python development? Here's how to get started:

Install cookiecutter: pip install cookiecutter
Generate your project: cookiecutter gh:huynguyengl99/cookiecutter-python
Choose your template and configure options
Start developing immediately!

Useful Links

📂 Repository: github.com/huynguyengl99/cookiecutter-python
📖 Documentation: Full README with detailed setup instructions
🐛 Issues: Report bugs or request features
💡 Discussions: Share your use cases and get help

Conclusion

These cookiecutter templates represent hundreds of hours of research, testing, and refinement to create the ideal starting point for Python projects. Whether you're building a simple CLI tool or a complex Django API, you can focus on your core business logic instead of wrestling with configuration.

The templates embody modern Python best practices:

Type safety with comprehensive type checking
Code quality with automated formatting and linting
Testing excellence with comprehensive test setups
Production readiness with Docker and deployment configs
Developer experience with helpful scripts and documentation

Give them a try for your next project. I'm confident they'll save you hours of setup time and help you ship better code faster.

What are you waiting for? Let's build something amazing together! 🚀

Have you used these templates? Found them helpful? Have suggestions for improvements? I'd love to hear about your experience! Drop a comment below or open an issue on GitHub.

Easy Multilingual Page (Scales to Over 50 Languages)

Huy Nguyen — Fri, 21 Jun 2024 14:17:13 +0000

Hi all,

So, a few months back, I kicked off a project to create a multilingual website. Let me tell you, managing content in multiple languages is a real headache. It’s doable if you’re only dealing with 3 or 4 languages, but once you hit 30+, it’s a whole different ball game trying to keep everything updated and consistent.

To make life easier, I decided to bring in a CMS for my Astro-based site. I dug into frameworks like Wagtail and Strapi since I’m familiar with Python and Node.js, but man, those were tough nuts to crack. Plus, I wanted a centralized place for all my content instead of having to jump between different pages for each language. This way, I can easily compare content across languages without getting lost.

And here's the kicker: ChatGPT is shaping up to be a solid translator, but no one’s integrated it into any frameworks yet. So, with my Django and Astrowind know-how (Astrowind is this cool UI built on Astro and Tailwind, and it’s open source), I whipped up a small Django library/framework to act as a headless CMS. I also tweaked the Astrowind source to pull in content from the CMS with multilingual support.

Check out this demo if you wanna see it in action:
https://django-astrowind.netlify.app/en

Not much has changed from the original Astrowind (https://astrowind.vercel.app/), except now you can switch languages on the fly, and the content is managed through a CMS with hashed caching.

If you’re interested in using it, I’ve put together a tutorial to help you set it up on your own machine:
https://django-headless-cms.readthedocs.io/en/latest/quick-start.html

Hope you dig it. And I hope my little project can be of some help to you guys.

DEV Community: Huy Nguyen

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

Setup

1. Messages are Pydantic models

2. One handler per message

3. Wire it into FastAPI

4. Try it

5. The docs page

6. A typed TypeScript client

7. Mistakes become compile errors

8. The rename

9. Scale it out with Redis

10. Bonus: a Python client for another service

Where to go from here

Agent Graph Engineering, Part 3: Contracts, Not Conventions

Table of contents

The failure mode, which is not a crash

Types are the stance, generation is the machinery

Three seams, three generators

What generation actually preserves

The schema can also build the interface

What it costs

When you do not need any of this

Agent Graph Engineering, Part 2: Split the Services, and Know Where the Line Goes

What the merged version looks like

The test you cannot write

The cut is about state, not layers

Two threads, not one

Updates travel one way

The boundary is generated, not hand-written

What it costs

When not to do this

Agent Graph Engineering, Part 1: The Stack, and Why Each Piece Is There

The one idea

Why this is worth your time

The stack

Python

Pydantic AI, for the agent

LangGraph, for the flow

chanx, for transport and contract

OpenTelemetry, into a trace store you own

pytest with a mocked LLM

Evals with real models

Keep business logic out of the agent

What we build

What it costs

Agent Graph Engineering, Part 0: I Needed This Before It Had a Name

Where I was

The state machine itch

Finding the pieces

Why this outlives the frameworks

What this series is

Benchmarking Python API Frameworks with Real Workloads: FastAPI vs Litestar vs DRF vs Django Ninja vs Django Bolt

The core design decisions

Setup

Results

The gap collapses once you hit the database

What authentication actually costs

Memory and CPU

A word on Django Bolt

Feedback I have NOT addressed yet

Takeaways

Streaming Agents with Pydantic AI over Typed WebSockets

The big picture

The agent: tools that know they're dangerous

The message contract

Streaming structured output: snapshots into deltas

Broadcast, don't send: background runs and groups

Beyond the socket: HTTP, workers, other frameworks

Human-in-the-loop: the approval round trip

The gotcha: tell the model the harness has it covered

AsyncAPI → TypeScript: the client writes itself

Persistence: history as a typed artifact

Testing the whole flow without an LLM

Production notes

Takeaways

Links

🚀 Supercharge Your FastAPI WebSockets with Channel Layers & Group Messaging

The Pain Points We've All Hit

What Fast Channels Delivers