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 upstreampivianpmon first use (requires Node ≥22.19 + npm). To use an existing CLI instead, putpi(oromp) onPATH, or pointPiConfig.binary/install_dirat 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.87.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 varapi_keyresolves to (provider-mapped viaPROVIDER_API_KEY_ENV, falling back to<PROVIDER>_API_KEY, thenANTHROPIC_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 onlyenv_passthrough(PATH,HOME,LANG,LC_ALL,TERM,TZ) — so a stray*_API_KEYin your shell never leaks into the agent. The configuredapi_keyis then injected under exactly one provider variable. - Private
HOME.home_diroverridesHOMEfor the child. This is the load-bearing isolation lever: omp derives its config root and logs from$HOME/.omp, soagent_diralone isn't enough — withouthome_dirthe agent still writes logs into your real~/.omp(verified empirically against omp 15.7.3). Sethome_dirand 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_themesgive a clean, reproducible agent. - No MCP autoconnect (
no_mcp=True, default). Underrpc-uithe agent otherwise auto-connects MCP servers it finds in the environment;no_mcpstrips those trigger vars. offline(--offline+PI_OFFLINE=1),skip_version_check(PI_SKIP_VERSION_CHECK=1, default True) andtelemetry=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.
An existing managed copy is upgraded when its version differs from pi_version and
auto_install=True. A pi already found on PATH (or set via an absolute binary)
is used as-is; update that installation separately if needed.
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 installtakes 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=Falseand providebinary/install_dirpointing 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 emitPERMISSIONrequests at all. To make a human-in-the-loop (AskHost) flow meaningful, run the CLI in a stricter mode — pass--approval-mode write(oralways-ask) viaPiConfig.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(),
)
Tasks in remote Git repositories
Use start_repository_task with a trusted Git URL selected by the host. It clones the
repository, creates a per-job worktree, and starts pi in that worktree. No changes are
published until the host explicitly calls merge(); discard() deletes the local clone
without changing the remote. For SSH, pass a private-key path; otherwise Git uses the host's credentials.
from pathlib import Path
from pidriver import PiClient, PiConfig, start_repository_task
client = PiClient(PiConfig())
session, workspace = await start_repository_task(
client, "Fix the failing tests", "git+ssh://git@github.com/org/project.git",
root=Path("/srv/pi-jobs"), job_id="task-123",
ssh_key=Path("/srv/keys/project_ed25519"),
known_hosts=Path("/srv/keys/known_hosts"), # optional: pin host keys to this file
)
# Drain the session's events, then close it before finalizing the workspace.
async for event in session:
...
await session.aclose()
print(workspace.diff()) # review before deciding
state = workspace.to_state() # save for approval after a host restart if needed
# workspace = GitRepositoryWorkspace.from_state(state)
workspace.merge() # commits, merges and pushes to the cloned default branch
workspace.cleanup() # remove the local clone after successful push
# Or: workspace.discard() # no changes to origin
If the remote branch has moved, the push is rejected with MergeConflict; the local
workspace remains available for inspection. Never accept arbitrary repository URLs from
untrusted requests: the caller must enforce an allowlist. Empty repositories and a
second job with the same ID are rejected. SSH also accepts standard ssh:// and
git@host:path URLs. The key path (not its contents) is saved by to_state() for
later pushes; the key must still exist then. SSH uses strict host-key checking and
non-interactive authentication. Pass known_hosts=Path(...) to pin host keys to that
file alone (UserKnownHostsFile=<file>, GlobalKnownHostsFile=/dev/null; the path is
saved by to_state() too) — useful in containers where ssh's own ~/.ssh isn't yours
to manage; otherwise configure ssh's default known_hosts on the host in advance.
Workspace jail
Run an agent loose inside a project without letting it run anywhere on disk. resolve_project
maps a user-supplied name to a real directory only if it is explicitly allow-listed and lives
physically under root — rejecting path separators and symlink escapes.
from pidriver import resolve_project, list_candidate_projects, ProjectError
try:
cwd = resolve_project("myrepo", root=PROJECTS_ROOT, allowlist=("myrepo", "docs"))
except ProjectError as exc:
... # empty / not allow-listed / not a dir / escapes root (incl. via symlink)
# Offer a picker: every immediate sub-dir of a root, flagged git / selected.
projects = list_candidate_projects(PROJECTS_ROOT, allowlist=("myrepo",)) # list[ProjectDir]
Model discovery
list_models runs pi --list-models under the same isolation env the agent uses and parses the
table into typed ModelInfo rows; parse_models is the pure parser (handy for tests / cached
output). When a provider is set the result is narrowed to it.
from pidriver import list_models, parse_models
models = await list_models(config, provider="anthropic") # list[ModelInfo]
for m in models:
print(m.id, m.context) # "anthropic/claude-opus-4-8" "200K"
Provider OAuth login
Some providers authenticate via OAuth (Claude Pro/Max, Codex, …) instead of an API key. PiOAuthLogin
drives a bundled Node helper that reuses pi's own AuthStorage (no version skew), writing
credentials into the isolated <agent_dir>/auth.json. Requires node on PATH and a resolvable pi
install.
from pidriver import PiOAuthLogin
login = PiOAuthLogin(config) # config.agent_dir receives auth.json
info = await login.start("anthropic") # {"session", "url", "instructions"}
# user opens info["url"], authorizes, copies the code:
await login.submit_code(info["session"], code) # accepts a bare code or a callback URL
valid_extension_source(src) validates a pi install argument, and resolve_node_modules(config)
locates pi's bundled Node libraries (package identity from config.npm_package, so the omp fork
works too) — both exported for hosts building their own extension/auth tooling.
Web instance service
The web subsystem is an optional layer over the same driver core. pidriver stays
dependency-free by default; install the web extra only for the ASGI server:
uv pip install -e ".[web]"
uv run pidriver-web --store pidriver.sqlite3 \
--projects-root /srv/projects --allow-project acme
Core pieces:
PiInstanceSupervisordrains everyPiSessionin one background task, persists event history, broadcasts events to subscribers, and exposesprompt/respond/cancel/stop.PiInstanceStorestores snapshots and event logs in SQLite. On service restart, previously live subprocesses are markedstalebecause they cannot be reattached.PiWebAppis a small ASGI app with JSON endpoints and SSE streaming.
Useful routes:
| Route | Purpose |
|---|---|
GET / |
Minimal dashboard for launching and watching instances. |
GET /api/projects |
List project-jail candidates. |
GET /api/models |
List models visible to this isolated pi config. |
POST /api/instances |
Start an instance from {prompt, project, model, approval_policy}. |
GET /api/instances / GET /api/instances/{id} |
List or inspect snapshots. |
GET /api/instances/{id}/events |
Read persisted event history. |
GET /api/instances/{id}/events/stream |
SSE stream for live events. |
POST /api/instances/{id}/prompt |
Send a follow-up prompt. |
POST /api/instances/{id}/respond |
Answer a pending interaction. |
POST /api/instances/{id}/cancel / POST /api/instances/{id}/stop |
Interrupt or terminate. |
GET /api/auth/providers / POST /api/auth/{provider}/start |
OAuth helper endpoints. |
Development
uv pip install -e ".[dev]"
uv run pytest
uv run mypy src
uv run ruff check
License
MIT © Grigory Bakunov
Metadata
Release files for pidriver 0.4.2
For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.
Source distribution (sdist)
| File | Size | Uploaded | Provenance | Actions |
|---|---|---|---|---|
| pidriver-0.4.2.tar.gz | 166.5 kB | Without provenance | Details for pidriver-0.4.2.tar.gz |
Built distribution (wheel)
| File | Interpreter | ABI | Platform | Actions Reset |
|---|---|---|---|---|
| pidriver-0.4.2-py3-none-any.whl | Python 3 | none | any | Details for pidriver-0.4.2-py3-none-any.whl |
Total release size: 251.5 kB
Release files / pidriver-0.4.2.tar.gz
| Download URL | pidriver-0.4.2.tar.gz |
|---|---|
| Size | 166.5 kB |
| Tags | Source |
|
SHA-256 checksum How to use checksums |
33f5c460b6e0b92a77c9c25c4a9b7494484b319285d4c5c9f2ec76fc2902ffe6
|
|
BLAKE2b-256 checksum How to use checksums |
8c1b5b864c58c06b0a11854e7abf80b22e998f5428abb3a5b0d4cf5890e50730
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
No |
| Uploaded via |
uv/0.12.19 {"installer":{"name":"uv","version":"0.12.19","subcommand":["publish"]},"python":null,"implementation":{"name":null,"version":null},"distro":{"name":"macOS","version":null,"id":null,"libc":null},"system":{"name":null,"release":null},"cpu":null,"openssl_version":null,"setuptools_version":null,"rustc_version":null,"ci":null}
|
Release files / pidriver-0.4.2-py3-none-any.whl
| Download URL | pidriver-0.4.2-py3-none-any.whl |
|---|---|
| Size | 85.0 kB |
| Tags | Python 3 |
|
SHA-256 checksum How to use checksums |
643e5568fdcafd7e9892556a26e6b4fc68122f28e0e9167214b003a67f7d3580
|
|
BLAKE2b-256 checksum How to use checksums |
eac8d119f237793f76214e3ae815ca9034f1120b5f34c3b17fcfbf4430fb000d
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
No |
| Uploaded via |
uv/0.12.19 {"installer":{"name":"uv","version":"0.12.19","subcommand":["publish"]},"python":null,"implementation":{"name":null,"version":null},"distro":{"name":"macOS","version":null,"id":null,"libc":null},"system":{"name":null,"release":null},"cpu":null,"openssl_version":null,"setuptools_version":null,"rustc_version":null,"ci":null}
|
Security for pidriver 0.4.2
Provenance
Provenance describes where a file came from. On PyPI, provenance is shared via attestations, which provide a verifiable record of the build or publishing details. View details, limitations and caveats.
No provenance found
None of the files in this release were uploaded with provenance metadata.
A note on package security
PyPI is a repository for distributing packages, not a service for auditing them. While third-party security researchers and our staff do their best to detect and remove malicious packages, PyPI cannot guarantee the quality, security or suitability of any given package.
To learn more about protecting yourself from malicious actors, visit our package security documentation.