agenttoolkit

agenttoolkit provides one provider-neutral definition for tools exposed to LLM agents. Define a tool once — schema, availability, metadata, and execution logic — and expose it to OpenAI, Anthropic, or any other provider without duplicating definitions.

It intentionally contains no application-specific tools and no agent loop: it is a building block, not a framework.

Table of contents

• Features
• Installation
• Quickstart
• Defining tools
• Dependency injection with ToolContext
• Conditional availability and descriptions
• Driving an agent loop
• Results and errors
• Middleware
• Merging registries
• Filesystem and shell primitives
• Skills
• Development
• License

Features

• Registration through a @tools.action decorator — no hand-written JSON Schema, for either plain function signatures or Pydantic models.
• Runtime metadata (effects, status, tags, custom fields) and an requires_approval flag, kept out of the model-facing schema but readable by the host loop that dispatches calls.
• Context-based dependency injection (Inject[T]) so tools can receive application services without the model ever seeing them.
• Conditional tool availability and dynamic, context-aware descriptions.
• Sync and async tool execution behind a single async API.
• A composable, opt-in middleware pipeline with provided error-boundary and logging middleware; no middleware is installed implicitly.
• Thin, dependency-free schema adapters for OpenAI and Anthropic tool-call formats.
• Tool implementations return their natural Python values; the execution pipeline does not impose an application-specific result envelope.
• Async filesystem and shell ports with local, Docker, and Bubblewrap implementations for common agent capabilities.
• Local Agent Skills discovery and progressive loading, compatible with the SKILL.md convention.

Installation

uv add mm-agenttoolkit

Requires Python 3.13+. On 3.13, modules that use forward references need from __future__ import annotations, since lazy annotation evaluation (PEP 649) is only native starting with 3.14.

Quickstart

This is the shape of code you actually write and run — define tools with the decorator, hand their schema to the model, execute whichever call it makes, and feed the result back:

from pydantic import BaseModel, Field

from agenttoolkit import (
    CallLoggingMiddleware,
    ErrorBoundaryMiddleware,
    Inject,
    ToolContext,
    Tools,
    ToolSchemaFormat,
)


class SearchParams(BaseModel):
    query: str = Field(description="What to search for")
    limit: int = Field(default=5, ge=1, le=20)


class SearchClient:
    async def search(self, query: str, limit: int) -> list[str]:
        return [query] * limit


tools = Tools(
    context=ToolContext(SearchClient()),
    middleware=(
        ErrorBoundaryMiddleware(),
        CallLoggingMiddleware(),
    ),
)


@tools.action(
    "Search the connected knowledge base.",
    params=SearchParams,
    status=lambda params: f"Searching for {params.query}...",
)
async def search(params: SearchParams, client: Inject[SearchClient]) -> list[str]:
    return await client.search(params.query, params.limit)


# 1. Send the schema to the model.
schema = tools.get_schema(ToolSchemaFormat.ANTHROPIC)

# 2. The model asks to call "search" with {"query": "tool middleware"}.
result: object = await tools.execute(
    "search", {"query": "tool middleware"}
)

# 3. Serialize the value and feed it back to the model.

Here the application explicitly enables error handling and call logging. An unknown tool name, invalid arguments, or an exception inside the tool then comes back as an agent-readable "Tool failed: ..." string, while the full exception is logged.

Defining tools

The @tools.action(...) decorator is the entire surface most code touches. Parameters come from a plain function signature or, for validation and richer schemas, a Pydantic model passed as params=:

@tools.action("Add two integers.")
def add(a: int, b: int) -> int:
    return a + b


class RefundParams(BaseModel):
    order_id: str
    amount: float = Field(gt=0, description="Amount to refund, in USD")


@tools.action(
    "Issue a refund for an order.",
    params=RefundParams,
    status=lambda params: (
        f"Refunding {params.amount} for order {params.order_id}..."
    ),
    tags=["billing", "write"],
    requires_approval=True,
    metadata={"owner": "billing-team"},
)
def refund(params: RefundParams, client: Inject[BillingClient]) -> str:
    client.refund(params.order_id, params.amount)
    return "refunded"

