Anthropic SDK Track Part 4: Multi-Agent Orchestration

1. Hub-and-Spoke Coordinator

Imagine you need an AI that can research a topic, analyze data, and write a report — but a single agent trying to do everything would be overwhelmed. The solution is a hub-and-spoke architecture: one “coordinator” agent acts as the manager, breaking work into pieces and delegating to specialized “subagent” workers. Each subagent focuses on one job (researching, analyzing, or writing) and reports back to the coordinator, who assembles the final answer.

This pattern is the foundation of production multi-agent systems. It keeps each agent focused, makes errors easier to diagnose (since subagents are isolated), and allows parallel execution for speed.

1.1 The Coordinator’s Role

flowchart TD
    U["User Request"] --> C["Coordinator Agent"]
    C -->|"Decompose & Delegate"| S1["Research Subagent"]
    C -->|"Decompose & Delegate"| S2["Analysis Subagent"]
    C -->|"Decompose & Delegate"| S3["Writing Subagent"]
    S1 -->|"Structured Results"| C
    S2 -->|"Structured Results"| C
    S3 -->|"Structured Results"| C
    C -->|"Evaluate gaps"| C
    C -->|"Final synthesis"| R["Response to User"]
                        

The coordinator has three primary responsibilities:

1. Task decomposition — breaking the user’s request into smaller, subagent-appropriate pieces
2. Delegation — dispatching each piece to a specialist with all the context they need
3. Result aggregation — evaluating completeness, spotting gaps, re-delegating if needed, and assembling the final answer

Here’s a complete, runnable coordinator. The key insight: we give Claude a dispatch_subagent tool — when it wants to delegate work, it calls that tool with a role and prompt. Our code intercepts the call and runs a separate Claude instance (the subagent):

# Multi-Agent Coordinator — Complete Runnable Example
# Requires: pip install anthropic
# Set env var: ANTHROPIC_API_KEY=sk-ant-...

import anthropic
import json

# Initialize the Anthropic client (reads ANTHROPIC_API_KEY from environment)
client = anthropic.Anthropic()


def execute_tool(name: str, input_data: dict) -> dict:
    """Simulate tool execution. In production, this calls real APIs."""
    print(f"  [Tool: {name}] input: {json.dumps(input_data)[:80]}")
    return {
        "status": "success",
        "result": f"Subagent '{input_data.get('role', 'unknown')}' completed task"
    }


def run_agent(user_message: str, tools: list, system: str = "") -> str:
    """
    Generic agentic loop (from Part 3).
    Sends a message to Claude, executes any tool calls, and repeats
    until Claude returns a final text response (stop_reason='end_turn').
    """
    messages = [{"role": "user", "content": user_message}]

    for _ in range(10):  # Safety cap to prevent infinite loops
        response = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=4096,
            system=system,
            tools=tools,
            messages=messages
        )

        # If Claude is done talking, return its text
        if response.stop_reason == "end_turn":
            return "\n".join(
                b.text for b in response.content if b.type == "text"
            )

        # If Claude wants to use tools, execute them and feed results back
        if response.stop_reason == "tool_use":
            messages.append({"role": "assistant", "content": response.content})
            tool_results = []
            for block in response.content:
                if block.type == "tool_use":
                    result = execute_tool(block.name, block.input)
                    tool_results.append({
                        "type": "tool_result",
                        "tool_use_id": block.id,
                        "content": json.dumps(result)
                    })
            messages.append({"role": "user", "content": tool_results})

    return "[max iterations reached]"

def coordinator_agent(user_request: str) -> str:
    """
    Hub-and-spoke coordinator that manages subagent execution.
    The coordinator receives a user request, breaks it into subtasks,
    and delegates each one using the 'dispatch_subagent' tool.
    """

    # System prompt tells Claude HOW to be a coordinator
    coordinator_system = """You are a research coordinator. Your job is to:
1. Decompose the user's research request into specific, non-overlapping subtasks
2. Delegate each subtask to an appropriate subagent via the 'dispatch_subagent' tool
3. Evaluate results for gaps or contradictions
4. If gaps exist, dispatch additional targeted research
5. Synthesize all findings into a coherent final answer

Rules:
- Each subagent operates with isolated context (no shared memory)
- Pass ALL necessary context explicitly in the subagent prompt
- Partition scope to minimize duplication between subagents
- Evaluate completeness before synthesizing"""

    # The dispatch_subagent tool — this is what Claude calls to delegate work
    tools = [{
        "name": "dispatch_subagent",
        "description": "Dispatch a research subtask to a specialized subagent. "
                       "The subagent has no access to prior conversation — "
                       "pass all required context in the prompt.",
        "input_schema": {
            "type": "object",
            "properties": {
                "role": {
                    "type": "string",
                    "description": "Subagent role: 'researcher', 'analyst', or 'fact_checker'"
                },
                "prompt": {
                    "type": "string",
                    "description": "Complete task description with all necessary context"
                },
                "required_output": {
                    "type": "string",
                    "description": "Expected output format"
                }
            },
            "required": ["role", "prompt", "required_output"]
        }
    }]

    # Run the coordinator through our agentic loop
    return run_agent(user_request, tools, coordinator_system)


# --- Run it ---
result = coordinator_agent("Compare Python and Rust for CLI tools in 2 sentences.")
print(f"Coordinator result: {result[:200]}")

1.2 Isolated Subagent Context

Here’s the single most important rule of multi-agent systems: subagents don’t remember anything the coordinator knows. Each subagent starts with a blank slate. If the coordinator learned that the customer’s name is “Alice” five messages ago, the subagent has no idea unless you explicitly tell it.

