Home › Claude › Agent › Orchestrate

🎼 Orchestrate: coordinating a team of agents

Some goals are too big or too varied for one agent. Orchestration puts a coordinator in charge: it plans the work, hands each part to a specialist, keeps track of progress, checks the results and puts everything together.

• Beginner
• ⏱ 30 min
• General workflow pattern

What is it?

Agent orchestration means coordinating several tools, skills, agents and workflows to complete one larger goal. A central orchestrator, usually Claude with planning instructions, decides what needs doing, who does each part, and in what order, then checks and combines the results.

Orchestrator (coordinator)

The "project manager" agent that plans, delegates, monitors and combines.

Worker / specialised agent

An agent with one focused role, such as researcher, writer or reviewer, often with its own tools and skills.

Multi-agent system

A system where several agents work together on one goal.

Task dependency

When a task needs another task's result first.

Shared context

Background information that every agent receives, such as product facts or brand voice.

Validation

Checking a result is correct and complete before using it.

Token

A small chunk of text (roughly ¾ of a word). AI usage is measured and paid for in tokens.

Single agent and multi-agent systems compared

Why is it important?

• Big goals have many parts, such as research, strategy, writing and review, and each benefits from focused instructions and tools.
• Focus: each worker only sees what it needs, so it isn't distracted by everyone else's details.
• Speed: independent parts can run in parallel.
• Quality control: the orchestrator can validate each result and send weak work back.

Real-world analogy

How does it work?

1. Understand the goal: the orchestrator receives the user's goal and shared context.
2. Plan: it breaks the goal into tasks, picks a specialist for each and notes the dependencies.
3. Assign: tasks whose dependencies are done are sent to workers, in parallel when they're independent.
4. Monitor: it tracks status (waiting, running, done or failed) and the budget.
5. Validate: each result is checked. Weak results are retried with feedback.
6. Combine: results are merged into the final response.
7. Approve: anything risky, such as publishing, goes to a human (see Human in the Loop).

Workflow diagram

Simple overview

Detailed: the marketing workflow (with dependencies)

Try it: run the orchestrator

Practical example: an AI marketing workflow

Goal: "Create launch content for our new e-scooter, Glide 2."

1. Orchestrator plans three tasks: research (no dependencies), strategy (needs research), drafts (needs strategy).
2. Research agent summarises competitor notes: pricing, messages and gaps.
3. Strategy agent turns the research into three content angles, for example "longest battery in its class".
4. Copywriter agent drafts three social posts using the angles and the brand voice from shared context.
5. Validator checks the facts against the product sheet and the tone against the brand voice. One post claims "fastest scooter ever", which is unsupported, so it goes back to the copywriter with that feedback.
6. Human approval: the marketing lead reviews the final three posts and approves them. Nothing is published before this.

Code example

Runnable but simplified Python for learning, not production-ready. It needs pip install anthropic, the ANTHROPIC_API_KEY environment variable on your computer or server, and a notes.txt file with product facts, brand voice and competitor notes.

# orchestrator.py: plan, delegate, validate, then ask a human. A simplified teaching version.
import asyncio
import json
import pathlib
import anthropic

client = anthropic.AsyncAnthropic()
MODEL = "claude-opus-5-5"
MAX_CALLS = 15      # hard budget: the whole system stops if it would go over
calls = 0

async def ask(system: str, prompt: str, schema: dict | None = None):
    """One Claude call. If a schema is given, the reply is guaranteed JSON of that shape."""
    global calls
    calls += 1
    if calls > MAX_CALLS:
        raise RuntimeError("Budget exceeded. Stopping to protect costs.")
    extra = {"output_config": {"format": {"type": "json_schema", "schema": schema}}} if schema else {}
    response = await client.messages.create(
        model=MODEL, max_tokens=16000, system=system,
        messages=[{"role": "user", "content": prompt}], **extra,
    )
    text = "".join(b.text for b in response.content if b.type == "text")
    return json.loads(text) if schema else text

# The specialised workers: each is Claude with a focused role (a system prompt).
SPECIALISTS = {
    "researcher": "You are a market researcher. Use only the shared context. Be factual and brief.",
    "strategist": "You are a content strategist. Turn research into 3 clear content angles.",
    "writer": "You are a copywriter. Friendly, plain language. Max 60 words per post. No unsupported claims.",
}

PLAN_SCHEMA = {
    "type": "object",
    "properties": {"tasks": {"type": "array", "items": {
        "type": "object",
        "properties": {
            "id": {"type": "string"},
            "agent": {"type": "string", "enum": list(SPECIALISTS)},
            "instructions": {"type": "string"},
            "depends_on": {"type": "array", "items": {"type": "string"}},
        },
        "required": ["id", "agent", "instructions", "depends_on"],
        "additionalProperties": False,
    }}},
    "required": ["tasks"],
    "additionalProperties": False,
}
CHECK_SCHEMA = {
    "type": "object",
    "properties": {"ok": {"type": "boolean"}, "problems": {"type": "string"}},
    "required": ["ok", "problems"],
    "additionalProperties": False,
}

