SubAgentTool — bind a stateless sub-agent to an executor¶

Orqest agents often build compound tools: LLM-facing callables that delegate a structured decision to a stateless sub-agent, execute the decision against a real system (MCP pipeline, sandbox, HTTP API, …), commit the result back onto long-lived session state, and optionally refine the result when a quality check fails.

Before SubAgentTool, every consumer hand-wrote that pipeline — tens of lines per tool, with hand-rolled refinement loops, manual state mutations, and inconsistent reporting of "did a refinement fire?". SubAgentTool captures the pattern in a single class so consumers only write the domain-specific executor, state updater, and (optional) evaluator.

Minimal example¶

import asyncio

from pydantic import BaseModel

from orqest.agents import BaseAgent
from orqest.compound import SubAgentTool
from orqest.compound.sub_agent_tool import SubAgentEvalResult


class GeoOutput(BaseModel):
    code: str


class State:
    def __init__(self):
        self.mesh_code = ""
        self.quality = 0.0


# Sub-agent (stateless; produces a structured output)
class GeometryAgent(BaseAgent[State, GeoOutput]):
    async def _run_implementation(self, state: State, **kwargs) -> GeoOutput:
        # In real code this calls an LLM; here we fake it.
        note = kwargs.get("note", "")
        return GeoOutput(code=f"// mesh for: {note}")


async def run_pipeline(out: GeoOutput, state: State) -> dict:
    # Imagine MCP calls here producing a quality score
    return {"code": out.code, "quality": 0.8}


def update_state(result: dict, state: State) -> None:
    state.mesh_code = result["code"]
    state.quality = result["quality"]


def check_quality(result: dict) -> SubAgentEvalResult:
    return SubAgentEvalResult(passed=result["quality"] >= 0.5)


def refine_prompt(result: dict, prompt: str) -> str:
    return f"{prompt}\n\nPREVIOUS QUALITY {result['quality']} — improve."


tool = SubAgentTool(
    agent=GeometryAgent(...),
    executor=run_pipeline,
    state_updater=update_state,
    evaluator=check_quality,
    max_refinements=1,
    build_refinement_prompt=refine_prompt,
)


async def main() -> None:
    state = State()
    outcome = await tool.run(state, prompt="square plate")
    print(outcome.result, outcome.refined, outcome.iterations)

When to reach for this¶

You have a structured sub-agent that produces a plan/code/spec, and a deterministic executor that acts on that spec.
You want quality-based refinement without hand-rolling a retry loop.
You want best-effort semantics: a failed refinement keeps the original result rather than propagating an exception.
You compose with run_with_retry at the outer tool boundary to handle exceptions separately from quality failures.

Semantics — what `SubAgentTool` does and doesn't do¶

Does run one (agent → executor → state updater) cycle per pass.
Does fire up to max_refinements additional passes if an evaluator fails and a refinement-prompt builder is supplied.
Does commit state after every pass so downstream readers see the best result reached, even if a later refinement throws.
Does not wrap exceptions — raise from the sub-agent or executor to surface them to the caller (usually inside a run_with_retry block that enriches the prompt on failure).
Does not fire HookRunner events by itself. Wire the outer compound-tool boundary with EventBusPublishHook for observability.

Reference¶

SubAgentTool ¶

SubAgentTool(
    agent: BaseAgent,
    executor: Callable[[Any, StateT], Awaitable[ResultT]],
    state_updater: Callable[[ResultT, StateT], None],
    *,
    evaluator: Callable[[ResultT], SubAgentEvalResult]
    | None = None,
    max_refinements: int = 0,
    build_refinement_prompt: Callable[[ResultT, str], str]
    | None = None,
    name: str | None = None,
)

Compose a stateless sub-agent with an executor and optional refinement.

Usage — implement three domain functions and hand them to the tool:

executor(agent_output, state) -> Awaitable[ResultT] — turns the sub-agent's structured output into a real-world outcome (e.g. run the MCP pipeline and return a payload).
state_updater(result, state) -> None — mutates the long- lived session state with whatever needs to persist.
evaluator(result) -> SubAgentEvalResult (optional) — decides whether the result is good enough or a refinement cycle should fire.