This is by design — it prevents “context pollution” (where irrelevant details from one task leak into another) and keeps each subagent laser-focused. But it means you must be deliberate about what information you pass.

This is the most common source of multi-agent bugs — subagents that fail because critical context was not passed. Here’s how to do it right:

# Isolated Subagent Spawning — Context Must Be Explicit
# Requires: pip install anthropic

import anthropic

client = anthropic.Anthropic()


def spawn_subagent(role: str, prompt: str, tools: list = None) -> str:
    """
    Spawn an isolated subagent with its own fresh context.
    The subagent knows NOTHING except what's in 'prompt'.
    """

    # Each role gets a focused system prompt
    role_systems = {
        "researcher": (
            "You are a focused researcher. Search for specific facts "
            "and return structured findings with sources."
        ),
        "analyst": (
            "You are a data analyst. Analyze the provided data "
            "and identify patterns, anomalies, and insights."
        ),
        "fact_checker": (
            "You are a fact checker. Verify claims against "
            "provided sources. Flag contradictions."
        ),
    }

    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=4096,
        system=role_systems.get(role, "You are a helpful assistant."),
        tools=tools or [],
        # Note: only ONE message — the subagent has NO prior history
        messages=[{"role": "user", "content": prompt}]
    )

    return response.content[0].text


# --- Example: BAD vs GOOD context passing ---

# ❌ BAD: The subagent has no idea what "competitor" or "pricing" refers to
bad_prompt = "Research the competitor's pricing."
# Result: vague, unhelpful response because critical details are missing

# ✅ GOOD: Everything the subagent needs is in the prompt itself
good_prompt = """Research Acme Corp's pricing for their enterprise API product.

Context: We are comparing pricing models for our Series A pitch deck.
We are a B2B SaaS company in the developer tools space.

Required output format:
- Pricing tiers (name, price, features)
- Free tier details
- Enterprise contact-sales details
- Public case studies mentioning pricing

Sources: Use only publicly available information from acme.com and tech press."""

# Run the good version
result = spawn_subagent("researcher", good_prompt)
print(f"Research findings: {result[:200]}...")

1.3 Iterative Refinement Loops

A simple coordinator sends work out and accepts whatever comes back. A production-quality coordinator is smarter: it checks whether the results actually answer the original question, identifies gaps (“we know pricing but not market share”), and dispatches follow-up research. This iterative loop is what separates robust systems from fragile one-shot pipelines.

The pattern works like this: after each round of subagent results, the coordinator asks itself “Is this enough to give a complete answer?” If not, it figures out what’s missing and sends out more targeted subagents:

import anthropic
import json

client = anthropic.Anthropic()

def coordinator_with_refinement(request: str, max_rounds: int = 3) -> str:
    """Coordinator with iterative gap-detection and re-delegation."""

    messages = [{"role": "user", "content": request}]
    all_findings = []

    for round_num in range(max_rounds):
        # Coordinator evaluates current state and decides next action
        eval_prompt = f"""Current findings so far:
{json.dumps(all_findings, indent=2) if all_findings else 'None yet — this is the first round.'}

Original request: {request}

Evaluate:
1. Are there gaps in the findings?
2. Are there contradictions that need fact-checking?
3. Is the coverage sufficient to synthesize a complete answer?

If gaps exist, dispatch additional subagent(s). If complete, synthesize the final answer."""

        response = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=4096,
            system="You are a research coordinator. Evaluate findings and identify gaps.",
            tools=coordinator_tools,
            messages=[{"role": "user", "content": eval_prompt}]
        )

        if response.stop_reason == "end_turn":
            # Coordinator decided findings are complete — return synthesis
            return response.content[0].text

        # Process new subagent dispatches and collect results
        for block in response.content:
            if block.type == "tool_use" and block.name == "dispatch_subagent":
                result = spawn_subagent(block.input["role"], block.input["prompt"])
                all_findings.append({
                    "round": round_num + 1,
                    "role": block.input["role"],
                    "result": result
                })

    # Final synthesis after max rounds
    return synthesize_findings(all_findings, request)

2. Agent Tool & Subagent Spawning

In Section 1, we built coordinators “from scratch” using custom tool definitions and our own dispatch_subagent function. This works, but it’s like building a web server from raw sockets — educational, but not what you’d do in production.

The Agent tool (formerly “Task”, renamed in Claude Code v2.1.63) is a built-in, SDK-level mechanism provided by the Claude Agent SDK. When you include "Agent" in a coordinator’s allowed_tools list, the SDK automatically:

1. Exposes an “Agent” tool to the coordinator — Claude can call it like any other tool
2. Spawns a new subagent when the coordinator calls Agent — with its own isolated context window, system prompt, and tool permissions
3. Returns the subagent’s final output back to the coordinator as the tool result

Analogy: Think of the Agent tool like a “delegate” button in a project management app. The coordinator clicks “Delegate”, fills in who should do the work and what they need to know, and the system handles everything else — creating the task, assigning the worker, collecting the result, and reporting back.

2.1 How the Agent Tool Actually Works

Here’s the exact flow, step by step:

sequenceDiagram
    participant C as Coordinator Agent
    participant SDK as Claude Agent SDK
    participant S as Subagent (new context)

    C->>SDK: tool_use: Agent(subagent_type, prompt)
    SDK->>S: Spawns new agent with isolated context
    Note over S: Fresh context window
Own system prompt
Own tools subset
    S->>S: Executes autonomously (may use tools)
    S->>SDK: Returns final text result
    SDK->>C: tool_result with subagent output
    C->>C: Evaluates result, decides next step
                        

