Agent Guide¶

The Agent module provides a high-level, batteries-included interface that combines the context pipeline with any LLM provider. It handles streaming chat, automatic tool use loops, memory management, and agentic RAG -- all through a fluent builder API.

Overview¶

The Agent class wraps three systems into a single entry point:

1. ContextPipeline -- assembles system prompts, memory, and retrieval context
2. LLM Provider -- streams responses via the LLMProvider protocol with automatic retries
3. Tool loop -- executes tools and feeds results back to the model

User message
    |
    v
ContextPipeline.build()  -->  system + messages
    |
    v
LLM Provider (streaming)
    |
    v
Tool use?  -- yes --> execute tools --> feed back --> loop
    |
    no
    v
Yield text chunks

Quick Start¶

from anchor import Agent

agent = (
    Agent(model="claude-haiku-4-5-20251001")
    .with_system_prompt("You are a helpful assistant.")
)

for chunk in agent.chat("What is context engineering?"):
    print(chunk, end="", flush=True)

Constructor¶

Agent(
    model: str = "claude-haiku-4-5-20251001",
    *,
    api_key: str | None = None,
    llm: LLMProvider | None = None,
    fallbacks: list[str] | None = None,
    max_tokens: int = 16384,
    max_response_tokens: int = 1024,
    max_rounds: int = 10,
)

Fluent Configuration¶

All configuration methods return self for chaining:

from anchor import Agent, MemoryManager, InMemoryEntryStore

memory = MemoryManager(store=InMemoryEntryStore())

agent = (
    Agent(model="claude-haiku-4-5-20251001")
    .with_system_prompt("You are a helpful coding assistant.")
    .with_memory(memory)
)

with_system_prompt¶

agent.with_system_prompt(prompt: str) -> Agent

Sets the system prompt. Clears any previous system prompt and registers the new one with the underlying pipeline.

with_memory¶

agent.with_memory(memory: MemoryManager) -> Agent

Attaches a MemoryManager for conversation history and persistent facts. The agent automatically records user messages, assistant responses, and tool calls in memory.

with_tools¶

agent.with_tools(tools: list[AgentTool]) -> Agent

Adds tools (additive -- multiple calls accumulate tools). Each AgentTool is exposed to the model during the tool-use loop.

with_skill / with_skills¶

agent.with_skill(skill: Skill) -> Agent
agent.with_skills(skills: list[Skill]) -> Agent

Registers one or more Skills. Always-loaded skills have their tools available immediately. On-demand skills are advertised in a discovery prompt and activated via the activate_skill meta-tool.

Chat Methods¶

Synchronous Streaming¶

for chunk in agent.chat("Explain quantum computing"):
    print(chunk, end="", flush=True)

chat() returns an Iterator[str] that yields text chunks as they arrive. If the model calls tools, the agent executes them and feeds results back automatically, continuing until a final text response or max_rounds is reached.

Async Streaming¶

async for chunk in agent.achat("Explain quantum computing"):
    print(chunk, end="", flush=True)

achat() is the async counterpart. It uses pipeline.abuild() and async iteration over the streaming API.

Event Stream¶

chat()/achat() are text-only projections of a richer stream. stream()/astream() expose the full tool-use loop as one ordered sequence of typed events (AgentEvent, discriminated by .type) — built for UIs that need more than text:

from anchor import (
    RoundStarted, TextDelta, ToolStarted, ToolFinished,
    RoundFinished, TurnFinished,
)

async for event in agent.astream("Find and summarize the report"):
    match event:
        case RoundStarted(round=r, max_rounds=n):
            status.update(f"round {r + 1}/{n}")
        case TextDelta(text=text):
            log.write(text)
        case ToolStarted(tool_call_id=cid, name=name):
            spinners[cid] = show_spinner(name)
        case ToolFinished(tool_call_id=cid, is_error=err):
            spinners.pop(cid).done(error=err)
        case RoundFinished(usage=usage):
            meter.add(usage.prompt_tokens, usage.completion_tokens)
        case TurnFinished(diagnostics=diag):
            status.update(f"done ({diag.stopped_by})")

The event vocabulary: TurnStarted, RoundStarted, TextDelta, ToolStarted/ToolFinished (correlated by tool_call_id — in the async path, consecutive read_only tool calls run concurrently and writes run alone, with finish events arriving live in completion order), CompactionStarted/CompactionFinished, RoundFinished (with per-round RoundUsage), UsageLimitReached (a usage limit was crossed; see below), and a terminal TurnFinished carrying the final text and TurnDiagnostics.

Tool failures surface as ToolFinished(is_error=True), not exceptions; only turn-level failures (provider, MCP) raise. Events forwarded from a subagent's run arrive flat in the same stream with parent_tool_call_id set to the invoking tool call — top-level events leave it None, which is also how the text projections filter subagent text out.

Usage Limits¶

with_usage_limits caps what the whole run may spend — the turn and every subagent it spawns debit one shared pool. The loop-level complement to the pipeline's TokenBudget:

from anchor import UsageLimits

