Skip to content

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.

Registration is the last point at which a child can be refused, so it is where the block-edit lockdown lands for children the parent did not build. Every child passes through here -- specs the parent constructed and pre-built SubAgent(agent=...) instances alike -- which makes it the one place the guarantee covers every construction path.

Source code in symfonic/agent/subagents/registry.py 
35
36
37
38
39
40
41
42
43
44
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}")
        # Raises: a pre-built child holding a block-edit tool cannot be
        # sanitised after construction, and registering it anyway would
        # leave the caller trusting a lock that did not hold.
        assert_no_block_edit_surface(sa.name, sa.agent)
        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, parent_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

The id the child runs under. Minted here when the caller names none, so the value is known rather than left for the child to invent privately -- parent_run_id below has to point at something.

 None
parent_run_id str | None

v9.12.0 (issue #64) the run delegating this task. Recorded as the child's lineage; never passed as the child's own run_id. A child that ran under its parent's id would share every run-keyed map with it, and ConversationMetricsCollector.on_agent_end pops six of those -- so a child finishing mid-turn would tear down its still-running parent's tenant and conversation entries, and the parent's remaining token usage would be attributed to nobody and skipped by the budget tracker.

 None

Raises:

Type Description
StorageError

If name is not a registered sub-agent.
Source code in symfonic/agent/subagents/registry.py 
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
async def run(
    self,
    name: str,
    agent_input: Any,
    *,
    parent_scope: Any = None,
    agent_depth: int = 1,
    run_id: str | None = None,
    parent_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: The id the child runs under. Minted here when the caller
            names none, so the value is *known* rather than left for the
            child to invent privately -- ``parent_run_id`` below has to
            point at something.
        parent_run_id: v9.12.0 (issue #64) the run delegating this task.
            Recorded as the child's lineage; never passed as the child's
            own ``run_id``. A child that ran under its parent's id would
            share every run-keyed map with it, and
            ``ConversationMetricsCollector.on_agent_end`` *pops* six of
            those -- so a child finishing mid-turn would tear down its
            still-running parent's tenant and conversation entries, and the
            parent's remaining token usage would be attributed to nobody
            and skipped by the budget tracker.

    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}")
    child_run_id = run_id or uuid.uuid4().hex[:12]
    if parent_run_id:
        self._record_lineage(sa, child_run_id, parent_run_id)
    return await sa.agent.run(
        query=str(agent_input),
        scope=parent_scope,
        run_id=child_run_id,
        agent_depth=agent_depth,
    )