symfonic.agent.subagents.tool

tool

Delegation tools that wire a SubAgentRegistry into the parent palette.

create_delegation_tools

create_delegation_tools(registry: SubAgentRegistry, resolve_scope: Callable[[], Any], resolve_depth: Callable[[], int], max_depth: int, record_delegation: Callable[[str], None] | None = None, resolve_run_id: Callable[[], str | None] | None = None) -> list[StructuredTool]

Build the run_agent + list_agents tools for a parent agent.

Parameters:

Name	Type	Description	Default
registry	SubAgentRegistry	The child-agent store.	required
resolve_scope	Callable[[], Any]	Returns the active tenant scope, or None if unset.	required
resolve_depth	Callable[[], int]	Returns the parent's current agent depth (0 at top level).	required
max_depth	int	max_agent_depth ceiling; delegation past it is refused gracefully (returned as a tool message, not an exception).	required
record_delegation	Callable[[str], None] | None	v9.1.0 (issue #29) optional callback invoked with the sub-agent name on each successful delegation so the engine can stamp AgentResponse.delegated_to. Called only after the child run returns (a refused/unknown/failed delegation is not recorded).	None
resolve_run_id	Callable[[], str | None] | None	v9.12.0 (issue #64) returns the active run's correlation id, or None outside a run. It is handed to SubAgentRegistry.run as the child's parent, never as the child's own run_id: the registry mints the child a fresh id and records the ancestry, which is what :meth:~symfonic.agent.subagents.registry.SubAgentRegistry.run explains. Before this the correlation hook was dead -- the registry accepted a run_id no caller passed, so every delegated run was correlated to nothing. A callable rather than a value for the same reason resolve_scope and resolve_depth are: the tool is bound once at construction and runs inside whatever turn calls it. Optional, so an unwired caller keeps the shipped behaviour.	None

Source code in src/symfonic/agent/subagents/tool.py

def create_delegation_tools(
    registry: SubAgentRegistry,
    resolve_scope: Callable[[], Any],
    resolve_depth: Callable[[], int],
    max_depth: int,
    record_delegation: Callable[[str], None] | None = None,
    resolve_run_id: Callable[[], str | None] | None = None,
) -> list[StructuredTool]:
    """Build the ``run_agent`` + ``list_agents`` tools for a parent agent.

    Args:
        registry: The child-agent store.
        resolve_scope: Returns the active tenant scope, or ``None`` if unset.
        resolve_depth: Returns the parent's current agent depth (0 at top level).
        max_depth: ``max_agent_depth`` ceiling; delegation past it is refused
            gracefully (returned as a tool message, not an exception).
        record_delegation: v9.1.0 (issue #29) optional callback invoked with
            the sub-agent name on each successful delegation so the engine can
            stamp ``AgentResponse.delegated_to``. Called only after the child
            run returns (a refused/unknown/failed delegation is not recorded).
        resolve_run_id: v9.12.0 (issue #64) returns the active run's
            correlation id, or ``None`` outside a run. It is handed to
            ``SubAgentRegistry.run`` as the child's **parent**, never as the
            child's own ``run_id``: the registry mints the child a fresh id and
            records the ancestry, which is what
            :meth:`~symfonic.agent.subagents.registry.SubAgentRegistry.run`
            explains. Before this the correlation hook was dead -- the registry
            accepted a ``run_id`` no caller passed, so every delegated run was
            correlated to nothing.

            A callable rather than a value for the same reason
            ``resolve_scope`` and ``resolve_depth`` are: the tool is bound once
            at construction and runs inside whatever turn calls it. Optional,
            so an unwired caller keeps the shipped behaviour.
    """
    roster = _roster_text(registry)

    async def run_agent(name: str, task: str) -> str:
        """Delegate a self-contained task to a named sub-agent."""
        child_depth = resolve_depth() + 1
        if child_depth > max_depth:
            return (
                f"Delegation refused: maximum sub-agent depth ({max_depth}) "
                "reached; cannot delegate further."
            )
        if name not in registry.names():
            return f"Unknown sub-agent {name!r}. Available:\n{roster}"
        scope = resolve_scope()
        # Resolved beside the scope and outside the ``try`` on purpose: a
        # resolver that raised inside it would be reported to the parent's
        # model as "Sub-agent 'x' failed", which is a wiring fault filed as a
        # child's death.
        parent_run_id = resolve_run_id() if resolve_run_id is not None else None
        try:
            resp = await registry.run(
                name,
                task,
                parent_scope=scope,
                agent_depth=child_depth,
                parent_run_id=parent_run_id,
            )
        except Exception as exc:  # noqa: BLE001 - contract: surface, don't raise
            # Per the tool contract, a child failure is returned to the parent
            # model as a tool message rather than crashing the parent run.
            return f"Sub-agent {name!r} failed: {exc}"
        if record_delegation is not None:
            record_delegation(name)
        return getattr(resp, "final_response", None) or ""

    async def list_agents() -> str:
        """List the sub-agents available for delegation."""
        return f"Available sub-agents:\n{roster}"

    run_agent_tool = StructuredTool.from_function(
        coroutine=run_agent,
        name="run_agent",
        description=(
            "Delegate a self-contained task to a specialized sub-agent and "
            "get its result back. Pass the sub-agent `name` and a complete "
            "`task` description — the sub-agent does NOT see this conversation, "
            "so include everything it needs.\nAvailable sub-agents:\n" + roster
        ),
    )
    list_agents_tool = StructuredTool.from_function(
        coroutine=list_agents,
        name="list_agents",
        description=(
            "List the sub-agents available for delegation via run_agent."
        ),
    )
    return [run_agent_tool, list_agents_tool]