Sub-Agents

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

"""LangGraph agent backing the Sub-Agents demo.Demonstrates multi-agent delegation with a visible delegation log.A top-level "supervisor" LLM orchestrates three specialized sub-agents,exposed as tools:  - `research_agent`  — gathers facts  - `writing_agent`   — drafts prose  - `critique_agent`  — reviews draftsEach sub-agent is a full `create_agent(...)` under the hood. Everydelegation appends an entry to the `delegations` slot in shared agentstate so the UI can render a live "delegation log" as the supervisorfans work out and collects results. This is the canonical LangGraphsub-agents-as-tools pattern, adapted to surface delegation events tothe frontend via CopilotKit's shared-state channel.This is the FastAPI variant — the graph is exported and registered in`langgraph.json`. Identical agent topology to the langgraph-pythonreference; only the server framework differs."""import uuidfrom operator import addfrom typing import Annotated, Literal, TypedDictfrom langchain.agents import AgentState as BaseAgentState, create_agentfrom langchain.tools import ToolRuntime, toolfrom langchain_core.messages import HumanMessage, ToolMessagefrom langchain_openai import ChatOpenAIfrom langgraph.types import Commandfrom copilotkit import CopilotKitMiddleware# ---------------------------------------------------------------------------# Shared state# ---------------------------------------------------------------------------class Delegation(TypedDict):    id: str    sub_agent: Literal["research_agent", "writing_agent", "critique_agent"]    task: str    status: Literal["running", "completed", "failed"]    result: strclass AgentState(BaseAgentState):    """Shared state. `delegations` is rendered as a live log in the UI.    `delegations` uses `operator.add` as its channel reducer so concurrent    tool calls within a single supervisor turn each contribute their own    entry. Without a reducer, parallel `tool_calls` would each read the    same snapshot and the channel would last-write-wins, silently dropping    every delegation but one from the UI log.    """    delegations: Annotated[list[Delegation], add]# ---------------------------------------------------------------------------# Sub-agents (real LLM agents under the hood)# ---------------------------------------------------------------------------# Each sub-agent is a full-fledged `create_agent(...)` with its own# system prompt. They don't share memory or tools with the supervisor —# the supervisor only sees their return value._sub_model = ChatOpenAI(model="gpt-4o-mini")_research_agent = create_agent(    model=_sub_model,    tools=[],    system_prompt=(        "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 = create_agent(    model=_sub_model,    tools=[],    system_prompt=(        "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 = create_agent(    model=_sub_model,    tools=[],    system_prompt=(        "You are an editorial critique sub-agent. Given a draft, give "        "2-3 crisp, actionable critiques. No preamble."    ),)def _invoke_sub_agent(agent, task: str) -> str:    """Run a sub-agent on `task` and return its final message content."""    result = agent.invoke({"messages": [HumanMessage(content=task)]})    messages = result.get("messages", [])    if not messages:        return ""    return str(messages[-1].content)def _delegation_command(    sub_agent: str,    task: str,    status: Literal["completed", "failed"],    result: str,    tool_call_id: str,) -> Command:    """Build a Command that appends a single new delegation entry.    Emits ONLY the new entry under `delegations`. The channel reducer    (`operator.add` on `AgentState.delegations`) extends the existing    list, so parallel tool calls within one supervisor turn each    contribute their own entry instead of clobbering each other via a    last-write-wins read-modify-write.    """    entry: Delegation = {        "id": str(uuid.uuid4()),        "sub_agent": sub_agent,  # type: ignore[typeddict-item]        "task": task,        "status": status,        "result": result,    }    return Command(        update={            "delegations": [entry],            "messages": [                ToolMessage(                    content=result,                    tool_call_id=tool_call_id,                )            ],        }    )def _delegate(    sub_agent_name: str,    agent,    task: str,    tool_call_id: str,) -> Command:    """Invoke a sub-agent and turn the outcome into a Command.    Wrapped in try/except so that a sub-agent LLM failure (rate limit,    transport error, missing API key, etc.) is recorded as a `failed`    delegation entry and surfaced to the supervisor as a ToolMessage,    instead of propagating and crashing the supervisor turn. The    user-facing `result` is scrubbed to the exception class name only;    full details are captured server-side via the standard logging path    when the exception is re-raised at the caller's discretion (we do    not re-raise here — recovery is the supervisor's job).    """    try:        result = _invoke_sub_agent(agent, task)        return _delegation_command(            sub_agent_name, task, "completed", result, tool_call_id        )    except Exception as exc:  # noqa: BLE001 - intentional broad catch        # Keep the message generic; class name only, no exception args        # (which can contain prompts, keys, or other sensitive data).        message = (            f"sub-agent call failed: {exc.__class__.__name__} "            f"(see server logs for details)"        )        return _delegation_command(            sub_agent_name, task, "failed", message, tool_call_id        )# ---------------------------------------------------------------------------# Supervisor tools (each tool delegates to one sub-agent)# ---------------------------------------------------------------------------# Each @tool wraps a sub-agent invocation. The supervisor LLM "calls"# these tools 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 a ToolMessage the supervisor can# read on its next step.@tooldef research_agent(task: str, runtime: ToolRuntime) -> Command:    """Delegate a research task to the research sub-agent.    Use for: gathering facts, background, definitions, statistics.    Returns a bulleted list of key facts.    """    return _delegate("research_agent", _research_agent, task, runtime.tool_call_id)@tooldef writing_agent(task: str, runtime: ToolRuntime) -> Command:    """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`.    """    return _delegate("writing_agent", _writing_agent, task, runtime.tool_call_id)@tooldef critique_agent(task: str, runtime: ToolRuntime) -> Command:    """Delegate a critique task to the critique sub-agent.    Use for: reviewing a draft and suggesting concrete improvements.    """    return _delegate("critique_agent", _critique_agent, task, runtime.tool_call_id)# ---------------------------------------------------------------------------# Supervisor (the graph we export)# ---------------------------------------------------------------------------graph = create_agent(    model=ChatOpenAI(model="gpt-4o-mini"),    tools=[research_agent, writing_agent, critique_agent],    middleware=[CopilotKitMiddleware()],    state_schema=AgentState,    system_prompt=(        "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. 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."    ),)

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

