python-agent-core

A minimalist, event-driven agent loop with cooperative preemption and unified backends in Python.

Tip

🤖 Pairing with an AI Assistant?
Want a 10-second overview? Copy-paste this seed prompt into Claude Code, Antigravity, Copilot, or ChatGPT:

Analyze https://github.com/akatzmann/py-agent-core. 
1. What is this tool, what is a typical use-case, and do I need it?
2. How does it differ from standard agent frameworks?
3. Give me an open-ended assessment based on the actual codebase/structure (not just README marketing): Does it deliver on its claimed USP?
Keep your answer concise so I can digest it in 10 seconds.

---

⚡ The Core Value: Why Cooperative Preemption?

Traditional frameworks treat ChatCompletion calls as monolithic black boxes. If a user tries to cancel or override the agent mid-sentence, the runtime is blocked until the generation finishes or the task is crudely killed.

py-agent-core handles preemption at the transport layer. By checking an interrupt flag during token consumption, it severs the network socket instantly when triggered—saving latency, tokens, and allowing immediate cognitive redirection.

               THE COOPERATIVE PREEMPTION LOOP
               
  [User App]                     [PyAgent]                    [LLM API]
      │                              │                             │
      │ ─── agent.run_loop() ──────► │                             │
      │                              │ ─── Stream Request ───────► │
      │                              │                             │
      │ ◄── event: text_delta ────── │ ◄── Token Stream ────────── │
      │ ◄── event: text_delta ────── │ ◄── Token Stream ────────── │
      │                              │                             │
      │ ─── agent.interrupt() ─────► │                             │
      │                              │ ─── stream.close() ────────►X (Socket Closed)
      │ ◄── event: interrupted ───── │                             │

---

🔍 How It Compares

Note

🏛️ Architecture Heritage
py-agent-core adapts the minimalist event-loop architecture of pi-agent-core (the TypeScript engine powering systems like OpenClaw) into idiomatic, async Python.

Architectural Dimension	py-agent-core	Big-Box Frameworks
🔄 Control Paradigm	⚡ Stream of Events (Async Generator) Yields raw events (text_delta, tool_start, etc.)	📥 Callback Managers Hides execution behind custom routers
🛑 Stream Preemption	✅ Surgical Transport-Level Abort Closes socket connection mid-generation instantly	❌ Thread Kill / No Cancel Must wait for the LLM to finish speaking
🧩 Tool Specifications	✅ Auto-parsed Typehints & Docstrings Decorate standard Python functions with @tool	⚠️ Manual Declarations Requires complex Pydantic schemas or JSON
🎯 Model Resilience	✅ Dynamic Parameter Fallback Gracefully falls back when reasoning vs sampling parameters clash	❌ SDK Schema Crash Crashes if a parameter is not supported by the model
🌳 Agent Hierarchy	✅ Nested Worker Trees Instantiate child agents directly inside parent tools	❌ Complex Graph Configs Requires custom routers and state channels
📦 Footprint & Bloat	✅ Zero-Dependency Core Minimalist loop code you can read and fit in your head	❌ Sprawling Class Hierarchy Hundreds of helper wrappers and utilities

---

Quickstart (Run 100% Locally)

1. Install

pip install python-agent-core

(For full local development environment setup instructions, refer to the Installation Guide.)

2. Define a Tool

Write standard Python functions. Their signatures, type hints, and docstrings compile into tool schemas automatically.

from py_agent_core import tool

@tool
def calculate_factorial(n: int) -> int:
    """Compute the factorial of n.
    Args:
        n: The target integer.
    """
    return 1 if n <= 1 else n * calculate_factorial(n - 1)

3. Run the Event Loop (Ollama)

Consume the stream as structured events in real time:

import asyncio
from py_agent_core import PyAgent, OllamaBackend

async def main():
    # Runs completely offline using a local model (llama3)
    backend = OllamaBackend(model="llama3")
    agent = PyAgent(backend=backend, tools=[calculate_factorial])
    
    async for event in agent.run_loop("Calculate the factorial of 5"):
        if event.type == "text_delta":
            print(event.content, end="", flush=True)
        elif event.type == "tool_start":
            print(f"\n[Tool Started: {event.content}]")
        elif event.type == "tool_end":
            print(f"\n[Tool Finished: {event.content['result']}]")

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

---

🛠️ Swapping LLM Backends

Switching between provider endpoints requires zero changes to your agent logic:

OpenAI / Azure OpenAI

