`aimu.aio`¶

Async surface. Mirrors the sync API one-for-one: same class names, different namespace. See how-to: use async for usage patterns and explanation: async design for why the surface is shaped this way.

Differences from the sync surface:

Every run(), chat(), generate() is async def.
Streaming returns AsyncIterator[StreamChunk] (consume with async for).
Parallel and concurrent_tool_calls=True use asyncio.TaskGroup instead of ThreadPoolExecutor.
In-process providers (AsyncHuggingFaceClient, AsyncLlamaCppClient) wrap an existing sync client; calling aio.client(HuggingFaceModel.X) directly raises.

Top-level¶

aimu.aio.chat `async` ¶

chat(user_message: str, *, model: Union[str, Model, None] = None, system: Optional[str] = None, generate_kwargs: Optional[dict] = None, stream: bool = False, images: Optional[list] = None, include: Optional[Iterable[Union[str, StreamingContentType]]] = None, thinking: Optional[Union[bool, str]] = None, events: Optional['EventSink'] = None) -> Union[str, AsyncIterator[StreamChunk]]

One-shot async chat: builds a fresh client, sends one message, returns the response.

Example::

text = await aio.chat("Summarize this", model="anthropic:claude-sonnet-4-6")

async for chunk in await aio.chat("Tell me a story", model="ollama:qwen3.5:9b", stream=True):
    if chunk.is_text():
        print(chunk.content, end="")

Parameters:

Name	Type	Description	Default
`thinking`	`Optional[Union[bool, str]]`	Optional thinking control. `None` (default) leaves the provider's own behavior untouched. `False` disables reasoning and selects the model's instruct-mode sampling profile; `True` enables it at the model's default effort; `"low"`/`"medium"`/`"high"` sets the effort level. A model that cannot honour the request logs a warning and continues, so models stay swappable; an unrecognised value raises `ValueError`.	`None`
`events`	`Optional['EventSink']`	Optional event sink (see :mod:`aimu.events`); attach it to see the `ModelTurnStarted` / `ModelTurnFinished` events this one-shot call emits.	`None`

aimu.aio.client ¶

client(model: Union[str, Model, Any, None] = None, *, system: Optional[str] = None, events: Optional['EventSink'] = None, **kwargs: Any) -> AsyncModelClient

Construct an :class:AsyncModelClient from a model string, enum, or existing sync client.

For in-process providers (HuggingFace, LlamaCpp), pass an existing sync client to avoid loading model weights twice::

sync_client = aimu.client(HuggingFaceModel.LLAMA_70B)
async_client = aio.client(sync_client)

When model is omitted, a default is resolved from AIMU_LANGUAGE_MODEL or an already-available local model. The async path probes only Ollama and local OpenAI-compatible servers (an hf: default would need an explicit sync-client wrap).

Parameters:

Name	Type	Description	Default
`events`	`Optional['EventSink']`	Optional event sink (see :mod:`aimu.events`). Attach it to see the `ModelTurnStarted` / `ModelTurnFinished` events every `chat()` / `generate()` call on the returned client emits.	`None`

aimu.aio.AsyncModelClient ¶

AsyncModelClient(model: Union[Model, ModelSpec, str, Any], **kwargs: Any)

Bases: AsyncBaseModelClient

Public factory for async provider-backed model clients.

Accepts a provider Model enum member, a "provider:model_id" string, or for in-process providers, an existing sync client to wrap.

Examples::

# Cloud providers (separate sync/async clients are cheap)
client = AsyncModelClient("anthropic:claude-sonnet-4-6")
client = AsyncModelClient(OllamaModel.QWEN_3_8B)

# In-process providers: wrap an existing sync client to share weights
sync_client = aimu.client(HuggingFaceModel.LLAMA_70B)
async_client = AsyncModelClient(sync_client)

last_usage `property` `writable` ¶

last_usage: Optional[dict]

Token usage of the most recent non-streaming response, or None.

last_output_truncated `property` `writable` ¶

last_output_truncated: bool

Whether the most recent response was cut off at an output limit rather than finishing.

last_structured `property` `writable` ¶

last_structured

Validated object from the most recent schema= call, or None (populated after a streamed structured call is fully consumed; mirrors :attr:last_usage).

last_request `property` `writable` ¶

