fastapi-rpckit

Define your realtime API once in Python. Get the server, the contract, and typed clients for Python and TypeScript — none of which can drift apart.

Agents, browser automation, live dashboards and voice need more than request/response over HTTP: server-pushed events, binary streams, one long-lived connection. So the JSON-RPC envelope gets hand-written, the dispatch table grows by hand, and the frontend client is maintained separately — until the two disagree in production.

fastapi_rpckit makes the Python definition the single source of truth:

fastapi-rpckit is the FastAPI-first WebSocket RPC successor to pyrpckit, developed in the Canon workspace. FastAPI is a required dependency. REST and SSE stay with FastAPI/Canon and fastapi-streams. The existing dispatch engine, OpenRPC contracts, binary/raw channels, and client generators are retained. The old repository remains available.

Version 0.1.0 is the initial successor release; see the migration guide.

The idea

You define each operation once, on the server:

@tasks.method()
async def create(params: CreateTask, store: Inject[TaskStore]) -> Task:
    return await store.create(params.title)


@tasks.event()
async def updated(store: Inject[TaskStore]) -> AsyncIterator[TaskUpdated]:
    async for task in store.watch():
        yield TaskUpdated(task=task)

One command turns that into an OpenRPC document and clients in both languages:

fastapi-rpckit generate --config rpcgen.toml

And your frontend gets the whole API fully typed — no schema written twice, no client kept in sync by hand, no stringly-typed method names:

const task = await client.tasks.create({ title: "Ship 0.6" }); // Task

for await (const update of client.tasks.updated()) {           // TaskUpdated
  render(update.task);
}

Run --check in CI and a definition that outgrew its clients fails the build instead of shipping.

Install

uv add fastapi-rpckit
uv add "fastapi-rpckit[codegen]"  # client generation
uv add "fastapi-rpckit[otel]"     # OpenTelemetry integration

Python 3.12–3.14. FastAPI and Pydantic are required; Dishka, codegen, and OpenTelemetry remain optional extras.

Serve with FastAPI

from fastapi import FastAPI
from fastapi_rpckit import RpcChannel, RpcService, create_router

greetings = RpcChannel("greetings")


@greetings.method("hello")
async def hello() -> str:
    return "Hello"


rpc = RpcService()
rpc.socket("/rpc", channels=(greetings,))
app = FastAPI()
app.include_router(create_router(rpc))

For feature-level mounting on an existing APIRouter, use RpcRoutes. For Canon/Dishka composition, see the three-transport example. Both reuse the same WebSocket runtime; no second RPC router model is introduced.

Quickstart

Channels group related operations and provide their namespace; a service mounts them on a socket. Their dispatch logic can also be called directly:

from fastapi_rpckit import Inject, RpcChannel, RpcModel, RpcService, RpcSuccess


class CreateTask(RpcModel):
    title: str


class Task(RpcModel):
    id: int
    title: str


class TaskStore:
    def __init__(self) -> None:
        self._tasks: list[Task] = []

    async def create(self, title: str) -> Task:
        task = Task(id=len(self._tasks) + 1, title=title)
        self._tasks.append(task)
        return task


tasks = RpcChannel("tasks")


@tasks.method()
async def create(params: CreateTask, store: Inject[TaskStore]) -> Task:
    """Create a task."""
    return await store.create(params.title)


app = RpcService(version=1)
app.socket("/rpc", channels=(tasks,))


async def test_create() -> None:
    server = tasks.create_server(context=TaskStore())
    response = await server.handle(
        {
            "jsonrpc": "2.0",
            "id": 1,
            "method": "tasks.create",
            "params": {"title": "Ship 0.6"},
        }
    )
    assert isinstance(response, RpcSuccess)
    assert response.result == Task(id=1, title="Ship 0.6")

tasks.create is the wire name, the docstring becomes the contract summary, and Inject[TaskStore] is resolved on the server — it never appears in the public schema.

Operations on a channel are implemented by the server unless you say otherwise: @tasks.method(), @tasks.event(), @tasks.subscription(), and @tasks.stream() cover the usual client-to-server API. Only when the server also calls into the client do you name both sides, tasks.server and tasks.client — see Client methods.

Documentation

Read the hardening stack and SDK migration guide for all 15 ticket PRs, changed defaults, and integration steps.

• Services and channels — methods, namespaces, parameter styles, sockets, protocol versions
• Dependency injection — Inject[T], resolvers, scopes, Dishka
• Connections and events — the connect hook, rejecting handshakes, server-pushed events, limits
• Typed errors — stable codes, typed details, generated exception classes
• Binary streams — upload, download, and bidirectional byte streams beside JSON-RPC
• Raw sockets — proxy native protocols such as CDP with generated socket clients
• Contract and clients — rpcgen.toml, the CLI, the shape of generated clients
• Transports — FastAPI, custom sockets, testing

Examples

examples/ holds standalone runnable scripts, and examples/generated_clients contains real generated Python and TypeScript output you can read before installing anything.

Development

uv sync --all-groups
uv run ruff check .
uv run ruff format .
uv run pytest