If max_refinements > 0 and the evaluator rejects the first result, the tool calls build_refinement_prompt(result, prompt) to construct a new prompt and reruns the sub-agent + executor cycle. Refinement exceptions are caught and the best prior result is kept — matching numatics-ai v2's "best effort refinement" rule.

The tool does NOT wrap retry-on-exception; for that, wrap the run call in :func:orqest.agents.retry.run_with_retry.

Configure the compound flow.

Parameters:

Name	Type	Description	Default
`agent`	`BaseAgent`	A stateless :class:`~orqest.agents.base_agent.BaseAgent` whose `.run()` takes `(state, **kwargs)` and returns a structured output (typically a Pydantic model).	required
`executor`	`Callable[[Any, StateT], Awaitable[ResultT]]`	Async function that runs the real-world action using the sub-agent's output and returns a serializable result payload.	required
`state_updater`	`Callable[[ResultT, StateT], None]`	Synchronous mutator that writes the result back onto the state object.	required
`evaluator`	`Callable[[ResultT], SubAgentEvalResult] \| None`	Optional quality check. When provided with `max_refinements > 0`, a failing evaluation triggers a refinement cycle.	`None`
`max_refinements`	`int`	Number of refinement cycles allowed after the first pass. `0` disables refinement.	`0`
`build_refinement_prompt`	`Callable[[ResultT, str], str] \| None`	`(result, original_prompt) -> new_prompt`. Required when `max_refinements > 0`.	`None`
`name`	`str \| None`	Optional tool name used in :class:`SubAgentResult` and error messages. Defaults to `agent.agent_name`.	`None`

Raises:

Type	Description
`ValueError`	If `max_refinements > 0` but `evaluator` or `build_refinement_prompt` is `None`.

Source code in orqest/compound/sub_agent_tool.py

def __init__(
    self,
    agent: BaseAgent,
    executor: Callable[[Any, StateT], Awaitable[ResultT]],
    state_updater: Callable[[ResultT, StateT], None],
    *,
    evaluator: Callable[[ResultT], SubAgentEvalResult] | None = None,
    max_refinements: int = 0,
    build_refinement_prompt: Callable[[ResultT, str], str] | None = None,
    name: str | None = None,
) -> None:
    """Configure the compound flow.

    Args:
        agent: A stateless :class:`~orqest.agents.base_agent.BaseAgent`
            whose ``.run()`` takes ``(state, **kwargs)`` and returns
            a structured output (typically a Pydantic model).
        executor: Async function that runs the real-world action
            using the sub-agent's output and returns a
            serializable result payload.
        state_updater: Synchronous mutator that writes the result
            back onto the state object.
        evaluator: Optional quality check. When provided with
            ``max_refinements > 0``, a failing evaluation triggers
            a refinement cycle.
        max_refinements: Number of refinement cycles allowed after
            the first pass. ``0`` disables refinement.
        build_refinement_prompt: ``(result, original_prompt) ->
            new_prompt``. Required when ``max_refinements > 0``.
        name: Optional tool name used in :class:`SubAgentResult`
            and error messages. Defaults to ``agent.agent_name``.

    Raises:
        ValueError: If ``max_refinements > 0`` but ``evaluator`` or
            ``build_refinement_prompt`` is ``None``.
    """
    if max_refinements > 0 and (
        evaluator is None or build_refinement_prompt is None
    ):
        raise ValueError(
            "SubAgentTool with max_refinements > 0 requires both "
            "evaluator and build_refinement_prompt."
        )

    self._agent = agent
    self._executor = executor
    self._state_updater = state_updater
    self._evaluator = evaluator
    self._max_refinements = max_refinements
    self._build_refinement_prompt = build_refinement_prompt
    self.name = name or getattr(agent, "agent_name", "sub_agent_tool")

run `async` ¶

run(
    state: StateT,
    prompt: str,
    *,
    use_enriched: bool = False,
    **agent_kwargs: Any,
) -> SubAgentResult[ResultT]

Execute one (agent + executor + state-update) cycle, then optionally refine up to max_refinements times.

The first result is always committed to state before refinement begins — so a failed refinement leaves the original result in place (best-effort semantics). Extra keyword arguments are forwarded verbatim to agent.run.

Parameters:

