Sub-Agents

Decompose work across multiple specialized agents with a visible delegation log.

"""Agno agent backing the Sub-Agents demo.Mirrors `langgraph-python/src/agents/subagents.py` and`google-adk/src/agents/subagents_agent.py`.A supervisor Agno agent delegates work to three specialized sub-agents(research / writing / critique) exposed as tools. Each delegationappends an entry to `session_state["delegations"]` so the UI can rendera live delegation log via `useAgent({ updates: [OnStateChanged] })`.Each sub-agent is itself a full `Agent(...)` with its own system prompt— the supervisor only sees the sub-agent's final text response. This isthe canonical Agno multi-agent pattern, surfaced to the frontend viashared state."""from __future__ import annotationsimport loggingimport uuidfrom typing import Anyimport dotenvfrom agno.agent.agent import Agentfrom agno.models.openai import OpenAIChatfrom agno.run import RunContextdotenv.load_dotenv()logger = logging.getLogger(__name__)# ---------------------------------------------------------------------------# Sub-agents (real Agno agents under the hood)# ---------------------------------------------------------------------------_SUB_MODEL_ID = "gpt-4o-mini"# Each sub-agent is a full Agno `Agent(...)` with its own system prompt.# They don't share memory or tools with the supervisor — the supervisor# only sees their final text response, which is returned via the# delegation tool below._research_agent = Agent(    model=OpenAIChat(id=_SUB_MODEL_ID, timeout=120),    description="Research sub-agent.",    instructions=(        "You are a research sub-agent. Given a topic, produce a concise "        "bulleted list of 3-5 key facts. No preamble, no closing."    ),)_writing_agent = Agent(    model=OpenAIChat(id=_SUB_MODEL_ID, timeout=120),    description="Writing sub-agent.",    instructions=(        "You are a writing sub-agent. Given a brief and optional source "        "facts, produce a polished 1-paragraph draft. Be clear and "        "concrete. No preamble."    ),)_critique_agent = Agent(    model=OpenAIChat(id=_SUB_MODEL_ID, timeout=120),    description="Critique sub-agent.",    instructions=(        "You are an editorial critique sub-agent. Given a draft, give "        "2-3 crisp, actionable critiques. No preamble."    ),)def _invoke_sub_agent(sub_agent: Agent, task: str) -> str:    """Run a sub-agent on `task` and return its final message content."""    result = sub_agent.run(input=task)    content = getattr(result, "content", None)    if isinstance(content, str):        return content.strip()    if content is None:        return ""    return str(content).strip()# ---------------------------------------------------------------------------# Shared-state helpers# ---------------------------------------------------------------------------def _append_delegation(    run_context: RunContext,    *,    sub_agent: str,    task: str,    status: str,    result: str,) -> str:    """Append a delegation entry and return its id."""    if run_context.session_state is None:        run_context.session_state = {}    delegations = list(run_context.session_state.get("delegations") or [])    entry_id = str(uuid.uuid4())    delegations.append(        {            "id": entry_id,            "sub_agent": sub_agent,            "task": task,            "status": status,            "result": result,        }    )    run_context.session_state["delegations"] = delegations    return entry_iddef _update_delegation(    run_context: RunContext,    *,    entry_id: str,    status: str,    result: str,) -> None:    """Mutate the delegation entry with `entry_id` in shared state.    If the entry has gone missing (another part of the system replaced    `session_state["delegations"]`), log loudly and skip rather than    appending a synthetic entry. Mirrors the conservative behavior used    in the google-adk reference.    """    if run_context.session_state is None:        run_context.session_state = {}    delegations = list(run_context.session_state.get("delegations") or [])    for entry in delegations:        if entry.get("id") == entry_id:            entry["status"] = status            entry["result"] = result            run_context.session_state["delegations"] = delegations            return    logger.warning(        "subagents: delegation entry %s missing on update — final %s "        "state (result_length=%d) will not be rendered",        entry_id,        status,        len(result),    )def _delegate(    run_context: RunContext,    *,    sub_agent_name: str,    sub_agent: Agent,    task: str,) -> dict[str, Any]:    """Common delegation flow: append running entry → invoke → update final."""    entry_id = _append_delegation(        run_context,        sub_agent=sub_agent_name,        task=task,        status="running",        result="",    )    try:        result = _invoke_sub_agent(sub_agent, task)    except Exception as exc:  # noqa: BLE001 — sub-agent transport can fail anywhere        logger.exception("subagents: sub-agent %s failed", sub_agent_name)        # Surface only the exception class to the supervisor / frontend —        # provider error strings can carry URLs / request IDs / partial        # credentials. The full traceback stays in server logs.        message = (            f"sub-agent call failed: {exc.__class__.__name__} "            "(see server logs for details)"        )        _update_delegation(            run_context, entry_id=entry_id, status="failed", result=message        )        return {"status": "failed", "error": message}    _update_delegation(        run_context, entry_id=entry_id, status="completed", result=result    )    return {"status": "completed", "result": result}# ---------------------------------------------------------------------------# Supervisor tools (each tool delegates to one sub-agent)# ---------------------------------------------------------------------------# Each function is a tool exposed to the supervisor agent. The supervisor# LLM "calls" these to delegate work; each call synchronously runs the# matching sub-agent, records the delegation into shared state, and# returns the sub-agent's output as the tool result the supervisor reads# on its next step.def research_agent(run_context: RunContext, task: str) -> dict[str, Any]:    """Delegate a research task to the research sub-agent.    Use for: gathering facts, background, definitions, statistics.    Returns {status, result} on success or {status: "failed", error} on    sub-agent failure.    """    return _delegate(        run_context,        sub_agent_name="research_agent",        sub_agent=_research_agent,        task=task,    )def writing_agent(run_context: RunContext, task: str) -> dict[str, Any]:    """Delegate a drafting task to the writing sub-agent.    Use for: producing a polished paragraph, draft, or summary. Pass    relevant facts from prior research inside `task`. Same return shape    as research_agent.    """    return _delegate(        run_context,        sub_agent_name="writing_agent",        sub_agent=_writing_agent,        task=task,    )def critique_agent(run_context: RunContext, task: str) -> dict[str, Any]:    """Delegate a critique task to the critique sub-agent.    Use for: reviewing a draft and suggesting concrete improvements.    Same return shape as research_agent.    """    return _delegate(        run_context,        sub_agent_name="critique_agent",        sub_agent=_critique_agent,        task=task,    )# ---------------------------------------------------------------------------# Supervisor (the agent we export)# ---------------------------------------------------------------------------_SUPERVISOR_INSTRUCTION = (    "You are a supervisor agent that coordinates three specialized "    "sub-agents to produce high-quality deliverables.\n\n"    "Available sub-agents (call them as tools):\n"    "  - research_agent: gathers facts on a topic.\n"    "  - writing_agent: turns facts + a brief into a polished draft.\n"    "  - critique_agent: reviews a draft and suggests improvements.\n\n"    "For most non-trivial user requests, delegate in sequence: "    "research -> write -> critique. Pass the relevant facts/draft "    "through the `task` argument of each tool. Each tool returns a dict "    "shaped {status: 'completed' | 'failed', result?: str, error?: str}. "    "If a sub-agent fails, surface the failure briefly to the user "    "(don't fabricate a result) and decide whether to retry. Keep your "    "own messages short — explain the plan once, delegate, then return a "    "concise summary once done. The UI shows the user a live log of "    "every sub-agent delegation, including the in-flight 'running' state.")agent = Agent(    model=OpenAIChat(id=_SUB_MODEL_ID, timeout=120),    tools=[research_agent, writing_agent, critique_agent],    description="Supervisor agent coordinating research / writing / critique sub-agents.",    instructions=_SUPERVISOR_INSTRUCTION,    tool_call_limit=10,)