When the coordinator invokes the Agent tool, the SDK receives a tool_use block and spawns a fresh subagent conversation. Here is what that looks like in the streamed messages:

# Detecting Agent tool invocation in the message stream
# The SDK emits ToolUseBlock with name="Agent" (or "Task" in older versions)

# What appears in the coordinator's response:
agent_tool_call = {
    "type": "tool_use",
    "id": "toolu_01XYZ...",
    "name": "Agent",                   # Built-in tool (renamed from "Task" in v2.1.63)
    "input": {
        "subagent_type": "web-researcher",  # Which AgentDefinition to use
        "prompt": (                         # Full task + context for the subagent
            "Search for the current market share of electric vehicles in Europe. "
            "Focus on 2024-2025 data from reputable sources (IEA, Bloomberg NEF). "
            "Return: market share %, top 3 manufacturers, year-over-year growth."
        )
    }
}

# The SDK automatically:
# 1. Looks up "web-researcher" in the agents dict
# 2. Creates a NEW conversation with:
#    - system prompt from AgentDefinition.prompt
#    - tools restricted to AgentDefinition.tools
#    - model from AgentDefinition.model (or inherits parent's)
#    - The "prompt" field as the user message
# 3. Runs that conversation to completion (its own autonomous loop)
# 4. Returns the subagent's final text as the tool_result to coordinator
# 5. Subagent's intermediate tool calls stay isolated (never reach coordinator)

2.2 AgentDefinition Configuration

An AgentDefinition is the “job description” for each subagent. It defines: what this agent does (description), how it behaves (prompt), what tools it can access (tools), and what model to use. The coordinator uses the description field to decide which agent is the right fit for each subtask.

# AgentDefinition Reference — Claude Agent SDK
# Requires: pip install claude-agent-sdk

from claude_agent_sdk import AgentDefinition

# Each field maps to the AgentDefinition class:
agents = {
    "web-researcher": AgentDefinition(
        # description: tells Claude WHEN to invoke this subagent
        # Make it specific — Claude matches tasks to agents via this field
        description=(
            "Searches the web for current information including recent events, "
            "pricing data, company news, and public statistics. "
            "Do NOT use for internal codebases or private documents."
        ),

        # prompt: the subagent's system prompt (replaces coordinator's entirely)
        # This is the ONLY behavioral instruction the subagent sees
        prompt=(
            "You are a web researcher specializing in accurate, sourced information.\n\n"
            "## Rules\n"
            "- Always cite source URLs for every claim\n"
            "- Distinguish between facts and opinions\n"
            "- If a source seems unreliable, say so explicitly\n"
            "- Return findings as structured JSON: {findings: [{claim, source_url, confidence}]}\n\n"
            "## Output Format\n"
            "Return a JSON object with 'findings' array and 'summary' string."
        ),

        # tools: the subagent can ONLY use these (security boundary!)
        # A web researcher should NOT have Write or Bash access.
        tools=["WebSearch", "WebFetch"],

        # model: can differ per agent (use cheaper models for simple tasks)
        # Accepts aliases: "sonnet", "opus", "haiku", or full model IDs
        model="sonnet",
    ),

    "code-analyst": AgentDefinition(
        description=(
            "Analyzes source code for architecture, bugs, performance issues, and patterns. "
            "Has read-only access to the codebase. Cannot modify files."
        ),
        prompt=(
            "You are a senior software engineer performing code review.\n"
            "- Reference specific file paths and line numbers\n"
            "- Categorize issues: bug, security, performance, style\n"
            "- Suggest concrete fixes with code snippets"
        ),
        tools=["Read", "Grep", "Glob"],  # Read-only tools only
        model="sonnet",
    ),

    "doc-writer": AgentDefinition(
        description=(
            "Writes technical documentation from research findings. "
            "Use AFTER research is complete — needs findings as input."
        ),
        prompt=(
            "You are a technical writer. Create clear, well-structured documentation.\n"
            "- Use markdown formatting\n"
            "- Include code examples where relevant\n"
            "- Write for a developer audience"
        ),
        tools=["Read", "Write"],  # Can write files (but nothing else)
        model="haiku",  # Cheaper model — writing is less complex
    ),
}

print(f"Defined {len(agents)} specialist agents:")
for name, defn in agents.items():
    print(f"  {name}: tools={defn.tools}, model={defn.model}")

Complete AgentDefinition Fields (all optional fields unless noted):

2.3 Coordinator Setup with Agent Tool (Claude Agent SDK)

The Claude Agent SDK (claude_agent_sdk) provides a production-ready way to run multi-agent systems. You define subagents via AgentDefinition objects and the SDK automatically exposes them through the Agent tool (renamed from “Task” in v2.1.63). The coordinator sees a tool whose description lists all available subagents and delegates automatically.

# Multi-Agent Coordinator with Claude Agent SDK
# Requires: pip install claude-agent-sdk
# If your environment has trouble spawning subprocesses, run from a normal .py script.
# Set env var: ANTHROPIC_API_KEY=sk-ant-...

import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions, AgentDefinition