Configure your target endpoints using the official SDK client objects:

from openai import AsyncOpenAI
from py_agent_core import OpenAIBackend

client = AsyncOpenAI(api_key="sk-...")
backend = OpenAIBackend(client=client, model="gpt-6.1-sol", api_mode="auto")

Modern /v1/responses API & Frontier Models

OpenAIBackend and AzureOpenAIBackend support OpenAI's official /v1/responses endpoint alongside legacy /v1/chat/completions:

• Active Reasoning with Function Tools: Native reasoning models (e.g., gpt-6.1-sol, o- series) execute function tools concurrently with active thinking traces via /v1/responses.
• Zero Model-Name Guessing: Protocol selection never branches on arbitrary model strings. In api_mode="auto", requests negotiate endpoint capabilities dynamically:
  • Optimistically connects to /v1/responses; gracefully falls back to /v1/chat/completions if the server returns 404/405 or unsupported endpoint status.
  • Dynamically promotes to /v1/responses if /v1/chat/completions rejects function tools due to active reasoning conflicts.
• Explicit Protocol Control: Set api_mode="responses" to enforce the responses endpoint or api_mode="chat_completions" for legacy proxies/servers (e.g., vLLM, LiteLLM).
• Thinking Trace Mapping: Configured thinking_level ("low", "medium", "high") maps directly to reasoning={"effort": ...} on /v1/responses and reasoning_effort on /v1/chat/completions.

Offline / Mock Testing

Simulate streaming completion and tool execution completely offline with zero LLM API dependency:

from py_agent_core import DummyBackend

backend = DummyBackend(lorem_text="hello world", chunk_delay=0.01)

---

💡 Executable Examples & Demos

We provide a comprehensive set of executable scripts in the examples/ directory. Each example can run completely offline out-of-the-box using the local DummyBackend.

🟢 Core Concepts (Beginner)

• Hello Agent: Bare-minimum streaming and logging run.
• Structured Streaming: Buffering and parsing structured JSON from token streams.
• Search Watchdog: Cooperative preemption via an external watchdog timer/timeout.
• Rhetoric Speaker: Real-time user input preemption mid-monologue.
• Hierarchical Assistant: Running nested sub-agents within tools to construct coordinating agent trees.

🟡 Production Patterns (Advanced)

• Streaming Guardrails: Stream middleware redacting PII and preempting on toxic words.
• Interactive TUI Chat: Full terminal TUI splitting user typing from token streams.
• Self-Healing Coder: Subprocess code execution with iterative error-traceback correction.
• Parallel Agent Swarm: Running multiple independent agents concurrently.
• Advanced Agent Features: Complete showcase of subscriptions, before/after hooks, parallel tool execution, context pruning pipelines, and active steering queues.

---

🛡️ Responsible Usage & Safety Disclaimer

python-agent-core is a foundational engine for event-driven agent loops. When embedding it in production or autonomous systems, keep the following operational considerations in mind:

• 💰 Token Spending & Cost Control: Autonomous multi-turn agent loops and recursive worker trees (Hierarchical Assistant, Agent Swarm) consume LLM completion and reasoning tokens on every iteration. Unchecked or infinite loops can quickly incur significant API costs. Always configure spending limits and quotas in your provider consoles (OpenAI, Azure), enforce max turn limits in your application logic, and test complex flows offline with DummyBackend or OllamaBackend.
• ⚠️ Tool Execution & Sandboxing: Functions decorated with @tool run with the full permissions of the host Python process. When providing tools that perform file I/O, execute shell commands, or query databases, implement input validation, sandboxing, or interactive confirmation steps before executing destructive operations.
• 🛑 Leverage Cooperative Preemption: Always retain access to agent.interrupt() or an AbortSignal in your UI or orchestrator so users or external watchdogs can sever network sockets mid-token and halt tool runs immediately.
• 📄 Warranty & Liability: Provided "as is" under the Apache-2.0 License. Contributors and maintainers assume no liability for LLM provider expenses incurred, runaway execution loops, or unintended side effects caused by user-defined tool implementations.

---

Running Tests

Verify the entire test suite including backend formatting correctness, tool execution, and preemption flows:

PYTHONPATH=. pytest

Contributing

We practice AI-first, spec-driven development using the OpenSpec framework. For all non-trivial changes, please refer to our Contributing Guide to set up your environment and author change specifications.

License

python-agent-core is open-source software licensed under the Apache-2.0 License.