last_request: Optional[Any]

The payload of the most recent request, post-adaptation, or None. See the sync :attr:~aimu.models.model_client.ModelClient.last_request for the shape-by-provider note.

Hierarchy¶

aimu.aio.AsyncRunner ¶

Bases: ABC

Abstract base for every concrete async agent and workflow.

messages `abstractmethod` `property` ¶

messages: MessageHistory

Message histories of all sub-runners, keyed by runner name.

run `abstractmethod` `async` ¶

run(task: str, generate_kwargs: Optional[dict[str, Any]] = None, stream: bool = False, images: Optional[list] = None) -> Union[str, AsyncIterator[StreamChunk]]

Run asynchronously (stream=False) or streaming (stream=True).

as_tool ¶

as_tool(*, name: Optional[str] = None, description: Optional[str] = None) -> Callable

Wrap this async runner as an async @tool-style callable: await tool(task).

Async mirror of :meth:aimu.agents.base.Runner.as_tool. The returned callable is an async def delegating to await self.run(task), so the @tool decorator marks it __tool_is_async__ = True and the async agent loop awaits it directly.

Agents¶

aimu.aio.Agent `dataclass` ¶

Agent(model_client: AsyncBaseModelClient, system_message: Optional[str] = None, name: Optional[str] = None, tools: list[Callable] = list(), max_iterations: int = 10, continuation_prompt: str = DEFAULT_CONTINUATION_PROMPT, reset_messages_on_run: bool = False, final_answer_prompt: Optional[str] = None, deps: Optional[Any] = None, tool_approval: Optional[Callable] = None, thinking: Optional[Union[bool, str]] = None, events: Optional[EventSink] = None, compaction: Optional[Callable[[list[dict]], list[dict]]] = None, concurrent_tool_calls: bool = False)

Bases: _AgentLoopMixin, AsyncRunner

Async equivalent of :class:aimu.agents.Agent.

Calls await model_client.chat() repeatedly until the model produces a turn without invoking tools, or max_iterations real model calls have been made by the loop. On exhausting that cap with a tool call still pending, one forced wrap-up turn (tools disabled) runs after the cap to guarantee a final answer -- see aimu.agents.Agent's docstring for the full degenerate-turn handling this driver shares byte-for-byte with the sync one. That wrap-up call is the one exception: it is never counted against max_iterations.

Quick start::

from aimu.tools import tool
from aimu import aio

@tool
async def fetch(url: str) -> str:
    """Fetch the contents of a URL."""
    import httpx
    async with httpx.AsyncClient() as c:
        return (await c.get(url)).text[:500]

client = aio.client("anthropic:claude-sonnet-4-6")
agent = aio.Agent(client, "You are a helpful assistant.", tools=[fetch])
print(await agent.run("Fetch example.com"))

run `async` ¶

run(task: str, generate_kwargs: Optional[dict[str, Any]] = None, stream: bool = False, images: Optional[list] = None, tools: Optional[list[Callable]] = None, deps: Optional[Any] = None, tool_approval: Optional[Callable] = None, schema: Optional[type] = None, thinking: Optional[Union[bool, str]] = None, events: Optional[EventSink] = None, compaction: Optional[Callable[[list[dict]], list[dict]]] = None) -> Union[str, Any, AsyncIterator[StreamChunk]]

Run the async agentic loop. images attach only to the initial turn.

The loop makes at most self.max_iterations real model calls (the initial turn plus every continuation/tool-follow-up turn), then, if a tool call is still pending at that cap, one additional forced wrap-up call (tools disabled) to guarantee a final answer -- that one call is deliberately not counted against max_iterations. Identical to the sync driver's definition; see :meth:aimu.agents.Agent.run.