None of status, tags, requires_approval, or metadata are visible to the model — they never appear in the generated JSON Schema. They exist for the host loop that dispatches the call:

• status — a human-readable status message, either a plain string or a callable taking the parsed params. Prefer a callable: the parameter type is inferred from params, giving type checking and IDE navigation. Render it with tool.format_status(args) (e.g. to show "Refunding 20.0 for order o-123..." while the call runs).
• tags — a frozenset[str] for grouping or filtering tools, readable as tool.tags.
• metadata — an arbitrary read-only mapping for anything else the host application needs, readable as tool.extra.
• requires_approval — readable as tool.requires_approval; check it before calling tools.execute(...) if the action needs user confirmation first. agenttoolkit does not enforce approval itself.

tool = tools.get("refund")
tool.tags                # frozenset({"billing", "write"})
tool.extra["owner"]      # "billing-team"
tool.requires_approval   # True
tool.format_status({"order_id": "o-123", "amount": 20.0})
# "Refunding 20.0 for order o-123..."

A callable status gets its field access checked statically by the IDE or type checker; a plain string is rendered as-is.

Prefer tools.action(...) for registering tools. Direct registration is an internal implementation detail.

Dependency injection with ToolContext

ToolContext carries application services that tools need but that should never appear in the model-facing schema. Wrap a parameter in Inject[T] and it is resolved from context at call time instead of being part of the argument schema:

context = ToolContext(SearchClient(), some_other_service)
tools = Tools(context=context)

The context is known up front, so it belongs in the constructor. Dependencies that only materialise later go through context.provide(...) on the instance you already handed over; a context that differs per request goes through the context= argument on execute(...), get_schema(...) and create_action_model(...), which leaves the registry context untouched.

ToolContext.resolve(T) returns the most recently provided instance of type T (or a subclass), searching in reverse insertion order. Useful mutators:

context.provide(extra_service)  # append more dependencies
context.without(SearchClient)   # drop instances of a type
context.clear()                 # remove everything

If an Inject[T] parameter has no default and no matching dependency is found in context, execution returns an agent-readable failure rather than silently passing None.

Conditional availability and descriptions

Use provided(...) and requires(...) to expose a tool only when its dependency is present (and, optionally, satisfies a predicate). Predicates compose with &, |, and ~:

from agenttoolkit import provided, requires


@tools.action(
    "Issue a refund (admin only).",
    available_when=provided(BillingClient)
    & requires(UserInfo, predicate=lambda user: user.is_admin),
)
def refund(order_id: str, amount: float) -> str: ...

Use description_from_context(...) when a tool's description itself should depend on context (e.g. embedding a resolved account name), with a fallback for when the dependency isn't provided:

from agenttoolkit import description_from_context

description = description_from_context(
    BankingClient,
    render=lambda client: f"Look up the balance for {client.account_name}.",
    fallback="Look up account balance.",
)


@tools.action(description)
def balance() -> float: ...

Driving an agent loop

Tools.get_schema(...) returns the schema for every tool available in the active (or a given) context; Tools.execute(...) dispatches a model-produced call:

openai_schemas = tools.get_schema(ToolSchemaFormat.OPENAI)
anthropic_schemas = tools.get_schema(ToolSchemaFormat.ANTHROPIC)

result = await tools.execute("search", {"query": "tool middleware"}, context=context)

A typical loop confirms approval-gated tools before executing, and reports status while a call is in flight:

tool = tools.get(name)
if tool is not None and tool.requires_approval and not confirm(name, arguments):
    result = "Tool failed: Declined by user"
else:
    print(tool.format_status(arguments) if tool else name)
    result = await tools.execute(name, arguments, context=context)

Iterating a Tools instance yields the underlying Tool objects — every registered one, gated or not. Filter with tool.is_available(context) to print a catalog of what a given context actually exposes (tool.name, tool.resolve_description(context), tool.tags, ...).

Results and errors

Tool functions return their natural Python value: text, numbers, collections, Pydantic models, or None. Tools.execute() passes that value through unchanged:

class WeatherResult(BaseModel):
    city: str
    temp_c: float


