⚡ tokenloop

A protocol-driven, model-agnostic agent loop for Python. AgentLoop runs the prompt → model → tool-call → model loop for you — streaming, retries, human-in-the-loop pauses, history compaction, and session persistence included — while every swappable piece (provider, tool, guardrail, approver, compressor, session store) is a plain typing.Protocol, not a base class you're forced to inherit from.

tokenloop is also an active, ongoing parity port of a companion Rust crate — much of the module layout is deliberately file-for-file with its Rust counterpart (core/loop.py ↔ agent/agent.rs, core/stream.py ↔ agent/stream.rs, core/registry.py ↔ run/registry.rs, and so on), which is why you'll find "Python counterpart of ..." in a lot of the source docstrings. A few Rust features are deliberately not ported yet — see Deliberately not ported below.

Philosophy

• Protocol-driven, not inheritance-driven. LLMProvider, Tool, Guardrail, Approver, Compressor, SessionService, ToolHook — every one of these is a structural Protocol. Anything with the right shape works; you never subclass a framework base class to plug something in.
• AgentLoop is stateless; Session isn't. Build one AgentLoop, reuse it for every conversation and every concurrent run. All per-conversation state — history, app/user/tmp state, past runs — lives in a Session, loaded and saved through a SessionService keyed by session_id. Nothing conversational is ever held on the AgentLoop object itself.
• Two different lifetimes, two different objects. RunContext is the live object threaded through hooks and tools during one run()/ astream() call — it's never persisted. RunRecord is what actually gets written to Session.runs once a run finishes: an immutable summary, not a working object.
• Every event tells you who and when. All 20+ event types share one BaseEvent (run_id, session_id, agent_id, created_at) — full run lifecycle (RunStarted/RunPaused/RunError/RunCompleted), per-turn model events, and per-tool-call events, all correlatable.

Features

• Core loop — AgentLoop runs a real, working turn loop (astream(); run() just drains it and returns the final message): automatic retry with exponential backoff on 408/409/429/5xx, best-effort history compaction, a RunControlService for pausing/resuming/cancelling a run by run_id, and a working interactive CLI (agent.as_cli()) for local dev/testing.
• Providers — LLMProvider is a Protocol; ships with working AnthropicProvider and OpenAIProvider implementations, both lazily imported so you only pay for the SDK you actually use.
• Tools — FunctionTool builds a JSON Schema straight from a plain Python function's signature (stdlib inspect/typing reflection, no Pydantic). Tool calls in one turn dispatch concurrently. Ships with four builtin tools: update_state, BashTool, Read, and FileStore.
• Human-in-the-Loop — a tool can pause a run for Confirmation, InputRequired, or ExternalExecution; an Approver can ALLOW/DENY/ AUDIT any tool call; RunMode.INTERACTIVE vs NON_INTERACTIVE controls whether pausing is even allowed; full resume support built in.
• Image input — attach ImageBase64/ImageUrl sources to a user turn (image_from_file() reads and encodes a local file); both providers serialize them into their own content-block format, and context compaction accounts for their token cost.
• Guardrails — pre_hooks validate the user's message before the model is ever called; post_hooks validate the assistant's response. Both also get a fire-and-forget on_completion for logging/eval that never adds latency to the run.
• Compression — SimpleCompressor trims oldest messages under a token budget (tool-call/result pairing preserved); HistoryWindow separately caps the persisted session history by message or run count.
• Skills — disk-based, compatible with Anthropic's Agent Skills layout (a SKILL.md per skill); every discovered skill resolves into a single invoke_skill meta-tool rather than one tool per skill.
• MCP — a client-side bridge (MCPToolProvider/MCPManager) adapts any MCP server's tools onto the same Tool protocol as everything else.
• Sessions — SessionService protocol, with an InMemorySessionService included (single-process — bring your own for multi-replica deployments).
• State — a standalone State utility with delta tracking and an RFC 6902 JSON-Patch diff(), for building your own state-sync pipeline (not wired into RunContext/the AG-UI bridge, which stay plain dicts and full-snapshot-only today).
• Streaming & observability — every run emits a typed LoopEvent stream; an AGUIBridge translates it into AG-UI protocol events for frontend consumption, and a zero-dependency SpanCollector turns it into spans for whatever tracing backend you want to export to.

Installation

pip install tokenloop

ag-ui-protocol, anthropic, mcp, and openai are currently all unconditional dependencies — one install gets you both providers and the AG-UI/MCP protocol surface. (No optional extras exist yet.)

Quickstart

import asyncio
import os

from tokenloop import AgentLoop
from tokenloop.provider.anthropic import AnthropicProvider


async def main() -> None:
    agent = AgentLoop(
        AnthropicProvider(model=os.environ["ANTHROPIC_MODEL"]),
        instruction="You are a concise, helpful assistant.",
    )

    # Interactive REPL, good for trying things out:
    await agent.as_cli(None)

    # Or drive it programmatically:
    # message = await agent.run("session-1", "run-1", "What's 2 + 2?")
    # print(message.content)


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

Adding a tool is just a plain Python function:

def get_weather(city: str) -> str:
    """Get the current weather for a city."""
    ...

agent.add_tool(get_weather)

See examples/ for more — tool calling, session state, streaming events, and a skills-based agent.

Testing

uv run pytest -v      # 35 passed, 0 failed, 0 skipped
uv run ruff check     # all checks pass
uv run python examples/smoke_test.py   # offline, no API key needed

The test suite covers HITL resume (all three pause kinds), a HITL-skipped tool not aborting the rest of its batch, RunControlService (pause/resume/cancel), image serialization for both providers, and State's delta-tracking/diff behavior. examples/smoke_test.py exercises the tool-call round trip and a blocking guardrail end-to-end against a fake provider — no network or API key required.

Documentation

There's no separate docs site yet — this README is it, plus:

• docs/HITL_PARITY.md — a deep dive into the HITL/RunControlService design and its parity with the Rust crate.
• Repository
• Releases

Deliberately not ported (yet)

A few things the companion Rust crate has that this port doesn't, by explicit scope decision (see the relevant source docstrings):

• LLM-summarizing compaction — today's SimpleCompressor only trims old messages, it doesn't ask the model to summarize them.
• Hook-triggered HITL pauses — only a tool's own execution body can raise a pause today; a ToolHook.before_call cannot.
• Inline skill-script execution sharing the Bash engine.
• Rust's {key}-style prompt templating.