How to Build Multi-Agent Automation Systems with Claude API (Step-by-Step)

import os
import anthropic
from dotenv import load_dotenv

# Load environment variables from .env file
load_dotenv()

# Initialize the Anthropic client
client = anthropic.Anthropic(api_key=os.environ.get("ANTHROPIC_API_KEY"))

# Confirm the connection works with a simple ping
def test_connection():
    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=64,
        messages=[{"role": "user", "content": "Say 'connected' and nothing else."}]
    )
    print(response.content[0].text)

if __name__ == "__main__":
    test_connection()

import os
import anthropic
from dotenv import load_dotenv

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

# Tool definitions tell the orchestrator what agents are available
# and what arguments each agent expects
AGENT_TOOLS = [
    {
        "name": "research_agent",
        "description": (
            "Calls a specialized research agent to gather detailed factual information "
            "about a company, topic, or entity. Returns raw research notes."
        ),
        "input_schema": {
            "type": "object",
            "properties": {
                "query": {
                    "type": "string",
                    "description": "The specific research question or company name to investigate."
                },
                "depth": {
                    "type": "string",
                    "enum": ["brief", "standard", "deep"],
                    "description": "How thorough the research should be."
                }
            },
            "required": ["query", "depth"]
        }
    },
    {
        "name": "summarization_agent",
        "description": (
            "Calls a specialized summarization agent to distill raw research notes "
            "into a clean, structured business report."
        ),
        "input_schema": {
            "type": "object",
            "properties": {
                "raw_content": {
                    "type": "string",
                    "description": "The raw research notes or content to summarize."
                },
                "format": {
                    "type": "string",
                    "enum": ["bullet_points", "paragraph", "executive_summary"],
                    "description": "The desired output format for the summary."
                }
            },
            "required": ["raw_content", "format"]
        }
    }
]

if __name__ == "__main__":
    print(f"Loaded {len(AGENT_TOOLS)} agent tools:")
    for tool in AGENT_TOOLS:
        print(f"  - {tool['name']}: {tool['description'][:60]}...")

import os
import anthropic
from dotenv import load_dotenv
from tools import AGENT_TOOLS

load_dotenv()

class OrchestratorAgent:
    """
    Central coordination agent that delegates tasks to specialized worker agents.
    Uses Claude's tool use feature to route subtasks intelligently.
    """

    # Model is set once at the class level for easy updates
    MODEL = "claude-sonnet-4-6"
    MAX_TOKENS = 4096
    SYSTEM_PROMPT = """You are a senior research coordinator. Your job is to:
1. Receive a business research request from the user
2. Use the research_agent tool to gather raw information
3. Use the summarization_agent tool to turn that raw information into a clean report
4. Return the final formatted report to the user

Always call research_agent first, then summarization_agent with the results.
Be methodical and thorough. Do not skip steps."""

    def __init__(self):
        self.client = anthropic.Anthropic(api_key=os.environ.get("ANTHROPIC_API_KEY"))
        self.conversation_history = []

    def run(self, user_request: str) -> str:
        """
        Entry point for the orchestrator. Takes a plain-English request
        and returns a completed research report.
        """
        print(f"\n[Orchestrator] Received request: {user_request}")

        # Start the conversation with the user's request
        self.conversation_history = [
            {"role": "user", "content": user_request}
        ]

        # Hand off to the execution loop
        return self._execute()

    def _execute(self) -> str:
        """
        Internal execution method. Sends the conversation to Claude
        and returns the final text response after all tool calls resolve.
        """
        while True:
            response = self.client.messages.create(
                model=self.MODEL,
                max_tokens=self.MAX_TOKENS,
                system=self.SYSTEM_PROMPT,
                tools=AGENT_TOOLS,
                messages=self.conversation_history
            )

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

            # If Claude is done and has no more tool calls, return the final text
            if response.stop_reason == "end_turn":
                final_text = next(
                    (block.text for block in response.content if hasattr(block, "text")),
                    "No response generated."
                )
                return final_text

            # If Claude wants to use tools, process them
            if response.stop_reason == "tool_use":
                # Add Claude's response (with tool calls) to history
                self.conversation_history.append({
                    "role": "assistant",
                    "content": response.content
                })
                # Process tool calls and continue the loop
                self._handle_tool_calls(response.content)
                continue

            # Catch any unexpected stop reasons
            break

        return "Orchestration ended unexpectedly."

    def _handle_tool_calls(self, content_blocks):
        """
        Processes all tool_use blocks in Claude's response.
        Calls the appropriate worker agent for each tool request.
        """
        tool_results = []

        for block in content_blocks:
            if block.type == "tool_use":
                print(f"[Orchestrator] Delegating to: {block.name} | Input: {block.input}")
                result = self._dispatch_to_agent(block.name, block.input)

                tool_results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": result
                })

        # Add all tool results back to the conversation so Claude can continue
        self.conversation_history.append({
            "role": "user",
            "content": tool_results
        })

    def _dispatch_to_agent(self, agent_name: str, agent_input: dict) -> str:
        """
        Routes a tool call to the correct worker agent function.
        Returns the agent's response as a plain string.
        """
        if agent_name == "research_agent":
            from agents import run_research_agent
            return run_research_agent(agent_input["query"], agent_input["depth"])

        if agent_name == "summarization_agent":
            from agents import run_summarization_agent
            return run_summarization_agent(agent_input["raw_content"], agent_input["format"])

        return f"Error: Unknown agent '{agent_name}'"


