agent_runtime — Python port of Claude Code's agent spawning architecture

A clean, minimal Python implementation of the agent spawning patterns we reverse-engineered from Claude Code. Focuses on the architecture, not feature completeness:

• Async-generator-based agent loop (recursion = subagents)
• AbortController with hierarchical, GC-safe cancellation
• Module-level agent registry with push/drain mailbox
• Bounded concurrency via asyncio.Semaphore (coming in Step 2)
• Extension points wired in from day 1 (compaction, hooks, etc.)

Built on LangChain. Message types (HumanMessage, AIMessage, ToolMessage, SystemMessage), the chat model (BaseChatModel), and tools (BaseTool) all come from langchain_core. The provider seam is langchain_anthropic.ChatAnthropic by default; swap in ChatOpenAI, ChatGoogleGenerativeAI, etc. without touching agent code.

Status: Step 1 — foundation only

The architecture is in place; the dispatcher and tools aren't wired up yet (Step 2). Files:

agent_runtime/
├── __init__.py        # public API surface
├── types.py           # ToolUseContext, QueryParams (rest is LangChain)
├── cancel.py          # AbortController + hierarchical propagation
├── registry.py        # AGENT_REGISTRY + inbox push/drain
├── query.py           # THE agent loop
└── session.py         # Session (multi-turn session facade)

Roadmap

• Step 1 ✅ Foundation (LangChain-based)
• Step 2 ⬜ Tool dispatcher (run_tools) + first runnable example
• Step 3 ⬜ AgentTool (the spawn!) + push-notification flow
• Step 4 ⬜ TaskStopTool — cancellation tool
• Step 5 ⬜ Compaction infrastructure
• Step 6 ⬜ Hooks (PreToolUse, PostToolUse, Stop)
• Later ⬜ Tombstones, depth caps, budget, persistence, telemetry

Setup

pip install -r requirements.txt
export ANTHROPIC_API_KEY=...

Minimal use (once Step 2 is done)

import asyncio
from langchain_anthropic import ChatAnthropic
from langchain_core.tools import tool
from agent_runtime import Session
from langchain_core.messages import AIMessageChunk

@tool
def get_time() -> str:
    """Return the current time."""
    from datetime import datetime
    return datetime.now().isoformat()

async def main():
    engine = Session(
        llm=ChatAnthropic(model="claude-sonnet-4-5"),
        tools=[get_time],
    )
    async for event in engine.submit_message("What time is it?"):
        if isinstance(event, AIMessageChunk):
            print(event.content, end="", flush=True)

asyncio.run(main())

Python version

Requires Python 3.11+ for asyncio.TaskGroup, asyncio.timeout, modern type-hint syntax (X | Y, dict[K, V]).