pi-python

Unofficial project: this repository is not affiliated with, endorsed by, or maintained by the official pi project or its maintainers.

Python port of pi-agent-core from the pi project, with LangChain replacing pi-ai for LLM calls.

The loop semantics, event protocol, and tool execution are faithful ports of pi; LangChain is only a StreamFn boundary adapter — never a replacement for the agent loop.

This repository is a monorepo containing:

Python packages (PyPI):

• pi-agent-core-lc / pi_agent_core: lightweight core loop, messages, tools, adapters.
• pi-agent-harness-lc / pi_agent_harness: harness runtime, sessions, queues, hooks, compaction, skills/templates, and local execution env.
• pi-agent-cli-lc / pi_agent_cli: standard ACP coding agent (python -m pi_agent_cli, headless -p).

Rust TUI (beta — prebuilt binary in GitHub Releases):

• zypi: full-screen terminal UI for interactive coding-agent sessions. Vendored fork of grok-build (Apache-2.0), with all x.ai/* vendor extensions removed. See TUI & Code-Agent Guide for details.

Status

Features

Core runtime (Phase 1)

• Agent — stateful agent with steering / follow-up queues, abort, subscribe() event barrier
• agent_loop — tool execution loop with pi-compatible event protocol (parallel/sequential tools, terminate semantics, should_stop_after_turn, prepare_next_turn)
• LangChain StreamFn adapter (langchain_stream) for OpenAI / Anthropic / DeepSeek-style / any init_chat_model provider
• Mock stream for tests — the whole suite runs without API keys

Production hardening (Phase 2 / 2.5)

• Cross-provider message replay (transform_messages: tool-call id normalization, thinking downgrade, image stripping)
• Usage & cost tracking (accumulated across stream chunks, correct for all three real-world reporting shapes) + CostCalculator
• Thinking/reasoning: provider param mapping (reasoning_effort / Anthropic thinking budgets), streamed thinking_delta events, Anthropic signature replay, DeepSeek-style reasoning_content
• Stream-level retries before the first token (exponential backoff + jitter, Retry-After aware)
• Runaway protection: max_turns (raises MaxTurnsExceededError) and per-tool tool_timeout
• Guardrail hooks: before_llm_call (with a ContextBudget token signal — the compaction hook point), after_llm_call (tripwire on raise), on_agent_end
• Observability: on_payload / on_response hooks; every event carries run_id / turn_id
• Granular stream events: text_start/end, thinking_start/end, toolcall_start/end (plus deltas)
• Structured output: response_schema (pydantic model or JSON schema) → parsed AssistantMessage.structured_output
• Tool-result images: Anthropic native blocks, user-message fallback elsewhere, stripped when supports_images=False
• OpenAI-compatible gateways via Model.base_url (SiliconFlow, vLLM, DeepSeek, ...)

Tool ecosystem (P6)

• Built-in coding tools (pi_agent_core.coding_tools): read / bash / edit / write / grep / find / ls with pi-faithful truncation notices and details payloads
• LangChain BaseTool → AgentTool adapter (from_langchain_tool) — MCP tools via langchain-mcp-adapters work out of the box

Coding Agent CLI (Phase 4 — beta)

• pi_agent_cli: standard ACP coding agent over AgentHarness — no x.ai/* vendor extensions
• Headless mode: python -m pi_agent_cli -p "prompt" for scripting and CI
• /new, /resume, /quit session commands; @ local directory listing
• pi-aligned system prompt engine with tool contributions, AGENTS.md context files, <available_skills> format

TUI — zypi (Phase 4 — beta)

• Full-screen Rust terminal UI for interactive coding-agent sessions
• Spawns the Python ACP agent (python -m pi_agent_cli) as the backend
• Session management, markdown rendering, Mermaid diagrams, MCP support
• Prebuilt Linux x86_64 binary in GitHub Releases; build from source for other platforms
• See TUI & Code-Agent Guide for setup and usage

FrontierHarness Evaluation (30-task benchmark)

• 21 Terminal-Bench + 9 DeepSWE industrial tasks; Windows local + WSL2 Docker sandbox runtimes
• Side-by-side comparison with official Pi; full report: FRONTIER-HARNESS-30-EVAL-REPORT.md

Install

Python packages (PyPI)

pip install pi-agent-core-lc
pip install pi-agent-harness-lc    # optional: session/harness runtime
pip install pi-agent-cli-lc        # optional: ACP stdio / headless agent

# Optional LLM providers:
pip install pi-agent-core-lc[openai]      # ChatOpenAI
pip install pi-agent-core-lc[anthropic]   # ChatAnthropic
pip install pi-agent-core-lc[deepseek]    # ChatDeepSeek / OpenAI-compatible gateways
pip install pi-agent-core-lc[all]         # all providers + harness + cli

pip install -e ".[dev]"
pip install -e "./packages/pi-agent-harness"
pip install -e "./packages/pi-agent-cli"

TUI binary (zypi) — beta

Download the prebuilt Linux x86_64 binary from GitHub Releases:

# Download and install (Linux x86_64)
curl -L -o zypi https://github.com/zy1233/pi-python/releases/latest/download/zypi-linux-x86_64
chmod +x zypi
sudo mv zypi /usr/local/bin/

# Or build from source (requires Rust toolchain)
cd tui && cargo build --profile release-dist -p pi-pager-bin
# binary at target/release-dist/zypi

After installing, set up config and start the TUI — see TUI & Code-Agent Guide.

Quick start

import asyncio
from pi_agent_core import Agent, Model
from pi_agent_core.adapters import langchain_stream


async def main():
    agent = Agent(
        initial_state={
            "system_prompt": "You are a helpful assistant.",
            "model": Model(provider="openai", model_id="gpt-4o-mini"),
        },
        stream_fn=langchain_stream,
    )

    agent.subscribe(lambda event, signal: print(event.type))

    await agent.prompt("Hello!")
    await agent.wait_for_idle()


asyncio.run(main())

Requires OPENAI_API_KEY (or the env var for your provider).

OpenAI-compatible gateways

Point Model.base_url at any OpenAI-compatible endpoint. Use provider="deepseek" when the endpoint streams thinking as reasoning_content (SiliconFlow, DeepSeek, most vLLM gateways) so it surfaces as thinking_delta events instead of being dropped:

model = Model(
    provider="deepseek",
    model_id="Qwen/Qwen3-8B",
    base_url="https://api.siliconflow.cn/v1",
    context_window=32_000,
)
agent = Agent(initial_state={"model": model}, stream_fn=langchain_stream)
# pass the key explicitly or via env: DEEPSEEK_API_KEY

Structured output

from pydantic import BaseModel


class Person(BaseModel):
    name: str
    age: int


agent = Agent(
    initial_state={"model": model},
    stream_fn=langchain_stream,
    response_schema=Person,  # or a JSON schema dict
)
await agent.prompt("Invent a fictional person.")
await agent.wait_for_idle()

agent.messages[-1].structured_output  # {'name': ..., 'age': ...} (None if not parseable)

Schema instructions are injected into the system prompt for every provider; OpenAI-style providers additionally get native response_format enforcement. Streaming stays intact — the JSON text still flows as text_delta events.

Production configuration

from pi_agent_core import Agent, ContextBudget


def on_payload(payload: dict):  # outgoing request (pre-call)
    log.debug("LLM call", model=payload["model"], tools=len(payload["tools"]))


async def before_llm_call(context, budget: ContextBudget | None):
    if budget and budget.fraction > 0.8:
        return await my_compactor.compact(context)  # durably replaces loop context
    return None


def after_llm_call(context, message):
    if contains_pii(message):  # guardrail tripwire: raise to abort the run
        raise PiiDetected()


agent = Agent(
    initial_state={"model": model, "tools": tools},
    stream_fn=langchain_stream,
    max_turns=25,  # raises MaxTurnsExceededError -> agent.error_message
    tool_timeout=120.0,  # per tool call, seconds; times out into an error tool result
    max_retries=3,  # stream-level retries before the first token
    on_payload=on_payload,
    before_llm_call=before_llm_call,
    after_llm_call=after_llm_call,
)

Runnable version: examples/production_agent.py.

Built-in coding tools

A port of pi's coding-agent tool suite lives in pi_agent_core.coding_tools, decoupled from the core runtime (tools consume the loop; they are not part of it). Factories bind each tool to a working directory:

from pi_agent_core.coding_tools import create_coding_tools, create_read_only_tools

tools = create_coding_tools("/path/to/project")  # read / bash / edit / write
audit = create_read_only_tools("/path/to/project")  # read / grep / find / ls
agent = Agent(initial_state={"model": model, "tools": tools}, stream_fn=langchain_stream)

Groups mirror pi: create_coding_tools(cwd) (read/bash/edit/write — full file operations plus command execution) and create_read_only_tools(cwd) (read/grep/find/ls — inspection with a no-modification guarantee). Per-tool create_*_tool(cwd) factories and create_tool(name, cwd) / create_all_tools(cwd) cover custom mixes. Truncation limits are pi's (2000 lines / 50KB, grep lines capped at 500 chars) with actionable notices ("Use offset=N to continue") so the model can page through anything that was cut.

Using LangChain / MCP tools

Any LangChain BaseTool — including MCP tools produced by langchain-mcp-adapters — plugs into the same loop through the adapter:

from langchain_core.tools import tool
from pi_agent_core.adapters import from_langchain_tool, from_langchain_tools


@tool
def get_weather(city: str) -> str:
    """Get the current weather for a city."""
    return f"Sunny in {city}"


agent = Agent(
    initial_state={"model": model, "tools": [from_langchain_tool(get_weather)]},
    stream_fn=langchain_stream,
)

Parameter schemas come from tool_call_schema (LangChain-injected arguments are excluded); results normalize to pi content blocks (plain strings, content-block lists, base64 images); content_and_artifact artifacts land in details["artifact"]; tool exceptions bubble into is_error=True tool results that the model can react to.

Custom messages

AgentMessage is structurally open: anything with a string role flows through the loop and persists to the session (AgentMessageProtocol, a runtime-checkable Protocol in pi_agent_harness). The harness ships four custom roles — bashExecution, custom, branchSummary, compactionSummary — and harness_convert_to_llm decides how each role reaches the LLM; unknown roles are dropped at that boundary:

from pydantic import BaseModel
from pi_agent_harness import AgentMessageProtocol
from pi_agent_harness.messages import BashExecutionMessage, harness_convert_to_llm

# Built-in custom role: record a shell run in the session/context
bash = BashExecutionMessage(command="pytest -q", output="97 passed", exitCode=0, timestamp=0)
await harness.append_message(bash)  # persisted; replayed to the LLM as a user message


# Your own role: any object with `role: str` satisfies the protocol
class DeployNote(BaseModel):
    role: str = "deployNote"
    environment: str
    timestamp: int


assert isinstance(DeployNote(environment="staging", timestamp=0), AgentMessageProtocol)

To make a custom role visible to the LLM, wrap harness_convert_to_llm (or pass your own convert_to_llm to the core loop) and map it to a UserMessage — the same pattern the harness uses for its own roles.

Events

Agent events (all carry run_id and a 1-based turn_id):

agent_start → turn_start → message_start/end (user)
→ message_start(assistant) → message_update* → message_end(assistant)
→ [tool_execution_start/update/end → message_start/end (toolResult)]
→ turn_end → ... → agent_end

message_update wraps the granular assistant-stream events: text_start/delta/end, thinking_start/delta/end, toolcall_start/delta/end. End events carry the aggregate (content full text / complete tool_call block).

Test

uv run --extra dev --extra harness python -m pytest  # 418 tests, no API keys needed
ruff check . && ruff format --check .

Real-API validation (key via env only; skipped when unset):

Use .venv-test-real for pytest real-LLM tests; the smoke script can run from the main .venv.

Documentation

Concept mapping (pi → Python)

Roadmap

Completed

• Phase 1 (MVP loop) — done: Agent + agent_loop + event protocol + tool execution
• Phase 2 (production enhancements) — done: transform_messages, usage/cost, thinking/reasoning
• Phase 2.5 (core hardening) — done: retries, runaway protection, observability, granular events, guardrail hooks, ContextBudget, structured output, tool-result images
• Phase 3 (AgentHarness) — done: H1 session tree, H2 runtime, H3 compaction/tree navigation, H4 skills/templates/system prompt/LocalExecutionEnv (design doc)
• P6 (tool ecosystem) — done: all 7 built-in tools (read/bash/edit/write/grep/find/ls), group factories, and the LangChain BaseTool adapter (design doc)
• Phase 4 (Coding Agent CLI — beta) — done: tui/ vendored grok-build fork (Apache-2.0); pi_agent_cli standard ACP agent; TUI binary zypi; x.ai/* vendor RPCs removed; session management (/new /resume /quit); headless -p; config.toml + Windows notes (design doc, Windows, TUI guide)
• Phase 5 (Coding Agent prompt engine) — done: pi-aligned build_system_prompt + tool contributions + context files + <available_skills> (design doc)
• FrontierHarness Eval — done: 30-task benchmark (21 Terminal-Bench + 9 DeepSWE); Windows + WSL Docker runtimes; side-by-side Pi comparison (report)

Next

• Phase 6 (Extended integrations) — planned: native MCP client, git-aware context injection, multi-provider production testing matrix
• TUI platform expansion — planned: prebuilt binaries for macOS (arm64/x86_64) and Windows; auto-update support

License

Python packages (pi-agent-core-lc, pi-agent-harness-lc, pi-agent-cli-lc) are MIT.

The tui/ directory is a vendored fork of xai-org/grok-build and remains Apache License 2.0. See tui/LICENSE, tui/NOTICE, and tui/THIRD-PARTY-NOTICES. tui/ is not included in Python sdists or wheels.