agent.with_usage_limits(UsageLimits(
    total_tokens_limit=50_000,   # prompt+completion+cache, all rounds
    tool_calls_limit=20,
    cost_limit=2.50,             # USD (optional)
))

Crossing a limit never raises, at any level. The loop emits a UsageLimitReached event (scope="run" for the shared pool, scope="turn" for an agent's own per-turn limit), gives the model one wrap-up round (the final-round notice plus tool_choice="none"), and ends the turn with stopped_by="usage_limit" in TurnFinished.diagnostics / agent.last_turn. A subagent cut mid-run wraps up the same way and returns its partial result, marked [partial result] on the plain-text path; a structured-output child's partiality is visible in ChildTurn.diagnostics.stopped_by (a marker would corrupt the JSON). A child that starts on an already exhausted pool gets exactly one wrap-up round. Checks are post-hoc, so a run can overshoot by the rounds in flight (one per concurrently running agent) plus the bounded wrap-up calls. Rounds themselves are capped by max_rounds.

A subagent may carry its own narrower limits (SubagentDefinition(usage_limits=...), or with_usage_limits on an agent wrapped with as_tool); both its per-turn limits and the run pool are enforced, so the effective budget only narrows. Aggregated accounting lands in TurnDiagnostics.children and the run_total_tokens / run_total_tool_calls / run_total_cost_usd properties.

cost_limit is priced per round from, in order: the provider-reported billed cost (claude_cli sends it), the runtime-overridable anchor.llm.pricing.MODEL_PRICING table, and genai-prices when the [pricing] extra is installed. A model no source knows logs one warning and debits $0 — its tokens still count. In prompted output mode, run()/arun() hold one pool across the retry turns, so retries cannot re-arm the budget.

On providers that do not report usage on the stream (all but Anthropic today), per-round prompt_tokens/completion_tokens are estimated with the agent's tokenizer, so the limit is enforced consistently on every provider. Compaction summarization calls are not counted (the summary is counted in the next round's prompt).

Approval (Human-in-the-Loop)¶

Mark tools that need a human decision and register an approval callback — the tool call pauses until it resolves:

from anchor import ApprovalDecision, ApprovalRequest, tool

@tool(requires_approval=True)
def delete_records(table: str) -> str:
    """Delete all records from a table."""
    ...

async def approve(request: ApprovalRequest) -> ApprovalDecision:
    ok = await show_dialog(request.name, request.tool_input)
    return ApprovalDecision(
        approved=ok, reason=None if ok else "user declined — suggest a dry run",
    )

agent.with_tools([delete_records]).with_approval(approve)

A pre-hook can also route any call to the callback with HookResult(decision="ask"). Deny becomes an is_error tool result carrying the reason, so the model adjusts instead of retrying blindly; approve may rewrite the input via updated_input. A gated call with no callback configured fails closed. Async callbacks require astream()/achat() and may stay pending indefinitely — approval timeouts are the application's decision. Approval resolves before a call takes an execution slot, and calls batched together (consecutive read_only tools) run their callbacks concurrently — serialize inside the callback if your UI needs one prompt at a time. Durable pause (ending the turn with pending approvals and resuming later) is planned on top of the conversation store.

Structured Output¶

from pydantic import BaseModel

class Report(BaseModel):
    summary: str
    severity: int

report = agent.with_output_model(Report).run("Assess the incident.")