tools is a per-run override of the agent's configured self.tools; deps is a per-run override of the agent's self.deps (injected as ctx.deps into tools that declare a :class:~aimu.tools.ToolContext parameter); tool_approval is a per-run override of self.tool_approval (the gate run before each tool call, (name, arguments) -> bool, which may be a coroutine; deny appends a refusal tool message); schema makes the run a single structured-output turn returning a validated instance; thinking is a per-run override of self.thinking (the portable reasoning control), applied to every model turn the run makes; events is a per-run override of self.events (a callable taking one :class:~aimu.events.RunEvent), installed as the active sink for the run's duration via a scoped contextvars.ContextVar override. Safe across agents that share a model_client and run concurrently (e.g. every worker Agent in a :class:~aimu.agents.Parallel built via Parallel.from_client): each concurrently running agent's asyncio.Task gets its own independent copy of the ContextVar, so one cannot clobber another's. It is also scoped to this specific client (and whatever it delegates to or from), not to any client called while the scope is open: a different client called from inside a tool (e.g. a fresh client the tool builds for itself, as make_subagent_tool does) never receives this run's sink, on either surface and regardless of concurrent_tool_calls -- it falls back to its own self.events, so give it an explicit events= if it needs to report anywhere. The one case that still depends on the surface is a tool that calls the same client (e.g. reusing ctx.deps): unlike sync, concurrent_tool_calls=True dispatches async tools via asyncio.TaskGroup.create_task, which always copies the current context, so that reentrant call sees the override (attributed to this agent, since the attribution wrapper stamps any event that arrives without its own) -- sync's equivalent case (a fresh ThreadPoolExecutor thread with an empty context) does not -- see aimu.agents.Agent.run's docstring. Sequential tool dispatch (the default) sees it on both surfaces, since no thread/task boundary is crossed. compaction is a per-run override of self.compaction (a callable applied to the conversation before every model turn the run makes; see :mod:aimu.context), not used by the schema= structured-output path. See the sync :meth:aimu.agents.Agent.run for full semantics.

as_model_client ¶

as_model_client() -> AsyncBaseModelClient

Return an :class:AsyncBaseModelClient view of this agent.

Each await client.chat() runs the full agent loop.

aimu.aio.SkillAgent `dataclass` ¶

SkillAgent(model_client: AsyncBaseModelClient, system_message: Optional[str] = None, name: Optional[str] = None, tools: list[Callable] = list(), max_iterations: int = 10, continuation_prompt: str = DEFAULT_CONTINUATION_PROMPT, reset_messages_on_run: bool = False, final_answer_prompt: Optional[str] = None, deps: Optional[Any] = None, tool_approval: Optional[Callable] = None, thinking: Optional[Union[bool, str]] = None, events: Optional[EventSink] = None, compaction: Optional[Callable[[list[dict]], list[dict]]] = None, concurrent_tool_calls: bool = False, skill_manager: SkillManager = SkillManager(), script_env: Optional[dict[str, str]] = None)

Bases: Agent

Async :class:Agent with filesystem-discovered skill injection.

On first run (or after a message reset) the SkillAgent appends the skill catalog to its system message and surfaces the async skills server's tools (via aio.MCPClient.as_tools()) through :meth:_effective_tools, so the tool-loop engine advertises and dispatches them (the model can call activate_skill on demand).

script_env is host context handed to every skill script this agent runs, merged over the inherited environment by :func:aimu.skills.mcp.run_script_file. Without it the only way to tell a script something it cannot discover (where to write output, which account to send from) is a process-wide variable, which makes one agent's context every subprocess's context. It is a field rather than a build_skills_server argument the caller passes because this class builds that server itself, twice: on first run and again in :meth:reload_skills.

reload_skills `async` ¶

reload_skills() -> None

Rebuild the skills server from the (refreshed) manager and surface new tools now.

Re-snapshots the skills tools and re-injects the catalog. Because the tool-loop engine re-reads :meth:_effective_tools each round, a skill authored mid-run is advertised and dispatchable for the rest of the run. Call after writing a new skill/script (see :func:aimu.skills.make_skill_script_tool).

aimu.aio.OrchestratorAgent ¶

Bases: AsyncRunner, ABC

Async base for the orchestrator + worker-tools pattern.

Subclasses define worker :class:Agent instances and @tool-decorated dispatch functions in __init__, then call :meth:_init_orchestrator to wire everything up. Worker dispatch functions should be async def so the orchestrator's concurrent_tool_calls=True actually overlaps work.

For the simple case of dispatching to a fixed list of workers, use :meth:assemble to skip subclassing entirely.

assemble `classmethod` ¶

assemble(model_client: AsyncBaseModelClient, system_message: str, *, workers: list[AsyncRunner], name: str = 'orchestrator', concurrent_tool_calls: bool = True, final_answer_prompt: Optional[str] = None, events: Optional[EventSink] = None) -> 'OrchestratorAgent'