async def main():
    """Run a coordinator that delegates to specialist subagents."""

    # Define specialist subagents via AgentDefinition
    # The SDK exposes these as the "Agent" tool to the coordinator
    agents = {
        "web-researcher": AgentDefinition(
            # description: tells Claude WHEN to use this subagent
            description=(
                "Searches the web for current information. Use for recent events, "
                "pricing comparisons, news, and public data that changes over time."
            ),
            # prompt: the subagent's system prompt (its role and behavior)
            prompt="""You are a web research specialist. Search for current, accurate information.
- Cite sources with URLs when possible
- Note when information may be outdated
- Focus on factual data, not opinions
- Provide structured findings with clear sections""",
            # tools: what this subagent can do (scoped for security)
            tools=["WebSearch", "WebFetch", "Read"],
            # model: cheaper/faster model for simple research
            model="sonnet",
        ),

        "code-analyst": AgentDefinition(
            description=(
                "Analyzes source code for architecture, bugs, performance issues, and patterns. "
                "Read-only access — cannot modify files."
            ),
            prompt="""You are a senior software engineer performing code review.
- Reference specific file paths and line numbers
- Categorize issues: bug, security, performance, style
- Suggest concrete fixes with code snippets
- Be thorough but concise""",
            # Read-only tools: no Edit, Write, or Bash access
            tools=["Read", "Grep", "Glob"],
            model="sonnet",
        ),

        "doc-writer": AgentDefinition(
            description=(
                "Writes technical documentation from findings. "
                "Use AFTER research/analysis is complete — pass findings in the prompt."
            ),
            prompt="""You are a technical writer. Create clear, well-structured documentation.
- Use markdown formatting
- Include code examples where relevant
- Write for a developer audience
- Keep it concise and scannable""",
            tools=["Read", "Write"],
            # Use cheaper model for writing tasks
            model="haiku",
        ),
    }

    # Run the coordinator — it automatically delegates via the Agent tool
    # The SDK handles: tool registration, subagent spawning, context isolation,
    # parallel execution, and result collection
    async for message in query(
        prompt="Research the pros and cons of Rust vs Go for CLI tools, then write a comparison doc",
        options=ClaudeAgentOptions(
            # "Agent" in allowed_tools auto-approves subagent invocations
            # Without it, each invocation would need manual approval
            allowed_tools=["Read", "Grep", "Glob", "Agent"],
            # Pass your subagent definitions
            agents=agents,
        ),
    ):
        # Detect subagent invocations
        if hasattr(message, "content") and message.content:
            for block in message.content:
                if hasattr(block, "name") and block.name in ("Agent", "Task"):
                    agent_name = block.input.get("subagent_type", "unknown")
                    print(f"  → Delegating to: {agent_name}")

        # Print the final synthesized result
        if hasattr(message, "result"):
            print(f"\n{'='*60}")
            print(f"Final result:\n{message.result[:500]}...")


asyncio.run(main())

Key differences from the raw Client SDK approach (Section 1):

2.4 Managed Agents API (Create & Deploy)

In production, you register agents via the Managed Agents API under client.beta.agents. This is a beta feature requiring the managed-agents-2026-04-01 header (added automatically by the SDK). Registered agents get versioning, team metadata, and a persistent ID — decoupled from any specific codebase.

# Managed Agents API — Correct Python SDK Usage
# Requires: pip install anthropic
# Docs: https://platform.claude.com/docs/en/api/python/beta/agents

import os
from anthropic import Anthropic

client = Anthropic()  # reads ANTHROPIC_API_KEY from env


# --- CREATE an agent ---
# POST /v1/agents   (beta header added automatically)
agent = client.beta.agents.create(
    model="claude-sonnet-4-6",              # required: model string or ModelConfig
    name="customer-support-researcher",     # required: 1-256 chars
    description=(                           # optional: up to 2048 chars
        "Researches customer issues using internal knowledge base and CRM. "
        "Returns structured findings with customer_id, issue_category, and resolution."
    ),
    system=(                                # ← "system", NOT "system_prompt"
        "You are a customer support researcher. Look up customer accounts, "
        "search the knowledge base, and provide structured findings. "
        "Always include: customer_id, issue_category, relevant_articles, suggested_resolution."
    ),
    # tools: toolset config objects — NOT plain strings like ["bash", "read"]
    tools=[
        {
            "type": "agent_toolset_20260401",       # built-in tool bundle
            "default_config": {
                "enabled": True,
                "permission_policy": {"type": "always_ask"},  # or "always_allow"
            },
            "configs": [
                # Per-tool overrides (optional)
                {"name": "bash",  "enabled": False},   # disable bash for this agent
                {"name": "read",  "enabled": True, "permission_policy": {"type": "always_allow"}},
            ],
        }
    ],
    metadata={"team": "support", "environment": "production"},  # up to 16 k/v pairs
)

print(f"Created agent: {agent.id}")      # agent_011CZkYpogX7uDKUyvBTophP
print(f"Version: {agent.version}")       # 1 (starts at 1, increments on update)
print(f"Model: {agent.model.id}")        # claude-sonnet-4-6

# Managed Agents API — retrieve, list, update, archive
# Requires: pip install anthropic

from anthropic import Anthropic

client = Anthropic()
AGENT_ID = "agent_011CZkYpogX7uDKUyvBTophP"   # from create() response


# --- GET a specific agent (latest version) ---
# GET /v1/agents/{agent_id}
agent = client.beta.agents.retrieve(AGENT_ID)
print(f"Name: {agent.name}, version: {agent.version}")

# Get a specific historical version:
agent_v1 = client.beta.agents.retrieve(AGENT_ID, version=1)


# --- LIST agents (paginated) ---
# GET /v1/agents
page = client.beta.agents.list(limit=20)        # SyncPageCursor
for a in page:
    print(f"  {a.id}: {a.name} v{a.version}")

# Include archived agents:
page_all = client.beta.agents.list(include_archived=True)

