Orchestrate multiple AI agents that collaborate, delegate, and specialize to solve complex tasks.

Multi-Agent Systems

Orchestrate multiple AI agents that collaborate, delegate, and specialize to solve complex tasks.

---

Supervisor Pattern

A supervisor agent manages a pool of worker agents. It receives tasks, decides which worker to delegate to, and synthesizes results.

import anthropic

client = anthropic.Anthropic()

WORKERS = {
    "researcher": {
        "system": "You are a research agent. Given a topic, search for and "
                  "summarize relevant information. Be thorough and cite sources.",
        "tools": [web_search_tool, read_url_tool],
    },
    "coder": {
        "system": "You are a coding agent. Write, debug, and refactor code. "
                  "Always test your code before returning it.",
        "tools": [write_file_tool, run_command_tool, read_file_tool],
    },
    "writer": {
        "system": "You are a writing agent. Draft, edit, and polish text. "
                  "Match the requested tone and format precisely.",
        "tools": [write_file_tool, read_file_tool],
    },
}


def run_worker(worker_name: str, task: str) -> str:
    """Run a specialized worker agent on a task."""
    config = WORKERS[worker_name]
    messages = [{"role": "user", "content": task}]

    for _ in range(10):
        response = client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=4096,
            system=config["system"],
            tools=config["tools"],
            messages=messages,
        )

        if response.stop_reason == "end_turn":
            return extract_text(response)

        messages.append({"role": "assistant", "content": response.content})
        tool_results = execute_all_tools(response)
        messages.append({"role": "user", "content": tool_results})

    return "Worker did not complete within iteration limit."


def supervisor_agent(task: str) -> str:
    """Supervisor that delegates to specialized workers."""
    delegate_tool = {
        "name": "delegate_to_worker",
        "description": "Delegate a subtask to a specialized worker agent. "
                       "Available workers: researcher, coder, writer.",
        "input_schema": {
            "type": "object",
            "properties": {
                "worker": {
                    "type": "string",
                    "enum": ["researcher", "coder", "writer"],
                    "description": "Which worker to assign the subtask to.",
                },
                "subtask": {
                    "type": "string",
                    "description": "Clear description of what the worker should do.",
                },
            },
            "required": ["worker", "subtask"],
        },
    }

    messages = [{"role": "user", "content": task}]
    supervisor_system = (
        "You are a supervisor agent. Break complex tasks into subtasks "
        "and delegate them to specialized workers. Synthesize their results "
        "into a final answer. Available workers:\n"
        "- researcher: searches the web, gathers information\n"
        "- coder: writes and tests code\n"
        "- writer: drafts and polishes text"
    )

    for _ in range(15):
        response = client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=4096,
            system=supervisor_system,
            tools=[delegate_tool],
            messages=messages,
        )

        if response.stop_reason == "end_turn":
            return extract_text(response)

        messages.append({"role": "assistant", "content": response.content})
        results = []
        for block in response.content:
            if block.type == "tool_use":
                worker_result = run_worker(block.input["worker"], block.input["subtask"])
                results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": worker_result,
                })
        messages.append({"role": "user", "content": results})

    return "Supervisor did not complete."

---

Swarm Architecture

In a swarm, agents dynamically hand off conversations to each other. Each agent decides when it should pass control to a different specialist.

class SwarmAgent:
    def __init__(self, name: str, system: str, tools: list[dict],
                 handoff_targets: list[str]):
        self.name = name
        self.system = system
        self.tools = tools
        self.handoff_targets = handoff_targets


def create_handoff_tool(targets: list[str]) -> dict:
    return {
        "name": "transfer_to_agent",
        "description": f"Transfer this conversation to another agent. "
                       f"Available agents: {', '.join(targets)}. "
                       f"Use this when the request is outside your specialty.",
        "input_schema": {
            "type": "object",
            "properties": {
                "target_agent": {
                    "type": "string",
                    "enum": targets,
                    "description": "The agent to transfer to.",
                },
                "context": {
                    "type": "string",
                    "description": "Summary of conversation so far for the next agent.",
                },
            },
            "required": ["target_agent", "context"],
        },
    }


def run_swarm(agents: dict[str, SwarmAgent], initial_agent: str,
              task: str) -> str:
    """Run a swarm where agents can hand off to each other."""
    current_agent = initial_agent
    messages = [{"role": "user", "content": task}]
    visited = []

    for _ in range(20):
        agent = agents[current_agent]
        visited.append(current_agent)

        # Add handoff tool to agent's tools
        handoff_tool = create_handoff_tool(agent.handoff_targets)
        all_tools = agent.tools + [handoff_tool]

        response = client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=4096,
            system=agent.system,
            tools=all_tools,
            messages=messages,
        )

        if response.stop_reason == "end_turn":
            return extract_text(response)

        messages.append({"role": "assistant", "content": response.content})

        tool_results = []
        handoff = None
        for block in response.content:
            if block.type == "tool_use":
                if block.name == "transfer_to_agent":
                    handoff = block
                    tool_results.append({
                        "type": "tool_result",
                        "tool_use_id": block.id,
                        "content": f"Transferring to {block.input['target_agent']}.",
                    })
                else:
                    result = execute_tool(block.name, block.input)
                    tool_results.append({
                        "type": "tool_result",
                        "tool_use_id": block.id,
                        "content": result,
                    })

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

        # Handle handoff
        if handoff:
            current_agent = handoff.input["target_agent"]
            context_msg = (
                f"[Handoff from {visited[-1]}] "
                f"Context: {handoff.input['context']}"
            )
            messages = [{"role": "user", "content": context_msg}]

    return f"Swarm did not resolve. Agents visited: {visited}"