Build a ready-to-run async orchestrator from a list of worker runners.

Each worker becomes an async callable tool via :meth:AsyncRunner.as_tool. Workers may be any :class:AsyncRunner (an async Agent, a workflow, or a remote A2A agent), not just Agent instances.

restore ¶

restore(messages: list[dict]) -> None

Restore the inner orchestrator agent's state from a saved message list.

Workers are invoked as tools, so their own state is not part of the orchestrator's history; restore a worker directly if it needs resuming. See :meth:aimu.aio.Agent.restore for the full save/restore pattern.

Workflows¶

aimu.aio.Chain `dataclass` ¶

Chain(agents: list, name: str = 'chain')

Bases: AsyncRunner

Async prompt-chaining: each step's output feeds the next step's input.

Steps run sequentially (the pattern requires it). Each step may be an :class:aimu.aio.Agent or a nested async workflow.

from_client `classmethod` ¶

from_client(client: AsyncBaseModelClient, prompts: list[str], *, name: str = 'chain', events: Optional[EventSink] = None) -> Chain

Build a Chain from a single client and a list of step system_messages.

events is passed to every step's :class:aimu.aio.Agent, so one sink sees the whole pipeline with each event attributed to the step (agent) that produced it.

restore ¶

restore(messages: list[dict], *, step: int = 0) -> None

Restore a chain step's state from a saved message list.

step (keyword-only) selects which step's agent to restore (default 0); subsequent steps start fresh on the next run(). Raises IndexError if step is out of range. See :meth:aimu.aio.Agent.restore for the pattern.

aimu.aio.Router `dataclass` ¶

Router(routing_agent: Agent, handlers: dict[str, AsyncRunner], name: str = 'router', fallback: Optional[AsyncRunner] = None)

Bases: AsyncRunner

Async routing: classify the task, dispatch to a specialist handler.

from_client `classmethod` ¶

from_client(client: AsyncBaseModelClient, classifier_prompt: str, handlers: dict[str, AsyncRunner], *, fallback: Optional[AsyncRunner] = None, name: str = 'router', events: Optional[EventSink] = None) -> Router

Build a Router using client as the classifier with the given prompt.

events is passed to the classifier :class:aimu.aio.Agent this factory constructs. The handlers/fallback runners are supplied by the caller already built, so wire the same sink into them yourself if you want the whole dispatch attributed.

restore ¶

restore(messages: list[dict], *, route: Optional[str] = None) -> None

Restore one sub-runner's state from a saved message list.

route=None (default, keyword-only) restores the routing classifier; a route key restores that handler (raises KeyError listing the routes on a miss). Other sub-runners start fresh on the next run(). See :meth:aimu.aio.Agent.restore.

aimu.aio.Parallel `dataclass` ¶

Parallel(workers: list, name: str = 'parallel', aggregator: Optional[AsyncRunner] = None, separator: str = '\n\n---\n\n')

Bases: AsyncRunner

Async parallelization: run workers concurrently via asyncio.TaskGroup, aggregate.

Each worker receives the same task. An optional aggregator receives all worker outputs joined by separator. Without an aggregator, the joined output is returned.

Structured concurrency: if one worker raises, in-flight siblings are cancelled and an ExceptionGroup surfaces with all errors.

from_client `classmethod` ¶

from_client(client: AsyncBaseModelClient, worker_prompts: list[str], *, aggregator_prompt: Optional[str] = None, separator: str = '\n\n---\n\n', name: str = 'parallel', events: Optional[EventSink] = None) -> Parallel

Build a Parallel using client for all workers (and aggregator).