# Filter by creation date:
from datetime import datetime, timezone
page_recent = client.beta.agents.list(
    created_at_gte=datetime(2026, 5, 1, tzinfo=timezone.utc),
)


# --- UPDATE an agent (creates a new version) ---
# POST /v1/agents/{agent_id}
# IMPORTANT: version= must match the current server version (optimistic locking)
updated = client.beta.agents.update(
    AGENT_ID,
    version=agent.version,          # ← required; prevents concurrent overwrites
    system=(
        "You are a customer support researcher. "
        "Always respond in JSON. Include confidence scores."
    ),
    description="Updated: now returns JSON with confidence scores.",
    # Omit any field you don't want to change.
    # Send None or empty string to CLEAR an optional field.
)
print(f"Updated to version: {updated.version}")  # 2


# --- ARCHIVE (soft-delete, reversible) ---
# POST /v1/agents/{agent_id}/archive
archived = client.beta.agents.archive(AGENT_ID)
print(f"Archived at: {archived.archived_at}")

# List versions history:
versions_page = client.beta.agents.versions.list(AGENT_ID)
for v in versions_page:
    print(f"  v{v.version}: {v.updated_at}")

print("\nLifecycle: create → retrieve → list → update (versioned) → archive")
print("Update uses optimistic locking: version= must match server's current version.")

2.5 Explicit Context Passing

We’ve established that subagents start with a blank slate. So how do you give them what they need? Use a structured prompt builder that organizes context into clearly labeled sections (Task, Background, Constraints, Prior Findings, Output Format). This makes it easy for both the subagent and future developers to understand what’s being passed:

import anthropic
import json

client = anthropic.Anthropic()

def build_subagent_prompt(task: str, context: dict) -> str:
    """Build a well-structured prompt for a subagent with explicit context."""
    sections = [f"## Task\n{task}"]

    if "background" in context:
        sections.append(f"## Background\n{context['background']}")

    if "constraints" in context:
        sections.append(f"## Constraints\n" + "\n".join(f"- {c}" for c in context["constraints"]))

    if "prior_findings" in context:
        sections.append(f"## Prior Findings\n{json.dumps(context['prior_findings'], indent=2)}")

    if "output_format" in context:
        sections.append(f"## Required Output Format\n{context['output_format']}")

    return "\n\n".join(sections)

# Example: coordinator passing structured context to a subagent
prompt = build_subagent_prompt(
    task="Analyze the authentication flow in the user service and identify security vulnerabilities.",
    context={
        "background": "We're auditing a Node.js microservices application before SOC 2 compliance review.",
        "constraints": [
            "Focus only on authentication (not authorization)",
            "Flag: hardcoded secrets, missing rate limiting, weak token generation",
            "Ignore deprecation warnings"
        ],
        "prior_findings": [
            {"file": "auth/jwt.js", "issue": "Token expiry set to 30 days", "severity": "medium"}
        ],
        "output_format": "JSON array of {file, line, issue, severity, recommendation}"
    }
)

2.6 Parallel Subagent Spawning

Here’s a powerful optimization: when the coordinator asks for multiple subagents at the same time (multiple tool calls in one response), you can run them in parallel instead of one-by-one. If you have 3 research tasks that each take 10 seconds, sequential execution takes 30 seconds — parallel execution takes only 10 seconds.

Agent SDK (automatic parallel): When the coordinator issues multiple Agent tool calls in a single response, the SDK runs them in parallel automatically. No extra code needed — just include "Agent" in allowed_tools and the SDK handles the concurrency:

# Agent SDK — Automatic Parallel Subagent Execution
# When Claude calls Agent() multiple times in one response, the SDK runs them concurrently
# Requires: pip install claude-agent-sdk

import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions, AgentDefinition, AssistantMessage, ResultMessage


async def parallel_research():
    """The SDK auto-parallelizes multiple Agent calls in one turn."""

    agents = {
        "market-researcher": AgentDefinition(
            description="Research market trends, competitor pricing, and industry data.",
            prompt="Find accurate market data. Cite sources. Return structured JSON.",
            tools=["WebSearch", "WebFetch"],
            model="haiku",
        ),
        "tech-analyst": AgentDefinition(
            description="Analyze technical specifications and engineering trade-offs.",
            prompt="Evaluate technical options objectively. Compare with benchmarks.",
            tools=["WebSearch", "WebFetch", "Read"],
            model="sonnet",
        ),
        "risk-assessor": AgentDefinition(
            description="Identify risks, compliance issues, and failure modes.",
            prompt="Identify and categorize risks by severity (high/medium/low).",
            tools=["WebSearch", "Read"],
            model="haiku",
        ),
    }

    async for message in query(
        # Prompt explicitly requests parallel research on multiple topics
        prompt=(
            "I need a comprehensive analysis before choosing a cloud provider. "
            "Simultaneously research: (1) AWS vs GCP pricing for our workload, "
            "(2) performance benchmarks for managed Kubernetes, "
            "(3) compliance risks for healthcare data in each provider."
        ),
        options=ClaudeAgentOptions(
            allowed_tools=["Agent"],  # Agent tool = parallel subagent invocation
            agents=agents,
        ),
    ):
        if isinstance(message, AssistantMessage):
            agent_calls = [b for b in message.content if hasattr(b, "name") and b.name == "Agent"]
            if len(agent_calls) > 1:
                names = [c.input.get("subagent_type", "?") for c in agent_calls]
                print(f"[Parallel] Running {len(agent_calls)} subagents: {names}")

        if isinstance(message, ResultMessage) and message.subtype == "success":
            print(f"\n✅ Done in {message.num_turns} turns, ${message.total_cost_usd:.4f}")
            print(message.result[:500])


