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

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,
) -> 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).
    """
    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()
        try:
            resp = await registry.run(
                name, task, parent_scope=scope, agent_depth=child_depth
            )
        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]