ag2-subagent-delegation

name	ag2-subagent-delegation
description	Single-agent recursion and parallel fan-out within one AG2 `Agent` — auto-injected `run_subtask` / `run_subtasks(parallel=True)` (opt in via `tasks=TaskConfig(...)`) for self-delegation, and `Agent.as_tool()` as a lightweight no-hub way to call one named agent from inside another. Use when a single coordinator wants to break work into its own sub-tasks, fan out concurrent sub-tasks, or invoke a specialist agent as a tool. Covers context flow, recursion safety, and `persistent_stream` for sub-task history. **For two or more agents actually collaborating with a registry, durable channels, governance, or turn-taking, use `ag2-network-quickstart` instead** — the network is the standard multi-agent pattern in AG2.
license	Apache-2.0

Subagent delegation

When to use

• "Coordinator + specialists" — a parent agent should hand parts of a task to a research agent, math agent, etc.
• "Fan out then collect" — multi-part questions where each part is independent and parallel execution saves wall time.
• "Self-delegation" — one agent breaks complex work into focused sub-tasks for itself.

Two patterns

Pattern	Reach for it when	API
Auto-injected run_subtask / run_subtasks	Lightweight self-delegation, dynamic fan-out, parallel sub-questions	tasks=TaskConfig(...) on the parent
Agent.as_tool()	Distinct named delegates the LLM should reason about ("call the researcher", "call the writer")	Wrap a child Agent as a tool on the parent

The two compose — a coordinator can have both.

Pattern 1 — auto-injected run_subtasks

Subtask tools are off by default (tasks=False). Opt in with tasks=TaskConfig(...) and the agent gains:

• run_subtask(task: str) — one isolated sub-task agent.
• run_subtasks(tasks: list[str], parallel: bool = True) — fan out multiple in one tool call (default concurrent).

from ag2 import Agent, TaskConfig
from ag2.config import GeminiConfig

config = GeminiConfig(model="gemini-3-flash-preview")

coordinator = Agent(
    "coordinator",
    prompt=(
        "You answer multi-part questions by dispatching run_subtasks "
        "with parallel=True. Use one tool call with every sub-question "
        "packed into the 'tasks' list."
    ),
    config=config,
    tasks=TaskConfig(),  # opt in
)

reply = await coordinator.ask(
    "In one run_subtasks call, answer: "
    "(a) tallest waterfall, (b) Eiffel Tower year, (c) boiling point of nitrogen."
)

TaskConfig controls how the sub-task agents are built:

@dataclass
class TaskConfig:
    config: ModelConfig | None = None    # falls back to parent's config
    prompt: str = "You are a task agent..."
    include_tools: Iterable[str] | None = None   # None = inherit all parent tools
    exclude_tools: Iterable[str] = ()
    extra_tools: Iterable[Callable | Tool] = ()

Common shape — cheaper model for sub-tasks, narrow tool surface:

TaskConfig(
    config=worker_config,                  # smaller model
    prompt="You are a focused worker; one step only.",
    include_tools=["search", "fetch_url"], # don't expose `summarize` to children
)

Sub-task agents are built with tasks=False — they never gain run_subtask tools themselves. Recursive delegation is structurally impossible; no depth limit needed.

Pattern 2 — Agent.as_tool()

Expose a whole agent as a tool the LLM can name and call:

from ag2 import Agent
from ag2.config import AnthropicConfig

config = AnthropicConfig(model="claude-sonnet-4-6")

researcher = Agent("researcher", prompt="Provide concise factual findings.", config=config, tools=[search_tool])
writer     = Agent("writer", prompt="Turn research into clear prose.", config=config)

coordinator = Agent(
    "coordinator",
    prompt="First delegate research, then pass findings to the writer.",
    config=config,
    tools=[
        researcher.as_tool(description="Research a topic and return findings."),
        writer.as_tool(description="Write an article. Pass research notes in the context parameter."),
    ],
)

The coordinator's LLM sees task_researcher and task_writer. Each call has two parameters:

• objective (required) — what the sub-task should do.
• context (optional) — relevant info the parent wants to share.

as_tool() accepts:

Parameter	Description
description	Tool description shown to the LLM (required)
name	Override the default task_{agent.name}
stream	StreamFactory for custom sub-task streams (see below)
middleware	ToolMiddleware callables (e.g. approval_required)

