Introduction

In the evolving landscape of AI agents, single-agent systems often hit limits when handling multifaceted, distributed tasks. Enter hierarchical multi-agent systems (HMAS): structured teams where a supervisor agent orchestrates specialized worker agents. Paired with Claude's powerful reasoning and MCP (Model Context Protocol) servers, HMAS enable scalable, efficient workflows.

This guide dives deep into building HMAS using Claude Opus/Sonnet models via the Anthropic API and MCP servers for seamless inter-agent communication and persistent context. We'll cover design principles, implementation, and a real-world example: an automated market analysis pipeline. By the end, you'll have actionable code to deploy production-ready agent teams.

Why Hierarchical Multi-Agent Systems with Claude and MCP?

• Scalability: Decompose complex tasks into parallelizable subtasks handled by specialized agents.
• Efficiency: Claude's constitutional AI ensures reliable delegation; MCP servers provide low-latency context sharing without token bloat.
• Modularity: Swap agents easily for tasks like HR onboarding, sales lead gen, or engineering code reviews.
• Claude-Specific Advantages: Superior long-context handling (200K+ tokens) and tool-use make Claude ideal for orchestration.

MCP servers act as a lightweight protocol layer, allowing agents to publish/subscribe to shared contexts (e.g., JSON states, files) via WebSockets or HTTP. This extends Claude beyond single-session limits, enabling true multi-agent collaboration.

Word count so far: ~150

Prerequisites

• Anthropic API key (Opus recommended for supervisor, Sonnet/Haiku for workers).
• Python 3.10+.
• MCP server library (hypothetical; use pip install mcp-server or implement via FastAPI/WebSockets).
• Redis for shared state (optional, for production scaling).

Install dependencies:

pip install anthropic fastapi uvicorn websockets redis

Set env vars:

export ANTHROPIC_API_KEY=your_key_here

Word count: ~250

Step 1: Understanding and Setting Up MCP Servers

MCP defines a simple protocol:

• Publish: Agent posts context (JSON) to /publish/{topic}.
• Subscribe: Agent pulls latest from /subscribe/{topic}.
• Broadcast: Notify all subscribers.

Here's a basic MCP server using FastAPI:

# mcp_server.py
from fastapi import FastAPI, WebSocket
from typing import Dict, Any

app = FastAPI()
contexts: Dict[str, Dict[str, Any]] = {}

@app.post("/publish/{topic}")
def publish(topic: str, data: Dict[str, Any]):
    contexts[topic] = data
    return {"status": "published"}

@app.websocket("/ws/{topic}")
async def subscribe(websocket: WebSocket, topic: str):
    await websocket.accept()
    while True:
        if topic in contexts:
            await websocket.send_json(contexts[topic])
        await asyncio.sleep(1)

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

Run multiple instances for different agents: uvicorn mcp_server:app --port 8001 etc.

Pro Tip: Use Docker Compose for multi-MCP clusters.

Word count: ~450

Step 2: Defining the Agent Hierarchy

Design a 3-tier structure:

1. Supervisor (Claude Opus): Task decomposer, router, aggregator.
2. Worker Agents (Claude Sonnet/Haiku): Specialists (e.g., Researcher, Analyzer).
3. Leaf Tools: MCP-mediated external calls (e.g., web search).

Hierarchy flow:

• Supervisor receives task.
• Delegates via MCP publish to worker topics.
• Workers process, publish results.
• Supervisor subscribes, synthesizes.

Step 3: Implementing the Base Agent Class

import anthropic
import aiohttp
import asyncio
import json

from typing import Dict, Any, List

class ClaudeAgent:
    def __init__(self, name: str, model: str = "claude-3-5-sonnet-20240620", mcp_url: str = "ws://localhost:8000/ws"):
        self.client = anthropic.Anthropic()
        self.name = name
        self.model = model
        self.mcp_url = mcp_url

    async def run(self, task: str, context: Dict[str, Any] = None) -> Dict[str, Any]:
        # Fetch context from MCP
        context = await self._fetch_context(task)
        
        msg = [{"role": "user", "content": f"Task: {task}. Context: {json.dumps(context)}. Respond as {self.name}."}]
        
        response = self.client.messages.create(
            model=self.model,
            max_tokens=2000,
            messages=msg,
            temperature=0.1
        )
        result = {"output": response.content[0].text, "metadata": {}}
        
        # Publish back to MCP
        await self._publish_result(task, result)
        return result

    async def _fetch_context(self, topic: str) -> Dict[str, Any]:
        async with aiohttp.ClientSession() as session:
            async with session.get(f"http://localhost:8000/subscribe/{topic}") as resp:
                return await resp.json() if resp.status == 200 else {}

    async def _publish_result(self, topic: str, result: Dict[str, Any]):
        async with aiohttp.ClientSession() as session:
            await session.post(f"http://localhost:8000/publish/{topic}", json=result)