The default tool mode adds a synthetic final_result tool carrying the schema and forces tool_choice="any" — the model cannot stop in plain text; calling final_result with valid arguments ends the turn. Invalid arguments come back as an error tool result (the loop's own retry mechanic), bounded by max_output_retries; exhausting it ends the turn with stopped_by="output_missing" — stream()/chat() see that in the diagnostics, run() raises. Real tools still run in the same turn. The normalized JSON is also available in TurnFinished.output and agent.last_output. mode="prompted" injects the schema into the prompt instead and validates the text reply (the subagent mechanic) — the portable fallback for models without tool calling; it requires an agent without with_memory, since the schema and the retry turns would otherwise land in the persisted conversation.

Memory Tool¶

with_memory_tool(path) gives the model persistent memory compatible with Anthropic's memory_20250818 command set — on every provider:

agent.with_memory_tool("./agent-memory")

The model sees one memory tool (view / create / str_replace / insert / delete / rename) rooted at the virtual /memories directory, mapped into the given path with strict containment (path traversal is rejected). The memory protocol is appended to the system prompt. Pairs naturally with with_compaction/with_context_management: memory holds what must survive summarization.

Accessing the Last Result¶

After calling chat() or achat(), the full ContextResult is available:

for chunk in agent.chat("Hello"):
    pass

result = agent.last_result
print(result.diagnostics)

The @tool Decorator¶

The @tool decorator converts a plain function into an AgentTool with auto-generated JSON Schema from type hints:

from anchor import tool

@tool
def get_weather(city: str, units: str = "celsius") -> str:
    """Get the current weather for a city."""
    return f"Weather in {city}: 22 {units}"

The decorator extracts:

• name from fn.__name__ (override with name=)
• description from the first docstring paragraph (override with description=)
• input_schema from type hints (override with input_model=)

Parameterized Usage¶

from pydantic import BaseModel
from anchor import tool

class SearchInput(BaseModel):
    query: str
    max_results: int = 5

@tool(name="search", description="Search the knowledge base", input_model=SearchInput)
def search_kb(query: str, max_results: int = 5) -> str:
    """Search the knowledge base."""
    return f"Found {max_results} results for: {query}"

Async Tools¶

An async def under @tool (or as AgentTool.fn) is awaited: on the async path (achat/astream) with the tool's timeout, and from chat() in a fresh event loop. Calling chat() from inside a running loop cannot await it: that raises TypeError to the caller, the same as an async approval callback under stream().

@tool
async def fetch(url: str) -> str:
    """Fetch a page."""
    async with httpx.AsyncClient() as client:
        return (await client.get(url)).text

Three Tiers of Tool Creation¶

AgentTool Model¶

AgentTool is a frozen Pydantic model:

from anchor import AgentTool

my_tool = AgentTool(
    name="lookup",
    description="Look up a value",
    input_schema={
        "type": "object",
        "properties": {"key": {"type": "string"}},
        "required": ["key"],
    },
    fn=lambda key: f"Value for {key}",
)

Key methods:

• to_tool_schema() -- Provider-agnostic ToolSchema
• validate_input(tool_input) -- Returns (bool, str) tuple

Skills System¶

Skills group related tools into discoverable units with optional on-demand activation. This implements progressive tool disclosure -- the model starts with a small tool set and can activate more capabilities as needed.

Skill Model¶

from anchor import Skill

my_skill = Skill(
    name="data_analysis",
    description="Tools for analyzing datasets",
    instructions="Use these tools when the user asks about data trends.",
    tools=(analyze_tool, summarize_tool),
    activation="on_demand",  # or "always"
    tags=("analytics",),
)

Activation Modes¶

Always-loaded skills have their tools available from the first API round:

agent.with_skill(Skill(
    name="utils",
    description="Utility tools",
    tools=(calc_tool,),
    activation="always",
))

On-demand skills are advertised in a discovery prompt. The agent calls the auto-generated activate_skill meta-tool to make their tools available:

agent.with_skill(Skill(
    name="advanced_search",
    description="Advanced search with filters and facets",
    tools=(faceted_search_tool, filter_tool),
    activation="on_demand",
))

SkillRegistry¶

The SkillRegistry manages skill registration and activation state:

from anchor import SkillRegistry, Skill

registry = SkillRegistry()
registry.register(my_skill)

# Check status
registry.is_active("data_analysis")  # False (on_demand, not yet activated)

# Activate
skill = registry.activate("data_analysis")

# Get all active tools
tools = registry.active_tools()

Built-in Skills¶

memory_skill¶

Creates a skill with four CRUD tools for persistent user facts:

from anchor import Agent, MemoryManager, InMemoryEntryStore, memory_skill

memory = MemoryManager(store=InMemoryEntryStore())

agent = (
    Agent(model="claude-haiku-4-5-20251001")
    .with_memory(memory)
    .with_skill(memory_skill(memory))
)

The memory skill provides:

rag_skill¶

Creates an on-demand skill with a search_docs tool for agentic RAG:

from anchor import Agent, rag_skill

agent = (
    Agent(model="claude-haiku-4-5-20251001")
    .with_skill(rag_skill(retriever=my_retriever, embed_fn=my_embed_fn))
)

The model decides when to activate the skill and search documentation, making this agentic RAG -- retrieval timing is model-controlled.

Putting It All Together¶

from anchor import (
    Agent, MemoryManager, InMemoryEntryStore,
    memory_skill, rag_skill, tool,
)

# Custom tool
@tool
def calculate(expression: str) -> str:
    """Evaluate a math expression."""
    try:
        result = eval(expression)  # noqa: S307
        return str(result)
    except Exception as e:
        return f"Error: {e}"

# Setup
memory = MemoryManager(store=InMemoryEntryStore())

agent = (
    Agent(model="claude-haiku-4-5-20251001", max_rounds=5)
    .with_system_prompt("You are a helpful assistant with memory and tools.")
    .with_memory(memory)
    .with_tools([calculate])
    .with_skill(memory_skill(memory))
)

# Chat
for chunk in agent.chat("Remember that my favorite color is blue"):
    print(chunk, end="", flush=True)

Error Handling and Retries¶

The underlying BaseLLMProvider retries on transient errors with exponential backoff. Transient errors include:

• RateLimitError -- 429 responses (respects retry_after header)
• ServerError -- 5xx responses
• LLMTimeoutError -- request timeouts

Non-transient errors (AuthenticationError, ModelNotFoundError, ContentFilterError) raise immediately.

Tool execution errors are caught and returned as error messages to the model, allowing it to recover gracefully.

See the LLM Providers Guide for details on fallback chains and the error hierarchy.

See Also¶

• LLM Providers Guide -- multi-provider setup and fallbacks
• Pipeline Guide -- underlying context assembly
• Memory Guide -- memory management details
• Agent API Reference -- complete API signatures
• LLM API Reference -- provider protocol, models, and errors
• Formatters Guide -- output formatting