import uuidfrom operator import addfrom typing import Annotated, Literal, TypedDictfrom langchain.agents import AgentState as BaseAgentState, create_agentfrom langchain.tools import ToolRuntime, toolfrom langchain_core.messages import HumanMessage, ToolMessagefrom langchain_openai import ChatOpenAIfrom langgraph.types import Commandfrom copilotkit import CopilotKitMiddleware# ---------------------------------------------------------------------------# Shared state# ---------------------------------------------------------------------------class Delegation(TypedDict):    id: str    sub_agent: Literal["research_agent", "writing_agent", "critique_agent"]    task: str    status: Literal["running", "completed", "failed"]    result: strclass AgentState(BaseAgentState):    """Shared state. `delegations` is rendered as a live log in the UI.    `delegations` uses `operator.add` as its channel reducer so concurrent    tool calls within a single supervisor turn each contribute their own    entry. Without a reducer, parallel `tool_calls` would each read the    same snapshot and the channel would last-write-wins, silently dropping    every delegation but one from the UI log.    """    delegations: Annotated[list[Delegation], add]# ---------------------------------------------------------------------------# Sub-agents (real LLM agents under the hood)# ---------------------------------------------------------------------------# Each sub-agent is a full-fledged `create_agent(...)` with its own# system prompt. They don't share memory or tools with the supervisor —# the supervisor only sees their return value._sub_model = ChatOpenAI(model="gpt-4o-mini")_research_agent = create_agent(    model=_sub_model,    tools=[],    system_prompt=(        "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 = create_agent(    model=_sub_model,    tools=[],    system_prompt=(        "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 = create_agent(    model=_sub_model,    tools=[],    system_prompt=(        "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

