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"

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

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.

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=...).