Subagents

AG-UI carries subagent activity as first-class events. When an agent delegates, the child run arrives as a SUBAGENT_STARTED event plus message events stamped with a subagentRunId, and the adapter keeps that traffic out of the parent transcript: it becomes a card of its own, anchored to the tool call that spawned it. The running example is a trip planner whose orchestrator hands work to three specialists, and this guide walks the four files that make it work.

What the demo does

The Run tab shows the prebuilt <chat> composition in front of an orchestrator agent served over AG-UI. There are no starter suggestions here, so type a trip request. Give it the trip details up front, for example "Plan a trip from LAX to JFK, one adult, economy, departing next Tuesday morning and returning Friday evening", because a bare origin-and-destination request usually earns a clarifying question about dates before it delegates anything. If the request is ambiguous, for example it does not mention airports, the orchestrator asks a clarifying question before delegating.

Once it has both airports, the prompt tells it to call the task tool three times in a fixed order: research, then booking, then itinerary. Each dispatch appears inline in the transcript as a card carrying that specialist's own answer, streamed token by token inside the card rather than into the parent bubble. The card header shows the role name, the identifier of the tool call it belongs to, a status badge, and a message count.

While a specialist works, its card is expanded and the badge reads running. When the child finishes the badge flips to complete and the card collapses, and clicking the header opens it again. After the third card, the orchestrator writes its own summary of the plan as an ordinary assistant message.

How it is built

Four files carry the feature in the Code tab: a LangGraph graph with a single delegation tool, a FastAPI server that mounts the agent, an application config that registers it, and a component that mounts the chat. A fifth file next to the server, streaming/subagent_emitting_agent.py, holds the translation from the graph's private events into the protocol's subagent events.

Running a specialist as a child LLM

A subagent here is a plain model call with its own system prompt. What makes it visible is the callback handler passed alongside it, which taps the child's tokens as they arrive and carries the parent tool call identifier with them.

async def _run_subagent(
    role: str,
    task_description: str,
    system_prompt: str,
    tool_call_id: str,
) -> str:
    """Run a single subagent LLM, streaming its tokens through
    SubagentStreamHandler so they surface as `subagent_activity`
    `message_start` / `message` (per-token delta) events keyed by the parent
    tool_call_id."""
    llm = ChatOpenAI(model="gpt-5-mini", streaming=True)
    messages = [
        SystemMessage(content=system_prompt),
        HumanMessage(content=task_description),
    ]
    response = await llm.ainvoke(
        messages,
        config={"callbacks": [SubagentStreamHandler(tool_call_id)]},
    )
    content = response.content
    if isinstance(content, str):
        return content
    if isinstance(content, list):
        parts = [
            b.get("text", "")
            for b in content
            if isinstance(b, dict) and b.get("type") == "text"
        ]
        return "\n".join(parts)
    return ""

SubagentStreamHandler is an async LangChain callback handler: on the first token it dispatches a message_start custom event, and then one message event per token, each keyed by that tool call identifier.

The delegation tool

The orchestrator binds exactly one tool. InjectedToolCallId hands the tool body the identifier of its own call, and that identifier becomes the identity of the whole delegation: the events the tool dispatches carry it, and so does every event the handler emits. The body announces started before the child runs, finished after it returns, and error if it raises.

@tool
async def task(
    role: Literal["research", "booking", "itinerary"],
    task_description: str,
    tool_call_id: Annotated[str, InjectedToolCallId],
) -> ToolMessage:
    """Delegate a subtask to a specialized subagent.
 
    Roles:
      - research: gathers destination intel (airports, weather, conditions)
      - booking: finds flight options between origin and destination
      - itinerary: synthesizes a final trip plan combining research + bookings
 
    Args:
        role: One of "research", "booking", "itinerary".
        task_description: Plain-English description of what the subagent
            should do (e.g., "Gather info on LAX and JFK airports", or
            "Find morning flights from LAX to JFK").
 
    Returns:
        A stable tool message containing the subagent's final answer.
 
    The subagent run is surfaced to the UI as the protocol's standard
    subagent events: SUBAGENT_STARTED → attributed TEXT_MESSAGE_* per token →
    SUBAGENT_FINISHED (or SUBAGENT_ERROR), with ids derived from this tool's
    own call id (`<tool_call_id>-sub`).
    """
 
    async def _emit(payload: dict) -> None:
        try:
            await adispatch_custom_event(
                "subagent_activity", {"subagent_id": tool_call_id, **payload}
            )
        except Exception:
            pass
 
    prompts = {
        "research": _RESEARCH_PROMPT,
        "booking": _BOOKING_PROMPT,
        "itinerary": _ITINERARY_PROMPT,
    }
    system_prompt = prompts.get(role)
    if system_prompt is None:
        return f"Unknown role: {role}"
 
    await _emit({"phase": "started", "name": role})
    try:
        result = await _run_subagent(role, task_description, system_prompt, tool_call_id)
    except Exception as exc:
        # Surface the failure on the subagent card (SUBAGENT_ERROR), then
        # re-raise so the bridge's own tool-error path still runs.
        await _emit({"phase": "error", "message": f"{type(exc).__name__}: {exc}"})
        raise
    await _emit({"phase": "finished", "status": "complete"})
    return ToolMessage(
        id=tool_call_id,
        tool_call_id=tool_call_id,
        name="task",
        content=result,
    )

