Pi Agent Python SDK

Run Pi from Python. Stream responses, continue conversations, and control sessions.

A community Python SDK for Pi coding agent. Use synchronous or asynchronous clients with typed access to Pi's RPC commands and no third-party Python runtime dependencies. Pi runs as a subprocess and keeps its usual models, authentication, tools, extensions, skills, and configuration.

Requires Python 3.11+ and a separate Pi installation. The tested protocol baseline is Pi 0.85.1; see compatibility for version and platform scope.

Quick start · Streaming · Async · Configuration · RPC · Documentation

Quick start

Install Pi with Node.js 22.19.0 or newer:

npm install -g @earendil-works/pi-coding-agent@0.85.1
pi --version

Run pi once to configure your provider and model using Pi's normal setup. The SDK uses that configuration when it starts Pi.

Install the Python package:

python -m pip install pi-agent-python-sdk==0.1.0

The distribution is named pi-agent-python-sdk; import it as pi_agent.

from pi_agent import PiClient

with PiClient() as pi:
    result = pi.run("Explain the current project without changing files.")
    print(result.text)

The context manager starts and closes Pi. run() waits for the conversation to settle, including retries and queued follow-ups. The result includes finalized messages, session identity, elapsed time, and observed assistant usage.

What you can do

• Run and stream: get a final answer or consume text, thinking, and tool events.
• Keep a conversation: send follow-up prompts, resume sessions, fork, or clone.
• Control Pi: select models, adjust thinking, steer work, compact context, and call all 33 RPC commands in the pinned baseline.
• Integrate with your application: use sync or async clients, typed results, raw event dictionaries, and extension UI callbacks.

Streaming

from pi_agent import PiClient

with PiClient() as pi:
    with pi.stream("Explain this project's entry points without editing files.") as stream:
        for event in stream:
            if event.text_delta is not None:
                print(event.text_delta, end="", flush=True)
        result = stream.result()
    print(f"\nStop reason: {result.stop_reason}")

event.raw contains the full Pi event, including fields the SDK does not yet recognize. Keep the stream inside its context: leaving early clears queued input and aborts the work it owns. See errors and cancellation for handling timeouts and partial results.

Async usage

Use AsyncPiClient in applications that already run an event loop. Command arguments and results match the synchronous client.

import asyncio

from pi_agent import AsyncPiClient


async def main() -> None:
    async with AsyncPiClient() as pi:
        result = await pi.run("Explain the current project without changing files.")
        print(result.text)


if __name__ == "__main__":
    asyncio.run(main())

Async streaming uses async with pi.stream(...), async for event in stream, and await stream.result(). See the complete streaming example.

Configuration

Choose the project directory and whether to save the conversation:

from pi_agent import PiClient

with PiClient(cwd=".", no_session=True) as pi:
    print(pi.run("Describe this project without changing files.").text)

Leave no_session unset to preserve Pi's normal session persistence. Use provider= and model= to override Pi's configured model, or session= to open an existing session. See constructor options for environment overrides, executable paths, and deadlines.

Extensions are installed and configured through Pi. To load your own extension, pass extra_args=["--extension", "/absolute/path/to/your-extension.ts"] to either client. See using your own extensions for details and extension UI for an interactive example.

RPC access

The SDK speaks Pi's existing JSONL protocol over stdin/stdout:

Your Python application
    PiClient / AsyncPiClient
        pi --mode rpc
            Models · tools · extensions · sessions

Choose the interface that fits the work:

For example, inspect session state without starting a model run:

from pi_agent import PiClient

with PiClient() as pi:
    state = pi.get_state()
    print(state["sessionId"])

Python arguments use snake_case; wire dictionaries retain Pi's camelCase fields. prompt() acknowledges submission, which may be handled entirely by an extension. It does not wait for a completed answer.

One client owns one conversation at a time; create separate clients for independent conversations. After a low-level prompt submitted outside an owned run, use a fresh client for run() or stream() so events cannot be attributed to the wrong work. The SDK launches a new process and cannot attach to an existing Pi terminal session.

See RPC structure for the protocol mapping and module layout, and the command reference for every method.

Documentation

Examples use your configured Pi and may make provider calls. Development checks use an isolated local test provider; see CONTRIBUTING.md for setup and validation commands.

Maintainers: protocol discovery · maintenance · release checklist.

License

MIT