pidriver 0.0.2

Async Python driver for the pi coding agent (pi --mode rpc), built for isolated, headless project work.

pidriver

Async Python driver for the pi coding agent, built for isolated, headless project work. It drives any pi-compatible CLI in --mode rpc over its JSONL stdin/stdout protocol and exposes a small, typed Python API — with no third-party dependencies.

The recommended CLI is oh-my-pi (the omp binary); the original pi also works. Which one runs is just a PiConfig field (binary), so the same code drives either.

Embed an agent inside another application (a chat bot, a scheduler, a service) and have it develop a project autonomously, in an environment that is isolated from your personal pi configuration and secrets.

Start with PiClient for typed events and interaction handling; drop to the raw transport only when you need it.

Requirements

• Python 3.12+

• A pi-compatible CLI on PATH. Recommended — oh-my-pi (the omp binary) via Bun, or the official installer:

  bun install -g @oh-my-pi/pi-coding-agent   # provides the `omp` binary
  # or: curl -fsSL https://omp.sh/install | sh
  omp --version
  

  The original pi CLI works too: npm i -g @earendil-works/pi-coding-agent. Tell PiConfig which one to launch with binary ("omp" or "pi"); they share the --mode rpc protocol.

• A provider API key — passed explicitly through PiConfig, see Isolation.

Install

uv add pidriver          # once published
# or, from a checkout:
uv pip install -e ".[dev]"

Quick start

Configure a client once, then start() a session per task and iterate its typed events:

import asyncio, os
from pidriver import PiClient, PiConfig, AutoApprove, MessageDelta, ToolStart, AgentEnd

async def main():
    client = PiClient(PiConfig(
        binary="omp",             # which CLI to launch (oh-my-pi); "pi" also works
        provider="openai",
        model="gpt-4o",
        api_key=os.environ["OPENAI_API_KEY"],
        # Isolation defaults (scrubbed env, no host extensions/skills) are already on.
    ))

    # AutoApprove answers permission prompts itself, so the run is unattended.
    session = await client.start(
        "List the files here and summarize the project.",
        cwd="/srv/projects/acme",         # the project the agent works in
        interaction_handler=AutoApprove(),
    )
    async with session:
        async for event in session:
            match event:
                case MessageDelta(text=text):
                    print(text, end="", flush=True)
                case ToolStart(name=name, arguments=args):
                    print(f"\n[tool] {name} {args}")
                case AgentEnd(reason=reason):
                    print(f"\n[done: {reason}]")

asyncio.run(main())

Need the raw protocol instead? Drive SubprocessTransport directly — see examples/basic_prompt.py.

Architecture

pi --mode rpc speaks line-delimited JSON on stdin/stdout. pidriver is layered so each concern is swappable and testable in isolation:

   PiConfig ──► SubprocessTransport ──► pi --mode rpc
   (argv+env)   (JSONL framing only)    (child process)
        │              │
        │       raw JSON dicts (events + responses, undifferentiated)
        │              ▼
        │       PiClient/PiSession ── typed events, id-correlated commands,
        │                              interaction handling  ── respond()
        ▼
   PiSessionManager  ── registry · idle reaper · stop_all() over many sessions

Module	Role
pidriver.config	PiConfig — isolation knobs, to_argv() / to_env()
pidriver._transport	PiTransport protocol + SubprocessTransport (the swap boundary)
pidriver.events	the typed Event hierarchy + parse_event
pidriver.interaction	interaction handlers/policies (AskHost, AutoApprove, …)
pidriver.session	PiSession (back-compat alias AgentSession) — the live RPC session
pidriver.client	PiClient — starts sessions
pidriver.manager	PiSessionManager — registry, idle reaper, stop_all()
pidriver.usage	UsageTotals — token/cost accounting
pidriver.errors	exception hierarchy

Everything below is re-exported from the package root: from pidriver import ....

PiConfig

The single source of truth for how a pi process is spawned — binary, provider/model, tools, session, and isolation. Immutable (frozen dataclass). The transport consumes two derived outputs: to_argv() (the command line) and to_env() (the child environment).

PiConfig(provider="anthropic", model="claude-sonnet-4-6",
         api_key="sk-...", workspace="/path/to/project")

Binary & workspace

Field	Default	Purpose
binary	"pi"	Which pi-compatible CLI to launch — set to "omp" for oh-my-pi.
workspace	None	Project directory; becomes the subprocess cwd.

Model / provider

Field	Default	Purpose
provider	None	--provider (e.g. "anthropic", "openai").
model	None	--model.
thinking	None	--thinking (off/minimal/low/medium/high/xhigh).
api_key	None	Secret, injected into the provider's env var by to_env().
api_key_env	None	Override the env var name api_key is injected under.