What is this?#

Sub-agents are the canonical multi-agent pattern: a top-level supervisor LLM orchestrates one or more specialized sub-agents by exposing each of them as a tool. The supervisor decides what to delegate, the sub-agents do their narrow job, and their results flow back up to the supervisor's next step.

This is fundamentally the same shape as tool-calling, but each "tool" is itself a full-blown agent with its own system prompt and (often) its own tools, memory, and model.

When should I use this?#

Reach for sub-agents when a task has distinct specialized sub-tasks that each benefit from their own focus:

Research → Write → Critique pipelines, where each stage needs a different system prompt and temperature.
Router + specialists, where one agent classifies the request and dispatches to the right expert.
Divide-and-conquer — any problem that fits cleanly into parallel or sequential sub-problems.

The example below uses the Research → Write → Critique shape as the canonical example.

Setting up sub-agents#

Each sub-agent is a full create_agent(...) call with its own model, its own system prompt, and (optionally) its own tools. They don't share memory or tools with the supervisor; the supervisor only ever sees what the sub-agent returns.

subagents.py

from __future__ import annotationsimport loggingimport uuidfrom typing import Anyimport dotenvfrom agno.agent.agent import Agentfrom agno.models.openai import OpenAIChatfrom agno.run import RunContextdotenv.load_dotenv()logger = logging.getLogger(__name__)# ---------------------------------------------------------------------------# Sub-agents (real Agno agents under the hood)# ---------------------------------------------------------------------------_SUB_MODEL_ID = "gpt-4o-mini"# Each sub-agent is a full Agno `Agent(...)` with its own system prompt.# They don't share memory or tools with the supervisor — the supervisor# only sees their final text response, which is returned via the# delegation tool below._research_agent = Agent(    model=OpenAIChat(id=_SUB_MODEL_ID, timeout=120),    description="Research sub-agent.",    instructions=(        "You are a research sub-agent. Given a topic, produce a concise "        "bulleted list of 3-5 key facts. No preamble, no closing."    ),)_writing_agent = Agent(    model=OpenAIChat(id=_SUB_MODEL_ID, timeout=120),    description="Writing sub-agent.",    instructions=(        "You are a writing sub-agent. Given a brief and optional source "        "facts, produce a polished 1-paragraph draft. Be clear and "        "concrete. No preamble."    ),)_critique_agent = Agent(    model=OpenAIChat(id=_SUB_MODEL_ID, timeout=120),    description="Critique sub-agent.",    instructions=(        "You are an editorial critique sub-agent. Given a draft, give "        "2-3 crisp, actionable critiques. No preamble."    ),)