if __name__ == "__main__":
    orchestrator = OrchestratorAgent()
    result = orchestrator.run("Research the company Anthropic and give me an executive summary.")
    print("\n" + "="*60)
    print("FINAL REPORT:")
    print("="*60)
    print(result)

import os
import anthropic
from dotenv import load_dotenv

load_dotenv()

# Each worker agent shares the same client but uses a different system prompt
_client = anthropic.Anthropic(api_key=os.environ.get("ANTHROPIC_API_KEY"))
MODEL = "claude-sonnet-4-6"


def _format_agent_message(role: str, content: str) -> dict:
    """
    Helper to format a message dict for the Claude messages API.
    Keeps message construction consistent across all agents.
    """
    return {"role": role, "content": content}


def _parse_agent_response(response) -> str:
    """
    Extracts plain text from a Claude API response object.
    Returns a fallback string if no text content is found.
    """
    for block in response.content:
        if hasattr(block, "text"):
            return block.text.strip()
    return "Agent returned no text content."


def run_research_agent(query: str, depth: str) -> str:
    """
    Specialized agent for gathering research on a topic.
    Depth controls how comprehensive the output is.
    """
    depth_instructions = {
        "brief": "Write 2-3 sentences covering only the most essential facts.",
        "standard": "Write 3-5 paragraphs covering background, products/services, and key facts.",
        "deep": "Write a comprehensive deep-dive covering history, products, leadership, financials, and market position."
    }

    system_prompt = f"""You are a professional business research analyst.
Your job is to gather and present factual information clearly and objectively.
{depth_instructions.get(depth, depth_instructions['standard'])}
Focus on accuracy. If you're uncertain about a fact, say so clearly."""

    # Format the outgoing message for this agent
    messages = [
        _format_agent_message("user", f"Research the following: {query}")
    ]

    print(f"  [Research Agent] Running query: '{query}' at depth: {depth}")

    response = _client.messages.create(
        model=MODEL,
        max_tokens=2048,
        system=system_prompt,
        messages=messages
    )

    # Parse and return the response as a plain string
    result = _parse_agent_response(response)
    print(f"  [Research Agent] Returned {len(result)} characters of research.")
    return result