Name	Type	Description	Default
`state`	`StateT`	The agent's state object.	required
`prompt`	`str`	The user-facing prompt for the sub-agent.	required
`use_enriched`	`bool`	When `True`, runs the agent via :meth:`BaseAgent.run_enriched` and lifts the final-pass `confidence` / `uncertainty_targets` / `capability_boundary` onto the returned :class:`SubAgentResult`. The executor still receives the raw agent output (unwrapped from :class:`EnrichedOutput`). Default `False` preserves pre-metacognition behavior.	`False`
`**agent_kwargs`	`Any`	Forwarded to the agent.	`{}`

Source code in orqest/compound/sub_agent_tool.py

async def run(
    self,
    state: StateT,
    prompt: str,
    *,
    use_enriched: bool = False,
    **agent_kwargs: Any,
) -> SubAgentResult[ResultT]:
    """Execute one (agent + executor + state-update) cycle, then
    optionally refine up to ``max_refinements`` times.

    The first result is always committed to state before refinement
    begins — so a failed refinement leaves the original result in
    place (best-effort semantics). Extra keyword arguments are
    forwarded verbatim to ``agent.run``.

    Args:
        state: The agent's state object.
        prompt: The user-facing prompt for the sub-agent.
        use_enriched: When ``True``, runs the agent via
            :meth:`BaseAgent.run_enriched` and lifts the final-pass
            ``confidence`` / ``uncertainty_targets`` /
            ``capability_boundary`` onto the returned
            :class:`SubAgentResult`. The executor still receives the
            raw agent output (unwrapped from
            :class:`EnrichedOutput`). Default ``False`` preserves
            pre-metacognition behavior.
        **agent_kwargs: Forwarded to the agent.
    """

    async def _run_agent(state: StateT, **kw: Any) -> tuple[Any, dict[str, Any]]:
        """Returns ``(raw_output, enrichment_dict)``. The enrichment
        dict is empty when ``use_enriched`` is False or the agent
        has no protocol configured."""
        if use_enriched:
            enriched = await self._agent.run_enriched(state, **kw)
            return enriched.output, {
                "confidence": enriched.confidence,
                "uncertainty_targets": list(enriched.uncertainty_targets),
                "capability_boundary": enriched.capability_boundary,
            }
        return await self._agent.run(state, **kw), {}

    # --- First pass ---
    # Deliver the prompt both ways: as a user message on `state` (the
    # universal BaseAgent channel, when supported) and as a `note=`
    # kwarg (legacy — agents that read it directly still work).
    if hasattr(state, "add_message"):
        state.add_message("user", prompt)
    call_kwargs = {"note": prompt, **agent_kwargs}
    agent_output, last_enrichment = await _run_agent(state, **call_kwargs)
    first_result = await self._executor(agent_output, state)
    self._state_updater(first_result, state)

    iterations = 1
    current_result = first_result
    refined = False
    exit_reason = "passed"

    if (
        self._max_refinements > 0
        and self._evaluator is not None
        and self._build_refinement_prompt is not None
    ):
        eval_result = self._evaluator(current_result)
        while (
            not eval_result.passed
            and iterations <= self._max_refinements
        ):
            next_prompt = self._build_refinement_prompt(current_result, prompt)
            try:
                if hasattr(state, "add_message"):
                    state.add_message("user", next_prompt)
                refined_output, last_enrichment = await _run_agent(
                    state, note=next_prompt,
                )
                refined_result = await self._executor(
                    refined_output, state,
                )
                self._state_updater(refined_result, state)
                current_result = refined_result
                refined = True
                iterations += 1
                eval_result = self._evaluator(current_result)
            except Exception:
                # Best-effort refinement: exception keeps prior result.
                exit_reason = "refinement_failed_keep_original"
                break
        else:
            if not eval_result.passed:
                exit_reason = "max_refinements"

    return SubAgentResult(
        result=current_result,
        iterations=iterations,
        refined=refined,
        exit_reason=exit_reason,
        confidence=last_enrichment.get("confidence"),
        uncertainty_targets=last_enrichment.get("uncertainty_targets") or [],
        capability_boundary=bool(last_enrichment.get("capability_boundary", False)),
    )

SubAgentResult ¶

Outcome of a :meth:SubAgentTool.run invocation.

Captures the final result plus a short history of refinement attempts so consumers (metrics, trace spans, LLM tool returns) can surface "how many passes it took" without re-instrumenting.