asyncio.run(parallel_research())

Raw Client SDK (manual parallel): If using the raw anthropic SDK, you must detect multiple tool_use blocks and dispatch them with asyncio.gather() yourself:

# Raw anthropic SDK — Manual Parallel Execution with asyncio.gather
# Use when you cannot use the Agent SDK (API-only environments)

import anthropic
import asyncio

client = anthropic.Anthropic()


async def run_subagent_async(role: str, prompt: str) -> tuple[str, str]:
    """Async wrapper for a synchronous subagent call."""
    loop = asyncio.get_event_loop()
    result = await loop.run_in_executor(
        None,           # default thread pool
        lambda: spawn_subagent(role, prompt)  # synchronous function
    )
    return role, result


async def execute_parallel_subagents(tool_calls: list) -> list:
    """Execute multiple Agent tool calls in parallel using asyncio.gather."""
    tasks = [
        run_subagent_async(
            role=call["input"]["role"],
            prompt=call["input"]["prompt"],
        )
        for call in tool_calls
    ]
    # All subagents run concurrently — total time = max(individual times)
    results = await asyncio.gather(*tasks)

    return [
        {
            "type": "tool_result",
            "tool_use_id": tool_calls[i]["id"],
            "content": result[1],
        }
        for i, result in enumerate(results)
    ]


# Coordinator loop with parallel detection
def coordinator_loop_with_parallel(request: str) -> str:
    messages = [{"role": "user", "content": request}]

    while True:
        response = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=4096,
            tools=[{"name": "Agent", "description": "...", "input_schema": {...}}],
            messages=messages,
        )

        if response.stop_reason == "end_turn":
            return next(b.text for b in response.content if b.type == "text")

        if response.stop_reason == "tool_use":
            messages.append({"role": "assistant", "content": response.content})

            tool_calls = [
                {"id": b.id, "input": b.input}
                for b in response.content if b.type == "tool_use"
            ]

            # Detect multiple calls → run in parallel
            if len(tool_calls) > 1:
                tool_results = asyncio.run(execute_parallel_subagents(tool_calls))
            else:
                result_text = "placeholder"  # call your spawn function
                tool_results = [{"type": "tool_result", "tool_use_id": tool_calls[0]["id"], "content": result_text}]

            messages.append({"role": "user", "content": tool_results})

2.7 Built-In vs Custom Subagents

Even without defining custom subagents, the current SDK docs describe one built-in general-purpose subagent that Claude can invoke through the Agent tool. Everything else should be treated as either a custom programmatic agent or a filesystem-defined agent from .claude/agents/, not as a guaranteed SDK default.

# Built-in Subagents — Available without any AgentDefinition
# Requires: pip install claude-agent-sdk

import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions, ResultMessage


async def main():
    """Claude can delegate to the built-in general-purpose subagent when appropriate.
    Custom agents still need agents={} or filesystem definitions.
    """

    # Built-in general-purpose subagent: useful for delegated exploration or analysis
    # without defining a custom agent first.
    async for message in query(
        prompt="Find all API endpoints in this codebase and summarize their authentication patterns.",
        options=ClaudeAgentOptions(
            # Agent in allowed_tools enables subagent spawning
            allowed_tools=["Read", "Glob", "Grep", "Agent"],
            # No agents={} needed if Claude can use the built-in general-purpose subagent
        ),
    ):
        if isinstance(message, ResultMessage) and message.subtype == "success":
            print(message.result)

    # To disable all subagent spawning, block Agent entirely:
    # disallowed_tools=["Agent"]


asyncio.run(main())

2.8 What Subagents Inherit (and What They Don’t)

A subagent’s context window starts fresh (no parent conversation) but isn’t empty. The only channel from parent to subagent is the Agent tool’s prompt string:

Special cases:

• Explore & Plan skip CLAUDE.md files AND git status for speed/cost — they’re designed to be fast and cheap
• Forked subagents (experimental) inherit the FULL parent conversation — the exception that proves the rule
• The parent receives the subagent’s final message verbatim as the Agent tool result but may summarize it in its response

2.9 Resuming Subagents

By default, each subagent invocation creates a new instance with fresh context. To continue where a subagent left off (retaining its full conversation history, tool calls, and reasoning), you can resume it by capturing the session ID and agent ID:

# Resuming Subagents — Continue where a subagent left off
# Requires: pip install claude-agent-sdk

import asyncio
import json
import re
from claude_agent_sdk import query, ClaudeAgentOptions, AgentDefinition


def extract_agent_id(text: str) -> str | None:
    """Extract agentId from Agent tool result text."""
    match = re.search(r"agentId:\s*([a-f0-9-]+)", text)
    return match.group(1) if match else None


