How to Build a Multi-Agent Workflow Orchestrator with Claude API (Step-by-Step)

mkdir multi-agent-orchestrator
cd multi-agent-orchestrator
python -m venv venv
source venv/bin/activate   # Windows: venv\Scripts\activate
pip install anthropic python-dotenv

ANTHROPIC_API_KEY=sk-ant-your-key-here

import os
from anthropic import Anthropic
from dotenv import load_dotenv

load_dotenv()

client = Anthropic(api_key=os.environ.get("ANTHROPIC_API_KEY"))

response = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=64,
    messages=[{"role": "user", "content": "Say: setup confirmed"}]
)

print(response.content[0].text)

import os
from anthropic import Anthropic
from dotenv import load_dotenv

load_dotenv()

# The client is shared across all agents
client = Anthropic(api_key=os.environ.get("ANTHROPIC_API_KEY"))

# Tool schemas tell the Orchestrator what agents exist and how to call them
AGENT_TOOLS = [
    {
        "name": "researcher_agent",
        "description": (
            "Gathers and summarizes factual information on a topic. "
            "Use this when you need background research, market data, "
            "or factual context before analysis or writing."
        ),
        "input_schema": {
            "type": "object",
            "properties": {
                "task": {
                    "type": "string",
                    "description": "The specific research question or topic to investigate."
                }
            },
            "required": ["task"]
        }
    },
    {
        "name": "analyst_agent",
        "description": (
            "Analyzes data, findings, or text and produces structured insights, "
            "comparisons, or recommendations. Use after research is complete."
        ),
        "input_schema": {
            "type": "object",
            "properties": {
                "task": {
                    "type": "string",
                    "description": "The analysis task, including any data or context to analyze."
                }
            },
            "required": ["task"]
        }
    },
    {
        "name": "writer_agent",
        "description": (
            "Writes polished, audience-appropriate content such as reports, briefs, "
            "summaries, or email copy. Use after research and analysis are done."
        ),
        "input_schema": {
            "type": "object",
            "properties": {
                "task": {
                    "type": "string",
                    "description": "Writing task with full context, tone, and audience details."
                }
            },
            "required": ["task"]
        }
    }
]

import os
from anthropic import Anthropic
from dotenv import load_dotenv

load_dotenv()

client = Anthropic(api_key=os.environ.get("ANTHROPIC_API_KEY"))

MODEL = "claude-sonnet-4-6"


class ResearcherAgent:
    """Gathers factual context on any topic."""

    SYSTEM_PROMPT = (
        "You are a research specialist. Your only job is to gather and summarize "
        "factual information clearly and concisely. Do not analyze or editorialize. "
        "Return structured findings with clear section headers."
    )

    def run(self, task: str) -> str:
        response = client.messages.create(
            model=MODEL,
            max_tokens=1024,
            system=self.SYSTEM_PROMPT,
            messages=[{"role": "user", "content": task}]
        )
        return response.content[0].text


class AnalystAgent:
    """Interprets research and produces insights or recommendations."""

    SYSTEM_PROMPT = (
        "You are a senior analyst. You receive research findings or raw data and "
        "produce clear, structured analysis: key trends, risks, opportunities, and "
        "a concise recommendation. Be direct and specific — no filler."
    )

    def run(self, task: str) -> str:
        response = client.messages.create(
            model=MODEL,
            max_tokens=1024,
            system=self.SYSTEM_PROMPT,
            messages=[{"role": "user", "content": task}]
        )
        return response.content[0].text


class WriterAgent:
    """Produces polished final content from analysis and context."""

    SYSTEM_PROMPT = (
        "You are a professional business writer. You turn research summaries and "
        "analysis into clean, polished written content. Match the tone and format "
        "specified in the task. No corporate jargon — write like a smart human."
    )

    def run(self, task: str) -> str:
        response = client.messages.create(
            model=MODEL,
            max_tokens=1024,
            system=self.SYSTEM_PROMPT,
            messages=[{"role": "user", "content": task}]
        )
        return response.content[0].text


# Registry maps tool names to agent instances — used by the Orchestrator
AGENT_REGISTRY = {
    "researcher_agent": ResearcherAgent(),
    "analyst_agent":    AnalystAgent(),
    "writer_agent":     WriterAgent(),
}

import os
import json
from anthropic import Anthropic
from dotenv import load_dotenv
from tools import AGENT_TOOLS, client
from agents import AGENT_REGISTRY

load_dotenv()

MODEL = "claude-sonnet-4-6"

ORCHESTRATOR_SYSTEM = """
You are a workflow orchestrator. You receive complex goals and break them into
subtasks that you delegate to specialized agents using the tools available to you.

Follow this general sequence unless the task clearly requires otherwise:
1. Use researcher_agent to gather factual background.
2. Use analyst_agent to interpret the research findings.
3. Use writer_agent to produce the final output.

Always pass full context to each agent — they have no memory of prior steps.
When all subtasks are complete, synthesize and return the final result to the user.
"""


