Task Delegation - AG2
Task Delegation
Task delegation allows agents to delegate work to other agents through tool calling. The calling agent's LLM decides when and what to delegate, and each sub-task runs on its own isolated stream with independent history.
Why Use Subagents
Breaking work across multiple agents gives you:
- Separation of concerns — each agent has a focused prompt, tools, and config tuned for its role.
- Independent context — sub-tasks run on fresh streams, so history doesn't grow unboundedly.
- LLM-driven orchestration — the calling agent decides when to delegate, what context to pass, and how to use the result.
Note
When the LLM returns multiple tool calls in a single response, the framework dispatches them concurrently. Each concurrent sub-task gets its own copy of variables, so they don't interfere with each other.
Tip
For lightweight self-delegation where the parent doesn't need a named delegate, opt in to the auto-injected run_subtask / run_subtasks tools by passing tasks=TaskConfig(...) — see tasks= in The Agent Harness. Use Agent.as_tool() (below) when you want a distinct, purpose-named tool exposed to the LLM.
Subagents API
Use Agent.as_tool() to make one agent available as a tool for another.
<br> 1<br> 2<br> 3<br> 4<br> 5<br> 6<br> 7<br> 8<br> 9<br>10<br>11<br>12<br>13<br>14<br>15<br>16<br>17<br>18<br>19<br>20<br>21<br>22<br>23<br>24<br>25<br>26<br>27<br>28<br>29<br>30<br> |
<br>from autogen.beta import Agent<br>from autogen.beta.config import AnthropicConfig<br>config = AnthropicConfig("claude-sonnet-4-6")<br>researcher = Agent(<br> "researcher",<br> prompt="You are a thorough researcher. Provide concise factual findings.",<br> config=config,<br> tools=[search_tool],<br>)<br>writer = Agent(<br> "writer",<br> prompt="You are a skilled writer. Turn research into clear prose.",<br> config=config,<br>)<br>coordinator = Agent(<br> "coordinator",<br> prompt="First delegate research, then pass findings to the writer.",<br> config=config,<br> tools=[<br> researcher.as_tool(description="Research a topic and return findings."),<br> writer.as_tool(description="Write an article. Pass research notes in the context parameter."),<br> ],<br>)<br>reply = await coordinator.ask("Write a short article about the history of Python.")<br>print(await reply.content())<br> |
The coordinator's LLM sees two tools — task_researcher and task_writer — and calls them as needed. Each call spawns the target agent on a fresh stream, runs it to completion, and returns the result.
The calling agent's LLM sees a tool named task_{agent.name} with objective (required) and context (optional) parameters.
The context tool parameter is how the calling LLM shares relevant information with the sub-task:
<br>1<br>2<br>3<br>4<br> |
<br>task_writer(<br> objective="Write an article about Python's history",<br> context="Key findings: Created by Guido van Rossum in 1991. Named after Monty Python."<br>)<br> |
as_tool() accepts these parameters:
| Parameter | Type | Description |
|---|---|---|
description |
str |
Tool description shown to the LLM (required) |
name |
`str | None` |
stream |
`StreamFactory | None` |
middleware |
Iterable[ToolMiddleware] |
Tool middleware applied to the delegate tool (e.g., approval_required) |
You can also use subagent_tool() directly for more control:
<br>1<br>2<br>3<br>4<br>5<br>6<br>7<br>8<br>9<br> |
<br>from autogen.beta.tools.subagents import subagent_tool<br>coordinator = Agent(<br> "coordinator",<br> config=config,<br> tools=[<br> subagent_tool(researcher, description="Research a topic."),<br> ],<br>)<br> |
Self-Delegation
An agent can delegate to itself to break complex work into independent sub-tasks. Each sub-task runs as a fresh copy of the agent with its own stream and history.
<br> 1<br> 2<br> 3<br> 4<br> 5<br> 6<br> 7<br> 8<br> 9<br>10<br>11<br>12<br>13<br>14<br>15<br>16<br>17<br>18<br>19<br> |
<br>analyst = Agent(<br> "analyst",<br> prompt=(<br> "You have search and sub_task tools. "<br> "Only use sub_task when the task has clearly independent parts. "<br> "Otherwise handle it directly with search."<br> ),<br> config=config,<br> tools=[search_tool],<br>)<br>analyst.add_tool(<br> analyst.as_tool(<br> description="Break work into a focused sub-task for independent analysis.",<br> name="sub_task",<br> )<br>)<br>reply = await analyst.ask("Compare Python vs Rust for web APIs: performance, DX, and ecosystem.")<br> |
The analyst's LLM may call sub_task multiple times — one per aspect — then synthesise the results.
Sub-Task Streams
Default Behavior
By default, each sub-task creates a fresh MemoryStream. The sub-task's history is isolated — it doesn't carry over between invocations.
It means that subagent has no information about previous calls or results. It just sees the current call and the context.
| What | Behavior | Why |
|---|---|---|
| Dependencies | Copied | Isolated — child mutations don't affect parent |
| Variables | Copied; synced back on success | Concurrent-safe — user variable mutations propagate back |
| History | Fresh stream | Clean context — the LLM passes relevant info via context parameter |
| Depth counter | Incremented in child; excluded from sync-back | Internal bookkeeping — never leaks to parent |
| Agent prompt, tools, config | Inherited | The sub-agent brings its own capabilities |
Persistent Stream
persistent_stream() gives the same agent a consistent stream across multiple invocations within a context. The sub-task's history accumulates across calls rather than starting fresh each time:
<br>1<br>2<br>3<br>4<br>5<br>6<br> |
<br>from autogen.beta.tools.subagents import persistent_stream<br>researcher.as_tool(<br> description="Research a topic",<br> stream=persistent_stream(),<br>)<br> |
It stores the stream ID in context.dependencies keyed by f"ag:{agent.name}:stream" and reuses the parent stream's storage backend. This is useful when the sub-agent benefits from seeing its own prior work — for example, a researcher that should avoid repeating searches.
Custom Factory
For full control, pass any callable matching StreamFactory = Callable[[Agent, Context], Stream]:
<br> 1<br> 2<br> 3<br> 4<br> 5<br> 6<br> 7<br> 8<br> 9<br>10<br> |
<br>from autogen.beta import Agent, Context<br>from autogen.beta.streams.redis import RedisStream<br>def make_redis_stream(agent: Agent, ctx: Context) -> RedisStream:<br> return RedisStream(MY_REDIS_URL, prefix=f"ag2:sub:{agent.name}")<br>researcher.as_tool(<br> description="Research a topic",<br> stream=make_redis_stream,<br>)<br> |