async def main():
    agent_id = None
    session_id = None

    # First invocation — subagent analyzes the auth module
    async for message in query(
        prompt="Use the code-reviewer agent to review the authentication module.",
        options=ClaudeAgentOptions(
            allowed_tools=["Read", "Grep", "Glob", "Agent"],
            agents={
                "code-reviewer": AgentDefinition(
                    description="Expert code reviewer for security and quality.",
                    prompt="Analyze code quality and suggest improvements.",
                    tools=["Read", "Grep", "Glob"],
                )
            },
        ),
    ):
        # Capture session_id (needed to resume this session)
        if hasattr(message, "session_id"):
            session_id = message.session_id
        # Search for agentId in message content (appears in Agent tool results)
        if hasattr(message, "content"):
            content_str = json.dumps(message.content, default=str)
            extracted = extract_agent_id(content_str)
            if extracted:
                agent_id = extracted
        if hasattr(message, "result"):
            print("First review:", message.result[:200])

    # Second invocation — resume the SAME subagent with follow-up task
    # The subagent retains all its prior tool calls and reasoning
    if agent_id and session_id:
        async for message in query(
            prompt=f"Resume agent {agent_id} and now check for SQL injection vulnerabilities.",
            options=ClaudeAgentOptions(
                allowed_tools=["Read", "Grep", "Glob", "Agent"],
                agents={
                    "code-reviewer": AgentDefinition(
                        description="Expert code reviewer for security and quality.",
                        prompt="Analyze code quality and suggest improvements.",
                        tools=["Read", "Grep", "Glob"],
                    )
                },
                resume=session_id,  # Resume the same session
            ),
        ):
            if hasattr(message, "result"):
                print("Resumed review:", message.result[:200])


asyncio.run(main())

2.10 Troubleshooting Subagents

3. Enforcement Patterns

Multi-agent systems are powerful, but they need guardrails. How do you ensure your agent never processes a $10,000 refund without approval? Or that it always verifies the customer’s identity before accessing their account? There are two approaches, and understanding the difference is critical for both production systems and the CCA exam.

3.1 Programmatic Enforcement vs Prompt Guidance

Programmatic enforcement means your code blocks the action — it’s deterministic and impossible for the AI to bypass. Prompt-based guidance means you ask Claude to follow a rule in the system prompt — it works ~95% of the time but can fail under adversarial inputs or edge cases.

The rule of thumb: if violating the rule could cost money, break compliance, or harm users, use programmatic enforcement. If it’s about style or preference, prompting is fine.

Here’s the pattern: your code sits between Claude’s tool call and the actual execution. Claude says “I want to process a $800 refund” — your PolicyEnforcer checks the rules and blocks it before it ever happens:

# Programmatic Enforcement — Deterministic Policy Compliance
# This code guarantees rules are followed, regardless of what Claude outputs
# Requires: pip install anthropic

import anthropic
import json

client = anthropic.Anthropic()


class PolicyEnforcer:
    """Programmatic enforcement layer — deterministic compliance guarantees."""

    def __init__(self):
        self.verified_customer = None
        self.refund_limit = 500.0

    def check_prerequisites(self, tool_name: str, tool_input: dict) -> tuple:
        """Returns (allowed: bool, reason: str)."""

        # Prerequisite gate: must verify customer before order/refund operations
        if tool_name in ("lookup_order", "process_refund"):
            if self.verified_customer is None:
                return False, "Customer must be verified via get_customer before this operation"

        # Financial limit: block refunds exceeding threshold
        if tool_name == "process_refund":
            amount = tool_input.get("amount", 0)
            if amount > self.refund_limit:
                return False, f"Refund of ${amount} exceeds limit of ${self.refund_limit}. Escalate to human."

        return True, "OK"

    def on_tool_verified(self, tool_name: str, tool_input: dict, result: dict):
        """Update state after successful tool execution."""
        if tool_name == "get_customer" and result.get("verified"):
            self.verified_customer = result.get("customer_id")

def enforced_agent_loop(user_message: str, tools: list) -> str:
    """Agentic loop with programmatic enforcement."""
    enforcer = PolicyEnforcer()
    messages = [{"role": "user", "content": user_message}]

    while True:
        response = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=4096,
            tools=tools,
            messages=messages
        )

        if response.stop_reason == "end_turn":
            return "\n".join(b.text for b in response.content if b.type == "text")

        if response.stop_reason == "tool_use":
            messages.append({"role": "assistant", "content": response.content})
            tool_results = []

            for block in response.content:
                if block.type == "tool_use":
                    # ENFORCEMENT: Check prerequisites before execution
                    allowed, reason = enforcer.check_prerequisites(block.name, block.input)

                    if not allowed:
                        # Return error to Claude — it will adapt
                        tool_results.append({
                            "type": "tool_result",
                            "tool_use_id": block.id,
                            "content": json.dumps({"error": reason, "blocked": True}),
                            "is_error": True
                        })
                    else:
                        # Execute tool normally
                        result = execute_tool(block.name, block.input)
                        enforcer.on_tool_verified(block.name, block.input, result)
                        tool_results.append({
                            "type": "tool_result",
                            "tool_use_id": block.id,
                            "content": json.dumps(result)
                        })

            messages.append({"role": "user", "content": tool_results})

3.2 Prerequisite Gates

A prerequisite gate is like a bouncer at a club: “You can’t enter (process a refund) until you’ve shown ID (verified the customer).” The key insight is that this is enforced in code, not in the prompt. Even if Claude hallucinates and tries to skip verification, the gate blocks it.

The pattern uses a simple state tracker that records which prerequisites have been satisfied:

import anthropic
import json

# State tracking for prerequisite enforcement
class AgentState:
    """Tracks which prerequisites have been satisfied."""

    def __init__(self):
        self.satisfied = set()

    def mark_satisfied(self, prerequisite: str):
        self.satisfied.add(prerequisite)

    def check(self, required: list) -> tuple:
        missing = [r for r in required if r not in self.satisfied]
        if missing:
            return False, f"Missing prerequisites: {', '.join(missing)}"
        return True, "OK"

# Define prerequisite requirements
PREREQUISITES = {
    "lookup_order": ["customer_verified"],
    "process_refund": ["customer_verified", "order_confirmed"],
    "escalate_to_human": []  # No prerequisites
}

