Metadata-Version: 2.4
Name: pidriver
Version: 0.0.2
Summary: Async Python driver for the pi coding agent (pi --mode rpc), built for isolated, headless project work.
Project-URL: Homepage, https://github.com/bobuk/pidriver
Project-URL: Repository, https://github.com/bobuk/pidriver
Author-email: Grigory Bakunov <thebobuk@gmail.com>
License: MIT
License-File: LICENSE
Keywords: automation,coding-agent,llm,pi,rpc
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Typing :: Typed
Requires-Python: >=3.12
Provides-Extra: dev
Requires-Dist: mypy>=1.11; extra == 'dev'
Requires-Dist: pytest-asyncio>=0.24; extra == 'dev'
Requires-Dist: pytest>=8; extra == 'dev'
Requires-Dist: ruff>=0.6; extra == 'dev'
Description-Content-Type: text/markdown

# 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**](https://bun.sh) (the `omp` binary); the original
[`pi`](https://www.npmjs.com/package/@earendil-works/pi-coding-agent) 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`](#high-level-api-piclient--pisession) for typed events and interaction
handling; drop to the [raw transport](#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](https://bun.sh), or the official installer:

  ```sh
  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](#isolation).

## Install

```sh
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:

```python
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`](#transport) directly — see
[`examples/basic_prompt.py`](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).

```python
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](#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.

```python
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).

```python
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()`).

```python
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:

```python
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`

```python
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:

```python
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`](#events) 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](#answering-the-agent)). |
| `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:

```python
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()`:

```python
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`:

```python
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

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

## License

MIT © Grigory Bakunov
