Skip to content

aimu.tools

In-process @tool decorator and cross-process MCPClient.

Decorator

aimu.tools.tool 

tool(func: Callable) -> Callable

Mark a Python function as an AIMU tool.

Inspects the signature and docstring at decoration time and attaches an OpenAI-format tool spec to func.__tool_spec__. The function itself is unchanged and remains directly callable.

Each parameter must either have a type hint or a default value. Variadic parameters (*args / **kwargs) are not supported; declare each argument explicitly.

Supported parameter types: str, int, float, bool, list, dict, plus Optional[T] and T | None (which unwrap to the inner type). A Literal[...] parameter becomes a JSON Schema enum so the model sees the exact allowed values.

The docstring is parsed Google-style: the prose before the first section header (Args:, Returns:, ...) becomes the tool description, and an Args: / Arguments: / Parameters: section supplies per-parameter descriptions (each name: text entry, continuation lines allowed). Required vs. optional args are derived from default values.

A tool may be plain (def fn() -> T), async (async def fn() -> T), a generator (def fn(): yield ...; return T), or an async generator (async def fn(): yield ...). Generator and async-generator tools stream :class:~aimu.models.StreamChunk objects during execution; the agent forwards each yielded chunk through its own stream and treats the final yielded TOOL_CALLING chunk's content["response"] as the canonical tool result. The decorator sets discriminator attributes:

  • func.__tool_is_async__: True for async def or async def + yield.
  • func.__tool_is_streaming__: True for generator functions (sync or async).

Usage::

import aimu

@aimu.tool
def letter_counter(word: str, letter: str) -> int:
    """Count occurrences of a letter in a word."""
    return word.lower().count(letter.lower())

agent = Agent(client, tools=[letter_counter])

aimu.tools.ToolSignatureError

Bases: TypeError

Raised by @tool when a function signature can't be converted to a tool spec.

Argument validation

aimu.tools.coerce_tool_arguments 

coerce_tool_arguments(fn: Callable, arguments: dict) -> dict

Validate and lax-coerce model-supplied tool arguments against fn's type hints.

Returns a new dict of model-facing arguments coerced to their declared types ("5" -> 5, "true" -> True). Raises :class:ToolArgumentError with a single self-contained message when arguments are unknown, a required one is missing, or a value can't be coerced.

Callables not built by @tool (e.g. MCP as_tools() wrappers) carry no __tool_param_adapters__; their arguments pass through unchanged (the MCP server validates them).

aimu.tools.ToolArgumentError

Bases: ValueError

Raised when model-supplied tool-call arguments fail validation/coercion.

Caught at dispatch and surfaced to the model as a tool result so it can self-correct.

MCP client

aimu.tools.MCPClient 

MCPClient(config: Optional[dict] = None, server: Optional[FastMCP] = None, file: Optional[str] = None, url: Optional[str] = None, auth=None, headers: Optional[dict] = None)

Synchronous wrapper around an async FastMCP Client.

Uses anyio's start_blocking_portal() to run the FastMCP Client in a background thread with a properly initialized anyio event loop.