Prefer a fully-qualified model id (e.g. openai/gpt-4o, anthropic/claude-haiku-4-5) — bare fuzzy names can mis-route to the wrong backend.

Tools — tools (--tools allowlist), exclude_tools (--exclude-tools), no_tools, no_builtin_tools.

Session — session (False → --no-session), session_id, session_path (--session), session_name (--name), continue_session (--continue), system_prompt, append_system_prompt.

Isolation — see Isolation: agent_dir, session_dir, no_extensions, no_skills, no_context_files, no_prompt_templates, no_themes, offline, skip_version_check, inherit_env, env_passthrough.

Escape hatches — extra_args (appended to argv), extra_env (merged into the child env).

Methods:

• to_argv() -> list[str] — full command line, starting [binary, "--mode", "rpc", ...].
• to_env(base_environ=None) -> dict[str, str] — the child environment (scrubbed by default).
• api_key_env_name() -> str | None — which env var api_key resolves to (provider-mapped via PROVIDER_API_KEY_ENV, falling back to <PROVIDER>_API_KEY, then ANTHROPIC_API_KEY).

Isolation

This is the reason pidriver exists. By default a PiConfig keeps an agent that's developing a project from reading or mutating your global pi setup:

• Scrubbed environment (inherit_env=False, default). The child starts from an empty environment plus only env_passthrough (PATH, HOME, LANG, LC_ALL, TERM, TZ) — so a stray *_API_KEY in your shell never leaks into the agent. The configured api_key is then injected under exactly one provider variable.
• Private agent dir. agent_dir sets PI_CODING_AGENT_DIR, pointing pi at a private config directory instead of ~/.pi, so it loads no personal extensions/skills/settings.
• No host discovery (all default True): no_extensions, no_skills, no_context_files, no_prompt_templates, no_themes give a clean, reproducible agent.
• Separate session storage. session_dir (--session-dir) keeps session .jsonl files out of the default location.
• offline (--offline + PI_OFFLINE=1) and skip_version_check (PI_SKIP_VERSION_CHECK=1, default True) round out a hermetic run.

cfg = PiConfig(
    provider="anthropic", model="claude-sonnet-4-6", api_key=KEY,
    workspace="/srv/projects/acme",
    agent_dir="/var/lib/myapp/pi-home",     # private ~/.pi replacement
    session_dir="/var/lib/myapp/pi-sessions",
    extra_env={"HOME": "/var/lib/myapp"},   # see note below — also override HOME
    # inherit_env=False and no_* discovery flags are already the defaults.
)

Override HOME for full isolation. agent_dir/PI_CODING_AGENT_DIR alone is not enough: omp derives its log root from $HOME/.omp, and the default env_passthrough copies HOME from the host — so an agent can still write logs into your ~/.omp. To leave your personal setup byte-for-byte untouched, also point HOME at the private dir (via extra_env, as above) or drop it from env_passthrough. (Verified empirically against omp 15.7.3.)

Transport

SubprocessTransport(config, *, on_stderr=None, spawn=..., terminate_timeout=5.0)

A deliberately dumb transport: it spawns pi --mode rpc, writes JSON commands as JSONL lines, and yields parsed JSON dicts from stdout. It does not correlate request/response ids or interpret commands — that's the session layer's job. This thinness is what makes it the swap boundary (a future omp-rpc transport can satisfy the same PiTransport protocol).

await transport.start()                       # resolve pi binary + spawn
await transport.send({"type": "prompt", ...}) # write one JSONL command
obj = await transport.receive()               # next JSON dict, or None at EOF
async for obj in transport: ...               # iterate until EOF (single consumer)
await transport.aclose()                      # close stdin, SIGTERM→SIGKILL, reap
transport.pid, transport.returncode           # process introspection

Contract: exactly one consumer iterates at a time; receive() returns None at EOF; send/receive raise TransportClosedError before start() or after the process exits. on_stderr receives pi's diagnostic lines (never interleaved into the dict stream). Framing splits on \n only (a trailing \r is stripped), never on U+2028/U+2029, so JSON strings containing those survive intact.

PiTransport is the @runtime_checkable Protocol that SubprocessTransport implements — type against it and inject a fake transport in tests.

Session manager

PiSessionManager(factory=None, *, idle_timeout=1800.0, reap_interval=60.0, max_sessions=None, clock=...)

A registry + idle reaper + bulk shutdown for many concurrent sessions. Decoupled from the concrete session class — it depends only on the structural ManagedSession protocol (session_id, last_activity, aclose()).

