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.
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 ───── │ │
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 |
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 |
pip install git+https://github.com/akatzmann/py-agent-core.git(For full local development environment setup instructions, refer to the Installation Guide.)
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)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())Switching between provider endpoints requires zero changes to your agent logic:
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-4o")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)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.
- 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.
- 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.
Verify the entire test suite including backend formatting correctness, tool execution, and preemption flows:
PYTHONPATH=. pytestWe 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.
py-agent-core is open-source software licensed under the MIT License.