# In the agent loop, before executing any tool:
state = AgentState()

def execute_with_prerequisites(tool_name: str, tool_input: dict) -> dict:
    """Execute tool only if prerequisites are met."""
    required = PREREQUISITES.get(tool_name, [])
    allowed, reason = state.check(required)

    if not allowed:
        return {"error": reason, "suggestion": "Call get_customer first to verify identity"}

    result = execute_tool(tool_name, tool_input)

    # Update prerequisites based on results
    if tool_name == "get_customer" and result.get("verified"):
        state.mark_satisfied("customer_verified")
    elif tool_name == "lookup_order" and result.get("status"):
        state.mark_satisfied("order_confirmed")

    return result

3.3 Structured Handoffs for Human Escalation

Sometimes the agent hits a wall — a refund exceeds its authority, the customer is angry, or the situation is too complex. The agent needs to hand off to a human. But a bad handoff (“I can’t help, transferring you”) wastes the human’s time. A structured handoff packages everything the agent learned: what was tried, what failed, what the customer wants, and what the human should do next.

This is both a great UX pattern and a CCA exam requirement — handoffs must include context, attempted actions, and specific blockers:

import anthropic
import json
from datetime import datetime

client = anthropic.Anthropic()

def create_escalation_handoff(customer_id: str, reason: str, agent_actions: list) -> dict:
    """Create a structured handoff document for human agents."""
    return {
        "escalation": {
            "timestamp": datetime.utcnow().isoformat() + "Z",
            "customer_id": customer_id,
            "reason": reason,
            "priority": "high" if "explicit_request" in reason else "medium"
        },
        "context": {
            "actions_taken": agent_actions,
            "findings": [a["result"] for a in agent_actions if "result" in a],
            "blockers": [a for a in agent_actions if a.get("blocked")]
        },
        "recommended_next_steps": [
            "Review customer's account history",
            f"Address escalation reason: {reason}",
            "Follow up within 24 hours"
        ]
    }

# Example escalation output:
# {
#   "escalation": {"customer_id": "cust_123", "reason": "refund_exceeds_limit", ...},
#   "context": {"actions_taken": [...], "findings": [...], "blockers": [...]},
#   "recommended_next_steps": [...]
# }

4. Case Study: Multi-Agent Research Pipeline

Let’s put everything together in a realistic example. Imagine you’re building a research assistant that can investigate any topic by dispatching specialists: web researchers find information, document analysts extract structure, and fact-checkers verify claims. The coordinator manages the whole process, iterating until the answer is complete.

This example combines all three concepts from this article: hub-and-spoke coordination (Section 1), subagent spawning with isolated context (Section 2), and programmatic enforcement via a fact-checking gate (Section 3):

import anthropic
import json

client = anthropic.Anthropic()

RESEARCH_COORDINATOR_SYSTEM = """You are a research coordinator managing a team of specialists.

Your team:
- web_researcher: Searches the web for current information (has web_search tool)
- doc_analyst: Analyzes documents and extracts structured data (has read_document tool)
- fact_checker: Verifies claims against sources (has verify_fact tool)

Your process:
1. Decompose the research question into non-overlapping subtasks
2. Assign subtasks to appropriate specialists (avoid scope overlap)
3. Evaluate results — look for gaps, contradictions, unsupported claims
4. If gaps exist, dispatch targeted follow-up research
5. Have fact_checker verify key claims before final synthesis
6. Synthesize findings with proper source attribution

Rules:
- Each specialist has isolated context — pass ALL relevant info in their prompt
- Partition research scope clearly (e.g., researcher A covers market data, B covers tech trends)
- Return structured data separating CONTENT from METADATA (URLs, dates, confidence)
- If a specialist fails or times out, note the gap rather than fabricating content"""

RESEARCH_TOOLS = [
    {
        "name": "dispatch_specialist",
        "description": "Send a research task to a specialist. They return structured findings.",
        "input_schema": {
            "type": "object",
            "properties": {
                "specialist": {"type": "string", "enum": ["web_researcher", "doc_analyst", "fact_checker"]},
                "task": {"type": "string", "description": "Detailed task with all context needed"},
                "scope": {"type": "string", "description": "Clear scope boundary to prevent overlap"}
            },
            "required": ["specialist", "task", "scope"]
        }
    }
]

def run_research_pipeline(question: str) -> dict:
    """Multi-agent research pipeline with iterative refinement."""
    messages = [{"role": "user", "content": question}]
    findings = []

    for iteration in range(5):  # Safety cap — stop_reason drives termination
        response = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=4096,
            system=RESEARCH_COORDINATOR_SYSTEM,
            tools=RESEARCH_TOOLS,
            messages=messages
        )

        if response.stop_reason == "end_turn":
            return {
                "answer": "\n".join(b.text for b in response.content if b.type == "text"),
                "findings": findings,
                "iterations": iteration + 1
            }

        if response.stop_reason == "tool_use":
            messages.append({"role": "assistant", "content": response.content})

            tool_results = []
            for block in response.content:
                if block.type == "tool_use":
                    result = spawn_subagent(
                        role=block.input["specialist"],
                        prompt=block.input["task"]
                    )
                    findings.append({
                        "specialist": block.input["specialist"],
                        "scope": block.input["scope"],
                        "result": result
                    })
                    tool_results.append({
                        "type": "tool_result",
                        "tool_use_id": block.id,
                        "content": result
                    })

            messages.append({"role": "user", "content": tool_results})

    return {"answer": "Research incomplete after max iterations", "findings": findings}