import uuidfrom operator import addfrom typing import Annotated, Literal, TypedDictfrom langchain.agents import AgentState as BaseAgentState, create_agentfrom langchain.tools import ToolRuntime, toolfrom langchain_core.messages import HumanMessage, ToolMessagefrom langchain_openai import ChatOpenAIfrom langgraph.types import Commandfrom copilotkit import CopilotKitMiddleware# ---------------------------------------------------------------------------# Shared state# ---------------------------------------------------------------------------class Delegation(TypedDict):    id: str    sub_agent: Literal["research_agent", "writing_agent", "critique_agent"]    task: str    status: Literal["running", "completed", "failed"]    result: strclass AgentState(BaseAgentState):    """Shared state. `delegations` is rendered as a live log in the UI.    `delegations` uses `operator.add` as its channel reducer so concurrent    tool calls within a single supervisor turn each contribute their own    entry. Without a reducer, parallel `tool_calls` would each read the    same snapshot and the channel would last-write-wins, silently dropping    every delegation but one from the UI log.    """    delegations: Annotated[list[Delegation], add]# ---------------------------------------------------------------------------# Sub-agents (real LLM agents under the hood)# ---------------------------------------------------------------------------# Each sub-agent is a full-fledged `create_agent(...)` with its own# system prompt. They don't share memory or tools with the supervisor —# the supervisor only sees their return value._sub_model = ChatOpenAI(model="gpt-4o-mini")_research_agent = create_agent(    model=_sub_model,    tools=[],    system_prompt=(        "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 = create_agent(    model=_sub_model,    tools=[],    system_prompt=(        "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 = create_agent(    model=_sub_model,    tools=[],    system_prompt=(        "You are an editorial critique sub-agent. Given a draft, give "        "2-3 crisp, actionable critiques. No preamble."    ),)def _invoke_sub_agent(agent, task: str) -> str:    """Run a sub-agent on `task` and return its final message content."""    result = agent.invoke({"messages": [HumanMessage(content=task)]})    messages = result.get("messages", [])    if not messages:        return ""    return str(messages[-1].content)def _delegation_command(    sub_agent: str,    task: str,    status: Literal["completed", "failed"],    result: str,    tool_call_id: str,) -> Command:    """Build a Command that appends a single new delegation entry.    Emits ONLY the new entry under `delegations`. The channel reducer    (`operator.add` on `AgentState.delegations`) extends the existing    list, so parallel tool calls within one supervisor turn each    contribute their own entry instead of clobbering each other via a    last-write-wins read-modify-write.    """    entry: Delegation = {        "id": str(uuid.uuid4()),        "sub_agent": sub_agent,  # type: ignore[typeddict-item]        "task": task,        "status": status,        "result": result,    }    return Command(        update={            "delegations": [entry],            "messages": [                ToolMessage(                    content=result,                    tool_call_id=tool_call_id,                )            ],        }    )def _delegate(    sub_agent_name: str,    agent,    task: str,    tool_call_id: str,) -> Command:    """Invoke a sub-agent and turn the outcome into a Command.    Wrapped in try/except so that a sub-agent LLM failure (rate limit,    transport error, missing API key, etc.) is recorded as a `failed`    delegation entry and surfaced to the supervisor as a ToolMessage,    instead of propagating and crashing the supervisor turn. The    user-facing `result` is scrubbed to the exception class name only;    full details are captured server-side via the standard logging path    when the exception is re-raised at the caller's discretion (we do    not re-raise here — recovery is the supervisor's job).    """    try:        result = _invoke_sub_agent(agent, task)        return _delegation_command(            sub_agent_name, task, "completed", result, tool_call_id        )    except Exception as exc:  # noqa: BLE001 - intentional broad catch        # Keep the message generic; class name only, no exception args        # (which can contain prompts, keys, or other sensitive data).        message = (            f"sub-agent call failed: {exc.__class__.__name__} "            f"(see server logs for details)"        )        return _delegation_command(            sub_agent_name, task, "failed", message, tool_call_id        )# ---------------------------------------------------------------------------# Supervisor tools (each tool delegates to one sub-agent)# ---------------------------------------------------------------------------# Each @tool wraps a sub-agent invocation. The supervisor LLM "calls"# these tools 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 a ToolMessage the supervisor can# read on its next step.@tooldef research_agent(task: str, runtime: ToolRuntime) -> Command:    """Delegate a research task to the research sub-agent.    Use for: gathering facts, background, definitions, statistics.    Returns a bulleted list of key facts.    """    return _delegate("research_agent", _research_agent, task, runtime.tool_call_id)@tooldef writing_agent(task: str, runtime: ToolRuntime) -> Command:    """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`.    """    return _delegate("writing_agent", _writing_agent, task, runtime.tool_call_id)@tooldef critique_agent(task: str, runtime: ToolRuntime) -> Command:    """Delegate a critique task to the critique sub-agent.    Use for: reviewing a draft and suggesting concrete improvements.    """    return _delegate("critique_agent", _critique_agent, task, runtime.tool_call_id)

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 parent header shows how many sub-agents have been called and * whether the supervisor is still running. */export function DelegationLog({ delegations, isRunning }: DelegationLogProps) {  return (    <div      data-testid="delegation-log"      className="w-full h-full flex flex-col bg-white rounded-2xl shadow-sm border border-[#DBDBE5] overflow-hidden"    >      <div className="flex items-center justify-between px-6 py-3 border-b border-[#E9E9EF] bg-[#FAFAFC]">        <div className="flex items-center gap-3">          <span className="text-lg font-semibold text-[#010507]">            Sub-agent delegations          </span>          {isRunning && (            <span              data-testid="supervisor-running"              className="inline-flex items-center gap-1.5 px-2 py-0.5 rounded-full border border-[#BEC2FF] bg-[#BEC2FF1A] text-[#010507] text-[10px] font-semibold uppercase tracking-[0.12em]"            >              <span className="w-1.5 h-1.5 rounded-full bg-[#010507] animate-pulse" />              Supervisor running            </span>          )}        </div>        <span          data-testid="delegation-count"          className="text-xs font-mono text-[#838389]"        >          {delegations.length} calls        </span>      </div>      <div className="flex-1 overflow-y-auto p-4 space-y-3">        {delegations.length === 0 ? (          <p className="text-[#838389] italic text-sm">            Ask the supervisor to complete a task. Every sub-agent it calls will            appear here.          </p>        ) : (          delegations.map((d, idx) => {            const style = SUB_AGENT_STYLE[d.sub_agent];            return (              <div                key={d.id}                data-testid="delegation-entry"                className="border border-[#E9E9EF] rounded-xl p-3 bg-[#FAFAFC]"              >                <div className="flex items-center justify-between mb-2">                  <div className="flex items-center gap-2">                    <span className="text-xs font-mono text-[#AFAFB7]">                      #{idx + 1}                    </span>                    <span                      className={`inline-flex items-center gap-1 px-2 py-0.5 rounded-full text-[10px] font-semibold uppercase tracking-[0.1em] border ${style.color}`}                    >                      <span>{style.emoji}</span>                      <span>{style.label}</span>                    </span>                  </div>                  <span className="text-[10px] uppercase tracking-[0.12em] font-semibold text-[#189370]">                    {d.status}                  </span>                </div>                <div className="text-xs text-[#57575B] mb-2">                  <span className="font-semibold text-[#010507]">Task: </span>                  {d.task}                </div>                <div className="text-sm text-[#010507] whitespace-pre-wrap bg-white rounded-lg p-2.5 border border-[#E9E9EF]">                  {d.result}                </div>              </div>            );          })        )}      </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.