Pass exactly one of config, server, file, or url (a remote HTTP/SSE server). auth (a bearer-token string, the literal "oauth", or a configured FastMCP OAuth / httpx.Auth provider object) and headers apply only with url; a provider object is passed straight to the FastMCP Client (and can't be combined with headers). Connection errors are re-raised as :class:MCPConnectionError with the original exception chained.

close 

close() -> None

Close the MCP connection and release the blocking portal. Idempotent.

Call this when the client is held for the life of the process. Leaving teardown to __del__ does not work there: by the time the interpreter is finalizing, the portal's event loop is gone, and the cross-thread call that closes the connection blocks forever, hanging the process on exit. Closing explicitly happens while the portal can still answer.

ping 

ping() -> list

Verify the connection is alive by listing tools. Returns the tool list.

Raises :class:MCPConnectionError if the connection has been closed or the server is unreachable.

as_tools 

as_tools() -> list

Return this server's tools as @tool-style callables.

Each callable closes over this client, invokes :meth:call_tool cross-process, and returns the result's text content as a string. The callables carry __tool_spec__ (OpenAI format), __tool_is_async__ = False, and __tool_is_streaming__ = False, so they drop straight into client.tools or Agent(tools=...) and dispatch through the same path as @tool functions, no model_client.mcp_client reference needed::

mcp = MCPClient(server=my_server)
client.tools = builtin.web + mcp.as_tools()

The tool list is a snapshot taken now (one list_tools() round-trip); call as_tools() again to pick up server-side changes. Keep a reference to this MCPClient (or to the returned callables, which hold one) for the lifetime of the connection.

aimu.tools.MCPConnectionError

Bases: RuntimeError

Raised when an :class:MCPClient fails to establish or use a connection.

Built-in tools

The aimu.tools.builtin module ships ready-made @tool functions grouped by domain:

Group Tools
builtin.web get_weather, get_webpage, get_webpage_html, web_search, wikipedia
builtin.fs list_directory, read_file
builtin.compute calculate, execute_python, run_command
builtin.time get_current_date_and_time, convert_time
builtin.misc echo
builtin.ALL_TOOLS All of the above except execute_python and run_command (isolation, not containment; opt in via builtin.compute)

execute_python runs code in a fresh subprocess (its own hard timeout, crash isolation, and no access to this process's environment variables -- but it does not confine the filesystem or the network; see its docstring). execute_python_in_process is the explicit opt-in for trusted code where subprocess startup cost matters; it is not included in builtin.compute or builtin.ALL_TOOLS.

run_command runs a command line through /bin/sh -c (COMSPEC /c on Windows), sharing execute_python's subprocess supervisor rather than a separate implementation:

run_command(command, cwd="", timeout=30)

timeout is seconds, clamped to 600. It returns the exit code plus stdout and stderr labelled separately; a nonzero exit returns that output rather than an error string, since pytest exits 1 with the answer on stdout and git diff --exit-code exits 1 to mean "yes, there is a diff." Unlike execute_python, there is no memory cap: a 512 MB address-space limit breaks compilers and test suites, and imposing one on a shell child needs preexec_fn, which is neither portable nor safe alongside threads.

Not a security boundary. This is isolation, not containment, one step sharper than execute_python: the command reaches credentials sitting in files (a .env, ~/.aws/credentials) as the calling user, and process signalling is unconfined, so kill -9 against the host process is one command away. Gate it with tool_approval for untrusted callers and reach for a container when you need real containment.

Unlike execute_python, run_command is not added by make_tools(allow_code_execution=True): that flag names code execution, and widening it would hand a shell to every caller already passing it, so builtin.compute is the only route in. make_command_tool(env_passthrough=...) builds a variant whose child also sees the named environment variables, for callers that need gh or ssh to work; run_command itself is that factory called with no arguments, so no extra variable reaches the child beyond its default allowlist.

aimu.tools.builtin.echo 

echo(echo_string: str) -> str

Returns echo_string.

aimu.tools.builtin.get_current_date_and_time 

get_current_date_and_time(timezone: Optional[str] = None) -> str

Returns the current date and time, with its UTC offset and timezone.

Parameters:

Name Type Description Default
timezone Optional[str]

IANA timezone name (e.g. "Asia/Tokyo", "America/New_York") to report the time in. Omit for the local timezone.

 None

aimu.tools.builtin.convert_time 

convert_time(datetime_str: str, from_timezone: str, to_timezone: str) -> str

Converts a date and time from one timezone to another.

Both timezones must be IANA names ("America/Los_Angeles"), not abbreviations ("PST") or spoken names ("Pacific Time"). Example call: convert_time("2026-08-11T05:00:00", "America/Los_Angeles", "Europe/Zurich").

Handles daylight saving time, and flags times that a DST transition makes nonexistent or ambiguous.

Parameters:

Name Type Description Default
datetime_str str

Date and time, ISO 8601 preferred (e.g. "2026-11-02T15:00:00"). A 12-hour clock and an unpadded hour are also accepted ("2026-11-02 3:00 PM"). The date is required. If the time already carries a UTC offset, that offset is used and from_timezone is ignored.

 required
from_timezone str

IANA timezone name the time is given in (e.g. "Europe/Berlin").

 required
to_timezone str

IANA timezone name to convert to (e.g. "America/Denver").

 required

aimu.tools.builtin.get_weather 

get_weather(location: str) -> str

Returns the current weather for a given location.

Parameters:

Name Type Description Default
location str

City name or coordinates (e.g. "London", "48.8566,2.3522").

 required

aimu.tools.builtin.calculate 

calculate(expression: str) -> str

Evaluates a simple arithmetic expression and returns the result.

aimu.tools.builtin.execute_python 

execute_python(code: str) -> str

Execute Python code in a fresh subprocess and return its output.

NOT A SECURITY BOUNDARY: this is isolation, not containment. Running the code in a subprocess buys a hard timeout (an in-process exec could hang this process forever) and crash isolation (an ordinary crash or early exit in the code brings down only the child -- though nothing stops the code from reaching back out and taking this process down anyway, e.g. os.kill(os.getppid(), signal.SIGKILL), since process signalling isn't confined any more than the filesystem is; see below). It also buys no mutation of this process's imports or global state, a memory cap on Linux (best-effort on other POSIX platforms, absent on Windows -- a one-time warning is logged if the cap can't be applied, rather than failing silently), and no access to this process's environment variables: ANTHROPIC_API_KEY and anything else set there is invisible to the child.

That last point is about environment variables specifically, not credentials in general. It does not confine filesystem or network access: the child runs as the same user and can read, write, and make requests exactly as this process can, which means a .env file, ~/.aws/credentials, ~/.config/gh/hosts.yml, or any other credential sitting on disk is exactly as readable to the child as to this process's own user account. Treat any code reaching this tool as code you have chosen to run; gate it with tool_approval for untrusted callers, and reach for a container when you need real containment.

The child also runs with restricted builtins and an import allowlist, the same accident guard execute_python_in_process uses (stops accidents, not a determined attempt) -- kept identical between both backends so switching between them changes only where the code runs, not what it's allowed to do.

Captures stdout and the value of the last expression. Pre-imports math, statistics, json, re, itertools, functools, datetime, zoneinfo, and numpy/pandas/scipy/matplotlib when installed. For trusted code where the subprocess startup cost matters, see execute_python_in_process (same disclosed limits, weaker isolation).

Parameters:

Name Type Description Default
code str

Python code to execute.

 required

aimu.tools.builtin.execute_python_in_process 

execute_python_in_process(code: str) -> str

Execute Python code in this process and return its output. Explicit opt-in.

NOT A SECURITY BOUNDARY: this is isolation, not containment, and weaker isolation than execute_python (the default execute-code tool) -- code here shares this process's memory, imports, and environment variables (including any API keys sitting in os.environ), a hang blocks this process indefinitely, and a crash takes it down too. It runs with restricted builtins and an import allowlist, which stops accidents, not a determined attempt: a one-line expression reaches subprocess.Popen through the type hierarchy, and the filesystem through an allowlisted module's transitive attributes. Use this only for trusted code where the subprocess startup cost of execute_python matters; gate it with tool_approval otherwise, and reach for a container when you need real containment.

Captures stdout and the value of the last expression. Pre-imports math, statistics, json, re, itertools, functools, datetime, zoneinfo, and numpy/pandas/scipy/matplotlib when installed.

Parameters:

Name Type Description Default
code str

Python code to execute.

 required

aimu.tools.builtin.get_webpage 

get_webpage(url: str) -> str

Fetches a web page and returns its visible text content with HTML stripped.

When the page exposes a publication timestamp (in tags, JSON-LD, or a

Parameters:

Name Type Description Default
url str

The URL of the page to retrieve.

 required

aimu.tools.builtin.get_webpage_html 

get_webpage_html(url: str) -> str

Fetches a web page and returns its raw HTML markup (tags and all).

Use this when you need to see the page structure, e.g. to locate links, attributes, or form markup. For readable article text instead, use get_webpage. For inspecting and submitting forms with cookie/session persistence, use the tools from make_web_tools().

Note: this fetches server-rendered HTML only; it does not execute JavaScript, so pages built client-side (SPAs) will return their pre-render markup. Long pages are truncated to keep the output within a model's context window.

Parameters:

Name Type Description Default
url str

The URL of the page to retrieve.

 required 
web_search(query: str, num_results: int = 5, time_range: str = '', categories: str = '') -> str

Search the web using a SearXNG instance and return the top results.

Each result includes its publication date when the search engine reports one (shown as a "Published:" line), useful for judging how recent an article is.

Parameters:

Name Type Description Default
query str

The search query string.

 required
num_results int

Number of results to return (default 5).

 5
time_range str

Optional recency filter, one of "day", "week", "month", or "year". Use "day" to restrict results to roughly the last 24 hours (best for fresh news).

 ''
categories str

Optional SearXNG category filter, e.g. "news" to restrict to news engines (which report publication dates far more reliably than general web engines). Comma-separated for multiple, e.g. "news,science".

 ''

Set SEARXNG_BASE_URL env var to point to your SearXNG instance (or a .env file) (default: http://localhost:8080).

aimu.tools.builtin.wikipedia 

wikipedia(query: str) -> str

Fetches a Wikipedia article summary for the given query.

Parameters:

Name Type Description Default
query str

Article title or search phrase (e.g. "Albert Einstein", "general relativity").

 required

aimu.tools.builtin.list_directory 

list_directory(path: str) -> str

Lists files and subdirectories at the given path.

Parameters:

Name Type Description Default
path str

Directory path to list.

 required

aimu.tools.builtin.read_file 

read_file(path: str, max_lines: int = 2000) -> str

Reads a local file and returns its contents, capped at max_lines lines.

If the result says it was truncated, call again with a larger max_lines before drawing any conclusion from it: a partial document reads exactly like a complete one.

Parameters:

Name Type Description Default
path str

Path to the file to read.

 required
max_lines int

Maximum number of lines to return (default 2000).

 2000

Tool factories

Bind a tool to a specific resource (a memory store, a knowledge base) or policy (a command's environment allowlist) instead of a process-wide singleton.

aimu.tools.builtin.make_command_tool 

make_command_tool(*, env_passthrough: tuple[str, ...] = ()) -> Callable

Build a run_command whose child also sees the named environment variables.

env_passthrough is what keeps the capability real without the default being unsafe. The child's environment is an allowlist with no API keys in it, which is what stops run_command("env") lifting a credential into a model's context, and which also makes gh, ssh, and git push over ssh fail. A host that wants one of those working names the variable it needs, typically from its own configuration, so the allowance is the user's to grant rather than this library's to assume.

builtin.compute's run_command is this factory called with no arguments, rather than a separately defined twin. @tool freezes the docstring into __tool_spec__ at decoration time, so a closure with no docstring reaches a model with an empty description and a __doc__ assigned afterwards never lands; one definition is the only way to keep one security disclosure.

aimu.tools.builtin.make_memory_tools 

make_memory_tools(store)

Build store_memory, search_memories, and list_memories tools bound to store.

store may be any :class:aimu.memory.MemoryStore implementation: :class:~aimu.memory.SemanticMemoryStore (ChromaDB vector search), :class:~aimu.memory.DocumentStore (path-keyed), or a custom subclass.

Unlike the image/audio/speech tools, there is no env-var singleton for memory because the choice of store (ephemeral vs. persistent, which persist_path) is meaningful and should be explicit. Construct the store, pass it here, and add the returned list to the agent::

store = SemanticMemoryStore(persist_path="./.memory")
agent = Agent(client, tools=make_memory_tools(store) + builtin.web)

For cross-process or multi-agent memory, use the FastMCP servers in aimu.memory.mcp / aimu.memory.document_mcp instead.

aimu.tools.builtin.make_retrieval_tool 

make_retrieval_tool(store, *, n_results: int = 5)

Build a retrieve_context tool bound to store for retrieval-augmented agents.

store may be any :class:aimu.memory.MemoryStore (typically a :class:~aimu.memory.SemanticMemoryStore populated via :func:aimu.rag.ingest). The tool runs :func:aimu.rag.retrieve and returns the joined context, letting an agent fetch relevant background on demand::

from aimu.rag import ingest
store = SemanticMemoryStore()
ingest(store, my_documents)
agent = Agent(client, tools=[make_retrieval_tool(store)])

Like :func:make_memory_tools, there is no env-var singleton; the store (and what was ingested into it) is a meaningful, explicit choice.

aimu.tools.builtin.make_subagent_tool 

make_subagent_tool(model, *, system_message: str = DEFAULT_SUBAGENT_SYSTEM_MESSAGE, tools: Optional[list[Callable]] = None, agent_types: Optional[dict[str, dict]] = None, max_depth: int = 1, max_iterations: int = 10, concurrent_tool_calls: bool = True, deps: Any = None, tool_approval: Optional[Callable] = None, tool_name: str = 'spawn_subagent', events: Optional[EventSink] = None) -> Callable

Build a spawn_subagent tool that delegates subtasks to fresh, isolated sub-agents.

This is AIMU's answer to dynamic sub-agent spawning (as in Claude Code's Task tool): the LLM decides at runtime to hand an independent subtask to a brand-new :class:~aimu.agents.Agent with its own context, rather than choosing among a fixed roster. It is the dynamic complement to :class:~aimu.agents.OrchestratorAgent (which wires a known set of workers up front); both reduce to subagent.run(task) — the difference is who decides the roster.

Each invocation builds a fresh ModelClient(model) (its own message history — the :func:aimu.agents.prebuilt._base.make_workers isolation idiom), so concurrent spawns share no state. Parallelism is free: give the parent Agent concurrent_tool_calls=True and, when the model emits several spawn_subagent calls in one turn, they run concurrently (ThreadPoolExecutor). Genuine overlap is for cloud models — a single local model serializes on the GIL/CUDA, and a shared deps object handed to concurrent spawns must be thread-safe.

Two shapes, chosen by agent_types:

  • Generic (agent_types=None): the tool is spawn_subagent(task) — a fresh general-purpose sub-agent using system_message + tools.
  • Typed (agent_types given): the tool is spawn_subagent(agent_type, task) over a registry of named specialists (each value a dict with "system_message" and optional "tools" / "model" / "thinking" / "generate_kwargs" / "max_iterations", and nothing else -- an unrecognized spec key raises at factory-call time rather than being ignored, since an ignored key reads exactly like an applied one); the available names are listed in the tool description. An unknown agent_type, by contrast, is returned to the model as a tool result (self-correction), not raised: that one is the model's mistake to recover from, where a bad spec key is the programmer's. "thinking" takes the same values as :class:~aimu.agents.Agent's field and is read with .get(), so a spec omitting it leaves the spawned agent at None rather than inheriting anything from the caller: unlike "model", which falls back to the model this factory was built with, there is no factory-level thinking tier to fall back to. "generate_kwargs" is a dict assigned to the spawned client's default_generate_kwargs, so it applies to every request that sub-agent makes. Like "thinking" and unlike "model", an omitted key inherits nothing: there is no factory-level generation tier, so a caller with one default across a roster writes it into each spec. Only the keys a spec names are set, which matters because this tier sits above the model card in the precedence chain -- a filled-in default would shadow a card's own tuned profile. "max_iterations" is the spawned agent's tool-loop cap, and it is the one key besides "model" with a factory-level tier beneath it: an omitted key falls back to this factory's own max_iterations rather than inheriting nothing, which is what lets one default cover a roster without being written into each spec. It must be an int >= 1, checked at factory-call time, because bool is an int subclass (so True would read as a cap of 1) and a cap below 1 is a loop that makes no model call at all.

max_depth (default 1) is the recursion guard: it counts the caller's agent as level 1, so the default gives spawned sub-agents no spawn tool of their own. max_depth=2 lets one more level spawn, and so on; the nested tool is rebuilt with a decremented depth, so recursion is finite.

Parameters:

Name Type Description Default
model

A Model enum member, a "provider:model_id" string, or a BaseModelClient to clone the model from (a fresh client is built per spawn regardless — the live client is never shared).

 required
system_message str

Persona for generic sub-agents.

 DEFAULT_SUBAGENT_SYSTEM_MESSAGE
tools Optional[list[Callable]]

Tools each generic sub-agent receives (None = text-only).

 None
agent_types Optional[dict[str, dict]]

Optional registry that switches the tool to typed mode.

 None
max_depth int

Spawn levels permitted, counting the caller's agent as 1 (must be >= 1).

 1
max_iterations int

Tool-loop cap forwarded to each spawned agent.

 10
concurrent_tool_calls bool

Applied to spawned agents (so nested spawns overlap). The parent's own concurrency is set by its author.

 True
deps Any

ToolContext.deps passed to each spawned agent.

 None
tool_approval Optional[Callable]

Callback (name, arguments) -> bool (may be a coroutine) run before each of the sub-agent's tool calls; returning False appends a refusal instead of executing the tool. Matches :class:~aimu.agents.Agent/:meth:~aimu.agents.Agent.run's tool_approval semantics.

 None
tool_name str

Name of the produced tool (mint several differently-named spawn tools on one agent).

 'spawn_subagent'
events Optional[EventSink]

The sink each spawned child reports to. It has to be passed explicitly: a spawn builds a fresh client per call, which is outside the client family a caller's scoped per-run override reaches, so a delegated run is otherwise invisible to a caller measuring the whole turn. Set on the child :class:~aimu.agents.Agent (its own events field), not passed to its run.

 None

Example::

from aimu.tools.builtin import make_subagent_tool, web

spawn = make_subagent_tool("anthropic:claude-sonnet-4-6", tools=web)
agent = Agent(client, "Break the request into subtasks and spawn a sub-agent for each.",
              tools=[spawn], concurrent_tool_calls=True)
print(agent.run("Compare the GDP growth of France, Japan, and Brazil since 2019."))

aimu.tools.builtin.make_web_tools 

make_web_tools(*, session=None, timeout: int = 15, max_content_chars: int = 20000, user_agent: str = _DEFAULT_USER_AGENT)

Build find_forms and submit_form tools sharing a requests.Session.

The shared session preserves cookies across calls, so a GET-then-POST form flow works: find_forms scrapes hidden fields (including CSRF tokens) from the page, then submit_form echoes them back with the session's cookies intact. Pass these to an agent alongside the stateless get_webpage_html::

agent = Agent(client, tools=[get_webpage_html, *make_web_tools()])

Pass your own session to control its lifecycle (e.g. pre-set auth headers) or to share one across several tool sets; otherwise a fresh requests.Session is created.

Note: these fetch server-rendered HTML only and do not execute JavaScript, so JS-rendered (SPA) forms and anti-bot-protected pages are out of scope; a headless browser backend is a possible future addition.

submit_form performs writes (POST). To require confirmation before it runs, gate it via the tool_approval hook, e.g. Agent(..., tool_approval=policy) where the policy inspects the tool name (see docs/how-to/gate-tool-calls.md).