@tools.action("Get the current weather for a known city")
def get_weather(city: str) -> WeatherResult:
    temp_c = KNOWN_CITIES.get(city.lower())
    if temp_c is None:
        raise ValueError(f"Unknown city: {city!r}")
    return WeatherResult(city=city, temp_c=temp_c)

With ErrorBoundaryMiddleware enabled, exceptions from resolution, validation, middleware, dependency injection, or the tool itself are logged and converted to an agent-readable string prefixed with "Tool failed: ". Consequently tool implementations need no toolkit-specific result import. Without that middleware, exceptions propagate to the caller normally.

Because dispatch by name is dynamic and one registry can contain heterogeneous return types, the static return type of Tools.execute() is object. The application's provider adapter owns serialization and can add any provider-specific metadata at that boundary.

Middleware

Tools never installs middleware implicitly. Tool resolution and argument validation are core execution mechanics performed when the middleware chain invokes the call. With Tools() alone, execution exceptions propagate normally.

List every middleware an agent loop should use explicitly. For example, this enables the provided error boundary, call logging, and a timeout:

from agenttoolkit import (
    CallLoggingMiddleware,
    ErrorBoundaryMiddleware,
    ToolCall,
    ToolMiddleware,
)


class TimeoutMiddleware(ToolMiddleware):
    def __init__(self, seconds: float) -> None:
        self._seconds = seconds

    async def __call__(self, call: ToolCall, next):
        return await asyncio.wait_for(next(call), timeout=self._seconds)


tools = Tools(
    middleware=(
        ErrorBoundaryMiddleware(),
        CallLoggingMiddleware(),
        TimeoutMiddleware(5.0),
    ),
)

Custom middleware receives the raw ToolCall and wraps resolution, validation, dependency injection, and execution through next(call). call.raw_args contains the model-provided arguments; call.tool contains the registered tool when the name exists. This keeps every execution stage inside the explicit chain, allowing an installed error boundary to translate any failure for the agent.

Merging registries

Combine tools from multiple Tools instances — e.g. when composing a registry from several feature modules:

tools.merge(other_tools)               # raises on name collisions
tools.merge(other_tools, replace=True)  # other_tools wins on collisions

Filesystem and shell primitives

agenttoolkit.builtins contains raw async implementations rather than a predefined set of model-facing tools. Applications can use them directly, inject them through ToolContext, or expose only the operations appropriate for a particular agent.

from pathlib import Path

from agenttoolkit.builtins import (
    BindMount,
    CommandDefaults,
    DockerSandbox,
    LocalWorkspace,
    SandboxPolicy,
)

workspace = LocalWorkspace("./project")
await workspace.write_file("src/example.py", "print('hello')\n")

entries = await workspace.list_dir("src")
source = await workspace.read_file(entries[0].path)

output = workspace.root / "output"
output.mkdir(exist_ok=True)
cli_config = Path.home() / ".config" / "my-cli"

policy = SandboxPolicy.for_workspace(
    workspace.root,
    writable=True,
    enable_network_access=True,
)
sandbox = DockerSandbox(
    "my-cli:latest",
    defaults=CommandDefaults(working_directory=workspace.root),
    policy=policy,
    inherit_environment=("MY_CLI_TOKEN",),
    mounts=(
        BindMount.read_only(cli_config, "/home/agent/.config/my-cli"),
        BindMount.read_write(output, "/output"),
    ),
    user="host",
)
async with sandbox:
    result = await sandbox.execute("my-cli build --output /output")

The Workspace port provides read_file, write_file, edit_file, glob, list_dir, and stat. Exploration returns Entry values with a root-relative POSIX path, directory and symlink flags, size, and modification time. Local reads and writes are confined to the workspace root and bounded by a configurable file-size limit.

The CommandRunner port returns a common CommandResult from local and isolated backends. CommandDefaults configures the working directory, environment, timeout, and captured-output limit. SandboxPolicy is separate and contains only isolation requirements: readable and writable paths, network access, memory, process, and CPU limits. A sandbox backend must enforce every requested isolation setting or reject it.