Keep sub-agent system prompts narrow and focused. The point of this pattern is that each one does one thing well. If a sub-agent needs to know the whole user context to do its job, that's a signal the boundary is wrong.

Exposing sub-agents as tools#

The supervisor delegates by calling tools. Each tool is a thin wrapper around sub_agent.invoke(...) that:

Runs the sub-agent synchronously on the supplied task string.
Records the delegation into a delegations slot in shared agent state (so the UI can render a live log).
Returns the sub-agent's final message as a ToolMessage, which the supervisor sees as a normal tool result on its next turn.

subagents.py

from __future__ import annotationsimport loggingimport uuidfrom typing import Anyimport dotenvfrom agno.agent.agent import Agentfrom agno.models.openai import OpenAIChatfrom agno.run import RunContextdotenv.load_dotenv()logger = logging.getLogger(__name__)# ---------------------------------------------------------------------------# Sub-agents (real Agno agents under the hood)# ---------------------------------------------------------------------------_SUB_MODEL_ID = "gpt-4o-mini"# Each sub-agent is a full Agno `Agent(...)` with its own system prompt.# They don't share memory or tools with the supervisor — the supervisor# only sees their final text response, which is returned via the# delegation tool below._research_agent = Agent(    model=OpenAIChat(id=_SUB_MODEL_ID, timeout=120),    description="Research sub-agent.",    instructions=(        "You are a research sub-agent. Given a topic, produce a concise "        "bulleted list of 3-5 key facts. No preamble, no closing."    ),)_writing_agent = Agent(    model=OpenAIChat(id=_SUB_MODEL_ID, timeout=120),    description="Writing sub-agent.",    instructions=(        "You are a writing sub-agent. Given a brief and optional source "        "facts, produce a polished 1-paragraph draft. Be clear and "        "concrete. No preamble."    ),)_critique_agent = Agent(    model=OpenAIChat(id=_SUB_MODEL_ID, timeout=120),    description="Critique sub-agent.",    instructions=(        "You are an editorial critique sub-agent. Given a draft, give "        "2-3 crisp, actionable critiques. No preamble."    ),)def _invoke_sub_agent(sub_agent: Agent, task: str) -> str:    """Run a sub-agent on `task` and return its final message content."""    result = sub_agent.run(input=task)    content = getattr(result, "content", None)    if isinstance(content, str):        return content.strip()    if content is None:        return ""    return str(content).strip()# ---------------------------------------------------------------------------# Shared-state helpers# ---------------------------------------------------------------------------def _append_delegation(    run_context: RunContext,    *,    sub_agent: str,    task: str,    status: str,    result: str,) -> str:    """Append a delegation entry and return its id."""    if run_context.session_state is None:        run_context.session_state = {}    delegations = list(run_context.session_state.get("delegations") or [])    entry_id = str(uuid.uuid4())    delegations.append(        {            "id": entry_id,            "sub_agent": sub_agent,            "task": task,            "status": status,            "result": result,        }    )    run_context.session_state["delegations"] = delegations    return entry_iddef _update_delegation(    run_context: RunContext,    *,    entry_id: str,    status: str,    result: str,) -> None:    """Mutate the delegation entry with `entry_id` in shared state.    If the entry has gone missing (another part of the system replaced    `session_state["delegations"]`), log loudly and skip rather than    appending a synthetic entry. Mirrors the conservative behavior used    in the google-adk reference.    """    if run_context.session_state is None:        run_context.session_state = {}    delegations = list(run_context.session_state.get("delegations") or [])    for entry in delegations:        if entry.get("id") == entry_id:            entry["status"] = status            entry["result"] = result            run_context.session_state["delegations"] = delegations            return    logger.warning(        "subagents: delegation entry %s missing on update — final %s "        "state (result_length=%d) will not be rendered",        entry_id,        status,        len(result),    )def _delegate(    run_context: RunContext,    *,    sub_agent_name: str,    sub_agent: Agent,    task: str,) -> dict[str, Any]:    """Common delegation flow: append running entry → invoke → update final."""    entry_id = _append_delegation(        run_context,        sub_agent=sub_agent_name,        task=task,        status="running",        result="",    )    try:        result = _invoke_sub_agent(sub_agent, task)    except Exception as exc:  # noqa: BLE001 — sub-agent transport can fail anywhere        logger.exception("subagents: sub-agent %s failed", sub_agent_name)        # Surface only the exception class to the supervisor / frontend —        # provider error strings can carry URLs / request IDs / partial        # credentials. The full traceback stays in server logs.        message = (            f"sub-agent call failed: {exc.__class__.__name__} "            "(see server logs for details)"        )        _update_delegation(            run_context, entry_id=entry_id, status="failed", result=message        )        return {"status": "failed", "error": message}    _update_delegation(        run_context, entry_id=entry_id, status="completed", result=result    )    return {"status": "completed", "result": result}# ---------------------------------------------------------------------------# Supervisor tools (each tool delegates to one sub-agent)# ---------------------------------------------------------------------------# Each function is a tool exposed to the supervisor agent. The supervisor# LLM "calls" these to delegate work; each call synchronously runs the# matching sub-agent, records the delegation into shared state, and# returns the sub-agent's output as the tool result the supervisor reads# on its next step.def research_agent(run_context: RunContext, task: str) -> dict[str, Any]:    """Delegate a research task to the research sub-agent.    Use for: gathering facts, background, definitions, statistics.    Returns {status, result} on success or {status: "failed", error} on    sub-agent failure.    """    return _delegate(        run_context,        sub_agent_name="research_agent",        sub_agent=_research_agent,        task=task,    )def writing_agent(run_context: RunContext, task: str) -> dict[str, Any]:    """Delegate a drafting task to the writing sub-agent.    Use for: producing a polished paragraph, draft, or summary. Pass    relevant facts from prior research inside `task`. Same return shape    as research_agent.    """    return _delegate(        run_context,        sub_agent_name="writing_agent",        sub_agent=_writing_agent,        task=task,    )def critique_agent(run_context: RunContext, task: str) -> dict[str, Any]:    """Delegate a critique task to the critique sub-agent.    Use for: reviewing a draft and suggesting concrete improvements.    Same return shape as research_agent.    """    return _delegate(        run_context,        sub_agent_name="critique_agent",        sub_agent=_critique_agent,        task=task,    )

