Agentic Orchestration Basics

10 examples to get you started with agentic orchestration - 7 basic and 3 intermediate.

Prerequisites

Install the Python package: pip install claude-agent-sdk.
Set an API key in your environment: export ANTHROPIC_API_KEY=sk-ant-....
These examples assume you already know the shape of a single query() call; if not, start with the Claude Agent SDK basics first.
"Orchestration" here means application code (or a top-level agent) that decides when to call subagents and how to combine their results, not any single feature of the SDK itself.

Basic Examples

1. A Minimal Orchestrator: One Task, One Subagent

The smallest possible orchestrator: hand one task to one subagent and read back its result.

import asyncio
from claude_agent_sdk import query, AgentOptions, SubagentConfig
 
async def main():
    options = AgentOptions(
        allowed_tools=["file_edit"],
        subagents=[
            SubagentConfig(
                name="changelog-writer",
                description="Writes a CHANGELOG.md entry from a diff summary.",
                allowed_tools=["file_edit"],
            )
        ],
    )
 
    async for message in query(
        prompt="Use the changelog-writer subagent to add a CHANGELOG.md entry for this release.",
        options=options,
    ):
        if message.get("type") == "subagent_result":
            print(f"Result: {message['result']}")
 
asyncio.run(main())

The orchestrator here is a single top-level query() call; the model itself decides to invoke the one subagent it has.
This is the whole pattern: an orchestrating agent, a worker with its own scope, and a result that crosses back.
Nothing about this needs to be more complex until a real decomposition or specialization need shows up.

Related: Single-Agent Loops vs Multi-Agent Systems: A Mental Model - when this pattern is worth reaching for

2. Merging Two Subagent Results

Dispatch two subagents and combine their results into one output.

import asyncio
from claude_agent_sdk import query, AgentOptions, SubagentConfig
 
async def main():
    options = AgentOptions(
        subagents=[
            SubagentConfig(
                name="frontend-summary",
                description="Summarizes frontend changes in a PR.",
                allowed_tools=["file_edit"],
            ),
            SubagentConfig(
                name="backend-summary",
                description="Summarizes backend changes in a PR.",
                allowed_tools=["file_edit"],
            ),
        ],
    )
 
    results = {}
    async for message in query(
        prompt="Use both summary subagents, then write one combined PR description.",
        options=options,
    ):
        if message.get("type") == "subagent_result":
            results[message["subagent_name"]] = message["result"]
 
    print(results)
 
asyncio.run(main())

The orchestrating model decides to call both subagents and is responsible for combining their results into a single description.
Each subagent only sees its own slice of the task, not the other's output, unless the orchestrator explicitly passes it along.
This is the seed of every fan-out/merge pattern covered in depth later in this section.

3. A Fixed Prompt Chain

Run a fixed sequence of steps, each depending on the last.

import asyncio
from claude_agent_sdk import query, AgentOptions
 
async def summarize_then_translate(text: str, language: str) -> str:
    summary_chunks = []
    async for message in query(prompt=f"Summarize in three bullets:\n\n{text}"):
        if message.get("type") == "text":
            summary_chunks.append(message["text"])
    summary = "".join(summary_chunks)
 
    translated_chunks = []
    prompt = f"Translate this into {language}, keep it as three bullets:\n\n{summary}"
    async for message in query(prompt=prompt):
        if message.get("type") == "text":
            translated_chunks.append(message["text"])
    return "".join(translated_chunks)
 
print(asyncio.run(summarize_then_translate("...", "Spanish")))

Each step is a separate query() call whose prompt is built from the previous step's output.
There is no dynamic decision here about which step to run next; the sequence is fixed by the calling code.
This is a prompt chain, the simplest orchestration pattern, and often all a task needs when its steps are always the same and always in the same order.

Related: Prompt Chaining vs Routing: Choosing Your Orchestration Pattern - when a fixed chain is the right call

4. A Simple Router Between Two Paths

Pick which subagent to invoke based on the incoming task, instead of always running the same sequence.

import asyncio
from claude_agent_sdk import query, AgentOptions, SubagentConfig
 
