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.

Every operation on a channel is implemented by the server: @tasks.method(), @tasks.event(), @tasks.subscription(), and @tasks.stream() declare the whole API. Clients call methods on the server; the server never calls methods on the client. Push data to clients with events or subscriptions instead.

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
• Sockets — proxy native protocols such as CDP with typed protocol bindings
• 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