pidriver 0.1.0

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 default CLI is upstream pi (the pi binary); the oh-my-pi fork (omp) also works (set binary="omp", mode="rpc-ui"). Which one runs is just a PiConfig field, 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+ (no third-party runtime deps).
• A pi-compatible CLI — but you usually don't install one manually: pidriver can auto-install upstream pi via npm on first use (requires Node ≥22.19 + npm). To use an existing CLI instead, put pi (or omp) on PATH, or point PiConfig.binary / install_dir at it.
• 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(
        provider="openai",        # binary defaults to upstream "pi" (--mode rpc)
        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, mode & workspace

Field	Default	Purpose
binary	"pi"	Which pi-compatible CLI to launch — set to "omp" for oh-my-pi.
mode	"rpc"	Output mode → --mode. Upstream pi accepts text/json/rpc; rpc is the JSONL protocol this driver speaks. (rpc-ui is omp-only — set it explicitly when binary="omp".)
workspace	None	Project directory; becomes the subprocess cwd.

Binary auto-install — see Auto-install.

Field	Default	Purpose
install_dir	None	Where to npm install/look for a managed pi binary.
auto_install	True	npm install pi into install_dir if the binary can't be resolved.
npm_package	"@earendil-works/pi-coding-agent"	The npm package that provides pi.
pi_version	"0.78.1"	Pinned pi version to install (overridable).
extensions	()	Add-on sources (npm:…, git:…, path) installed via pi install before start.

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.

Approvals — auto_approve (--auto-approve, run tools without asking), approval_mode (--approval-mode <mode>, e.g. "always-ask"). Leave both unset to let your InteractionHandler decide.

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 (--system-prompt), append_system_prompt (--append-system-prompt, repeatable).

Isolation — see Isolation: home_dir, agent_dir, session_dir, package_dir, no_extensions, no_skills, no_context_files, no_prompt_templates, no_themes, no_mcp, offline, skip_version_check, telemetry, 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", mode, ...].
• 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 HOME. home_dir overrides HOME for the child. This is the load-bearing isolation lever: omp derives its config root and logs from $HOME/.omp, so agent_dir alone isn't enough — without home_dir the agent still writes logs into your real ~/.omp (verified empirically against omp 15.7.3). Set home_dir and your personal setup stays byte-for-byte untouched.
• Private state dirs. agent_dir → PI_CODING_AGENT_DIR (config/extensions/skills), session_dir → PI_CODING_AGENT_SESSION_DIR (session .jsonls), package_dir → PI_PACKAGE_DIR (installed packages) — each points pi at a private location instead of the defaults under ~/.omp.
• No host discovery (all default True): no_extensions, no_skills, no_context_files, no_prompt_templates, no_themes give a clean, reproducible agent.
• No MCP autoconnect (no_mcp=True, default). Under rpc-ui the agent otherwise auto-connects MCP servers it finds in the environment; no_mcp strips those trigger vars.
• offline (--offline + PI_OFFLINE=1), skip_version_check (PI_SKIP_VERSION_CHECK=1, default True) and telemetry=False (PI_TELEMETRY=0) round out a hermetic run.

home = Path("/var/lib/myapp/pi")
cfg = PiConfig(
    provider="anthropic", model="anthropic/claude-haiku-4-5", api_key=KEY,
    workspace="/srv/projects/acme",
    home_dir=home / "home",          # overrides $HOME — the key isolation lever
    agent_dir=home / "agent",        # PI_CODING_AGENT_DIR
    session_dir=home / "sessions",   # PI_CODING_AGENT_SESSION_DIR
    package_dir=home / "packages",   # PI_PACKAGE_DIR
    # inherit_env=False, no_* discovery, no_mcp, telemetry=False are already the defaults.
)

Auto-install

You don't have to install a CLI yourself. Upstream pi is an npm package, and pidriver can npm install it into a managed, isolated prefix on demand (requires Node ≥22.19 + npm on PATH — the pi bin is a Node shim, not a self-contained binary).

When the transport starts, it resolves the binary in order — explicit binary path → install_dir/bin/pi → PATH. If none resolves and auto_install=True (default), it runs npm install -g --prefix <install_dir> <npm_package>@<pi_version> and runs <install_dir>/bin/pi.

from pidriver import ensure_binary, install_pi, ensure_extensions  # also exported for explicit use

Extensions / skills / packages

Set PiConfig.extensions to a tuple of add-on sources and pidriver runs pi install <source> for each (into the isolated config dir, via PI_CODING_AGENT_DIR) before the agent starts — idempotently (sources already in the dir's settings.json are skipped; failures are logged, not fatal). Each source is one of npm:<pkg>[@version], git:github.com/<owner>/<repo>[@ref], an https:///ssh:// git url, or a local ./path. To actually load them, turn the relevant isolation flags off (e.g. no_extensions=False, no_skills=False).

PiConfig(extensions=("npm:pi-web-access", "git:github.com/owner/repo"),
         no_extensions=False, no_skills=False)

Notes:

• First npm install takes a few seconds; subsequent runs reuse the managed copy.
• Failures raise PiInstallError (missing npm/Node, install or version-verify failure).
• To skip auto-install, set auto_install=False and provide binary/install_dir pointing at an existing CLI.

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/omp executable can't be found or run.
PiInstallError	Auto-install failed (unsupported OS/arch, download/verify error).
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