The docstring is the model-facing contract, which is why the three roles are described there rather than only in the system prompt.

The orchestrator loop

The graph itself is the standard shape: a model node that may emit tool calls, a ToolNode that runs them, and an edge back to the model so the orchestrator can read one specialist's result before dispatching the next.

def build_subagents_graph():
    """Orchestrator LLM with a single `task` tool that dispatches subagents."""
    llm = ChatOpenAI(model="gpt-5-mini", streaming=True).bind_tools([task])
 
    async def orchestrator(state: MessagesState) -> dict:
        system_prompt = (PROMPTS_DIR / "subagents.md").read_text()
        messages = [SystemMessage(content=system_prompt)] + state["messages"]
        response = await llm.ainvoke(messages)
        return {"messages": [response]}
 
    def should_continue(state: MessagesState) -> str:
        last = state["messages"][-1]
        if hasattr(last, "tool_calls") and last.tool_calls:
            return "tools"
        return "generate_title"
 
    graph = StateGraph(MessagesState)
    graph.add_node("orchestrator", orchestrator)
    graph.add_node("tools", ToolNode([task]))
    graph.add_node("generate_title", generate_title)
    graph.set_entry_point("orchestrator")
    graph.add_conditional_edges(
        "orchestrator",
        should_continue,
        {"tools": "tools", "generate_title": "generate_title"},
    )
    graph.add_edge("tools", "orchestrator")
    graph.add_edge("generate_title", END)
    return graph.compile(checkpointer=MemorySaver())

The third node, generate_title, is unrelated to delegation; it summarizes the first user message into a thread title and swallows its own errors.

Serving the graph over AG-UI

The backend is a FastAPI application. add_langgraph_fastapi_endpoint from the ag-ui-langgraph package mounts an agent at a path that speaks the AG-UI event stream. The agent mounted here is a subclass of the package's LangGraphAgent rather than the class itself.

from fastapi import FastAPI
from ag_ui_langgraph import add_langgraph_fastapi_endpoint
from .graph import graph
from .streaming.subagent_emitting_agent import SubagentEmittingAgent
 
# SubagentEmittingAgent subclasses the ag-ui-langgraph bridge and wraps its
# run() generator to expand the graph's `subagent_activity` CUSTOM events into
# the protocol's standard SUBAGENT_STARTED / TEXT_MESSAGE_* (attributed via
# subagentRunId) / SUBAGENT_FINISHED / SUBAGENT_ERROR events, so the chat
# composition renders a live subagent card.
agent = SubagentEmittingAgent(name="subagents", graph=graph)
app = FastAPI(title="cockpit-ag-ui-subagents")
add_langgraph_fastapi_endpoint(app, agent, path="/agent")
 
 
@app.get("/ok")
def ok() -> dict:
    return {"ok": True}

The subclass wraps the bridge's run() generator, which is the async generator the endpoint consumes, and rewrites the stream as it passes through.

From custom events to subagent events

The graph cannot reach the wire directly. Its subagent_activity events travel out as AG-UI CUSTOM events, and the subclass consumes each one and yields typed protocol events in its place. Every other event passes through untouched. This translation lives in streaming/subagent_emitting_agent.py, next to the server rather than in the Code tab. Writing tid for the task call identifier, the expansion is:

One phase can therefore produce two events, which is why the seam is the run() generator rather than a one-event-in, one-event-out hook. Unknown phases are logged and dropped; a message phase with no text is dropped silently.

The agent provider

provideAgent() registers the agent once for the whole application, and it is the only provider the <chat> composition requires. Nothing about it is subagent-specific. The example passes a factory because it resolves its endpoint at runtime from the host that serves the demo. Your own application does not need the factory.

import { injectCockpitRuntimeConnection } from '@threadplane/cockpit-telemetry';
import { ApplicationConfig } from '@angular/core';
import { provideAgent } from '@threadplane/ag-ui';
 
export const appConfig: ApplicationConfig = {
  providers: [
    provideAgent(() => {
      const connection = injectCockpitRuntimeConnection();
      if (connection.adapter !== 'ag-ui') {
        throw new Error('incompatible runtime');
      }
      return {
        url: connection.url,
      };
    }),
  ],
};

Pass the URL of your AG-UI endpoint directly:

provideAgent({
  url: 'https://your-backend.example.com/agent',
});

The component

The component is the whole frontend of this example: one field holding the injected agent, passed to <chat>. The layout wrapper around it is presentation chrome for the demo, and your own application supplies its own.