async with PiSessionManager(idle_timeout=1800) as mgr:
    await mgr.register(session)        # add an already-started session
    # or: await mgr.create(...)        # build via the injected factory
    mgr.get(session_id)                # -> session (raises SessionNotFoundError)
    mgr.ids(); len(mgr); sid in mgr    # introspection
    await mgr.stop(session_id)         # close + drop one
    await mgr.stop_all()               # close all concurrently
# context exit stops the reaper and tears everything down

The background reaper closes sessions idle past idle_timeout (set None to disable); max_sessions caps the registry (register/create raise RuntimeError when exceeded).

Usage accounting

UsageTotals

An immutable token + cost tally that normalizes pi's two usage shapes into one addable value:

from pidriver import UsageTotals

a = UsageTotals.from_session_stats(stats_data)      # get_session_stats response
b = UsageTotals.from_assistant_usage(msg["usage"])  # per-message usage block
total = a + b                                        # aggregate with +
total.total_tokens          # input+output+cache_read+cache_write
total.with_cost(0.42)       # copy with cost replaced (stats may lack cost)
total.as_dict()             # plain dict for logging/serialization

Fields: input_tokens, output_tokens, cache_read_tokens, cache_write_tokens, cost_usd.

Errors

All exceptions derive from PiDriverError, so one except catches the family:

Exception	Raised when
PiDriverError	Base class for everything below.
PiNotFoundError	The pi executable can't be found or run.
PiStartError	The pi --mode rpc process failed to start.
PiProtocolError	A line from pi wasn't valid JSON / violated the contract (.raw).
PiCommandError	An RPC command returned success: false (.command, .error, .data).
PiTimeoutError	A response or awaited condition didn't arrive in time.
TransportClosedError	An op was attempted on a closed/exited transport.
SessionNotFoundError	A session id isn't registered with the manager (.session_id).

High-level API: PiClient / PiSession

The transport yields raw, undifferentiated JSON. The session layer adds the ergonomic surface this library is ultimately for: typed events, an auto-answer policy for the agent's questions, and a session object that plugs straight into PiSessionManager.

PiClient

client = PiClient(config: PiConfig, *,
                  transport_factory=None,            # build a transport from a config (tests)
                  interaction_handler=None)          # default handler for sessions it starts

session = await client.start(
    task: str, *,
    cwd: str | Path | None = None,        # project dir; overrides config.workspace
    config: PiConfig | None = None,       # per-session config override
    interaction_handler: InteractionHandler | None = None,   # default: client's, else AskHost
    resume: str | None = None,            # prior pi session path/id → --continue
    session_id: str | None = None,        # stable registry key (auto uuid4 if omitted)
    send_initial: bool = True,            # False → open without sending `task` yet
) -> PiSession                             # AgentSession is a back-compat alias of the same class

start() spawns the subprocess and (by default) sends task as the first prompt. Its signature is exactly what PiSessionManager(factory=...) expects, so you can wire them together:

manager = PiSessionManager(factory=client.start)
session = await manager.create("Add a healthcheck endpoint", cwd="/srv/proj")

PiSession

The live session (AgentSession is a back-compat alias for the same class) — async-iterate it for events, command it with the methods below. It's an async context manager (async with session: guarantees the subprocess is reaped) and satisfies ManagedSession, so it drops into PiSessionManager.

Member	Description
async for event in session / session.events()	Yields typed Event objects. Consuming twice raises.
await prompt(message)	Send a follow-up user turn (completes at the next AgentEnd).
await respond(request_id, value)	Answer an interaction (see below).
await cancel()	Interrupt the current turn (keeps the process alive).
await aclose()	Idempotently terminate the session and reap the subprocess.
session.pending	The current unanswered InteractionRequest, or None.
session.usage	Running UsageTotals accumulated from inline usage blocks.
session.ended	True once an AgentEnd has been seen.
session.session_id	Stable local id (registry key).
session.pi_session_id	The id pi reports in agent_start (pass to resume).

Events

Each RPC record maps to a frozen dataclass via parse_event. Unknown event types degrade to a bare Event (original payload in .raw) instead of raising — a new pi/omp release never crashes the driver. Every event carries .type (the raw tag) and .raw (the decoded dict).

