pi-agent-core

A stateful LLM agent framework for Python with tool execution, event streaming, steering/follow-up message queuing, and proxy transport.

Installation

uv add pi-agent-core

Overview

pi-agent-core provides a minimal, LLM-agnostic agent loop that handles the orchestration between your application and any LLM provider. You bring your own streaming function — the library handles state management, tool execution, event dispatch, mid-turn steering, and follow-up queuing.

Key Features

• LLM-agnostic — works with any provider via the StreamFn protocol
• Real-time event streaming — two-level event system for agent lifecycle and LLM streaming primitives
• Tool execution — define tools with JSON Schema parameters and async execute functions
• Steering & follow-up queues — interrupt mid-turn or queue messages for after completion
• Cancellation — cooperative cancellation via asyncio.Event
• Proxy transport — built-in SSE proxy client for routing through a backend server
• Fully typed — Pydantic models throughout with py.typed marker

Quick Start

import asyncio
from pi_agent_core import (
    Agent,
    AgentOptions,
    AgentEvent,
    AgentTool,
    AgentToolSchema,
    AgentToolResult,
    Model,
    TextContent,
)


# 1. Define your tools
async def greet(tool_call_id, params, cancel_event=None, on_update=None):
    name = params.get("name", "world")
    return AgentToolResult(content=[TextContent(text=f"Hello, {name}!")])


greet_tool = AgentTool(
    name="greet",
    description="Greet someone by name",
    parameters=AgentToolSchema(
        properties={"name": {"type": "string", "description": "Name to greet"}},
        required=["name"],
    ),
    execute=greet,
)


# 2. Implement a StreamFn for your LLM provider
# (see "Implementing a StreamFn" section below)
async def my_stream_fn(model, context, options):
    ...


# 3. Create and run the agent
agent = Agent(AgentOptions(stream_fn=my_stream_fn))
agent.set_model(Model(api="anthropic", provider="anthropic", id="claude-sonnet-4-20250514"))
agent.set_system_prompt("You are a helpful assistant.")
agent.set_tools([greet_tool])

# Subscribe to events
def on_event(event: AgentEvent):
    print(f"Event: {event.type}")

agent.subscribe(on_event)

# Send a prompt
asyncio.run(agent.prompt("Say hello to Alice"))

Architecture

Agent                     ← High-level stateful wrapper, subscriptions, queues
    ↓
agent_loop()              ← Core orchestration: prompt → stream → tools → steering loop
    ↓
StreamFn (user-provided)  ← You implement LLM streaming integration
    or
stream_proxy()            ← Built-in SSE proxy client as a StreamFn

Modules

Implementing a StreamFn

The library is LLM-agnostic. You provide a stream_fn(model, context, options) that returns a StreamResult — an async iterator of AssistantMessageEvent objects with an async result() method.

from pi_agent_core import (
    AssistantMessage,
    AssistantMessageEvent,
    StreamResult,
    StreamStartEvent,
    StreamTextStartEvent,
    StreamTextDeltaEvent,
    StreamTextEndEvent,
    StreamDoneEvent,
    TextContent,
    Model,
    AgentContext,
    SimpleStreamOptions,
)


class MyStream:
    """Implements the StreamResult protocol."""

    def __init__(self):
        self._events = []
        self._final = None

    def __aiter__(self):
        return self

    async def __anext__(self) -> AssistantMessageEvent:
        ...  # yield events from your LLM provider

    async def result(self) -> AssistantMessage:
        return self._final


async def my_stream_fn(
    model: Model,
    context: AgentContext,
    options: SimpleStreamOptions,
) -> StreamResult:
    stream = MyStream()
    # Start your LLM call, push events into the stream
    return stream

Event System

Agent Events (10 types)

Covers agent lifecycle, turns, messages, and tool execution:

agent_start, agent_end, turn_start, turn_end, message_start, message_update, message_end, tool_execution_start, tool_execution_update, tool_execution_end

Assistant Message Events (12 types)

Covers LLM streaming primitives consumed internally by the loop:

start, text_start, text_delta, text_end, thinking_start, thinking_delta, thinking_end, toolcall_start, toolcall_delta, toolcall_end, done, error

Steering & Follow-up Queues

Steering messages interrupt the agent mid-turn (skipping remaining tool calls):

agent.steer(UserMessage(content=[TextContent(text="Actually, use a different approach")]))

Follow-up messages trigger new turns after the current run completes:

agent.follow_up(UserMessage(content=[TextContent(text="Now summarize the results")]))

Both support "one-at-a-time" (default) or "all" dequeue modes.

Proxy Transport

For apps that route LLM calls through a backend server:

from pi_agent_core import Agent, AgentOptions, stream_proxy, ProxyStreamOptions

agent = Agent(AgentOptions(
    stream_fn=lambda model, context, options: stream_proxy(
        model, context,
        ProxyStreamOptions(
            **options.model_dump(),
            auth_token="your-auth-token",
            proxy_url="https://your-proxy.example.com",
        ),
    ),
))

Development

uv sync                     # Install dependencies
uv run pytest               # Run all tests
uv run pytest -v --tb=short # Verbose with short tracebacks
uv run ruff check .         # Lint
uv run ruff format .        # Format

Credits

This is a Python port of the TypeScript @mariozechner/pi-agent-core package from the pi-mono repository. The original TypeScript implementation by Mario Zechner provides the architecture, abstractions, and design that this library faithfully mirrors.

License

MIT