python-agent-core

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

⚡ 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

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.

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

py-agent-core is open-source software licensed under the MIT License.