oh-my-pi/python/omp-rpc

omp-rpc

Typed Python bindings for the omp --mode rpc protocol used by the coding agent.

This package wraps the newline-delimited JSON RPC transport exposed by the CLI and provides:

• typed command methods for the stable RPC surface
• typed startup options for common omp --mode rpc flags such as thinking level, tool selection, prompt appends, provider session IDs, and headless session toggles
• typed protocol models for state, bash results, compaction, and session stats
• a process-backed client that manages request correlation over stdio
• typed per-event listeners plus a typed catch-all notification hook
• helpers for collecting prompt runs and handling extension UI requests in manual or headless mode
• typed host-tool helpers so Python RPC owners can expose custom tools with JSON Schema metadata

Basic Usage

from omp_rpc import RpcClient

with RpcClient(provider="anthropic", model="claude-sonnet-4-5") as client:
    state = client.get_state()
    print(state.model.id if state.model else "no model")

    turn = client.prompt_and_wait("Reply with just the word hello")
    print(turn.require_assistant_text())

The wrapper also exposes the common RPC startup flags directly, so scripts do not need to build extra_args by hand:

from omp_rpc import RpcClient

with RpcClient(
    model="openrouter/anthropic/claude-sonnet-4.6",
    thinking="high",
    no_session=True,
    no_skills=True,
    no_rules=True,
    tools=("read", "edit", "write"),
    append_system_prompt="Focus on reproducible benchmark behavior.",
) as client:
    print(client.get_state().thinking_level)

For orchestration hosts, the wrapper also exposes typed event hooks and a simple way to seed todos before the first prompt:

from omp_rpc import MessageUpdateEvent, RpcClient

def on_message_update(event: MessageUpdateEvent) -> None:
    assistant_event = event.assistant_message_event
    if assistant_event.get("type") == "text_delta":
        print(assistant_event["delta"], end="", flush=True)

with RpcClient(model="openrouter/anthropic/claude-sonnet-4.6", no_session=True) as client:
    client.on_message_update(on_message_update)
    client.set_todos(
        [
            "Map the read and edit tool surface.",
            "Exercise the supported edit paths.",
            "Write concrete findings and gaps.",
        ]
    )
    client.prompt_and_wait("Evaluate the current tool behavior.")

set_todos() accepts either a flat list of todo strings/items or explicit phases, and get_state().todo_phases returns the typed current todo state.

By default the client runs:

omp --mode rpc

You can also point it at a custom command, which is useful inside this repo while developing against the Bun entrypoint:

from omp_rpc import RpcClient

with RpcClient(
    command=[
        "bun",
        "packages/coding-agent/src/cli.ts",
        "--mode",
        "rpc",
        "--provider",
        "anthropic",
        "--model",
        "claude-sonnet-4-5",
    ],
) as client:
    print(client.get_state().session_id)

Host-Owned Custom Tools

RPC hosts can expose custom tools to the agent with JSON Schema metadata. The Python helper keeps the wire format simple while still giving the handler a typed signature:

from typing import TypedDict

from omp_rpc import RpcClient, host_tool


class EchoArgs(TypedDict):
    message: str


def echo_host(args: EchoArgs, context) -> str:
    context.send_update(f"working:{args['message']}")
    return f"host:{args['message']}"


with RpcClient(
    no_session=True,
    custom_tools=(
        host_tool(
            name="echo_host",
            description="Echo a value from the Python host",
            parameters={
                "type": "object",
                "properties": {"message": {"type": "string"}},
                "required": ["message"],
                "additionalProperties": False,
            },
            execute=echo_host,
        ),
    ),
) as client:
    client.prompt_and_wait("Use the echo_host tool with the value hello")

If you want runtime conversion into a richer Python type, pass decode= to host_tool(...). That lets you keep the JSON Schema contract on the wire while parsing the incoming argument object into a dataclass or model in the handler.

Extension UI Requests

Extensions in RPC mode can ask the host for input. Those requests are available as typed ExtensionUiRequest instances:

request = client.next_ui_request(timeout=5.0)

if request.method == "confirm":
    client.send_ui_confirmation(request.id, True)
elif request.method in {"input", "editor"}:
    client.send_ui_value(request.id, "approved")

For non-interactive scripts, you can install a default headless policy instead of handling every request manually:

with RpcClient(model="anthropic/claude-sonnet-4-5") as client:
    client.install_headless_ui()
    turn = client.prompt_and_wait("needs ui-safe automation")
    print(turn.assistant_text)

That helper ignores passive UI notifications (notify, setStatus, setWidget, setTitle, set_editor_text), answers confirm with False, and cancels select/input/editor requests unless you provide explicit values.

Error Handling and Retained History

The client now surfaces more of the transport edge cases that the wire protocol allows:

• id-less parse and unknown-command failures are correlated back to the waiting request when they can be matched unambiguously
• late prompt / abort_and_prompt scheduling failures cause prompt_and_wait() and wait_for_idle() to raise instead of timing out
• unmatched background error responses are exposed through client.protocol_errors and client.on_protocol_error(...)
• listener exceptions no longer kill the stdout reader thread; they are exposed through client.listener_errors and client.on_listener_error(...)

For long-lived hosts, retained event and stderr history is bounded by default:

from omp_rpc import RpcClient

with RpcClient(max_event_history=20_000, max_stderr_chunks=256) as client:
    ...

If a single prompt streams more events than max_event_history allows, prompt_and_wait() raises a clear error so hosts can increase the limit instead of silently losing earlier events.

Prompt lifecycle collection is intentionally single-flight. Only one of prompt_and_wait(), wait_for_idle(), or collect_events() may be active at a time on a client instance. If a host needs concurrent orchestration, use separate RpcClient instances instead of overlapping lifecycle waiters on one session.

Text Helpers

assistant_text() and message_text() now return visible text blocks only. If a host explicitly needs reasoning text too, use the *_with_thinking helpers:

from omp_rpc import assistant_text, assistant_text_with_thinking

visible = assistant_text(message)
full = assistant_text_with_thinking(message)

Protocol Reference

The canonical wire protocol still lives in the repo at docs/rpc.md.