sub_agents

Level 2 · Tools and Skills — delegation: a parent agent hands work to a named child.

Delegation lets one agent call another as a tool. You declare a child as a SubAgent, pass it to the parent via SymfonicAgent(sub_agents=[...]), and the framework auto-registers run_agent / list_agents tools plus a concrete AgentStore. The parent's model can then emit run_agent('researcher', <task>) to delegate. This example drives the same SubAgentRegistry the engine builds internally, so the delegated call is deterministic without a live LLM.

• Lines: ~90
• Prerequisites: none (runs on MockModelProvider)
• Key concepts: SubAgent, SubAgentRegistry, sub_agents= wiring, run_agent / list_agents, depth guard

Run it

Installed via pip? Copy this example into your project with the CLI (no checkout needed):

pip install "symfonic-core[cli]"
symfonic examples add sub_agents
python -m sub_agents

Or, from a source checkout (run from the repo root):

python -m examples.sub_agents

Full code

"""Sub-agents: a parent hands work to a named child, and the child runs.

Usage: python -m examples.sub_agents

Public surface used, and nothing else::

    from symfonic import Agent
    from symfonic.capabilities.delegation import PrebuiltChild, delegated_children

Runs entirely offline. The previous version of this file said what it could
not show: "here we drive the same ``SubAgentRegistry`` the engine builds
internally so the flow is deterministic without a live LLM". It declared
children, built a registry, and called that registry by hand. The parent
never delegated, so nothing here could have caught a hand-off that did not
work -- and for a long time none did.

This one delegates. The model asks for ``run_agent``, the framework runs the
child, and the child's answer comes back as the tool result.

**Each child is its own agent.** Its provider, its instructions and its
model are the ones passed to *its* constructor -- which is why "give the
editor a cheaper model" is one argument on one line rather than a field on a
spec whose resolution rules you have to learn.

**These children hold no state, so they are handed over finished.**
``PrebuiltChild`` is right for a stateless specialist, a remote proxy or a
test double. A child with memory or any tenant-bound capability must be a
``ScopedChild`` instead: a finished agent is built with one scope, and
sharing one across tenants answers one tenant's delegation out of another
tenant's recollections.
"""

from __future__ import annotations

import asyncio
from typing import Any

from symfonic import Agent
from symfonic.capabilities.delegation import PrebuiltChild, delegated_children
from symfonic.core.testing import MockModelProvider

RESEARCHER = "Answer with one sourced fact and nothing else."
EDITOR = "Tighten prose without changing its meaning."

#: The editor runs on a cheaper model. One argument on one constructor --
#: which is the whole difference from a spec field whose resolution rules
#: against the parent you have to learn.
EDITOR_MODEL = "claude-haiku-4-5"


class AsksOnce:
    """A parent's model: one hand-off, then an answer.

    ``MockModelProvider`` repeats its scripted tool call every round, so a
    parent driven by it delegates until the plan's round cap. Two rounds is
    what a delegation looks like.
    """

    _symfonic_provider_family = "unknown"
    _symfonic_default_model = None

    def __init__(self, call: dict[str, Any], answer: str) -> None:
        self._call = call
        self._answer = answer

    def get_chat_model(self, config: Any) -> Any:  # noqa: ARG002
        from langchain_core.messages import AIMessage
        from langchain_core.outputs import ChatGeneration, ChatResult

        from symfonic.core.testing import MockChatModel

        call, answer = self._call, self._answer
        rounds = {"n": 0}

        class _Model(MockChatModel):
            def _generate(self, messages, stop=None, run_manager=None, **kwargs):  # noqa: ANN001, ANN202
                rounds["n"] += 1
                first = rounds["n"] == 1
                message = AIMessage(content="" if first else answer)
                if first:
                    message.tool_calls = [call]  # type: ignore[attr-defined]
                return ChatResult(generations=[ChatGeneration(message=message)])

            async def _agenerate(self, messages, stop=None, run_manager=None, **kw):  # noqa: ANN001, ANN202
                return self._generate(messages, stop, **kw)

        return _Model()

    def supports_thinking(self) -> bool:
        return False

    def supports_streaming(self) -> bool:
        return True

    def supports_forced_tool_choice(self, config: Any) -> bool:  # noqa: ARG002
        return True


def _mock(answer: str) -> Any:
    return MockModelProvider(response=answer)