async def handle_ticket(ticket_text: str) -> None:
    options = AgentOptions(
        subagents=[
            SubagentConfig(
                name="bug-triager",
                description="Triages bug reports: reproduces, labels severity.",
                allowed_tools=["bash", "file_edit"],
            ),
            SubagentConfig(
                name="feature-scoper",
                description="Scopes feature requests into a rough implementation plan.",
                allowed_tools=["file_edit"],
            ),
        ],
    )
    prompt = (
        f"Read this ticket, decide if it's a bug or a feature request, "
        f"and route it to the matching subagent:\n\n{ticket_text}"
    )
    async for message in query(prompt=prompt, options=options):
        print(message)
 
asyncio.run(handle_ticket("Login button does nothing on Safari 18."))

The routing decision, which subagent handles this ticket, is made by the model at runtime, not fixed in the calling code.
Both destinations are available in every call; only one (sometimes more) actually gets invoked, based on the ticket's content.
Routing suits variable input; a fixed chain would force every ticket through the same steps regardless of type.

5. Dispatching Independent Subagents in Parallel

Run several same-shaped subtasks concurrently instead of one after another.

import asyncio
from claude_agent_sdk import query, AgentOptions, SubagentConfig
 
async def audit_packages(packages: list[str]) -> None:
    subagents = [
        SubagentConfig(
            name=f"audit-{pkg}",
            description=f"Checks the {pkg} package for outdated dependencies.",
            allowed_tools=["bash"],
        )
        for pkg in packages
    ]
    options = AgentOptions(subagents=subagents)
    prompt = f"Run all audit subagents for {packages} and list findings per package."
 
    async for message in query(prompt=prompt, options=options):
        if message.get("type") == "subagent_result":
            print(f"[{message['subagent_name']}] {message['result']}")
 
asyncio.run(audit_packages(["billing", "auth", "search"]))

One subagent per package means each audit runs in its own context, with no shared state between them.
Because the packages are unrelated, the underlying loop can dispatch these subagent invocations concurrently rather than serially.
This is the pattern to reach for whenever a task is "do the same thing N times, on N independent inputs."

Related: Building Subagents for Parallel Research and Delegation - this pattern in depth

6. Capping Delegation Depth

Prevent a subagent from spawning its own subagents beyond one level.

from claude_agent_sdk import AgentOptions, SubagentConfig
 
worker = SubagentConfig(
    name="research-worker",
    description="Researches one topic; may not delegate further.",
    allowed_tools=["web_search"],
    # No `subagents` field on the worker itself: it has no delegation
    # capability of its own, so depth is capped at one level here.
)
 
options = AgentOptions(subagents=[worker])

The orchestrator can delegate to research-worker, but research-worker has no subagents of its own to delegate to.
Capping depth this way is a deliberate design choice, not an SDK default; nothing stops you from nesting subagents further if you configure it.
Uncapped depth makes cost and failure modes far harder to reason about as a system grows.

Related: Guardrails for Multi-Agent Systems: Bounding Cost and Scope - depth, tool access, and spend limits together

7. Retrying a Failed Subagent Call

Wrap a subagent invocation with a basic retry instead of failing the whole task on one bad attempt.

import asyncio
from claude_agent_sdk import query, AgentOptions, SubagentConfig
 
async def run_with_retry(prompt: str, options: AgentOptions, attempts: int = 3):
    last_error = None
    for attempt in range(attempts):
        try:
            results = []
            async for message in query(prompt=prompt, options=options):
                if message.get("type") == "subagent_result":
                    results.append(message["result"])
            return results
        except Exception as exc:
            last_error = exc
            await asyncio.sleep(2 ** attempt)  # backoff before retrying
    raise RuntimeError(f"Failed after {attempts} attempts") from last_error

A failed subagent call is treated like any other failure that can be retried, not a special case the orchestrator ignores.
Exponential backoff between attempts avoids hammering a flaky dependency immediately after it fails.
Real systems also need a limit on total attempts and a fallback for when retries are exhausted, covered in the error recovery article.

Related: Error Recovery and Retry Strategies in Agent Loops - retries, fallbacks, and circuit breakers

Intermediate Examples

8. Routing Plus Parallel Fan-Out

Combine a routing decision with a parallel dispatch once the path is chosen.

import asyncio
from claude_agent_sdk import query, AgentOptions, SubagentConfig
 