events is passed to every worker (and the aggregator, if any) this factory constructs. Every worker here shares one client, and workers run concurrently under asyncio.TaskGroup -- each worker's per-run sink is delivered through a scoped contextvars.ContextVar override (AsyncBaseModelClient._events_override / _effective_sink), not a mutation of client.events, so each worker's own asyncio.Task gets its own independent copy of the context it was created in and concurrent delivery is correctly attributed and ordered per worker (see tests/test_aio_workflow_parallel.py's concurrent-workers test, the async mirror of the sync one). A different client called from inside a worker's tool never receives that worker's sink, on either surface and regardless of concurrent_tool_calls; see aio.Agent.events for the one residual that does depend on the surface.

restore ¶

restore(messages: list[dict], *, worker: int = 0) -> None

Restore one worker's state from a saved message list.

worker (keyword-only) selects which worker by index (default 0). Other workers and the aggregator start fresh on the next run(). Raises IndexError if worker is out of range. See :meth:aimu.aio.Agent.restore.

aimu.aio.EvaluatorOptimizer `dataclass` ¶

EvaluatorOptimizer(generator: Agent, evaluator: Agent, name: str = 'evaluator_optimizer', max_rounds: int = 3, pass_keyword: str = 'PASS')

Bases: AsyncRunner

Async generate-evaluate-revise loop, identical semantics to sync version.

restore ¶

restore(messages: list[dict]) -> None

Restore the generator's state from a saved message list.

The evaluator starts fresh on the next round. See :meth:aimu.aio.Agent.restore.

aimu.aio.PlanExecuteEvaluator `dataclass` ¶

PlanExecuteEvaluator(planner: SkillAgent, executor: Agent, scorer: Scorer, criteria: Optional[str] = None, name: str = 'plan_execute_evaluator', max_rounds: int = 3, pass_threshold: float = 0.7, pass_keyword: Optional[str] = None)

Bases: AsyncRunner

Async plan → execute → evaluate → replan-on-fail loop.

Same semantics as :class:aimu.agents.PlanExecuteEvaluator but with async def run() and awaited delegation to planner/executor.

The scorer's score() is sync (it's a CPU/judge-call concern, not an AIMU-async concern). If you wire an LLM judge, that judge call blocks the event loop unless wrapped; use asyncio.to_thread from your scorer's score() if needed.

from_client `classmethod` ¶

from_client(client: AsyncBaseModelClient, *, judge_client: Optional[Any] = None, criteria: Optional[str] = None, executor_tools: Optional[list[Callable]] = None, skill_manager: Optional[SkillManager] = None, planner_system_message: Optional[str] = None, executor_system_message: Optional[str] = None, max_rounds: int = 3, pass_threshold: float = 0.7, pass_keyword: Optional[str] = None, name: str = 'plan_execute_evaluator', events: Optional[EventSink] = None) -> PlanExecuteEvaluator

Build a PlanExecuteEvaluator from a single async client.

