tw2_jsonrpc

tw2_jsonrpc is the Python sibling of the Rust tw2-jsonrpc experiment: one bidirectional JSON-RPC Peer, async-first transport code, named params only, and a small decorator API for registering local methods.

This package is peer-first: connect over TCP, attach to stdio, connect over WebSocket, launch a child process that speaks over stdio, or accept TCP and WebSocket peers with a server registry.

Client Call

import tw2_jsonrpc

async with await tw2_jsonrpc.connect_tcp("127.0.0.1", 8080) as peer:
    result = await peer.call("math.add", a=2, b=3)

Params are named only. Pass a dict or omit params:

await peer.call("math.add", {"a": 2, "b": 3})
await peer.call("math.add", a=2, b=3)
await peer.call("health")

Array params are rejected with InvalidParams.

Transports

All transports return a Peer. After that, calls and notifications feel the same.

peer = await tw2_jsonrpc.connect_tcp("127.0.0.1", 8080)
peer = await tw2_jsonrpc.connect_stdio()
peer = await tw2_jsonrpc.connect_ws("ws://127.0.0.1:8080/rpc")

Subprocesses

Use start_subprocess when the library should launch and own a child process that speaks newline-delimited JSON-RPC over stdio. This is stdio JSON-RPC with child-process lifecycle management, not a separate transport protocol:

peer = await tw2_jsonrpc.start_subprocess(["worker", "--stdio"])

start_process remains available as an alias.

Server

A Server owns methods that should be installed on every peer it creates. peer_cls is chosen at the transport boundary:

server = tw2_jsonrpc.Server()

@server.method("math.add")
async def add(a: int, b: int) -> int:
    return a + b

handle = await server.serve_tcp("127.0.0.1", 8080, peer_cls=tw2_jsonrpc.TcpPeer)
await handle.serve_forever()

Peer subclasses can still carry per-peer state and methods:

class AppPeer(tw2_jsonrpc.TcpPeer):
    @tw2_jsonrpc.method("session.ping")
    async def ping(self) -> str:
        return "pong"

handle = await server.serve_tcp("127.0.0.1", 8080, peer_cls=AppPeer)

Use a handler for per-connection setup:

async def on_peer(peer: AppPeer) -> None:
    @peer.method("session.echo")
    async def echo(text: str) -> str:
        return text

handle = await server.serve_tcp(
    "127.0.0.1",
    8080,
    peer_cls=AppPeer,
    handler=on_peer,
)

WebSocket serving and subprocess peers use the same server registry:

handle = await server.serve_ws("127.0.0.1", 8080, peer_cls=tw2_jsonrpc.WebSocketPeer)
peer = await server.connect_subprocess(["worker", "--stdio"], peer_cls=tw2_jsonrpc.SubprocessPeer)

The free serve_tcp and serve_ws helpers remain available when you do not need server-wide methods.

Peer Handlers

Runtime registration is available directly on a peer:

peer = await tw2_jsonrpc.connect_stdio()

@peer.method("math.add")
async def add(a: int, b: int) -> int:
    return a + b

The method name can default to the function name:

@peer.method
async def health() -> dict:
    return {"ok": True}

Peer subclasses can expose decorated methods:

class App(tw2_jsonrpc.StdioPeer):
    @tw2_jsonrpc.method("tools.echo")
    async def echo(self, text: str) -> str:
        return text

Subclass handlers receive self as the current peer, so callbacks and notifications can use the same object:

class App(tw2_jsonrpc.StdioPeer):
    @tw2_jsonrpc.method("callback.demo")
    async def callback_demo(self, name: str) -> dict:
        await self.notify("callback.ready", {"name": name})
        return {"ok": True}

You can also register or remove handlers explicitly:

peer.add_method(echo, "tools.echo")
peer.remove_method("tools.echo")

Notifications

Notifications are fire-and-forget. No result is returned.

await peer.notify("log.info", message="started")

Handlers live in the same method table. If the incoming JSON-RPC message omits id, no response is sent, even if the handler returns a value.

For readability, notification is available as an alias for method registration:

@peer.notification("log.info")
async def log_info(message: str) -> None:
    print(message)

Errors

Remote JSON-RPC error responses become JsonRpcError subclasses:

try:
    result = await peer.call("math.add", {"a": 1, "b": 2})
except tw2_jsonrpc.InvalidParams:
    ...
except tw2_jsonrpc.JsonRpcError:
    ...

Handlers can raise structured JSON-RPC errors:

raise tw2_jsonrpc.InvalidParams("field 'name' is required")

Local timeout and cancellation stay local:

result = await peer.call("slow.method", {"x": 1}, timeout=5.0)

If a call is cancelled or times out, the pending local waiter is removed. JSON-RPC 2.0 has no standard remote cancellation, so a later response for that id is ignored.

Mental Model

• Peer is a bidirectional JSON-RPC connection.
• await peer.call(...) sends a request and returns the decoded result.
• await peer.notify(...) sends a fire-and-forget message.
• @peer.method registers a request handler.
• Server registers methods shared by every peer it accepts or starts.
• @peer.notification is a readability alias for notification handlers.
• Only named params are supported.
• The shape is intentionally close to tw2-jsonrpc, but Python does not need a builder or proc macro.