Tool System¶

The executable logic is a plain async def handler. Tool is the glue that declares that function as an agent tool. It attaches the name and execution policy, and exposes the function signature and docstring to the model. A tool call is different. It is one request from the model to invoke that Tool.

Plain async function¶

from pathlib import Path
from axio import Tool


async def write_file(path: str, content: str) -> str:
    """Write content to a file at the given path."""
    Path(path).write_text(content)
    return f"Wrote {len(content)} bytes to {path}"


tool = Tool(name="write_file", handler=write_file)

Wrapping the handler in Tool makes it available to an Agent. The handler’s docstring becomes the description sent to the LLM. Function annotations are converted to a JSON schema object automatically. No decorators or schema registration are needed.

Use Annotated + Field to add descriptions, defaults, or numeric bounds:

from typing import Annotated
from axio import Tool, Field


async def search(
    query: Annotated[str, Field(description="Search query")],
    limit: Annotated[int, Field(default=10, ge=1, le=100)] = 10,
) -> str:
    """Search the knowledge base."""
    return f"results for {query!r} (limit={limit})"


tool = Tool(name="search", handler=search)
result_default = search.__defaults__
assert result_default == (10,)

Context injection¶

When a tool needs access to runtime state (a database connection, a sandbox object, etc.), use CONTEXT.get() inside the function and pass the value via Tool(context=...):

import asyncio
from typing import Annotated
from axio import Tool, CONTEXT, Field


async def search(
    query: Annotated[str, Field(description="Search query")],
    limit: Annotated[int, Field(default=10, ge=1, le=100)] = 10,
) -> str:
    """Search a list of documents."""
    documents: list[str] = CONTEXT.get()
    results = [s for s in documents if query.lower() in s.lower()]
    return "\n".join(results[:limit]) or "no results"


documents = ["Axio is async", "Pydantic is great", "Axio uses protocols"]
t = Tool(name="search", handler=search, context=documents)

result = asyncio.run(t(query="axio"))
assert "Axio" in result

Nested helpers that cannot receive arguments can also call CONTEXT.get():

import asyncio
from axio import Tool, CONTEXT


def helper() -> str:
    return str(CONTEXT.get())  # works even without an explicit argument


async def ping(msg: str) -> str:
    """Echo msg with context from ContextVar."""
    return f"{msg}:{helper()}"


t = Tool(name="ping", handler=ping, context="ctx-42")
assert asyncio.run(t(msg="hello")) == "hello:ctx-42"

Tools, instructions, and selection¶

Agent(tools=[...]) supplies a fixed base set. A tool provider can contribute additional tools at each iteration, and a loaded instruction can add tools needed for its procedure. The agent collects these sets, attaches its tool_guards, and passes the result to the Tool Selection.

agent.get_tools() returns the collected tools before selection. Only tools selected for the current model request can run. Two different tool objects with the same name are a configuration error; repeated references to the same object are allowed. A provider entry-point name does not prefix the tool names it supplies.

Integrations that own tools and instructions¶

A filesystem integration can supply file tools and the project’s conventions. An MCP integration needs a live connection as well as the tools discovered on that connection. A ToolProvider keeps each integration’s tools, guidance, and resource lifetime together. The transport still handles communication with the language model; a tool provider supplies what the agent can use to work.

For local project tools, install axio-tools-local and use its provider directly:

import asyncio
from pathlib import Path
from tempfile import TemporaryDirectory

from axio import Agent, MemoryContextStore
from axio.plugins import open_tool_providers
from axio.testing import StubTransport, make_text_response
from axio_tools_local.provider import LocalOptions, LocalToolProvider


async def main() -> None:
    with TemporaryDirectory() as directory:
        root = Path(directory)
        (root / "AGENTS.md").write_text("Use the project's formatting conventions.")
        provider = LocalToolProvider()
        options = LocalOptions(
            cwd=root,
            roots=(root,),
            isolated=True,
            instruction_files=(root / "AGENTS.md",),
        )
        async with open_tool_providers([provider], config=options):
            agent = Agent(
                system="Help with this project.",
                transport=StubTransport([make_text_response("Ready.")]),
                providers=[provider],
            )
            provider.bind(agent)
            assert "read_file" in {tool.name for tool in agent.get_tools()}
            assert await agent.run("Hello", MemoryContextStore()) == "Ready."


asyncio.run(main())

cwd controls relative tool paths; roots describes independent project roots. The explicit instruction file loads even in isolated mode. bind(agent) adds the local provider’s scoped instruction checks and prompt contribution. Bind each copied agent separately. The stub avoids a model request; replace it with your completion transport for real work.

Keep runs inside open_tool_providers(). It calls init() and registers cleanup before initialization, then closes providers in reverse order even after a partial startup, exception, or cancellation. An initialization exception propagates after cleanup. Separate contexts let independent integrations fail separately.

An agent reads all_tools and get_instructions() at each iteration, so it sees the provider’s current catalog after reload. The application owns reload and configuration. A copied list from collect_tools() is a snapshot: passing it as Agent(tools=...) does not keep the provider’s instructions or dynamic catalog.

For installed integrations, use discover_tool_providers() and select entries from its catalog. Constructing an Agent does not discover or open integrations. See Register and discover plugins for registration and diagnostics and MCP Server Tools for a provider backed by server sessions.

