`aimu.agents`¶

Agents and code-controlled workflows.

Hierarchy¶

aimu.agents.Runner ¶

Bases: ABC

Abstract base for every concrete agent and workflow in AIMU.

Concrete subclasses implement :meth:run and :attr:messages.

messages `abstractmethod` `property` ¶

messages: MessageHistory

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

For a leaf agent: {agent.name: model_client.messages}. For composite workflows: the per-runner dicts merged into one (via dict.update), so the result spans the full sub-tree regardless of nesting depth. Example::

router.messages
# {"router-classifier": [...], "code-handler": [...], "prose-handler": [...]}

Merge is by runner name: two sub-runners sharing a name collide and the one merged later wins (last-write). Give sub-runners distinct name=s if you need every history addressable.

Separately, when sub-runners share a single ModelClient (e.g. via Chain.from_config), they all reference the same messages list, so after a run every key points at that one shared (last step's) history. Use distinct clients per sub-runner to keep histories separate.

run `abstractmethod` ¶

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

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

as_tool ¶

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

Wrap this runner as a @tool-style callable: tool(task: str) -> str.

The returned callable runs :meth:run and returns its string result, so any agent or workflow can call this runner as a tool: drop it into Agent(tools=[...]) or an :class:OrchestratorAgent's worker list. Works for every concrete Runner (Agent, Chain, Router, a remote A2A agent, ...) since it only relies on run().

name defaults to self.name (sanitised to a valid identifier) or "runner". description defaults to the first line of self.system_message when present (Agent / SkillAgent), else a generic delegation string (workflows have no system_message).

aimu.agents.MessageHistory `module-attribute` ¶

MessageHistory = dict[str, list[dict]]

Agents¶

aimu.agents.Agent `dataclass` ¶

Agent(model_client: BaseModelClient, 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, Runner

A model client wrapped in an agentic loop.

model_client.chat() is a single turn: it issues one model request and, if the model requests tools, executes them and returns. This agent is the loop over that: it calls chat(task) and then chat() (no user message — a continuation turn on the current messages) until a turn makes no tool calls, or the loop has made max_iterations real model calls. max_iterations counts every call the bounded loop itself makes (the initial turn plus each continuation/tool-follow-up turn); the forced wrap-up call below is the one exception, since it runs after this cap and exists specifically to guarantee an answer once the cap is hit. No synthetic "continue" prompt is injected between successful tool rounds.

Tools are plain callables in tools=: functions decorated with @aimu.tools.tool for in-process tools, and/or MCPClient(...).as_tools() for cross-process FastMCP tools (each MCP tool becomes a callable). Mix them freely in one list, tools=builtin.web + mcp.as_tools().

When system_message is set or reset_messages_on_run is True, the agent clears model_client.messages and re-applies system_message before every run. This isolates state when a client is shared (e.g. inside a :class:Chain).

The loop guards against degenerate turns so a run never ends silently:

Empty turn. If a turn comes back with no content and no tool calls, the agent injects continuation_prompt (tools still enabled, so the model can resume a multi-step plan) and continues. These nudges are bounded by max_iterations and tagged so a UI can hide them.
Cap with tools pending. On exhausting max_iterations with a tool call still pending, the agent sends one forced wrap-up turn with tools disabled, so the model must synthesize an answer from the context it has gathered. final_answer_prompt customizes this wrap-up prompt; when unset a built-in default is used. This turn is not counted against max_iterations. A natural finish (a real answer) is unaffected. The pending calls are first answered with results stating they were not executed and why (logged at WARNING), since the wrap-up prompt is a user message and no provider accepts one on top of an unanswered tool call -- so the transcript stays valid to send, persist, and resume.
Still degenerate after wrap-up. If even the wrap-up yields no answer, the agent raises :class:~aimu.agents.DegenerateTurnError rather than returning empty output.

Quick start::

from aimu.tools import tool
from aimu.agents import Agent
import aimu

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

client = aimu.client("ollama:qwen3.5:9b")
agent = Agent(client, "You are a helpful assistant.", tools=[letter_counter])
print(agent.run("How many r's in strawberry?"))

run ¶

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, Iterator[StreamChunk]]

Run the 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. See the class docstring's "Cap with tools pending" note.

tools is a per-run override of the agent's configured self.tools: None (default) uses them, any other value (including [] to disable Python tools for this run) replaces them for every chat() call in the loop and is restored afterward.

deps is a per-run override of the agent's self.deps field, the value injected as ctx.deps into tools that declare a :class:~aimu.tools.ToolContext parameter.

tool_approval is a per-run override of the agent's self.tool_approval field, the gate run before each tool call ((name, arguments) -> bool; deny appends a refusal tool message). Defaults to approving everything.

schema (a dataclass or Pydantic v2 model) makes the run a single structured-output turn that returns a validated instance instead of looping with tools. Use it for an agent whose job is to return a typed object (e.g. a critic's verdict). It is mutually exclusive with the tool-calling loop. With stream=True the run yields :class:StreamChunks (thinking/generation) ending in a terminal DONE chunk carrying {"result": <object>}; the object is also on model_client.last_structured.

thinking is a per-run override of the agent's self.thinking field, the portable reasoning control (True/False/"low"/"medium"/"high"). None (default) uses the field. The effective value is applied to every model turn the run makes, including the continuation nudge and the forced wrap-up, so effort is uniform across the run. It is the public argument, so the model client validates it and warns once if its model cannot honour it; the agent itself makes no capability decisions.

events is a per-run override of the agent's self.events field: a callable taking one :class:~aimu.events.RunEvent (see :mod:aimu.events). None (default) uses the field. The resolved sink is installed as the active sink for the run's duration via a scoped contextvars.ContextVar override (restored afterward, even if the run raises), so the client's own turn events reach it alongside the loop's own :class:~aimu.events.RunStarted / :class:RunFinished / :class:ToolCalled / :class:ToolDenied, every one stamped with this agent's name and the current loop iteration. 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): the override lives in a per-execution-context ContextVar, not a mutation of the client, so each concurrently-running agent's OS thread gets its own independent copy and 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): concurrent_tool_calls=True dispatches sync tools via a plain ThreadPoolExecutor.submit() with no context copy, so that thread's empty context makes the override invisible there even though the client matches; aio.Agent.run(events=...)'s equivalent case is the opposite, since asyncio.TaskGroup.create_task always copies the current context -- see its 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 the agent's self.compaction field: a callable applied to the conversation before every model turn the run makes (see :mod:aimu.context for trim_messages / summarize_messages). None (default) uses the field. Not used by the schema= structured-output path, which makes a single model turn rather than running the tool loop.

as_model_client ¶

as_model_client() -> BaseModelClient

Return a :class:BaseModelClient view of this agent.

Each chat() call on the returned object runs the full agent loop, looping until the model stops calling tools. Use this only where an API expects a BaseModelClient. For direct use, call :meth:run instead.

from_config `classmethod` ¶

from_config(config: dict[str, Any], model_client: BaseModelClient) -> Agent

Create an Agent from a plain dict config.

Recognised keys: name, system_message, max_iterations, continuation_prompt, final_answer_prompt, thinking.

aimu.agents.SkillAgent `dataclass` ¶

SkillAgent(model_client: BaseModelClient, 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

An :class:Agent extended 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 skills server's tools (via MCPClient.as_tools()) through :meth:_effective_tools, so the tool-loop engine advertises and dispatches them (the model can call activate_skill to load full skill instructions before proceeding).

By default a fresh :class:SkillManager is created, scanning the standard search paths (.agents/skills/, .claude/skills/, ~/.agents/skills/, ~/.claude/skills/). Pass an explicit SkillManager to override.

Usage::

agent = SkillAgent(client, "Use available skills as needed.")
result = agent.run("Use the pdf-processing skill to extract pages.")

With explicit skill dirs::

agent = SkillAgent(client, skill_manager=SkillManager(skill_dirs=["./skills"]))

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 ¶

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 (and appears in the system prompt without starting a fresh conversation). Call after writing a new skill/script (see :func:aimu.skills.make_skill_script_tool).

from_config `classmethod` ¶

from_config(config: dict[str, Any], model_client: BaseModelClient) -> SkillAgent

Create a SkillAgent from a plain dict config.

Recognised keys: name, system_message, max_iterations, continuation_prompt, skill_dirs (omit to auto-discover).

aimu.agents.OrchestratorAgent ¶

Bases: Runner, ABC

Base class 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::

from aimu.tools import tool

class ResearchAgent(OrchestratorAgent):
    def __init__(self, client):
        researcher = Agent(client, "Research the topic.", name="researcher")

        @tool
        def research(topic: str) -> str:
            """Run the researcher on a topic."""
            return researcher.run(topic)

        self._init_orchestrator(
            client,
            name="research-orchestrator",
            system_message="Use the research tool to investigate.",
            tools=[research],
        )

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

assemble `classmethod` ¶

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

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

Each worker becomes a callable tool via :meth:Runner.as_tool; the orchestrator dispatches by name. Workers may be any :class:Runner (an :class:Agent, a Chain/Router/Parallel workflow, or a remote A2A agent), not just Agent instances; tool names/descriptions come from each worker's name and (when present) system_message.

Example::

researcher = Agent(client, "Research the topic.", name="researcher")
critic = Agent(client, "Critique the response.", name="critic")
orch = OrchestratorAgent.assemble(client, "Use both workers.",
                                  workers=[researcher, critic])
print(orch.run("Quantum computing"))

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:Agent.restore for the full save/restore pattern.

Workflows¶

aimu.agents.Chain `dataclass` ¶

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

Bases: Runner

Prompt Chaining pattern: agents run sequentially, output → input.

Each step's text output (concatenated GENERATING chunks) becomes the next step's task input. Steps may be :class:Agent instances or nested workflows.

Quick start::

chain = Chain.from_client(client, [
    "Break the task into 3 concrete steps.",
    "Execute each step and collect results.",
    "Polish the result into a final report.",
])
result = chain.run("Research top Python web frameworks.")

Direct construction (each step owns its own client/agent)::

chain = Chain(agents=[
    Agent(client_a, "Planner", name="planner"),
    Agent(client_b, "Executor", name="executor"),
])

from_client `classmethod` ¶

from_client(client: BaseModelClient, 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.

Each step gets its own :class:Agent with reset_messages_on_run=True so it clears the shared client's history and applies its own system_message. events is passed to every step's :class:Agent, so one sink sees the whole pipeline with each event attributed to the step (agent) that produced it.

run ¶

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

Run all agents sequentially. images are forwarded only to the first step.

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, the first step); subsequent steps start fresh on the next run(). Raises IndexError if step is out of range. See :meth:Agent.restore for the full save/restore pattern.

from_config `classmethod` ¶

from_config(configs: list[dict[str, Any]], client: BaseModelClient) -> Chain

Build a Chain from a list of agent config dicts and a single client.

Each step's Agent gets reset_messages_on_run=True so it clears the client's messages and applies its own system_message before running.

aimu.agents.Router `dataclass` ¶

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

Bases: Runner

Routing pattern: classify the task, dispatch to a specialist.

The routing_agent receives the task and must respond with a single route name. The Router dispatches to the matching handler (case-insensitive, whitespace-stripped). Handlers may be any :class:Runner (agent or nested workflow).

Quick start::

router = Router.from_client(
    client,
    classifier_prompt=(
        "Classify the task as one of: code, writing, math. "
        "Reply with only the category name."
    ),
    handlers={
        "code":    Agent(client, "You are a coder.", name="coder"),
        "writing": Agent(client, "You are a writer.", name="writer"),
        "math":    Agent(client, "You are a mathematician.", name="math"),
    },
    fallback=Agent(client, "Be helpful.", name="general"),
)

from_client `classmethod` ¶

from_client(client: BaseModelClient, classifier_prompt: str, handlers: dict[str, Runner], *, fallback: Optional[Runner] = 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: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.

run ¶

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

Classify the task and dispatch to the matched handler.

images are forwarded to the handler; the routing agent classifies on text only.

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:Agent.restore for the pattern.

from_config `classmethod` ¶

from_config(routing_config: dict[str, Any], handler_configs: dict[str, dict[str, Any]], client: BaseModelClient, fallback_config: Optional[dict[str, Any]] = None) -> Router

Build a Router from config dicts and a single client.

aimu.agents.Parallel `dataclass` ¶

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

Bases: Runner

Parallelization pattern: run workers concurrently, aggregate.

Each worker receives the same task. An optional aggregator receives all worker outputs joined by separator and produces the final result. Without an aggregator, the joined worker outputs are returned directly.

Workers run concurrently via ThreadPoolExecutor; results are collected in submission order. Workers may be any :class:Runner.

Quick start::

parallel = Parallel.from_client(
    client,
    worker_prompts=[
        "Analyze this from a security perspective.",
        "Analyze this from a performance perspective.",
        "Analyze this from a readability perspective.",
    ],
    aggregator_prompt="Synthesize the perspectives into one concise review.",
)

from_client `classmethod` ¶

from_client(client: BaseModelClient, 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 via ThreadPoolExecutor (Parallel._run_workers) -- each worker's per-run sink is delivered through a scoped contextvars.ContextVar override (BaseModelClient._events_override / _effective_sink), not a mutation of client.events, so each worker's own OS thread gets its own independent copy and concurrent delivery is correctly attributed and ordered per worker (see tests/test_workflow_parallel.py's concurrent-workers test). 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 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), mirroring Chain.restore's step. Other workers and the aggregator start fresh on the next run(). Raises IndexError if worker is out of range. See :meth:Agent.restore for the full save/restore pattern.

run ¶

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

Run workers concurrently then aggregate.

images are forwarded to every worker; the aggregator runs on text only.

aimu.agents.EvaluatorOptimizer `dataclass` ¶

EvaluatorOptimizer(generator: Agent, evaluator: Agent, name: str = 'evaluator_optimizer', max_rounds: int = 3, pass_keyword: str = 'PASS', stop_when: Optional[Callable[[Any], bool]] = None, verdict_schema: Optional[type] = None, passed_attr: str = 'passed', feedback_attr: str = 'feedback')

Bases: Runner

Evaluator-Optimizer pattern: iteratively improve via critic feedback.

The generator produces an initial response. The evaluator reviews it against the original task and decides whether it is accepted; if not, its feedback drives a revision. The loop stops when the evaluator accepts or max_rounds is reached.

Acceptance is decided by one of three mechanisms, in priority order:

stop_when: a predicate over the evaluator's output (the raw text, or the typed verdict when verdict_schema is set). The most flexible and robust option.
verdict_schema: a dataclass / Pydantic model the evaluator must return (via structured output). Acceptance reads its passed (bool) attribute and revision uses its feedback (str) attribute. Requires a structured-output-capable model; it raises on a malformed verdict rather than silently continuing.
pass_keyword (default): accept when this substring appears in the evaluator's text.

Usage::

eo = EvaluatorOptimizer(
    generator=Agent(client, "Write a clear, accurate explanation.", name="writer"),
    evaluator=Agent(client, "Reply PASS or REVISE: <feedback>.", name="critic"),
    max_rounds=4,
)
result = eo.run("Explain gradient descent.")

# Typed verdict instead of substring matching:
class Verdict(BaseModel):
    passed: bool
    feedback: str = ""
eo = EvaluatorOptimizer(generator=writer, evaluator=critic, verdict_schema=Verdict)

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:Agent.restore for the full save/restore pattern and system-message handling details.

run ¶

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

Run the generate-evaluate loop. Streaming yields only the final output.

aimu.agents.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: Runner

Plan → execute → evaluate → replan-on-fail loop.

A planner agent produces a plan for the task; an executor agent runs the plan with its tools; a pluggable :class:Scorer judges the output. On failure, the planner is invoked again with the prior round's feedback to produce a new plan (not a revision). The loop bounds at max_rounds and on exhaustion returns the highest-scoring attempt.

Two criteria modes:

User-supplied (criteria="..." at construction): the criteria is passed to the planner's prompt every round, and to the scorer as the row's reference field.
Planner-invented (criteria=None): the planner is asked to emit ## Evaluation criteria + ## Plan sections; the workflow parses both. The parsed criteria is passed to the scorer as reference.

Usage::

wf = PlanExecuteEvaluator.from_client(
    client=aimu.client("anthropic:claude-sonnet-4-6"),
    judge_client=aimu.client("openai:gpt-4o"),
    executor_tools=builtin.web + builtin.fs,
    criteria="The summary cites 3+ sources and is under 200 words.",
    max_rounds=3,
)
result = wf.run("Summarise the latest news on quantum computing.")

last_attempts `property` ¶

last_attempts: list[dict]

Per-round attempts from the most recent run().

Each entry has round, plan, criteria, output, score, feedback. Useful for introspection or post-hoc analysis.

run ¶

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

Run the plan-execute-evaluate loop. images are forwarded to the executor each round.

from_client `classmethod` ¶

from_client(client: BaseModelClient, *, judge_client: Optional[BaseModelClient] = 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 client and the common knobs.

planner is a :class:SkillAgent over client; pass skill_manager= to use task-type planning skills from disk.
executor is an :class:Agent over client with the given executor_tools.
scorer is an :class:LLMJudgeScorer over judge_client (defaults to client if not provided; a stronger separate model is recommended). Its criteria is the workflow's criteria if supplied, otherwise a generic fallback.
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.

For full control (e.g. a custom :class:Scorer), construct the dataclass directly.

A2A interop¶

Optional Agent2Agent protocol support (requires the a2a extra: pip install 'aimu[a2a]'). Available only when aimu.agents.HAS_A2A is True. See A2A vs MCP and Connect agents (A2A).

aimu.agents.RemoteAgent ¶

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

Bases: Runner

A remote A2A agent presented as a local synchronous Runner.

Construct via :meth:connect (it resolves the remote agent card first)::

remote = RemoteAgent.connect("http://localhost:9000")
print(remote.run("Summarise the news"))

# Composes like any other runner:
chain = Chain(agents=[local_agent, remote])
local = Agent(client, tools=[remote.as_tool()])

Holds a live async connection (an httpx.AsyncClient driven through an anyio portal); keep the instance alive for the connection's lifetime and let it be garbage collected (or call :meth:close) to tear the portal down.

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

name defaults to the remote agent card's name. Connection failures raise :class:A2AConnectionError with the original exception chained.

run ¶

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

Send task to the remote agent and return its text response.

stream=True yields the full response as a single GENERATING chunk followed by DONE; incremental token streaming over the portal is not supported on the sync surface (use aimu.aio.a2a.RemoteAgent for real streaming).

aimu.agents.serve_a2a ¶

serve_a2a(runner: Runner, *, 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 runner as an A2A agent over HTTP (blocking).

url is the externally reachable base URL advertised in the agent card; it defaults to http://{host}:{port}/. Extra kwargs pass through to uvicorn.run.

aimu.agents.build_a2a_app ¶

build_a2a_app(runner: Runner, *, 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 runner over A2A (does not run it).

aimu.agents.A2AConnectionError ¶

Bases: RuntimeError

Raised when a :class:RemoteAgent fails to connect to or call a remote A2A agent.

aimu.agents¶

Hierarchy¶

aimu.agents.Runner ¶

messages abstractmethod property ¶

run abstractmethod ¶

as_tool ¶

aimu.agents.MessageHistory module-attribute ¶

Agents¶

aimu.agents.Agent dataclass ¶

run ¶

as_model_client ¶

from_config classmethod ¶

aimu.agents.SkillAgent dataclass ¶

reload_skills ¶

from_config classmethod ¶

aimu.agents.OrchestratorAgent ¶

assemble classmethod ¶

restore ¶

Workflows¶

aimu.agents.Chain dataclass ¶

from_client classmethod ¶

run ¶

restore ¶

from_config classmethod ¶

aimu.agents.Router dataclass ¶

from_client classmethod ¶

run ¶

restore ¶

from_config classmethod ¶

aimu.agents.Parallel dataclass ¶

from_client classmethod ¶

restore ¶

run ¶

aimu.agents.EvaluatorOptimizer dataclass ¶

restore ¶

run ¶

aimu.agents.PlanExecuteEvaluator dataclass ¶

last_attempts property ¶

run ¶

from_client classmethod ¶

A2A interop¶

aimu.agents.RemoteAgent ¶

connect classmethod ¶

run ¶

aimu.agents.serve_a2a ¶

aimu.agents.build_a2a_app ¶

aimu.agents.A2AConnectionError ¶

`aimu.agents`¶

messages `abstractmethod` `property` ¶

run `abstractmethod` ¶

aimu.agents.MessageHistory `module-attribute` ¶

aimu.agents.Agent `dataclass` ¶

from_config `classmethod` ¶

aimu.agents.SkillAgent `dataclass` ¶

from_config `classmethod` ¶

assemble `classmethod` ¶

aimu.agents.Chain `dataclass` ¶

from_client `classmethod` ¶

from_config `classmethod` ¶

aimu.agents.Router `dataclass` ¶

from_client `classmethod` ¶

from_config `classmethod` ¶

aimu.agents.Parallel `dataclass` ¶

from_client `classmethod` ¶

aimu.agents.EvaluatorOptimizer `dataclass` ¶

aimu.agents.PlanExecuteEvaluator `dataclass` ¶

last_attempts `property` ¶

from_client `classmethod` ¶

connect `classmethod` ¶