For more control, use subagent_tool() directly:

from ag2.tools.subagents import subagent_tool

coordinator = Agent("coordinator", config=config, tools=[
    subagent_tool(researcher, description="Research a topic."),
])

Self-delegation via as_tool()

If you want a named self-delegate (sub_task instead of generic run_subtask), give an agent its own tool:

analyst = Agent(
    "analyst",
    prompt=(
        "You have search and sub_task tools. "
        "Only use sub_task when the task has clearly independent parts."
    ),
    config=config,
    tools=[search_tool],
)

analyst.add_tool(
    analyst.as_tool(
        description="Break work into a focused sub-task for independent analysis.",
        name="sub_task",
    )
)

Recursion safety

Self-delegation via as_tool() can recurse — the child has the same sub_task tool, so without a guard the LLM may chain calls indefinitely.

The simplest safe pattern is to prefer the auto-injected run_subtask / run_subtasks path for self-delegation. Sub-tasks spawned that way are constructed with tasks=False, so they have no run_subtask tools and recursion is structurally impossible.

If you genuinely need recursive as_tool() self-delegation, write a tool middleware that increments a depth counter in context.dependencies and short-circuits past a threshold. The subagents module exports subagent_tool, background_agent_tool, persistent_stream, and StreamFactory from ag2.tools.subagents — verify the current public surface there before relying on a built-in depth-limiting helper.

Sub-task streams

By default, each sub-task gets a fresh MemoryStream — its history is isolated and starts empty. Context flow:

What	Behaviour	Why
Dependencies	Copied (top-level shallow)	Isolated; treat dependencies as read-only inside subtasks
Variables	Copied; not synced back to the parent	Concurrent-safe — with siblings running via asyncio.gather, last-writer-wins would silently clobber values, so child mutations stay scoped to the child by design
History	Fresh stream	Clean context; relevant info passes via the context tool parameter
Tools	Inherited from parent (filtered by TaskConfig)	Sub-tasks need real capabilities to do work

persistent_stream()

When a sub-agent benefits from seeing its prior calls (e.g. avoid repeating searches), give it a stream that persists across invocations within the parent context:

from ag2.tools.subagents import persistent_stream

researcher.as_tool(
    description="Research a topic",
    stream=persistent_stream(),
)

Stores stream id in context.dependencies keyed by f"ag:{agent.name}:stream" and reuses the parent stream's storage backend.

Custom factory

from ag2 import Agent, Context
from ag2.streams.redis import RedisStream

def make_redis_stream(agent: Agent, ctx: Context) -> RedisStream:
    return RedisStream(MY_REDIS_URL, prefix=f"ag2:sub:{agent.name}")

researcher.as_tool(description="Research a topic", stream=make_redis_stream)

Going deeper

• Working starter: assets/research_squad.py (mirrors code_examples/05) — covers both run_subtasks(parallel=True) and Agent.as_tool(), with TaskStarted / TaskCompleted lifecycle events.
• Full reference: website/docs/user-guide/subagents.mdx.
• tasks= constructor knob (with KnowledgeConfig, etc.): website/docs/user-guide/agent_harness.mdx.

Common pitfalls

• Forgetting to opt in — tasks=False is the default. No TaskConfig, no run_subtask tools.
• Expecting sub-tasks to recurse with run_subtask — they can't. Sub-tasks themselves have tasks=False. If you need deeper trees, use Agent.as_tool() self-delegation with a manual depth-counter middleware (see "Recursion safety" above).
• Sharing mutable variables expecting them to merge — each sub-task copies the parent's variables, and mutations are never synced back to the parent (not even on success). Sibling mutations don't propagate either. Pass any result you need back through the sub-task's return value, not via shared variables.
• Treating dependencies as scoped per sub-task — only the top-level dict is copied. Mutable values inside it are still shared by reference. Treat dependencies as read-only inside sub-tasks.
• No description= on as_tool() — the LLM doesn't know when to call it. Required parameter.
• run_subtasks(parallel=False) when work is concurrent — defaults to True for a reason; only set False when later tasks depend on earlier results.
• Confusing task_{agent.name} collisions — pass name= to override if you want shorter names or distinct delegates of the same agent.