This is where CopilotKit's shared-state channel earns its keep: the supervisor's tool calls mutate delegations as they happen, and the frontend renders every new entry live.

Rendering a live delegation log#

On the frontend, the delegation log is just a reactive render of the delegations slot. Subscribe with useAgent({ updates: [UseAgentUpdate.OnStateChanged, UseAgentUpdate.OnRunStatusChanged] }), read agent.state.delegations, and render one card per entry.

delegation-log.tsx

/** * Live delegation log — renders the `delegations` slot of agent state. * * Each entry corresponds to one invocation of a sub-agent. The list grows * in real time as the supervisor fans work out to its children: the agno * supervisor's tools push entries with status `"running"` before invoking * the sub-agent, then flip them to `"completed"` or `"failed"` once the * sub-agent returns. */export function DelegationLog({ delegations, isRunning }: DelegationLogProps) {  return (    <div      data-testid="delegation-log"      className="h-full flex flex-col rounded-2xl bg-white border border-[#DBDBE5] shadow-sm"    >      <header className="px-5 py-3 border-b border-[#E9E9EF] flex items-center justify-between">        <div>          <h2 className="text-sm font-semibold text-[#010507]">            Sub-agent delegations          </h2>          <p className="text-xs text-[#838389] mt-0.5">            Each entry shows a sub-agent invocation by the supervisor.          </p>        </div>        <div className="flex items-center gap-2">          {isRunning && (            <span              data-testid="supervisor-running"              className="inline-flex items-center gap-1.5 px-2 py-0.5 rounded-full border border-amber-200 bg-amber-50 text-amber-700 text-[10px] font-semibold uppercase tracking-[0.12em]"            >              <span className="w-1.5 h-1.5 rounded-full bg-amber-600 animate-pulse" />              Supervisor running            </span>          )}          <span            data-testid="delegation-count"            className="text-xs font-mono text-[#838389]"          >            {delegations.length} calls          </span>        </div>      </header>      <div className="flex-1 overflow-y-auto p-4 space-y-3">        {delegations.length === 0 ? (          <p            data-testid="delegation-empty"            className="text-sm text-[#838389] italic"          >            No delegations yet. Ask the supervisor to plan a deliverable.          </p>        ) : (          delegations.map((d, idx) => (            <article              key={d.id}              data-testid="delegation-entry"              data-status={d.status}              className={`rounded-xl border p-3 ${ENTRY_BORDER_BY_STATUS[d.status]}`}            >              <div className="flex items-center gap-2 mb-2">                <span className="text-xs font-mono text-[#AFAFB7]">                  #{idx + 1}                </span>                <span                  className={`text-[11px] font-medium px-2 py-0.5 rounded-full border ${SUB_AGENT_BADGE_CLASS[d.sub_agent]}`}                >                  {SUB_AGENT_LABEL[d.sub_agent]}                </span>                <span                  className={`text-[11px] font-medium px-2 py-0.5 rounded-full border ${STATUS_BADGE_CLASS[d.status]}`}                >                  {STATUS_LABEL[d.status]}                </span>              </div>              <div className="text-xs text-[#57575B] mb-2">                <strong className="text-[#010507]">Task:</strong> {d.task}              </div>              {d.status === "running" ? (                <div className="flex items-center gap-2 text-xs text-amber-700">                  <span                    className="inline-block w-3 h-3 rounded-full border-2 border-amber-500 border-t-transparent animate-spin"                    aria-hidden                  />                  Sub-agent is working…                </div>              ) : (                <div                  className={`text-xs whitespace-pre-wrap ${                    d.status === "failed" ? "text-red-700" : "text-[#010507]"                  }`}                >                  {d.result}                </div>              )}            </article>          ))        )}      </div>    </div>  );}

The result: as the supervisor fans work out to its sub-agents, the log grows in real time, giving the user visibility into a process that would otherwise be a long opaque spinner.

Shared State — the channel that makes the delegation log live.
State streaming — stream individual sub-agent outputs token-by-token inside each log entry.