Event	Key fields	Meaning
AgentStart	session_id, model, cwd	The agent run has begun.
MessageDelta	text, channel	A streaming fragment of assistant (or thinking) text.
MessageComplete	text, channel	A full message at a turn boundary.
ToolStart	tool_call_id, name, arguments	The agent invoked a tool.
ToolUpdate	tool_call_id, name, output	Partial output while a tool runs.
ToolEnd	tool_call_id, name, result, is_error	A tool finished.
Usage	totals (a UsageTotals)	Token/cost for a step; also folded into session.usage.
AgentEnd	reason, final_text	The run finished (completed / cancelled / error / limit).
Error	message, code, fatal	An error (or a failed command ack) surfaced by pi.
InteractionRequest	request_id, kind, prompt, options, tool_call_id, default	The agent is blocked waiting for the host.
Event	type, raw	Base / fallback for unrecognized records.

channel is a Channel enum (ASSISTANT / THINKING).

Answering the agent

When the agent needs the host it emits an InteractionRequest whose kind is an InteractionKind:

InteractionKind	What the agent wants	Answer value
PERMISSION	Approve/deny a gated tool call (tool_call_id set)	a Decision, or True/False
QUESTION	A free-text answer	a str
CHOICE	Pick from options	the chosen str, or its int index

Decision values: ALLOW, ALLOW_ALWAYS, DENY. session.respond() takes the request id and coerces the value:

await session.respond(req.request_id, Decision.ALLOW)   # explicit decision
await session.respond(req.request_id, True)             # bool → allow / deny
await session.respond(req.request_id, "use postgres")   # free-text answer
await session.respond(req.request_id, 0)                # choice by index

An InteractionHandler — any async (request, session) -> InteractionResponse | None — can answer requests automatically. Returning None defers to the host (the request still surfaces through the event stream). Built-ins:

Handler	Behavior
AskHost()	Default. Never auto-answers — every request surfaces to your loop.
AutoApprove(allow=None, *, always=False)	Auto-approves PERMISSION (optionally filtered by an allow(request) predicate; always=True sends ALLOW_ALWAYS). Questions/choices are deferred — no safe default.
DenyAll()	Denies every PERMISSION; defers other kinds. A read-only sandbox.
chain(h1, h2, ...)	Composes handlers; first non-None response wins, else defers.

Build responses directly with InteractionResponse(req.request_id, value), InteractionResponse.allow(req.request_id, always=True), or .deny(req.request_id).

Surfacing permission prompts. omp defaults to --approval-mode yolo (auto-approve everything), so it won't emit PERMISSION requests at all. To make a human-in-the-loop (AskHost) flow meaningful, run the CLI in a stricter mode — pass --approval-mode write (or always-ask) via PiConfig.extra_args.

Interaction examples

AskHost — human in the loop. The default handler defers everything, so each request surfaces in your loop and you respond():

from pidriver import (
    PiClient, PiConfig, AskHost,
    MessageDelta, InteractionRequest, InteractionKind, Decision, AgentEnd,
)

client = PiClient(PiConfig(binary="omp", provider="openai", api_key=KEY))
session = await client.start("Refactor utils.py", cwd=project, interaction_handler=AskHost())

async with session:
    async for event in session:
        match event:
            case MessageDelta(text=text):
                print(text, end="", flush=True)

            case InteractionRequest(kind=InteractionKind.PERMISSION) as req:
                ok = input(f"\nAllow? {req.prompt} [y/N] ").lower() == "y"
                await session.respond(req.request_id, Decision.ALLOW if ok else Decision.DENY)

            case InteractionRequest(kind=InteractionKind.QUESTION) as req:
                await session.respond(req.request_id, input(f"\n{req.prompt} "))

            case InteractionRequest(kind=InteractionKind.CHOICE) as req:
                for i, opt in enumerate(req.options):
                    print(f"  {i}. {opt}")
                await session.respond(req.request_id, int(input("> ")))   # answer by index

            case AgentEnd(reason=reason):
                print(f"\n[{reason}]")

(AskHost is the default, so you can omit interaction_handler= for this behavior.)

AutoApprove — autonomous. The handler answers permission prompts itself, so an unattended run just consumes output. Free-text questions and choices are still deferred — if your task might trigger them, handle InteractionRequest in the loop too, or combine handlers with chain:

from pidriver import PiClient, PiConfig, AutoApprove, DenyAll, chain, AgentEnd

session = await client.start("Run the test suite and fix failures", cwd=project,
                             interaction_handler=AutoApprove())     # or AutoApprove(always=True)
async with session:
    async for event in session:
        if isinstance(event, AgentEnd):
            print(event.reason)

# Approve only safe (read-only) tools, deny the rest:
handler = chain(
    AutoApprove(allow=lambda r: "read" in r.prompt.lower()),
    DenyAll(),
)

Development

uv pip install -e ".[dev]"
uv run pytest
uv run mypy src
uv run ruff check

License

MIT © Grigory Bakunov