Multi-Agent System Setup

Knowledge

A single agent quickly hits its limits: It must plan, execute, and evaluate the quality of its own work -- all in one loop. In production, you split this responsibility across specialized agents. The result is a multi-agent system that's more robust, efficient, and controllable than a solo performer.

The Planner-Executor-Reviewer Architecture

The most proven architecture for production agents consists of three roles:

Why not a single agent? Three specialized agents are cheaper and better in total than a universal agent that must handle everything at once. The Planner is called only once (expensive, but rare), the Executor runs frequently (cheap), and the Reviewer ensures quality without needing the Executor's full context.

Inter-Agent Communication

Agents communicate via structured messages -- typically JSON objects with clear schemas. This isn't a free-text chat, but a defined protocol:

# Message format for inter-agent communication
from dataclasses import dataclass, field
from typing import Literal
from datetime import datetime

@dataclass
class AgentMessage:
    sender: Literal["planner", "executor", "reviewer"]
    receiver: Literal["planner", "executor", "reviewer"]
    message_type: Literal["plan", "task", "result", "feedback", "approval"]
    payload: dict
    timestamp: str = field(default_factory=lambda: datetime.now().isoformat())
    correlation_id: str = ""  # Links related messages together

Understanding

Complete Multi-Agent System in Python

The following example shows a production-ready multi-agent system. The Planner creates a plan, the Executor works through it, and the Reviewer checks the result.

import anthropic
import json
from dataclasses import dataclass, field, asdict
from typing import Literal
from datetime import datetime

client = anthropic.Anthropic()

@dataclass
class AgentMessage:
    sender: str
    receiver: str
    message_type: str
    payload: dict
    timestamp: str = field(default_factory=lambda: datetime.now().isoformat())

class ProductionAgent:
    def __init__(self, name: str, model: str, system_prompt: str):
        self.name = name
        self.model = model
        self.system_prompt = system_prompt
        self.message_log: list[AgentMessage] = []

    def process(self, task: str) -> str:
        """Processes a task and returns the response."""
        response = client.messages.create(
            model=self.model,
            max_tokens=2048,
            system=self.system_prompt,
            messages=[{"role": "user", "content": task}]
        )
        return response.content[0].text

class MultiAgentOrchestrator:
    def __init__(self):
        self.planner = ProductionAgent(
            name="planner",
            model="claude-opus-5",
            system_prompt=(
                "You are a planning agent. Analyze tasks and create "
                "structured plans in JSON format. Each step must be clear, "
                "atomic, and executable. Format: "
                '{"steps": [{"id": 1, "action": "...", "expected_output": "..."}]}'
            )
        )
        self.executor = ProductionAgent(
            name="executor",
            model="claude-haiku-4-5",
            system_prompt=(
                "You are an execution agent. You receive a single step "
                "and execute it precisely. Return only the result, "
                "no explanations."
            )
        )
        self.reviewer = ProductionAgent(
            name="reviewer",
            model="claude-sonnet-5",
            system_prompt=(
                "You are a review agent. Check results for correctness, "
                "completeness, and quality. Respond with JSON: "
                '{"approved": true/false, "feedback": "...", "score": 0-10}'
            )
        )

    def run(self, task: str) -> dict:
        # 1. Planner creates the plan
        plan_raw = self.planner.process(f"Create a plan for: {task}")
        plan = json.loads(plan_raw)

        results = []
        # 2. Executor works through each step
        for step in plan["steps"]:
            result = self.executor.process(
                f"Execute: {step['action']}\n"
                f"Expected output: {step['expected_output']}"
            )
            results.append({"step_id": step["id"], "result": result})

        # 3. Reviewer checks the overall result
        review = self.reviewer.process(
            f"Task: {task}\n"
            f"Plan: {json.dumps(plan)}\n"
            f"Results: {json.dumps(results)}"
        )

        return {
            "plan": plan,
            "results": results,
            "review": json.loads(review)
        }

# Usage
orchestrator = MultiAgentOrchestrator()
output = orchestrator.run("Analyze the performance of the API endpoints")

TypeScript Variant

import Anthropic from "@anthropic-ai/sdk";

const client = new Anthropic();

interface AgentMessage {
  sender: string;
  receiver: string;
  messageType: string;
  payload: Record<string, unknown>;
  timestamp: string;
}

interface PlanStep {
  id: number;
  action: string;
  expected_output: string;
}

class ProductionAgent {
  constructor(
    private name: string,
    private model: string,
    private systemPrompt: string,
    public messageLog: AgentMessage[] = []
  ) {}

  async process(task: string): Promise<string> {
    const response = await client.messages.create({
      model: this.model,
      max_tokens: 2048,
      system: this.systemPrompt,
      messages: [{ role: "user", content: task }],
    });
    const block = response.content[0];
    return block.type === "text" ? block.text : "";
  }
}

class MultiAgentOrchestrator {
  private planner: ProductionAgent;
  private executor: ProductionAgent;
  private reviewer: ProductionAgent;

  constructor() {
    this.planner = new ProductionAgent(
      "planner",
      "claude-opus-5",
      "You are a planning agent. Create structured plans as JSON."
    );
    this.executor = new ProductionAgent(
      "executor",
      "claude-haiku-4-5",
      "You are an execution agent. Execute steps precisely."
    );
    this.reviewer = new ProductionAgent(
      "reviewer",
      "claude-sonnet-5",
      "You are a review agent. Check results and respond as JSON."
    );
  }

  async run(task: string) {
    // 1. Create plan
    const planRaw = await this.planner.process(
      `Create a plan for: ${task}`
    );
    const plan = JSON.parse(planRaw);

    // 2. Execute steps
    const results = [];
    for (const step of plan.steps as PlanStep[]) {
      const result = await this.executor.process(
        `Execute: ${step.action}`
      );
      results.push({ stepId: step.id, result });
    }

    // 3. Review
    const review = await this.reviewer.process(
      `Task: ${task}\nResults: ${JSON.stringify(results)}`
    );

    return { plan, results, review: JSON.parse(review) };
  }
}

Apply

Reflect

A multi-agent system isn't automatically better than a single agent. The value comes from clear role separation, appropriate model selection per role, and structured communication. In practice, you start with a single agent and only split when you notice that a role is becoming too complex or costs are running too high.