Pi Agent Python SDK

Use your installed Pi coding agent as a Python API. Stream responses, continue conversations, and control sessions from Python.

A community Python SDK for Pi coding agent. It starts Pi as a subprocess and exposes the same agent runtime through a typed Python interface for RPC commands, streaming, and session control.

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

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

Why use this?

If you already use Pi from the terminal, this SDK lets you use the same agent from Python code. Instead of managing a separate implementation, you can start Pi in RPC mode and call it as a programmatic API from scripts, apps, services, or background jobs.

This is useful when you want to:

• Add Pi-powered coding or reasoning to a Python application.
• Automate project analysis, refactors, or repository workflows.
• Build a service or CLI around Pi without reimplementing the agent runtime.
• Stream text, thinking, tool, and session events in Python.
• Reuse your existing Pi configuration, models, tools, and extensions.

The SDK keeps Pi's usual setup and model selection, while exposing a typed Python interface for prompts, follow-ups, model control, and RPC access. It launches Pi as a subprocess and communicates over its JSONL protocol; it does not attach to an already-running interactive Pi terminal session.

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.

Quick start

Install Pi with Node.js 22.19.0 or newer:

npm install -g @earendil-works/pi-coding-agent@0.86.0
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 from PyPI:

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

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.

Continue the conversation with another call on the same client:

from pi_agent import PiClient

with PiClient() as pi:
    print(pi.run("Explain this project's entry points without editing files.").text)
    print(pi.run("Which of those entry points handles configuration?").text)

Pi keeps the conversation context. Each result contains only the messages and answer from that call; a call with no assistant output has empty text.

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 cleans up unfinished work. 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.

Models and thinking

The SDK uses Pi's configured model unless you pass provider= and model=. Pi determines which thinking levels that model supports; query them before choosing a level:

from pi_agent import PiClient

with PiClient() as pi:
    levels = pi.get_available_thinking_levels()
    print("Supported thinking levels:", levels)
    if "high" in levels:
        pi.set_thinking_level("high")
    print("Effective thinking level:", pi.get_state()["thinkingLevel"])

    result = pi.run("Analyse this project's architecture without changing files.")
    print(result.text)

Set the model and thinking level before starting run() or stream(). Pi can adjust unsupported level requests, so read get_state()["thinkingLevel"] for the effective value. The setting applies to subsequent prompts in the current session without changing global defaults. With AsyncPiClient, await the same methods. See the model and thinking commands for cycling levels and other controls.

The thinking level controls Pi's reasoning setting. Thinking output is separate: when Pi emits it, deltas are available in event.raw during streaming and finalized thinking blocks in result.messages. result.text contains the final assistant text, not thinking.

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.

Use one client per conversation and wait for each run() or stream to finish before starting the next. After using prompt() for your own event handling, use a fresh client for run() or stream() so results cannot include delayed events from earlier 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