class OrchestratorAgent:
    """
    Routes tasks to specialized sub-agents using Claude's tool use feature.
    Maintains a turn-by-turn conversation history so Claude can track progress.
    """

    def __init__(self):
        self.conversation: list[dict] = []

    def run(self, goal: str) -> str:
        print(f"\n🎯 Goal received: {goal}\n")

        # Seed the conversation with the user's goal
        self.conversation.append({"role": "user", "content": goal})

        # Delegation loop — runs until Claude stops calling tools
        turn = 0
        max_turns = 10  # Safety cap to prevent runaway loops

        while turn < max_turns:
            turn += 1
            print(f"--- Turn {turn} ---")

            response = client.messages.create(
                model=MODEL,
                max_tokens=4096,
                system=ORCHESTRATOR_SYSTEM,
                tools=AGENT_TOOLS,
                messages=self.conversation
            )

            print(f"Stop reason: {response.stop_reason}")

            # If Claude is done delegating, extract and return the final text
            if response.stop_reason == "end_turn":
                final_text = self._extract_text(response)
                print(f"\n✅ Orchestrator complete after {turn} turn(s).\n")
                return final_text

            # If Claude wants to call one or more tools, handle them
            if response.stop_reason == "tool_use":
                # Append Claude's full response (including tool_use blocks) to history
                self.conversation.append({
                    "role": "assistant",
                    "content": response.content
                })

                # Collect results for all tool calls in this turn
                tool_results = []
                for block in response.content:
                    if block.type == "tool_use":
                        result = self._dispatch_tool(block.name, block.input, block.id)
                        tool_results.append(result)

                # Return all tool results to Claude in a single user message
                self.conversation.append({
                    "role": "user",
                    "content": tool_results
                })

        return "Orchestration stopped: max turns reached without a final answer."

    def _dispatch_tool(self, tool_name: str, tool_input: dict, tool_use_id: str) -> dict:
        """Looks up the agent by tool name and runs it with the provided input."""
        print(f"  🔧 Calling tool: {tool_name}")
        print(f"     Input: {json.dumps(tool_input, indent=2)[:200]}")

        agent = AGENT_REGISTRY.get(tool_name)
        if agent is None:
            output = f"Error: No agent registered for tool '{tool_name}'"
        else:
            output = agent.run(tool_input["task"])

        print(f"     Output preview: {output[:150]}...\n")

        # tool_result must reference the tool_use_id from Claude's request
        return {
            "type": "tool_result",
            "tool_use_id": tool_use_id,
            "content": output
        }

    def _extract_text(self, response) -> str:
        """Pulls plain text out of Claude's final response content blocks."""
        parts = []
        for block in response.content:
            if hasattr(block, "text"):
                parts.append(block.text)
        return "\n".join(parts).strip()

import os
import sys
from anthropic import APIStatusError, APIConnectionError, RateLimitError
from dotenv import load_dotenv
from orchestrator import OrchestratorAgent

load_dotenv()


def run_workflow(goal: str) -> None:
    """Entry point for the multi-agent workflow."""
    orchestrator = OrchestratorAgent()

    try:
        result = orchestrator.run(goal)
        print("=" * 60)
        print("FINAL OUTPUT")
        print("=" * 60)
        print(result)

    except RateLimitError:
        print("Rate limit hit. Wait a moment and retry.")
        sys.exit(1)

    except APIConnectionError as e:
        print(f"Connection error: {e}. Check your network and API endpoint.")
        sys.exit(1)

    except APIStatusError as e:
        # Catches 400, 401, 500-level errors from the Anthropic API
        print(f"API error {e.status_code}: {e.message}")
        sys.exit(1)


if __name__ == "__main__":
    goal = (
        "Research the current Naples, Florida real estate market, "
        "analyze whether it's a good time for investors to buy, "
        "and write a concise 200-word investment brief for a potential buyer."
    )
    run_workflow(goal)

🎯 Goal received: Research the current Naples, Florida real estate market,
analyze whether it's a good time for investors to buy, and write a concise
200-word investment brief for a potential buyer.

--- Turn 1 ---
Stop reason: tool_use
  🔧 Calling tool: researcher_agent
     Input: {
  "task": "Research the current Naples, Florida real estate market including
  median home prices, inventory levels, days on market, and recent trends."
}
     Output preview: ## Naples, FL Real Estate Market — Research Summary

**Median Home Price:** ~$725,000 (Q1 2026)
**Active Listings:** Up 18% YoY, signaling improved inventory...

--- Turn 2 ---
Stop reason: tool_use
  🔧 Calling tool: analyst_agent
     Input: {
  "task": "Analyze these Naples FL market findings and assess whether conditions
  favor real estate investors in 2026: [research findings from turn 1]"
}
     Output preview: ## Investment Analysis — Naples FL (2026)

**Key Trends:** Inventory normalization after 2021-2023 seller's market peak.
Luxury segment ($1M+) showing 22% more days on market...

--- Turn 3 ---
Stop reason: tool_use
  🔧 Calling tool: writer_agent
     Input: {
  "task": "Write a 200-word investment brief for a potential real estate buyer
  in Naples, FL using this research and analysis: [combined context]"
}
     Output preview: **Naples, FL Real Estate Investment Brief — May 2026**

If you've been waiting for a better entry point into the Naples market, 2026
may be your window...

--- Turn 4 ---
Stop reason: end_turn

✅ Orchestrator complete after 4 turn(s).

============================================================
FINAL OUTPUT
============================================================
**Naples, FL Real Estate Investment Brief — May 2026**

If you've been waiting for a better entry point into the Naples market, 2026
may be your window. After the frenzied seller's market of 2021–2023, inventory
has climbed 18% year-over-year, giving buyers more leverage than they've had
in years. Median prices have stabilized near $725,000, with luxury properties
above $1M sitting longer — an opening for negotiation-savvy investors.

The fundamentals still favor long-term ownership. Naples continues to attract
high-net-worth relocations from the Northeast and Midwest, demand for rental
properties in the $3,000–$5,000/month range remains strong, and the area's
limited buildable land keeps supply naturally constrained.

Short-term risk: if interest rates climb above 7.5%, buyer demand could soften
further. Long-term outlook: positive. For investors with a 5-to-10-year
horizon and cash reserves to weather short dips, Naples remains one of
Florida's most resilient markets.

Recommended action: Target mid-range single-family homes ($600K–$900K) in
established neighborhoods like Pelican Bay or East Naples, where rental
yield and appreciation potential are both strong.