symfonic.agent.subagents.registry

registry

In-process AgentStore backed by child SymfonicAgent instances.

SubAgentRegistry

SubAgentRegistry(sub_agents: Sequence[SubAgent])

Concrete AgentStore that runs registered child agents in-process.

Satisfies the AgentStore protocol (list / read / run). The run method inherits the parent's tenant scope into the child and threads the (already incremented) agent depth so the child's own delegation guard can enforce max_agent_depth for nested delegation.

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

def __init__(self, sub_agents: Sequence[SubAgent]) -> None:
    self._agents: dict[str, SubAgent] = {}
    for sa in sub_agents:
        if sa.name in self._agents:
            raise ValueError(f"duplicate sub-agent name: {sa.name!r}")
        self._agents[sa.name] = sa

run async

run(
    name: str,
    agent_input: Any,
    *,
    parent_scope: Any = None,
    agent_depth: int = 1,
    run_id: str | None = None,
) -> Any

Run child name on agent_input; returns its AgentResponse.

Parameters:

Name	Type	Description	Default
name	str	Registered sub-agent name.	required
agent_input	Any	The delegated task (coerced to str for run).	required
parent_scope	Any	Tenant scope inherited from the parent run.	None
agent_depth	int	Depth to stamp on the child run (parent depth + 1).	1
run_id	str | None	Optional correlation id passed to the child.	None

Raises:

Type	Description
StorageError	If name is not a registered sub-agent.

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

async def run(
    self,
    name: str,
    agent_input: Any,
    *,
    parent_scope: Any = None,
    agent_depth: int = 1,
    run_id: str | None = None,
) -> Any:
    """Run child ``name`` on ``agent_input``; returns its ``AgentResponse``.

    Args:
        name: Registered sub-agent name.
        agent_input: The delegated task (coerced to ``str`` for ``run``).
        parent_scope: Tenant scope inherited from the parent run.
        agent_depth: Depth to stamp on the child run (parent depth + 1).
        run_id: Optional correlation id passed to the child.

    Raises:
        StorageError: If ``name`` is not a registered sub-agent.
    """
    sa = self._agents.get(name)
    if sa is None:
        raise StorageError(f"unknown sub-agent: {name!r}")
    return await sa.agent.run(
        query=str(agent_input),
        scope=parent_scope,
        run_id=run_id,
        agent_depth=agent_depth,
    )