`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.

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`
`builtin.misc`	`echo`, `get_current_date_and_time`
`builtin.ALL_TOOLS`	All of the above except `execute_python` (sandboxed REPL; opt in via `builtin.compute`)

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() -> str

Returns the current date and time.

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 sandboxed environment and return the output.

Captures stdout and the value of the last expression. Imports are limited to: math, statistics, json, re, itertools, functools, datetime, and numpy/pandas/scipy/matplotlib when installed. File system and subprocess access are not available.

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 element), it is prepended as a "Published:" line so the date isn't lost when the HTML is stripped.

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

aimu.tools.builtin.web_search ¶

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 = 200) -> str

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

Parameters:

Name	Type	Description	Default
`path`	`str`	Path to the file to read.	required
`max_lines`	`int`	Maximum number of lines to return (default 200).	`200`

Tool factories¶

Bind a tool to a specific resource (a memory store, a knowledge base) instead of a process-wide singleton.

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') -> 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"); the available names are listed in the tool description. An unknown agent_type is returned to the model as a tool result (self-correction), not raised.

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'`

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

aimu.tools¶

Decorator¶

aimu.tools.tool ¶

aimu.tools.ToolSignatureError ¶

Argument validation¶

aimu.tools.coerce_tool_arguments ¶

aimu.tools.ToolArgumentError ¶

MCP client¶

aimu.tools.MCPClient ¶

ping ¶

as_tools ¶

aimu.tools.MCPConnectionError ¶

Built-in tools¶

aimu.tools.builtin.echo ¶

aimu.tools.builtin.get_current_date_and_time ¶

aimu.tools.builtin.get_weather ¶

aimu.tools.builtin.calculate ¶

aimu.tools.builtin.execute_python ¶

aimu.tools.builtin.get_webpage ¶

aimu.tools.builtin.get_webpage_html ¶

aimu.tools.builtin.web_search ¶

aimu.tools.builtin.wikipedia ¶

aimu.tools.builtin.list_directory ¶

aimu.tools.builtin.read_file ¶

Tool factories¶

aimu.tools.builtin.make_memory_tools ¶

aimu.tools.builtin.make_retrieval_tool ¶

aimu.tools.builtin.make_subagent_tool ¶

aimu.tools.builtin.make_web_tools ¶

`aimu.tools`¶