async def handle_release(changed_files: list[str]) -> None:
    if len(changed_files) == 1:
        # Small change: route straight to one reviewer, no fan-out needed.
        options = AgentOptions(
            subagents=[SubagentConfig(
                name="reviewer",
                description="Reviews a single file change.",
                allowed_tools=["file_edit"],
            )]
        )
        prompt = f"Use the reviewer subagent on {changed_files[0]}."
    else:
        # Larger change: fan out one reviewer per file, in parallel.
        options = AgentOptions(
            subagents=[
                SubagentConfig(
                    name=f"reviewer-{i}",
                    description=f"Reviews {f}.",
                    allowed_tools=["file_edit"],
                )
                for i, f in enumerate(changed_files)
            ]
        )
        prompt = f"Use all reviewer subagents to review {changed_files} in parallel."
 
    async for message in query(prompt=prompt, options=options):
        print(message)
 
asyncio.run(handle_release(["src/orders/api.py", "src/orders/models.py"]))

The routing step decides the shape of the work (one reviewer vs. many) before any subagent runs.
Once routed to the fan-out path, the subtasks are independent and dispatched the same way as the parallel-audit example.
Mixing routing and fan-out is common in practice: routing picks the strategy, fan-out executes it.

9. A Circuit Breaker Around a Flaky Tool

Stop retrying a subagent whose underlying tool has failed repeatedly, instead of retrying forever.

import asyncio
import time
from claude_agent_sdk import query, AgentOptions
 
class CircuitBreaker:
    def __init__(self, failure_threshold: int = 3, reset_after: float = 60.0):
        self.failures = 0
        self.threshold = failure_threshold
        self.reset_after = reset_after
        self.opened_at: float | None = None
 
    def is_open(self) -> bool:
        if self.opened_at is None:
            return False
        if time.monotonic() - self.opened_at > self.reset_after:
            self.opened_at = None  # half-open: allow one attempt through
            self.failures = 0
            return False
        return True
 
    def record_failure(self) -> None:
        self.failures += 1
        if self.failures >= self.threshold:
            self.opened_at = time.monotonic()
 
    def record_success(self) -> None:
        self.failures = 0
        self.opened_at = None
 
breaker = CircuitBreaker()
 
async def call_flaky_subagent(prompt: str, options: AgentOptions):
    if breaker.is_open():
        raise RuntimeError("Circuit open: skipping call, dependency looked unhealthy")
    try:
        async for message in query(prompt=prompt, options=options):
            pass
        breaker.record_success()
    except Exception:
        breaker.record_failure()
        raise

The breaker tracks consecutive failures and stops issuing new calls once a threshold is crossed, instead of retrying a dependency that is clearly down.
After reset_after seconds it allows exactly one attempt through (half-open) to test whether the dependency has recovered.
This protects the rest of the orchestrator from spending time and tokens on calls that are very likely to fail again immediately.

10. Scoring Orchestrator Output Before Accepting It

Add a lightweight check that rejects an obviously bad merged result instead of returning it unchecked.

import asyncio
from claude_agent_sdk import query, AgentOptions, SubagentConfig
 
def looks_reasonable(result: str) -> bool:
    return len(result.strip()) > 20 and "TODO" not in result
 
async def orchestrate_with_check(prompt: str, options: AgentOptions) -> str:
    final_result = ""
    async for message in query(prompt=prompt, options=options):
        if message.get("type") == "subagent_result":
            final_result = message["result"]
 
    if not looks_reasonable(final_result):
        raise ValueError("Orchestrator result failed the sanity check")
    return final_result

looks_reasonable is a stand-in for whatever cheap, deterministic check makes sense for your task: length, required fields, absence of placeholder text.
Checking output before it reaches a caller catches a class of failures a retry alone would not, since the call itself succeeded, the content was just wrong.
This is a lightweight preview of the evaluation discipline covered in full in the production-readiness checklist.

Related: Evaluating Agent Quality: A Checklist for Production Readiness - scoring success rate, cost, and failure modes systematically

Stack versions: Written against the Claude model lineup current as of ~June 2026 - Claude Fable 5, Claude Opus 4.8, Claude Sonnet 5 (the default), and Claude Haiku 4.5 - and the Claude Agent SDK (latest release). Model names, SDK versions, and pricing move quickly - verify current specifics at platform.claude.com/docs before relying on them.

Agentic Orchestration Basics