Word count: ~750

Step 4: Building the Supervisor Agent

Extend for orchestration:

class SupervisorAgent(ClaudeAgent):
    def __init__(self):
        super().__init__("supervisor", "claude-3-opus-20240229")
        self.workers = {
            "research": ClaudeAgent("researcher"),
            "analyze": ClaudeAgent("analyzer"),
            "summarize": ClaudeAgent("summarizer", "claude-3-haiku-20240307")
        }

    async def orchestrate(self, goal: str) -> str:
        # Step 1: Decompose
        decomp = await self.run(f"Decompose goal into subtasks: {goal}")
        subtasks = decomp["output"].split("\
")  # Parse list

        results = []
        for subtask in subtasks[:3]:  # Limit for demo
            worker_type = self._route(subtask)
            topic = f"task-{hash(subtask) % 10000}"
            await self._publish_to_worker(topic, subtask, worker_type)
            
            # Wait and fetch
            worker_result = await self._wait_for_result(topic)
            results.append(worker_result)

        # Aggregate
        agg_prompt = f"Synthesize: {goal}. Subresults: {json.dumps(results)}"
        final = await self.run(agg_prompt)
        return final["output"]

    def _route(self, subtask: str) -> str:
        # Simple routing logic
        if "research" in subtask.lower(): return "research"
        # ...
        return "analyze"

    async def _publish_to_worker(self, topic: str, task: str, worker: str):
        data = {"task": task, "worker": worker}
        async with aiohttp.ClientSession() as session:
            await session.post(f"http://localhost:8000/publish/{topic}", json=data)

    async def _wait_for_result(self, topic: str, timeout=60) -> Dict:
        for _ in range(timeout):
            result = await self._fetch_context(topic)
            if result.get("status") == "done":
                return result
            await asyncio.sleep(5)
        raise TimeoutError("Worker timeout")

Word count: ~1100

Step 5: Real-World Example - Market Analysis Pipeline

Goal: Analyze competitor pricing for a SaaS product.

async def main():
    supervisor = SupervisorAgent()
    goal = "Analyze top 3 competitors' pricing for AI coding tools (Claude Code vs GitHub Copilot, Cursor). Recommend strategy."
    result = await supervisor.orchestrate(goal)
    print(result)

asyncio.run(main())

Expected Flow:

1. Supervisor decomposes: Research competitors, Analyze features/pricing, Summarize insights.
2. Researcher publishes scraped data to MCP topic pricing-research.
3. Analyzer pulls, computes deltas.
4. Summarizer crafts report.

Enhance workers with tools: Add tools=[web_search] via Anthropic's tool-use.

Output example:

Competitor Analysis

• Copilot: $10/user/mo
• Cursor: $20/user/mo
• Claude Code: Free tier + API Recommendation: Price at $15 with Opus integration.

Word count: ~1300

Best Practices and Scaling

• Error Handling: Wrap runs in try/except; supervisor retries failed workers.
• Rate Limits: Use async queues; Sonnet/Haiku for cost-efficiency.
• Persistence: Integrate Redis pub/sub with MCP for distributed deploys.
• Monitoring: Log to MCP /metrics topic; use Prometheus.
• Security: API keys per agent; MCP auth via JWTs.
• Optimization: Prompt engineering - Use XML tags for structured outputs.

# Structured output prompt example
SYSTEM_PROMPT = """<supervisor>
Respond in <plan>subtasks</plan><delegate>assignments</delegate> format."""

For enterprise: Deploy on Kubernetes; integrate with n8n for no-code triggers.

Common Pitfalls:

• Over-delegation: Limit to 5 subtasks.
• Context Loss: MCP TTL=300s.

Word count: ~1500

Conclusion

Hierarchical multi-agent systems with Claude and MCP unlock enterprise-grade AI orchestration. Start with this scaffold, iterate on your domain (e.g., Legal contract review: Parser > Reviewer > Drafter). Monitor Anthropic updates for agentic improvements like native multi-model support.

Experiment: Fork the code, add your workers. Share your builds in Claude Directory comments!

Total word count: ~1620