Tool dataclass¶

Every handler function is wrapped in a Tool frozen dataclass:

@dataclass(frozen=True, slots=True)
class Tool[T]:
    name: str
    handler: Callable[..., Awaitable[Any]]
    description: str = ""  # defaults to handler.__doc__
    guards: tuple[PermissionGuard, ...] = ()
    concurrency: int | None = None
    context: T = ...  # default: empty mapping
    schema: MappingProxyType[str, Any] = ...  # default: built from the handler
    access: AccessDescriptor | None = None
    permission_identity: str = ""

handler: An async def function. A fresh call is made per invocation with validated kwargs.
description: Defaults to handler.__doc__. Pass an explicit string to override.
guards: Guards run sequentially before the handler. Each receives the Tool object and the raw kwargs, and either returns a (possibly modified) kwargs dict (allow) or raises GuardError (deny).
concurrency: Limits parallel invocations of this tool via an asyncio.Semaphore.
context: Arbitrary runtime state available via CONTEXT.get() inside the handler. Use this to inject a database connection, a sandbox object, or any other state the handler needs without touching global state.
schema: The JSON Schema sent to the model. Built from the handler’s annotations when you do not pass one. An explicit schema also replaces the fields used for validation and default injection. It becomes the filter that strips keys the model invented.
access: Describes the operation’s effects for instruction preflight and permission policy. For example, file_access("write", "path") declares the file written by a handler. The descriptor must report all effects without performing them. Missing declarations mean unknown effects. See Declare what a tool will do.
permission_identity: Provider-supplied identity included in opaque-call approvals. Providers can distinguish differently configured instances of an otherwise identical tool, so permission for one server configuration does not authorize another.

Input schema¶

@property
def input_schema(self) -> JSONSchema:
    return copy.deepcopy(dict(self.schema))

Transports send this schema to the LLM so it knows how to call the tool.

Returning content blocks¶

handler is typed Callable[..., Awaitable[Any]], not Awaitable[str]. The widening is load-bearing. A handler may return a list of content blocks: TextBlock, ImageBlock, AudioBlock, VideoBlock. The agent puts them straight into the ToolResultBlock, so a tool that reads a screenshot hands the model pixels rather than a description of them. Anything else is stringified. The agent additionally re-emits media from a tool result as ImageOutput / AudioOutput / VideoOutput events so a harness can save it. See Stream Events.

Streaming tools¶

A tool whose output arrives over time - a shell command, a long build - can stream it. Attach an async generator yielding (key, text) pairs to the handler as .stream, and optionally a format_stream_result(chunks) for the finished result:

async def _shell_stream(command: str, timeout: int = 5):
    yield ("stdout", "compiling...\n")
    yield ("stderr", "[exit code: 0]")


shell.stream = _shell_stream
shell.format_stream_result = staticmethod(_format_records)

Tool.supports_streaming is then true. The agent dispatches through Tool.call_streaming(), emitting a ToolOutputDelta per chunk while the tool is still running. The model still sees one finished result, built by Tool.format_stream_result() from the chunks and their timings. The default is plain concatenation. key is the tool’s own channel name. axio-tools-local uses stdout and stderr.

Guards and validation run first either way. call_streaming() acquires the same semaphore, validates the same kwargs, and calls the same guards before the first chunk is produced.

Execution flow¶

        sequenceDiagram
    participant Agent
    participant Tool
    participant Guard
    participant Handler

    Agent->>Tool: __call__(**kwargs)
    Tool->>Tool: Acquire semaphore (if set)
    Tool->>Tool: Inject defaults + validate types/bounds
    loop For each guard
        Tool->>Guard: check(tool, **kwargs)
        Guard-->>Tool: kwargs (or raise GuardError)
    end
    Tool->>Handler: handler(**kwargs)
    Handler-->>Tool: result string
    Tool-->>Agent: result

The agent calls tool(**kwargs) with the input the model provided.
If the tool has a concurrency limit, it acquires the semaphore.
Missing fields with defaults are injected. Provided fields are validated (type, Literal, bounds).
Each guard in the guards tuple is called sequentially with the fully materialised kwargs.
Guards return a (possibly modified) kwargs dict to allow, or raise GuardError to deny.
The handler is called with the materialised kwargs (stray keys stripped unless handler accepts **kwargs).
Any exception from the handler is wrapped in HandlerError.

Exception hierarchy¶

AxioError
└── ToolError
    ├── GuardError    # Guard denied or crashed
    └── HandlerError  # Handler raised during execution

The agent catches both and wraps the error message in a ToolResultBlock with is_error=True, so the model can see what went wrong and retry or adjust its approach.

ToolSelector¶

The ToolSelector protocol allows a component to filter or rank the full set of available tools before each LLM call.

from collections.abc import Iterable
from typing import Any, Protocol, runtime_checkable
from axio.messages import Message
from axio import Tool


@runtime_checkable
class ToolSelector(Protocol):
    async def select(
        self,
        messages: Iterable[Message],
        tools: Iterable[Tool[Any]],
    ) -> Iterable[Tool[Any]]: ...

A selector is useful when you have a large tool catalogue and want to avoid sending every tool’s schema to the model on every turn. Embeddings or keyword matching can pick only the relevant tools.

Pass a ToolSelector implementation directly to Agent(selector=...).