10 examples to get you started with agentic orchestration - 7 basic and 3 intermediate.

Prerequisites

Install the Python package: pip install claude-agent-sdk.
Set an API key in your environment: export ANTHROPIC_API_KEY=sk-ant-....
These examples assume you already know the shape of a single query() call; if not, start with the Claude Agent SDK basics first.
"Orchestration" here means application code (or a top-level agent) that decides when to call subagents and how to combine their results, not any single feature of the SDK itself.

Basic Examples

1. A Minimal Orchestrator: One Task, One Subagent

The smallest possible orchestrator: hand one task to one subagent and read back its result.

import asyncio
from claude_agent_sdk import query, AgentOptions, SubagentConfig
 
async def main():
    options = AgentOptions(
        allowed_tools=["file_edit"],
        subagents=[
            SubagentConfig(
                name="changelog-writer",
                description="Writes a CHANGELOG.md entry from a diff summary.",
                allowed_tools=["file_edit"],
            )
        ],
    )
 
    async for message in query(
        prompt="Use the changelog-writer subagent to add a CHANGELOG.md entry for this release.",
        options=options,
    ):
        if message.get("type") == "subagent_result":
            print(f"Result: {message['result']}")
 
asyncio.run(main())

The orchestrator here is a single top-level query() call; the model itself decides to invoke the one subagent it has.
This is the whole pattern: an orchestrating agent, a worker with its own scope, and a result that crosses back.
Nothing about this needs to be more complex until a real decomposition or specialization need shows up.

Related: Single-Agent Loops vs Multi-Agent Systems: A Mental Model - when this pattern is worth reaching for

2. Merging Two Subagent Results

Dispatch two subagents and combine their results into one output.

import asyncio
from claude_agent_sdk import query, AgentOptions, SubagentConfig
 
async def main():
    options = AgentOptions(
        subagents=[
            SubagentConfig(
                name="frontend-summary",
                description="Summarizes frontend changes in a PR.",
                allowed_tools=["file_edit"],
            ),
            SubagentConfig(
                name="backend-summary",
                description="Summarizes backend changes in a PR.",
                allowed_tools=["file_edit"],
            ),
        ],
    )
 
    results = {}
    async for message in query(
        prompt="Use both summary subagents, then write one combined PR description.",
        options=options,
    ):
        if message.get("type") == "subagent_result":
            results[message["subagent_name"]] = message["result"]
 
    print(results)
 
asyncio.run(main())

The orchestrating model decides to call both subagents and is responsible for combining their results into a single description.
Each subagent only sees its own slice of the task, not the other's output, unless the orchestrator explicitly passes it along.
This is the seed of every fan-out/merge pattern covered in depth later in this section.

3. A Fixed Prompt Chain

Run a fixed sequence of steps, each depending on the last.

import asyncio
from claude_agent_sdk import query, AgentOptions
 
async def summarize_then_translate(text: str, language: str) -> str:
    summary_chunks = []
    async for message in query(prompt=f"Summarize in three bullets:\n\n{text}"):
        if message.get("type") == "text":
            summary_chunks.append(message["text"])
    summary = "".join(summary_chunks)
 
    translated_chunks = []
    prompt = f"Translate this into {language}, keep it as three bullets:\n\n{summary}"
    async for message in query(prompt=prompt):
        if message.get("type") == "text":
            translated_chunks.append(message["text"])
    return "".join(translated_chunks)
 
print(asyncio.run(summarize_then_translate("...", "Spanish")))

Each step is a separate query() call whose prompt is built from the previous step's output.
There is no dynamic decision here about which step to run next; the sequence is fixed by the calling code.
This is a prompt chain, the simplest orchestration pattern, and often all a task needs when its steps are always the same and always in the same order.

Related: Prompt Chaining vs Routing: Choosing Your Orchestration Pattern - when a fixed chain is the right call

4. A Simple Router Between Two Paths

Pick which subagent to invoke based on the incoming task, instead of always running the same sequence.

import asyncio
from claude_agent_sdk import query, AgentOptions, SubagentConfig
 
