pyrpckit

Build a pleasant, typed JSON-RPC API in Python.

pyrpckit lets you describe an API with small async functions, then use that same description to serve requests, publish an OpenRPC contract, and generate typed clients. It stays out of the way of your transport and dependency injection choices, so the API definition remains the easy part to read.

It is especially handy for WebSocket-style applications, where a client calls methods and the server can also send typed events over the same connection.

Contents

• Install
• Your first API
• Use it from your application
• Send typed events
• Create a contract and clients
• FastAPI and Dishka
• Errors
• Development

Install

uv add pyrpckit

Add extras only when you need them:

uv add "pyrpckit[codegen]"          # client generation
uv add "pyrpckit[fastapi,dishka]"   # WebSocket and Dishka helpers

Python 3.12 or newer is required.

Your first API

Start with a WebSocket channel and a normal async function. Methods declared on the same channel share one socket. RpcModel gives request and response data a consistent JSON shape; its field names are automatically available in camelCase on the wire.

from fastapi import APIRouter, FastAPI, WebSocket

from pyrpckit import Inject, RpcChannel, RpcModel
from pyrpckit.fastapi import serve


class OpenPage(RpcModel):
    url: str


class Page(RpcModel):
    title: str
    url: str


router = APIRouter(prefix="/sessions/{session_id}")
browser = RpcChannel(
    name="browser-control",
    namespace="browser",
    tags=("navigation",),
)


@browser.method()
async def open_page(
    params: OpenPage,
    navigation: Inject[BrowserNavigation],
) -> Page:
    page = await navigation.open(params.url)
    return Page(title=page.title, url=page.url)


@router.websocket("/control")
async def control_endpoint(websocket: WebSocket) -> None:
    await serve(browser, websocket, resolver=resolver)


app = FastAPI()
app.include_router(router)

That is the API. Clients see a browser.open_page method that accepts OpenPage and returns Page. Your application sees the BrowserNavigation service it already knows how to provide.

Inject[...] marks a server-side dependency. It never becomes part of the public JSON-RPC request or the generated contract. A handler can have one positional RpcModel parameter (or none) plus any number of injected services.

You can choose a more descriptive wire name without changing the Python function name:

@browser.method("history.back")
async def go_back(navigation: Inject[BrowserNavigation]) -> None:
    await navigation.back()

Use it from your application

For a custom transport, create the same channel directly and give its server a resolver for your own services. The core library does not prescribe a web framework or DI container.

from pyrpckit import RpcChannel


channel = RpcChannel(name="browser", namespace="browser")


@channel.method()
async def ping() -> str:
    return "pong"


server = channel.server(resolver=resolver)
response = await server.handle_json(request_body)

handle_json() takes a JSON request (or batch) and returns JSON ready to send back. If your transport already decoded the request, use handle() instead.

Each call gets its own dependency scope by default. This makes request-scoped resources such as database sessions simple to clean up. Keep work that outlives a call in a service designed to own that longer lifetime.

Send typed events

For server-initiated updates, declare an event source alongside the methods it belongs to. The WebSocket runtime starts it once per connection and sends each yielded value as a JSON-RPC notification.

from collections.abc import AsyncIterator
from typing import Literal


class PageChanged(RpcModel):
    type: Literal["page.changed"] = "page.changed"
    url: str


@browser.event("event", payload=PageChanged)
async def page_events(
    events: Inject[BrowserEvents],
) -> AsyncIterator[PageChanged]:
    async with events.subscribe() as stream:
        async for event in stream:
            yield PageChanged(url=event.url)

Event payloads are validated before they are sent and are included in the generated client types. Literal type fields also become discriminated event unions in supported clients.

Create a contract and clients

The OpenRPC document is the portable description of your API. Declare public WebSocket URLs explicitly, including any deployment prefixes. Keep them in sync with your FastAPI endpoints; URL variable names are preserved exactly.

from pyrpckit import RpcContract, ServerVariable


contract = RpcContract.from_channels(
    channels=[browser],
    title="Browser API",
    server_urls={
        "browser-control": "wss://api.example.com/sessions/{sessionId}/control",
    },
    variables={
        "sessionId": ServerVariable(default="demo-session"),
    },
)

pyrpckit schema browser.api:contract --output schema/browser.openrpc.json
pyrpckit generate schema/browser.openrpc.json \
  --language python \
  --output src/browser_client \
  --package browser_client \
  --client-name BrowserClient

Use --language typescript to generate a TypeScript client from the same document.

FastAPI and Dishka

Use a normal FastAPI router and one WebSocket endpoint per channel. FastAPI resolves path parameters and dependencies before serve() accepts the socket. Pass an object or a mapping of dependency types to values as context; these values and the WebSocket are injectable into methods and events on that socket. FastAPI owns cleanup of its dependencies, including dependencies using yield.

from dataclasses import dataclass

from fastapi import APIRouter, Depends, FastAPI, WebSocket

from pyrpckit import Inject, RpcChannel
from pyrpckit.fastapi import serve


@dataclass(frozen=True)
class BrowserConnection:
    session_id: str
    user_id: str


router = APIRouter(prefix="/sessions/{session_id}")
control = RpcChannel(name="control", namespace="browser")


async def browser_connection(
    websocket: WebSocket,
    session_id: str,
    user: User = Depends(current_user),
) -> BrowserConnection:
    return BrowserConnection(session_id=session_id, user_id=user.id)


@control.method()
async def current_url(connection: Inject[BrowserConnection]) -> str:
    return await lookup_url(connection.session_id)


@router.websocket("/control")
async def control_endpoint(
    websocket: WebSocket,
    context: BrowserConnection = Depends(browser_connection),
) -> None:
    await serve(control, websocket, context=context, resolver=resolver)


app = FastAPI()
app.include_router(router)

If you use Dishka, pass its adapter as the resolver:

from pyrpckit.dishka import DishkaResolver
from pyrpckit.fastapi import serve


@router.websocket("/dishka-control")
async def dishka_endpoint(websocket: WebSocket) -> None:
    await serve(control, websocket, resolver=DishkaResolver(container))

Each socket opens a Dishka SESSION; individual calls open REQUEST scopes. The core package has no FastAPI or Dishka dependency.

serve() also accepts error_mapper, max_concurrency (default 32), max_queue_size (default 128), and subprotocol. Call it on an unaccepted socket. For contracts, declare matching subprotocols with RpcContract.from_channels(..., subprotocols={"control": "jsonrpc"}). Contract export freezes channel definitions; all channels must share a protocol version. Configure tags and version directly on each RpcChannel.

Migrating from RpcAPIRouter: replace it with APIRouter, create channels with RpcChannel, and register explicit endpoints calling serve(). Replace @router.connection() / @channel.connection() with ordinary FastAPI dependencies and pass their results as context. Replace router.contract() with RpcContract.from_channels() and explicit server_urls.

Errors

Declare the errors a caller can handle, then raise them naturally in the handler:

from pyrpckit import RpcError


class PageNotFound(RpcError):
    code = -32004
    message = "Page not found"


class PageId(RpcModel):
    id: str


@browser.method(errors=(PageNotFound,))
async def get_page(
    params: PageId,
    navigation: Inject[BrowserNavigation],
) -> Page:
    page = await navigation.get(params.id)
    if page is None:
        raise PageNotFound()
    return Page(title=page.title, url=page.url)

Known errors become clear JSON-RPC responses and are recorded in the contract. Invalid requests, unknown methods, invalid parameters, and unexpected failures are handled as standard JSON-RPC errors. Notifications do not receive a reply.

Development

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