def run_summarization_agent(raw_content: str, format: str) -> str:
    """
    Specialized agent for condensing research into a structured report.
    Format controls the output style.
    """
    format_instructions = {
        "bullet_points": "Format the output as clean bullet points grouped by category.",
        "paragraph": "Format the output as flowing prose paragraphs.",
        "executive_summary": (
            "Format as a professional executive summary with sections: "
            "Overview, Key Facts, and Takeaways."
        )
    }

    system_prompt = f"""You are a professional business writer specializing in executive communications.
Your job is to take raw research notes and produce a polished, readable report.
{format_instructions.get(format, format_instructions['executive_summary'])}
Be concise. Remove redundancy. Keep every sentence purposeful."""

    # Format the incoming raw content as the user message
    messages = [
        _format_agent_message(
            "user",
            f"Summarize the following research content:\n\n{raw_content}"
        )
    ]

    print(f"  [Summarization Agent] Summarizing {len(raw_content)} characters in '{format}' format.")

    response = _client.messages.create(
        model=MODEL,
        max_tokens=1024,
        system=system_prompt,
        messages=messages
    )

    result = _parse_agent_response(response)
    print(f"  [Summarization Agent] Returned {len(result)} characters.")
    return result


if __name__ == "__main__":
    # Quick standalone test for each agent
    research = run_research_agent("Anthropic AI company", "standard")
    print("\n--- Research Output ---")
    print(research[:300], "...")

    summary = run_summarization_agent(research, "executive_summary")
    print("\n--- Summary Output ---")
    print(summary)

import os
from dotenv import load_dotenv
from orchestrator import OrchestratorAgent

load_dotenv()

def main():
    """
    Main entry point for the multi-agent research system.
    Demonstrates the full orchestration loop with tool use.
    """
    print("="*60)
    print("Naples AI — Multi-Agent Research System")
    print("="*60)

    # Initialize the orchestrator
    orchestrator = OrchestratorAgent()

    # Define the tasks to run — swap these for real-world use cases
    tasks = [
        "Research the company Anthropic and give me an executive summary report.",
    ]

    for i, task in enumerate(tasks, 1):
        print(f"\n[Task {i}] {task}")
        print("-" * 60)

        result = orchestrator.run(task)

        print("\n" + "="*60)
        print(f"COMPLETED REPORT — Task {i}")
        print("="*60)
        print(result)
        print()

if __name__ == "__main__":
    main()

============================================================
Naples AI — Multi-Agent Research System
============================================================

[Task 1] Research the company Anthropic and give me an executive summary report.
------------------------------------------------------------

[Orchestrator] Received request: Research the company Anthropic...
[Orchestrator] Stop reason: tool_use
[Orchestrator] Delegating to: research_agent | Input: {'query': 'Anthropic AI company', 'depth': 'standard'}
  [Research Agent] Running query: 'Anthropic AI company' at depth: standard
  [Research Agent] Returned 1842 characters of research.
[Orchestrator] Stop reason: tool_use
[Orchestrator] Delegating to: summarization_agent | Input: {'format': 'executive_summary', 'raw_content': '...'}
  [Summarization Agent] Summarizing 1842 characters in 'executive_summary' format.
  [Summarization Agent] Returned 612 characters.
[Orchestrator] Stop reason: end_turn

============================================================
COMPLETED REPORT — Task 1
============================================================
**Executive Summary: Anthropic**

**Overview**
Anthropic is an AI safety company founded in 2021 by former OpenAI researchers...

**Key Facts**
- Founded by Dario Amodei, Daniela Amodei, and others from OpenAI
- Flagship model: Claude, optimized for safety and helpfulness
- Raised over $7 billion in funding as of 2024
- Headquartered in San Francisco, CA

**Takeaways**
Anthropic is a leading safety-focused AI lab competing directly with OpenAI and Google DeepMind...

anthropic.BadRequestError: tool_use_id 'toolu_01XYZ' does not match any tool use block

# Wrong — hardcoded ID will not match
tool_results.append({"type": "tool_result", "tool_use_id": "my-custom-id", "content": result})

# Correct — use the ID from the block Claude returned
tool_results.append({"type": "tool_result", "tool_use_id": block.id, "content": result})

anthropic.BadRequestError: messages: roles must alternate between "user" and "assistant"