import { Component } from '@angular/core';
import { ChatComponent } from '@threadplane/chat';
import { injectAgent } from '@threadplane/ag-ui';
import { ExampleChatLayoutComponent } from '@threadplane/example-layouts';
 
/**
 * Subagents demo — research-delegation over the AG-UI transport.
 *
 * Retrieves the agent with injectAgent() (provided by provideAgent /
 * provideFakeAgent) and passes it to the prebuilt <chat> composition. No
 * subagent-specific wiring is needed in the component: when the orchestrator
 * dispatches a `task` tool call, the backend emits the protocol's standard
 * SUBAGENT_STARTED / TEXT_MESSAGE_* (attributed via subagentRunId) /
 * SUBAGENT_FINISHED events, the @threadplane/ag-ui reducer projects them onto
 * `agent.subagents()`, and <chat> renders each dispatch inline as a persistent
 * `chat-subagent-card`.
 *
 * Demonstrates the chat-runtime decoupling: same <chat> composition as the
 * LangGraph cockpit, AG-UI runtime instead of LangGraph.
 */
@Component({
  selector: 'app-subagents',
  standalone: true,
  imports: [ChatComponent, ExampleChatLayoutComponent],
  template: `
    <example-chat-layout>
      <chat main [agent]="agent" class="flex-1 min-w-0" />
    </example-chat-layout>
  `,
})
export class SubagentsComponent {
  protected readonly agent = injectAgent();
}

There is no subagent wiring in the component at all, because the composition reads the projection the adapter maintains.

How a card finds its tool call

The adapter does the attribution before any component sees it. SUBAGENT_STARTED creates an entry keyed by subagentRunId, recording the event's name and storing parentToolCallId as the entry's tool call identifier. Any TEXT_MESSAGE_START, TEXT_MESSAGE_CONTENT, TEXT_MESSAGE_END, TOOL_CALL_START, TOOL_CALL_ARGS, TOOL_CALL_END, or TOOL_CALL_RESULT event that carries a subagentRunId is routed into that entry instead of the parent message list. An attributed event that arrives before its SUBAGENT_STARTED creates the entry rather than being dropped, and the later SUBAGENT_STARTED fills in the identity.

Those entries are projected onto agent.subagents(), a Map of runtime-neutral Subagent wrappers. Each wrapper exposes toolCallId, name, and signals for status(), messages(), toolCalls(), and state(). SUBAGENT_FINISHED moves the status to complete, SUBAGENT_ERROR moves it to error, and a finished event whose outcome is suspended deliberately leaves the status at running, because a suspended run resumes under the same identifier.

The anchoring happens one level lower, in <chat-tool-calls>, which the chat composition renders for each message. It indexes agent.subagents() by the wrapper's toolCallId rather than by the map key, because adapters key their maps differently. When a tool call in the message matches a wrapper, that call renders as a <chat-subagent-card> instead of an ordinary tool-call card, and it never merges into a group with the calls next to it.

What crosses the wire

One delegation round, captured against this backend with the deltas and the bridge's own step and RAW events elided, looks like this:

TOOL_CALL_START       toolCallId: call_CN5…, toolCallName: task
TOOL_CALL_ARGS        toolCallId: call_CN5…, delta: …
TOOL_CALL_END         toolCallId: call_CN5…
SUBAGENT_STARTED      subagentRunId: call_CN5…-sub, name: research, parentToolCallId: call_CN5…
TEXT_MESSAGE_START    messageId: call_CN5…-sub-m1, subagentRunId: call_CN5…-sub
TEXT_MESSAGE_CONTENT  messageId: call_CN5…-sub-m1, delta: "L", subagentRunId: call_CN5…-sub
TEXT_MESSAGE_END      messageId: call_CN5…-sub-m1, subagentRunId: call_CN5…-sub
SUBAGENT_FINISHED     subagentRunId: call_CN5…-sub, outcome: { type: success }
TOOL_CALL_RESULT      toolCallId: call_CN5…, content: "LAX: Central Terminal Area …"

Two details are worth keeping. The tool call is fully announced — start, arguments, end — before the tool body runs, because TOOL_CALL_END marks the end of the argument stream rather than the end of execution; the card therefore always attaches to a call the adapter already knows about. And the delegation window nests between TOOL_CALL_END and TOOL_CALL_RESULT, so the specialist's text and the tool result that repeats it are separate things on the wire.

Cross-adapter parity

The consumer Angular code is the same; moving this component to a LangGraph deployment changes one import.

- import { injectAgent } from '@threadplane/ag-ui';
+ import { injectAgent } from '@threadplane/langgraph';

Both adapters project their own delegation traffic onto the same subagents() signal of runtime-neutral wrappers, and <chat> renders whichever it is given. What differs is how the delegation is identified: this example emits protocol subagent events explicitly from a bridge subclass, while the LangGraph adapter recognizes a dispatch tool by name and matches the child run's own namespace.

What's Next