async def handle_ticket(ticket_text: str) -> None:
    options = AgentOptions(
        subagents=[
            SubagentConfig(
                name="bug-triager",
                description="Triages bug reports: reproduces, labels severity.",
                allowed_tools=["bash", "file_edit"],
            ),
            SubagentConfig(
                name="feature-scoper",
                description="Scopes feature requests into a rough implementation plan.",
                allowed_tools=["file_edit"],
            ),
        ],
    )
    prompt = (
        f"Read this ticket, decide if it's a bug or a feature request, "
        f"and route it to the matching subagent:\n\n{ticket_text}"
    )
    async for message in query(prompt=prompt, options=options):
        print(message)
 
asyncio.run(handle_ticket("Login button does nothing on Safari 18."))

The routing decision, which subagent handles this ticket, is made by the model at runtime, not fixed in the calling code.
Both destinations are available in every call; only one (sometimes more) actually gets invoked, based on the ticket's content.
Routing suits variable input; a fixed chain would force every ticket through the same steps regardless of type.

5. Dispatching Independent Subagents in Parallel

Run several same-shaped subtasks concurrently instead of one after another.

import asyncio
from claude_agent_sdk import query, AgentOptions, SubagentConfig
 
async def audit_packages(packages: list[str]) -> None:
    subagents = [
        SubagentConfig(
            name=f"audit-{pkg}",
            description=f"Checks the {pkg} package for outdated dependencies.",
            allowed_tools=["bash"],
        )
        for pkg in packages
    ]
    options = AgentOptions(subagents=subagents)
    prompt = f"Run all audit subagents for {packages} and list findings per package."
 
    async for message in query(prompt=prompt, options=options):
        if message.get("type") == "subagent_result":
            print(f"[{message['subagent_name']}] {message['result']}")
 
asyncio.run(audit_packages(["billing", "auth", "search"]))

One subagent per package means each audit runs in its own context, with no shared state between them.
Because the packages are unrelated, the underlying loop can dispatch these subagent invocations concurrently rather than serially.
This is the pattern to reach for whenever a task is "do the same thing N times, on N independent inputs."

Related: Building Subagents for Parallel Research and Delegation - this pattern in depth

6. Capping Delegation Depth

Prevent a subagent from spawning its own subagents beyond one level.

from claude_agent_sdk import AgentOptions, SubagentConfig
 
worker = SubagentConfig(
    name="research-worker",
    description="Researches one topic; may not delegate further.",
    allowed_tools=["web_search"],
    # No `subagents` field on the worker itself: it has no delegation
    # capability of its own, so depth is capped at one level here.
)
 
options = AgentOptions(subagents=[worker])

The orchestrator can delegate to research-worker, but research-worker has no subagents of its own to delegate to.
Capping depth this way is a deliberate design choice, not an SDK default; nothing stops you from nesting subagents further if you configure it.
Uncapped depth makes cost and failure modes far harder to reason about as a system grows.

Related: Guardrails for Multi-Agent Systems: Bounding Cost and Scope - depth, tool access, and spend limits together

7. Retrying a Failed Subagent Call

Wrap a subagent invocation with a basic retry instead of failing the whole task on one bad attempt.

import asyncio
from claude_agent_sdk import query, AgentOptions, SubagentConfig
 
async def run_with_retry(prompt: str, options: AgentOptions, attempts: int = 3):
    last_error = None
    for attempt in range(attempts):
        try:
            results = []
            async for message in query(prompt=prompt, options=options):
                if message.get("type") == "subagent_result":
                    results.append(message["result"])
            return results
        except Exception as exc:
            last_error = exc
            await asyncio.sleep(2 ** attempt)  # backoff before retrying
    raise RuntimeError(f"Failed after {attempts} attempts") from last_error

A failed subagent call is treated like any other failure that can be retried, not a special case the orchestrator ignores.
Exponential backoff between attempts avoids hammering a flaky dependency immediately after it fails.
Real systems also need a limit on total attempts and a fallback for when retries are exhausted, covered in the error recovery article.

Related: Error Recovery and Retry Strategies in Agent Loops - retries, fallbacks, and circuit breakers

Intermediate Examples

8. Routing Plus Parallel Fan-Out

Combine a routing decision with a parallel dispatch once the path is chosen.

import asyncio
from claude_agent_sdk import query, AgentOptions, SubagentConfig
 