def build_children(provider_for: Any = None) -> list[PrebuiltChild]:
    """Two specialists, each with its own persona and its own model.

    ``provider_for`` is a seam, not configuration: it takes ``(name, answer)``
    and returns a provider, so a test can hand each child one that records
    the ``model_name`` it was asked for. An example whose only evidence for
    "the editor runs on a cheaper model" is the line that says so has no
    evidence -- ``Agent`` exposes nothing about what it was built with.
    """
    make = provider_for or (lambda name, answer: _mock(answer))  # noqa: ARG005
    researcher = Agent(
        make("researcher", "Rye flour peaked in March 2024."),
        instructions=RESEARCHER,
    )
    editor = Agent(
        make("editor", "Rye peaked in March 2024."),
        instructions=EDITOR,
        model=EDITOR_MODEL,
    )
    return [
        PrebuiltChild(
            name="researcher",
            agent=researcher,
            description="Deep research over the price archive.",
            when_to_use="Whenever a question needs a sourced number.",
        ),
        PrebuiltChild(
            name="editor", agent=editor, description="Tightens prose."
        ),
    ]


async def main() -> None:
    children = build_children()
    capability = delegated_children(children, max_depth=2)

    parent = Agent(
        AsksOnce(
            {
                "name": "run_agent",
                "args": {"name": "researcher", "task": "When did rye peak?"},
                "id": "d1",
            },
            "Research says rye peaked in March 2024.",
        ),
        instructions="Delegate research questions; answer everything else yourself.",
        capabilities=[capability],
    )

    result = await parent.run("When did rye flour prices peak?")

    print("Children the parent can reach:")
    for child in children:
        print(f"  - {child.name}: {child.when_to_use or child.description}")

    call = next(c for c in (result.tool_calls or []) if c.name == "run_agent")
    print()
    print(f"Delegated  : {call.name}({call.arguments})")
    print(f"Child said : {call.result}")
    print(f"Parent said: {result.text}")
    print(f"Parent msgs: {[m.role for m in result.messages]}")
    print()
    print("The child's transcript is not in that list. Containment is the")
    print("reason to delegate: a specialist that dumped its scratch work into")
    print("the parent's context would cost more than doing the work inline.")


if __name__ == "__main__":
    asyncio.run(main())

Step by step

1. Build the child agent

def build_researcher() -> SymfonicAgent:
    return SymfonicAgent(
        model_provider=MockModelProvider(response="..."),
        config=FrameworkConfig(
            domain=DomainTemplate(name="researcher", description="..."),
            agent=AgentConfig(model=ModelConfig(max_tokens=512, temperature=0.4)),
        ),
    )

A child is just another SymfonicAgent — its own provider, persona (DomainTemplate), and model config. Capping max_tokens=512 keeps a delegated answer short; the framework default (16k) is large for a helper. The child can have its own tools and memory too.

2. Declare the delegation contract

researcher_sa = SubAgent(
    name="researcher",
    agent=researcher,
    description="Background research specialist.",
    when_to_use="Explain / define / compare questions with no dedicated tool.",
)

SubAgent wraps the child with the metadata the parent's model needs to decide whether to delegate: name (how it's addressed), description (what it does), and when_to_use (the routing hint the model reads). Write when_to_use carefully — it is the model's only guidance on when to hand off.

3. Wire children into the parent

parent = SymfonicAgent(
    model_provider=MockModelProvider(response="..."),
    config=FrameworkConfig(domain=DomainTemplate(name="assistant", ...)),
    sub_agents=[researcher_sa],
)

Passing sub_agents=[...] is all it takes. Before the graph compiles, the engine auto-registers two tools — run_agent(name, task) and list_agents() — and a concrete AgentStore backed by your children. The parent's model now sees these tools and can delegate on its own.

4. Delegate a call

registry = SubAgentRegistry([researcher_sa])
response = await registry.run("researcher", "Why is the sky blue?")
print(f"Child answered: {response.final_response}")

With a real provider you'd just ask the parent a question and its model would emit run_agent('researcher', ...) when appropriate. To keep this demo deterministic, we build the same SubAgentRegistry the engine builds internally and invoke the child directly — the flow is identical, minus the model's routing decision.

Depth guard

Delegation is recursive-capable (a child can have its own children), so the framework enforces a depth guard to prevent runaway chains. A child cannot delegate infinitely; the guard caps the delegation depth. See the Sub-agents guide for tuning it.

What to try next

• Full delegation guide with a live model → Sub-agents (Delegation)
• Give the parent or child real tools → Writing Tools

See also

• The Agent
• Sub-agents guide