async def run_task(task: dict, results: dict, context: str) -> str:
    """Give one task to its specialist, then validate the result (max 2 tries)."""
    earlier = "\n\n".join(f"## Result of {d}\n{results[d]}" for d in task["depends_on"])
    prompt = f"{task['instructions']}\n\n## Shared context\n{context}\n\n{earlier}"
    for attempt in (1, 2):
        output = await ask(SPECIALISTS[task["agent"]], prompt)
        check = await ask(
            "You are a strict reviewer. Check the output against the task and the shared context.",
            f"Task: {task['instructions']}\n\nShared context:\n{context}\n\nOutput:\n{output}",
            CHECK_SCHEMA,
        )
        if check["ok"]:
            return output
        print(f"  [retry] {task['id']} failed validation: {check['problems']}")
        prompt += f"\n\nA reviewer found these problems. Fix them: {check['problems']}"
    raise RuntimeError(f"Task '{task['id']}' failed validation twice. A human needs to look.")

async def orchestrate(goal: str, context: str) -> dict:
    plan = await ask(
        "You are an orchestrator. Split the goal into 2-4 tasks for these agents: "
        + ", ".join(SPECIALISTS) + ". In depends_on, list the ids of tasks that must finish first.",
        goal, PLAN_SCHEMA,
    )
    tasks = {t["id"]: t for t in plan["tasks"]}
    print("Plan:", [(t["id"], t["agent"], t["depends_on"]) for t in tasks.values()])

    results = {}
    while len(results) < len(tasks):
        # A task is ready when everything it depends on has a result.
        ready = [t for t in tasks.values()
                 if t["id"] not in results and all(d in results for d in t["depends_on"])]
        if not ready:
            raise RuntimeError("The plan has a loop or an unknown dependency.")
        outputs = await asyncio.gather(*(run_task(t, results, context) for t in ready))  # parallel
        for t, out in zip(ready, outputs):
            results[t["id"]] = out
            print(f"[done] {t['id']} ({t['agent']})")
    return results

async def main():
    goal = "Create launch content for our new e-scooter 'Glide 2': research competitors, choose angles, draft 3 social posts."
    context = pathlib.Path("notes.txt").read_text()
    results = await orchestrate(goal, context)
    final = list(results.values())[-1]           # the last task to finish (the drafts)
    print("\n=== Final content ===\n" + final)
    print(f"\n(Claude calls used: {calls}/{MAX_CALLS})")

    # Human in the loop: nothing is published without approval.
    if input("\nApprove for publishing? Type 'yes' to approve: ").strip().lower() == "yes":
        print("Approved. Handed to the publishing team.")
    else:
        print("Not approved. Nothing was published.")

if __name__ == "__main__":
    asyncio.run(main())

Expected output (wording and plan will vary):

Plan: [('research', 'researcher', []), ('strategy', 'strategist', ['research']), ('drafts', 'writer', ['strategy'])]
[done] research (researcher)
[done] strategy (strategist)
  [retry] drafts failed validation: Post 2 claims "fastest scooter ever", which the product sheet does not support.
[done] drafts (writer)

=== Final content ===
1. Go further on one charge: Glide 2's 60 km battery ...

(Claude calls used: 9/15)

Approve for publishing? Type 'yes' to approve: yes
Approved. Handed to the publishing team.

Step-by-step explanation

1. ask() is the single entry point for every Claude call. It counts calls and stops at the budget, which protects you from runaway costs and loops.
2. SPECIALISTS defines the workers. Here each "agent" is Claude with a different system prompt. Real systems would also give each one its own tools and skills.
3. PLAN_SCHEMA forces the orchestrator's plan into valid JSON: a list of tasks, each with an agent (only from our list), instructions and depends_on.
4. run_task() passes the shared context and only the results this task depends on. That is how agents communicate: through the orchestrator, not by chatting with each other.
5. Each output is checked by a reviewer call using CHECK_SCHEMA (result validation). Failures are retried once with the problems as feedback, then the task stops for a human.
6. In orchestrate(), the while loop finds every task whose dependencies are done and runs those tasks in parallel with asyncio.gather.
7. If no task is ever ready, the plan has a loop or a missing task, so the program stops instead of waiting forever.
8. main() ends with a human approval step. Publishing only happens on an explicit "yes".

Communication, shared context, dependencies and validation

Handling failures, loops, conflicts and costs

When one agent is enough, and when orchestration adds value

Anthropic's advice is to start with the simplest solution and add complexity only when it clearly improves results.

Use cases

• Marketing campaigns: research, strategy, content, review and approval
• Deep research: many sub-questions explored in parallel, then combined into a report
• Software tasks: plan, change several files with specialised agents, then test and review
• Due diligence: financial, legal and market analysis by separate specialists

Advantages and limitations

Best practices

• Start with one agent, and move to orchestration only when you can see the benefit.
• Give every worker a clear role, clear inputs and a clear output format.
• Make the plan structured (JSON) so code can check it.
• Set hard limits on calls, retries and time.
• Validate every result, and log every step for debugging.
• Put a human approval step before anything public, costly or irreversible.

Knowledge check

Mini project

Key takeaways

• Orchestration means a coordinator plans, delegates, monitors, validates and combines.
• Specialised agents with focused context do better on big, varied goals.
• Manage dependencies, shared context, validation, loops and budgets explicitly.
• It costs much more than a single agent, so use it only when it adds real value.
• It's a general pattern. Claude products such as Claude Code and the Agent SDK offer subagents that help you apply it.

Further learning

• Building effective agents, the "Orchestrator-workers" section (Anthropic engineering blog)
• How we built our multi-agent research system (Anthropic engineering blog)
• Subagents in Claude Code (official docs)
• Claude Agent SDK overview (official docs)
• Structured outputs for machine-readable plans (official docs)