# Setup
agents = {
    "triage": SwarmAgent(
        name="triage",
        system="You are a triage agent. Determine what the user needs and "
               "transfer to the appropriate specialist.",
        tools=[],
        handoff_targets=["sales", "support", "billing"],
    ),
    "sales": SwarmAgent(
        name="sales",
        system="You are a sales agent. Help with product questions and purchases.",
        tools=[search_products_tool, create_order_tool],
        handoff_targets=["triage", "billing"],
    ),
    "support": SwarmAgent(
        name="support",
        system="You are a support agent. Help resolve technical issues.",
        tools=[search_knowledge_base_tool, create_ticket_tool],
        handoff_targets=["triage", "billing"],
    ),
    "billing": SwarmAgent(
        name="billing",
        system="You are a billing agent. Help with invoices and payments.",
        tools=[lookup_invoice_tool, process_refund_tool],
        handoff_targets=["triage", "support"],
    ),
}

result = run_swarm(agents, "triage", "I need to return a defective product")

---

Agent-to-Agent Communication via Shared State

When agents need to collaborate on a shared artifact (like a document or codebase), use a shared state store.

import threading
from dataclasses import dataclass, field


@dataclass
class SharedState:
    """Thread-safe shared state for multi-agent collaboration."""
    artifacts: dict[str, str] = field(default_factory=dict)
    messages: list[dict] = field(default_factory=list)
    _lock: threading.Lock = field(default_factory=threading.Lock)

    def write_artifact(self, key: str, value: str, author: str):
        with self._lock:
            self.artifacts[key] = value
            self.messages.append({
                "author": author,
                "action": "write",
                "key": key,
                "preview": value[:100],
            })

    def read_artifact(self, key: str) -> str:
        with self._lock:
            return self.artifacts.get(key, "")

    def post_message(self, author: str, message: str):
        with self._lock:
            self.messages.append({"author": author, "message": message})

    def get_messages_since(self, index: int) -> list[dict]:
        with self._lock:
            return self.messages[index:]


def make_shared_tools(state: SharedState, agent_name: str) -> list[dict]:
    """Create tools that let an agent interact with shared state."""
    return [
        {
            "name": "write_shared_artifact",
            "description": "Write or update a shared artifact that other agents can read.",
            "input_schema": {
                "type": "object",
                "properties": {
                    "key": {"type": "string", "description": "Artifact name."},
                    "content": {"type": "string", "description": "Artifact content."},
                },
                "required": ["key", "content"],
            },
        },
        {
            "name": "read_shared_artifact",
            "description": "Read a shared artifact written by another agent.",
            "input_schema": {
                "type": "object",
                "properties": {
                    "key": {"type": "string", "description": "Artifact name."},
                },
                "required": ["key"],
            },
        },
        {
            "name": "send_message_to_agents",
            "description": "Post a message visible to all other agents.",
            "input_schema": {
                "type": "object",
                "properties": {
                    "message": {"type": "string"},
                },
                "required": ["message"],
            },
        },
    ]

---

Pipeline Pattern

For sequential workflows, chain agents in a pipeline where each agent's output feeds the next.

def pipeline(task: str, stages: list[dict]) -> str:
    """Run agents in sequence, each building on the previous output."""
    current_input = task

    for stage in stages:
        agent_name = stage["name"]
        system = stage["system"]
        tools = stage.get("tools", [])

        prompt = (
            f"Previous stage output:\n{current_input}\n\n"
            f"Your task: {stage['instruction']}"
        )

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

        for _ in range(10):
            response = client.messages.create(
                model="claude-sonnet-4-20250514",
                max_tokens=4096,
                system=system,
                tools=tools,
                messages=messages,
            )

            if response.stop_reason == "end_turn":
                current_input = extract_text(response)
                break

            messages.append({"role": "assistant", "content": response.content})
            results = execute_all_tools(response)
            messages.append({"role": "user", "content": results})

        print(f"[Pipeline] {agent_name} completed.")

    return current_input


# Usage
stages = [
    {
        "name": "researcher",
        "system": "You are a research agent.",
        "instruction": "Research this topic and produce a structured outline with key facts.",
        "tools": [web_search_tool],
    },
    {
        "name": "writer",
        "system": "You are a writing agent.",
        "instruction": "Turn this research into a well-written article.",
        "tools": [],
    },
    {
        "name": "editor",
        "system": "You are an editing agent. Fix grammar, improve clarity, tighten prose.",
        "instruction": "Edit and polish this article. Return the final version.",
        "tools": [],
    },
]

article = pipeline("Write an article about quantum computing in 2026", stages)

---

Choosing a Pattern

Pattern	Use When	Trade-off
Supervisor	Tasks need dynamic delegation to specialists	Supervisor is a bottleneck; extra API calls
Swarm	Conversational routing with many specialists	Complex handoff logic; hard to debug
Pipeline	Sequential processing stages	Inflexible ordering; no backtracking
Shared State	Agents collaborate on the same artifact	Coordination overhead; potential conflicts

Start with the simplest pattern (pipeline or single supervisor) and add complexity only when the task demands it.