Only backends that own persistent resources expose a lifecycle. DockerSandbox supports open()/close() and async with, starts one container, tunnels every command through docker exec, and removes the container afterwards. This preserves container state and avoids paying container startup latency for every command. If a Docker command times out, the sandbox removes the container to guarantee that no detached process keeps running; call open() again before continuing.

DockerSandbox enforces all sandbox resource limits. BubblewrapSandbox supports filesystem and network isolation and rejects unsupported resource limits. LocalShellRunner executes trusted commands directly and accepts no SandboxPolicy, so it makes no isolation claim.

DockerSandbox also supports named bind mounts and an explicit allowlist of host environment variables. BindMount.read_write(...) writes directly back to the host. inherit_environment fails fast when a requested variable is missing and forwards its name without embedding the secret value in the generated Docker arguments. On POSIX hosts, user="host" maps the container process to the host UID and GID so generated files remain owned by the developer. Use environment={...} on CommandDefaults or env={...} on execute(...) for explicit values and per-call overrides.

Skills

Local Agent Skills are discovered from directories containing one subdirectory per skill, each with a SKILL.md file using YAML frontmatter (name, description, and optional license, compatibility, metadata, allowed-tools) followed by Markdown instructions:

skills/
  internet-research/
    SKILL.md
    references/
      guide.md
    scripts/
      search.py

name must be 1–64 lowercase letters, numbers, or hyphens, and must match its parent directory name.

from agenttoolkit import Skills

skills = Skills.from_dir("./skills")

# Render the compact skill listing for the agent's system prompt.
system_prompt = f"You are helpful.\n\n{skills.render_prompt()}"

# Progressive loading returns full instructions and relative resource paths.
loaded = skills.load("internet-research")
system_prompt += f"\n\n{loaded.instructions}"

# Re-scan the configured directories after skills are added or removed.
changes = skills.refresh()
print(changes.added, changes.updated, changes.removed)

# The application decides which general filesystem and process tools to expose.
guide = read_file(loaded.directory / "references/guide.md")
output = await run_process(
    ["python", "scripts/search.py", "python packaging"],
    cwd=loaded.directory,
)

Skills.from_dir accepts multiple directories; a skill discovered later overrides one with the same name from an earlier directory (logged as a warning). SKILL.md is re-parsed from disk on each load, so instructions can be edited without restarting the process. refresh() rebuilds the registry from the configured directories, picking up added, changed, and removed skills. If discovery fails, the previous registry remains available.

refresh() returns an immutable SkillChanges value containing the registry revision and the added, updated, and removed skill names. refresh_if_changed() first compares a lightweight fingerprint of the SKILL.md paths, modification times, and sizes, avoiding parsing when no skill document changed.

Agents that can write their own skills can attach SkillRefreshMiddleware. By default it checks the registry after every tool call, silently and without touching the tool's own result. The check uses a lightweight fingerprint, so unchanged skill documents are not reparsed:

from agenttoolkit import (
    CallLoggingMiddleware,
    ErrorBoundaryMiddleware,
    SkillRefreshMiddleware,
)

tools = Tools(
    context=ToolContext(skills),
    middleware=(
        ErrorBoundaryMiddleware(),
        CallLoggingMiddleware(),
        SkillRefreshMiddleware(),
    ),
)

The registry is resolved from the call's ToolContext, not captured at construction — a per-call context= argument refreshes the registry actually in use, and the middleware is a no-op when the context holds no Skills.

Invalid skill edits are not activated and the previous registry remains available. Applications that embed skills.render_prompt() in model context should render that dynamic portion again before each model invocation.

load() returns an immutable LoadedSkill containing name, instructions, the absolute skill directory, and sorted relative resources. Resource reading, process execution, timeouts, sandboxing, and permissions deliberately belong to the application's general filesystem and process tools instead of the Skills API. Skill directories and their scripts must still be treated as trusted code.

Development

Install the locked development environment and run all quality checks:

uv sync --locked
uv run --locked ruff check .
uv run --locked pytest

The test command measures branch coverage for agenttoolkit and fails below 90%. Dependabot groups Python dependency updates into one weekly pull request; the same CI matrix validates every update on Python 3.13 and 3.14.

See CONTRIBUTING.md for the full contribution workflow and conventions.

License

MIT — see LICENSE.md.