async def handle_release(changed_files: list[str]) -> None:
    if len(changed_files) == 1:
        # Small change: route straight to one reviewer, no fan-out needed.
        options = AgentOptions(
            subagents=[SubagentConfig(
                name="reviewer",
                description="Reviews a single file change.",
                allowed_tools=["file_edit"],
            )]
        )
        prompt = f"Use the reviewer subagent on {changed_files[0]}."
    else:
        # Larger change: fan out one reviewer per file, in parallel.
        options = AgentOptions(
            subagents=[
                SubagentConfig(
                    name=f"reviewer-{i}",
                    description=f"Reviews {f}.",
                    allowed_tools=["file_edit"],
                )
                for i, f in enumerate(changed_files)
            ]
        )
        prompt = f"Use all reviewer subagents to review {changed_files} in parallel."
 
    async for message in query(prompt=prompt, options=options):
        print(message)
 
asyncio.run(handle_release(["src/orders/api.py", "src/orders/models.py"]))

The routing step decides the shape of the work (one reviewer vs. many) before any subagent runs.
Once routed to the fan-out path, the subtasks are independent and dispatched the same way as the parallel-audit example.
Mixing routing and fan-out is common in practice: routing picks the strategy, fan-out executes it.

9. A Circuit Breaker Around a Flaky Tool

Stop retrying a subagent whose underlying tool has failed repeatedly, instead of retrying forever.

import asyncio
import time
from claude_agent_sdk import query, AgentOptions
 
class CircuitBreaker:
    def __init__(self, failure_threshold: int = 3, reset_after: float = 60.0):
        self.failures = 0
        self.threshold = failure_threshold
        self.reset_after = reset_after
        self.opened_at: float | None = None
 
    def is_open(self) -> bool:
        if self.opened_at is None:
            return False
        if time.monotonic() - self.opened_at > self.reset_after:
            self.opened_at = None  # half-open: allow one attempt through
            self.failures = 0
            return False
        return True
 
    def record_failure(self) -> None:
        self.failures += 1
        if self.failures >= self.threshold:
            self.opened_at = time.monotonic()
 
    def record_success(self) -> None:
        self.failures = 0
        self.opened_at = None
 
breaker = CircuitBreaker()
 
async def call_flaky_subagent(prompt: str, options: AgentOptions):
    if breaker.is_open():
        raise RuntimeError("Circuit open: skipping call, dependency looked unhealthy")
    try:
        async for message in query(prompt=prompt, options=options):
            pass
        breaker.record_success()
    except Exception:
        breaker.record_failure()
        raise

The breaker tracks consecutive failures and stops issuing new calls once a threshold is crossed, instead of retrying a dependency that is clearly down.
After reset_after seconds it allows exactly one attempt through (half-open) to test whether the dependency has recovered.
This protects the rest of the orchestrator from spending time and tokens on calls that are very likely to fail again immediately.

10. Scoring Orchestrator Output Before Accepting It

Add a lightweight check that rejects an obviously bad merged result instead of returning it unchecked.

import asyncio
from claude_agent_sdk import query, AgentOptions, SubagentConfig
 
def looks_reasonable(result: str) -> bool:
    return len(result.strip()) > 20 and "TODO" not in result
 
async def orchestrate_with_check(prompt: str, options: AgentOptions) -> str:
    final_result = ""
    async for message in query(prompt=prompt, options=options):
        if message.get("type") == "subagent_result":
            final_result = message["result"]
 
    if not looks_reasonable(final_result):
        raise ValueError("Orchestrator result failed the sanity check")
    return final_result

looks_reasonable is a stand-in for whatever cheap, deterministic check makes sense for your task: length, required fields, absence of placeholder text.
Checking output before it reaches a caller catches a class of failures a retry alone would not, since the call itself succeeded, the content was just wrong.
This is a lightweight preview of the evaluation discipline covered in full in the production-readiness checklist.

Related: Evaluating Agent Quality: A Checklist for Production Readiness - scoring success rate, cost, and failure modes systematically

Stack versions: Written against the Claude model lineup current as of ~June 2026 - Claude Fable 5, Claude Opus 4.8, Claude Sonnet 5 (the default), and Claude Haiku 4.5 - and the Claude Agent SDK (latest release). Model names, SDK versions, and pricing move quickly - verify current specifics at platform.claude.com/docs before relying on them.

Agentic Orchestration Basics

Search Documentation

Agentic Orchestration Basics