The judge_client for the scorer may be sync or async (the scorer is sync; if it's an :class:LLMJudgeScorer it expects a sync client).

events is passed to both the planner and executor, so one sink sees the whole plan -> execute round attributed to whichever produced it. It is not passed to the scorer: an :class:LLMJudgeScorer is not a :class:Runner and has no events field to receive it.

Tools¶

aimu.aio.MCPClient ¶

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

Async wrapper around a FastMCP Client.

Use the connect() classmethod factory to construct + connect in one await::

mcp = await MCPClient.connect(server=my_fastmcp_server)
try:
    tools = await mcp.get_tools()
    result = await mcp.call_tool("foo", {"x": 1})
finally:
    await mcp.aclose()

connect `async` `classmethod` ¶

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

Construct and connect in one await. Returns the live instance.

Pass exactly one source: config, server, file, or url (a remote HTTP/SSE server). auth (a bearer-token string, "oauth", or a configured FastMCP OAuth / httpx.Auth provider object) and headers apply only with url.

ping `async` ¶

ping() -> list

Verify the connection is alive by listing tools.

get_tools `async` ¶

get_tools() -> list[dict]

Return tools in OpenAI function-calling format.

as_tools `async` ¶

as_tools() -> list

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

Async mirror of :meth:aimu.tools.MCPClient.as_tools. Each callable is an async def that awaits :meth:call_tool and returns the result's text content; it carries __tool_spec__, __tool_is_async__ = True, and __tool_is_streaming__ = False, so it drops into client.tools / aio.Agent(tools=...) and the async dispatcher awaits it directly::

mcp = await aio.MCPClient.connect(server=my_server)
agent = aio.Agent(client, tools=await mcp.as_tools())

The list is a snapshot (one list_tools() round-trip); call again to refresh.

aclose `async` ¶

aclose() -> None

Close the underlying connection. Idempotent.

aimu.aio.tools.builtin.make_async_subagent_tool ¶

make_async_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', observer: Optional[SubagentObserver] = None, events: Optional[EventSink] = None) -> Callable

Async twin of :func:aimu.tools.builtin.make_subagent_tool.

Produces an async def spawn_subagent tool (__tool_is_async__=True) that builds a fresh, isolated :class:aimu.aio.Agent per call and awaits its run. Parallelism is free: give the parent :class:aimu.aio.Agent concurrent_tool_calls=True and multiple spawn calls in one turn overlap under an asyncio.TaskGroup. See the sync docstring for the full contract (generic vs typed mode, the per-spec "model" / "thinking" / "generate_kwargs" / "max_iterations" keys, max_depth recursion guard, unknown-agent_type handling, and the tool_approval gate forwarded to every spawned sub-agent).

In-process providers (HuggingFace, LlamaCpp) are wrapped per spawn via a fresh sync client (the aio surface can't construct them from an enum); the process weight cache prevents reloading weights.

Passing observer (a :class:SubagentObserver) switches each spawn to a streamed child run and reports it as it happens, without making this a streaming tool (which would disable the parent's concurrent dispatch). Nested spawns inherit it.

events is 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 Agent rather than passed to its run, which covers the observed path too (_run_observed calls run itself).

aimu.aio.tools.builtin.SubagentObserver ¶

Bases: Protocol

Display hook for one sub-agent spawn, so a front end can show its work as it happens.

Passing an observer to :func:make_async_subagent_tool switches the spawn to a streamed child run: every chunk is forwarded here while the tool's return value stays the child's final answer. The spawn tool itself remains a plain (non-streaming) tool, so concurrent spawns still overlap under the parent's concurrent_tool_calls. Callbacks are display-only; an exception raised by one is logged and swallowed rather than failing the spawn.

Attaching an observer is therefore not purely additive: an observed spawn issues its model calls through the provider's streaming request path, where an unobserved one uses the non-streaming path. The answer is the same either way, but any behavior that differs between a provider's two request paths applies to observed spawns.

spawned `async` ¶

spawned(spawn_id: str, agent_type: Optional[str], task: str) -> None

A sub-agent has been built for task. agent_type is None in generic (untyped) mode.

chunk `async` ¶

chunk(spawn_id: str, chunk: StreamChunk) -> None

One chunk from the child's streamed run.

finished `async` ¶

finished(spawn_id: str, result: str, error: Optional[BaseException]) -> None

The spawn ended. result is the final (or partial, on failure) generated text, and error is the exception that ended it, including a CancelledError.

Personal assistant¶

Primitives for building an always-on assistant. See how-to: build a personal assistant.

aimu.aio.Channel ¶

Bases: ABC

Async transport: receive inbound messages, send replies.

Subclasses that own a live connection (network adapters) should expose an async connect() classmethod factory, mirroring aimu.aio.MCPClient.connect.

receive `abstractmethod` ¶

receive() -> AsyncIterator[ChannelMessage]

Yield inbound messages until the channel closes (an async generator).

send `abstractmethod` `async` ¶

send(content: Union[str, AsyncIterator[StreamChunk]], *, reply_to: Optional[ChannelMessage] = None) -> None

Send a reply.

content is either a finished string or an AsyncIterator[StreamChunk] to relay incrementally. reply_to carries routing for multi-recipient adapters (which chat to answer); single-user adapters like the CLI ignore it.

aclose `async` ¶

aclose() -> None

Release any resources. Default no-op so simple adapters need not override.

aimu.aio.ChannelMessage `dataclass` ¶

ChannelMessage(text: str, sender: Optional[str] = None, channel: Optional[str] = None, images: Optional[list] = None, metadata: dict[str, Any] = dict())

A transport-level message in or out of a channel.

Distinct from LLM conversation state (which stays list[dict] in OpenAI format): text and images map directly onto agent.run(task, images=...).

Attributes:

Name	Type	Description
`text`	`str`	The message text.
`sender`	`Optional[str]`	Adapter-defined sender id (a chat id, `"cli"`); used by `send(reply_to=)`.
`channel`	`Optional[str]`	Adapter name, e.g. `"cli"` or `"telegram"`.
`images`	`Optional[list]`	Optional images to forward to a vision-capable agent.
`metadata`	`dict[str, Any]`	Raw adapter payload, opaque to the assistant loop.

aimu.aio.CLIChannel ¶

CLIChannel(*, prompt: str = '> ', stream_thinking: bool = True, stream_tools: bool = True)

Bases: Channel

Read user input from stdin and write replies to stdout.

Single local user, so send(reply_to=...) is accepted and ignored. Streaming replies are written token-by-token as they arrive. stream_thinking / stream_tools (both on by default) relay the model's reasoning and tool calls as labelled lines alongside the answer; set either False to drop that content before it reaches the terminal.

aimu.aio.WebChannel ¶

WebChannel(websocket: Any, *, stream_thinking: bool = True, stream_tools: bool = True)

Bases: Channel

Bridges one browser WebSocket onto the Channel ABC.

The app's server pump task calls :meth:feed for each inbound frame and feed(None) on disconnect; :meth:receive ends on that sentinel, which lets the agent loop tear down cleanly. Replies are sent as JSON frames (see the module docstring for the protocol).

feed `async` ¶

feed(text: Optional[str]) -> None

Enqueue an inbound frame; None is the end-of-stream sentinel.

send_frame `async` ¶

send_frame(frame: dict) -> None

Send one JSON frame to the browser, swallowing errors once the socket has closed.

The single point through which every frame reaches the socket, so subclasses add their own frame types by calling this rather than touching the socket directly. Once closed (e.g. a proactive push racing a disconnect), a late frame is dropped instead of crashing the sending task.

aimu.aio.Scheduler ¶

Scheduler()

Run interval and one-shot async jobs concurrently until stopped.

Usage::

scheduler = Scheduler()
scheduler.every(60, check_inbox, name="inbox")
scheduler.at(5, lambda: channel.send("Welcome!"))
await scheduler.run()   # blocks until scheduler.stop()

A job that raises is logged and, for interval jobs, the loop continues on the next tick (one misbehaving reminder must not tear down the daemon). Only :meth:stop unwinds the run loop.

every ¶

every(seconds: float, callback: Job, *, name: Optional[str] = None, first_delay: Optional[float] = None) -> str

Register a recurring job firing every seconds. Returns the job id.

first_delay overrides the initial wait before the first fire (defaults to seconds). If the scheduler is already running, the job starts immediately.

at ¶

at(delay_seconds: float, callback: Job, *, name: Optional[str] = None) -> str

Register a one-shot job firing once after delay_seconds. Returns the job id.

cancel ¶

cancel(name: str) -> bool

Cancel a registered job by id. Returns True if it existed.

stop ¶

stop() -> None

Signal :meth:run to cancel all jobs and return.

run `async` ¶

run() -> None

Run all registered jobs concurrently, blocking until :meth:stop is called.

A single-use run: if :meth:stop was already called (even before run), the loop returns immediately. This avoids a lost-stop race when a sibling task signals stop before the run loop is scheduled. Use a fresh :class:Scheduler to run again.

A2A interop¶

Async twin of aimu.agents.a2a (requires the a2a extra). aimu.aio.a2a.RemoteAgent uses the a2a-sdk async client natively (no anyio portal) and supports incremental message/stream streaming.

aimu.aio.RemoteAgent ¶

RemoteAgent(client: A2AClient, httpx_client: AsyncClient, name: str, card: Any)

Bases: AsyncRunner

A remote A2A agent presented as a local asynchronous AsyncRunner.

Construct via the async :meth:connect::

remote = await RemoteAgent.connect("http://localhost:9000")
print(await remote.run("Summarise the news"))
async for chunk in await remote.run("Summarise", stream=True):
    ...

connect `async` `classmethod` ¶

connect(url: str, *, name: Optional[str] = None, agent_card_path: str = DEFAULT_AGENT_CARD_PATH, timeout: float = 60.0) -> 'RemoteAgent'

Resolve the remote agent card at url and return a connected RemoteAgent.

run `async` ¶

run(task: str, generate_kwargs: Optional[dict[str, Any]] = None, stream: bool = False, images: Optional[list] = None) -> Union[str, AsyncIterator[StreamChunk]]

Send task to the remote agent; return its text (or a chunk stream).

aimu.aio.serve_a2a ¶

serve_a2a(runner: AsyncRunner, *, host: str = '127.0.0.1', port: int = 9000, url: Optional[str] = None, name: Optional[str] = None, description: Optional[str] = None, skills: Optional[list[AgentSkill]] = None, **uvicorn_kwargs: Any) -> None

Serve an async runner as an A2A agent over HTTP (blocking).

aimu.aio.build_a2a_app ¶

build_a2a_app(runner: AsyncRunner, *, url: str, name: Optional[str] = None, description: Optional[str] = None, skills: Optional[list[AgentSkill]] = None, agent_card_path: str = DEFAULT_AGENT_CARD_PATH)

Build the Starlette ASGI app that serves an async runner over A2A (does not run it).

aimu.aio¶

Top-level¶

aimu.aio.chat async ¶

aimu.aio.client ¶

aimu.aio.AsyncModelClient ¶

last_usage property writable ¶

last_output_truncated property writable ¶

last_structured property writable ¶

last_request property writable ¶

Hierarchy¶

aimu.aio.AsyncRunner ¶

messages abstractmethod property ¶

run abstractmethod async ¶

as_tool ¶

Agents¶

aimu.aio.Agent dataclass ¶

run async ¶

as_model_client ¶

aimu.aio.SkillAgent dataclass ¶

reload_skills async ¶

aimu.aio.OrchestratorAgent ¶

assemble classmethod ¶

restore ¶

Workflows¶

aimu.aio.Chain dataclass ¶

from_client classmethod ¶

restore ¶

aimu.aio.Router dataclass ¶

from_client classmethod ¶

restore ¶

aimu.aio.Parallel dataclass ¶

from_client classmethod ¶

restore ¶

aimu.aio.EvaluatorOptimizer dataclass ¶

restore ¶

aimu.aio.PlanExecuteEvaluator dataclass ¶

from_client classmethod ¶

Tools¶

aimu.aio.MCPClient ¶

connect async classmethod ¶

ping async ¶

get_tools async ¶

as_tools async ¶

aclose async ¶

aimu.aio.tools.builtin.make_async_subagent_tool ¶

aimu.aio.tools.builtin.SubagentObserver ¶

spawned async ¶

chunk async ¶

finished async ¶

Personal assistant¶

aimu.aio.Channel ¶

receive abstractmethod ¶

send abstractmethod async ¶

aclose async ¶

aimu.aio.ChannelMessage dataclass ¶

aimu.aio.CLIChannel ¶

aimu.aio.WebChannel ¶

feed async ¶

send_frame async ¶

aimu.aio.Scheduler ¶

every ¶

at ¶

cancel ¶

stop ¶

run async ¶

A2A interop¶

aimu.aio.RemoteAgent ¶

connect async classmethod ¶

run async ¶

aimu.aio.serve_a2a ¶

aimu.aio.build_a2a_app ¶

`aimu.aio`¶

aimu.aio.chat `async` ¶

last_usage `property` `writable` ¶

last_output_truncated `property` `writable` ¶

last_structured `property` `writable` ¶

last_request `property` `writable` ¶

messages `abstractmethod` `property` ¶

run `abstractmethod` `async` ¶

aimu.aio.Agent `dataclass` ¶

run `async` ¶

aimu.aio.SkillAgent `dataclass` ¶

reload_skills `async` ¶

assemble `classmethod` ¶

aimu.aio.Chain `dataclass` ¶

from_client `classmethod` ¶

aimu.aio.Router `dataclass` ¶

from_client `classmethod` ¶

aimu.aio.Parallel `dataclass` ¶

from_client `classmethod` ¶

aimu.aio.EvaluatorOptimizer `dataclass` ¶

aimu.aio.PlanExecuteEvaluator `dataclass` ¶

from_client `classmethod` ¶

connect `async` `classmethod` ¶

ping `async` ¶

get_tools `async` ¶

as_tools `async` ¶

aclose `async` ¶

spawned `async` ¶

chunk `async` ¶

finished `async` ¶

receive `abstractmethod` ¶

send `abstractmethod` `async` ¶

aclose `async` ¶

aimu.aio.ChannelMessage `dataclass` ¶

feed `async` ¶

send_frame `async` ¶

run `async` ¶

connect `async` `classmethod` ¶

run `async` ¶