Building Multi-Agent Orchestration Systems with Claude Code

 1# Create project directory
 2mkdir claude-orchestration
 3cd claude-orchestration
 4
 5# Set up Python environment
 6python -m venv venv
 7source venv/bin/activate  # Windows: venv\Scripts\activate
 8
 9# Install dependencies
10pip install anthropic crewai langchain python-dotenv
11
12# Create project structure
13mkdir -p agents/{pm,design,backend,frontend} shared_state logs

 1"""
 2Configuration for multi-agent orchestration system
 3"""
 4
 5import os
 6from pathlib import Path
 7from dotenv import load_dotenv
 8
 9load_dotenv()
10
11class Config:
12    # API Configuration
13    ANTHROPIC_API_KEY = os.getenv("ANTHROPIC_API_KEY")
14    MODEL = "claude-sonnet-4.5"  # or claude-opus-4.5 for complex tasks
15
16    # Project paths
17    PROJECT_ROOT = Path(__file__).parent
18    AGENTS_DIR = PROJECT_ROOT / "agents"
19    SHARED_STATE_DIR = PROJECT_ROOT / "shared_state"
20    LOGS_DIR = PROJECT_ROOT / "logs"
21
22    # Agent configuration
23    AGENTS = {
24        "pm": {
25            "name": "Product Manager",
26            "prompt_file": AGENTS_DIR / "pm" / "prompt.txt",
27            "output_files": ["requirements.md", "tasks.json"],
28            "priority": 1
29        },
30        "design": {
31            "name": "Designer",
32            "prompt_file": AGENTS_DIR / "design" / "prompt.txt",
33            "output_files": ["design-spec.md", "wireframes.md"],
34            "priority": 2,
35            "depends_on": ["pm"]
36        },
37        "backend": {
38            "name": "Backend Engineer",
39            "prompt_file": AGENTS_DIR / "backend" / "prompt.txt",
40            "output_files": ["api-contract.json", "schema.sql"],
41            "priority": 2,
42            "depends_on": ["pm"]
43        },
44        "frontend": {
45            "name": "Frontend Engineer",
46            "prompt_file": AGENTS_DIR / "frontend" / "prompt.txt",
47            "output_files": ["components-plan.md", "state-management.md"],
48            "priority": 3,
49            "depends_on": ["design", "backend"]
50        }
51    }
52
53    # Orchestration settings
54    MAX_ITERATIONS = 5
55    PARALLEL_EXECUTION = True
56    ENABLE_VALIDATION = True
57
58    # Shared state configuration
59    STATE_FILE = SHARED_STATE_DIR / "state.json"
60    COMMUNICATION_LOG = LOGS_DIR / "communication.log"
61
62config = Config()

  1"""
  2Agent Manager - Handles individual agent execution
  3"""
  4
  5import json
  6import logging
  7from pathlib import Path
  8from typing import Dict, List, Optional, Any
  9from datetime import datetime
 10
 11from anthropic import Anthropic
 12from config import config
 13
 14logging.basicConfig(level=logging.INFO)
 15logger = logging.getLogger(__name__)
 16
 17
 18class Agent:
 19    """Represents a single specialized agent"""
 20
 21    def __init__(self, agent_id: str, agent_config: Dict[str, Any]):
 22        self.id = agent_id
 23        self.name = agent_config["name"]
 24        self.prompt_file = agent_config["prompt_file"]
 25        self.output_files = agent_config["output_files"]
 26        self.priority = agent_config.get("priority", 10)
 27        self.depends_on = agent_config.get("depends_on", [])
 28        self.client = Anthropic(api_key=config.ANTHROPIC_API_KEY)
 29
 30        # Load system prompt
 31        with open(self.prompt_file, 'r') as f:
 32            self.system_prompt = f.read()
 33
 34    def execute(self, context: Dict[str, Any]) -> Dict[str, Any]:
 35        """
 36        Execute agent with given context
 37
 38        Args:
 39            context: Dictionary containing user request and shared state
 40
 41        Returns:
 42            Dictionary with execution results and outputs
 43        """
 44        logger.info(f"🚀 Executing {self.name} agent...")
 45
 46        # Build user message with context
 47        user_message = self._build_user_message(context)
 48
 49        # Call Claude API
 50        try:
 51            response = self.client.messages.create(
 52                model=config.MODEL,
 53                max_tokens=4096,
 54                system=self.system_prompt,
 55                messages=[
 56                    {"role": "user", "content": user_message}
 57                ]
 58            )
 59
 60            # Extract response content
 61            response_text = response.content[0].text
 62
 63            # Parse and save outputs
 64            outputs = self._process_outputs(response_text)
 65
 66            result = {
 67                "agent_id": self.id,
 68                "agent_name": self.name,
 69                "status": "success",
 70                "outputs": outputs,
 71                "timestamp": datetime.now().isoformat()
 72            }
 73
 74            logger.info(f"✅ {self.name} completed successfully")
 75            return result
 76
 77        except Exception as e:
 78            logger.error(f"❌ {self.name} failed: {str(e)}")
 79            return {
 80                "agent_id": self.id,
 81                "agent_name": self.name,
 82                "status": "error",
 83                "error": str(e),
 84                "timestamp": datetime.now().isoformat()
 85            }
 86
 87    def _build_user_message(self, context: Dict[str, Any]) -> str:
 88        """Build user message with all relevant context"""
 89
 90        message_parts = [
 91            "# Task Context",
 92            f"\n## User Request\n{context.get('user_request', 'No request provided')}",
 93        ]
 94
 95        # Include outputs from dependent agents
 96        if self.depends_on:
 97            message_parts.append("\n## Inputs from Other Agents")
 98            for dep_agent in self.depends_on:
 99                dep_outputs = context.get('agent_outputs', {}).get(dep_agent, {})
100                if dep_outputs:
101                    message_parts.append(f"\n### From {dep_agent} Agent:")
102                    for file_name, content in dep_outputs.items():
103                        message_parts.append(f"\n#### {file_name}\n```\n{content}\n```")
104
105        # Include shared state
106        shared_state = context.get('shared_state', {})
107        if shared_state:
108            message_parts.append("\n## Current Shared State")
109            message_parts.append(f"```json\n{json.dumps(shared_state, indent=2)}\n```")
110
111        return "\n".join(message_parts)
112
113    def _process_outputs(self, response_text: str) -> Dict[str, str]:
114        """Extract and save outputs from agent response"""
115
116        outputs = {}
117
118        # Simple parsing - look for file content in response
119        # In production, you'd want more sophisticated parsing
120        for output_file in self.output_files:
121            # Try to find content between markers
122            start_marker = f"# {output_file}"
123            if start_marker in response_text:
124                # Extract content after marker until next file or end
125                start_idx = response_text.find(start_marker)
126                content = response_text[start_idx:]
127
128                # Find end (next file marker or end of response)
129                next_file = None
130                for other_file in self.output_files:
131                    if other_file != output_file:
132                        next_marker = f"# {other_file}"
133                        if next_marker in content[len(start_marker):]:
134                            next_idx = content.find(next_marker, len(start_marker))
135                            content = content[:next_idx]
136                            break
137
138                # Save to shared state
139                output_path = config.SHARED_STATE_DIR / output_file
140                output_path.parent.mkdir(parents=True, exist_ok=True)
141
142                with open(output_path, 'w') as f:
143                    f.write(content)
144
145                outputs[output_file] = content
146                logger.info(f"  📄 Saved {output_file}")
147
148        return outputs
149
150
151class AgentManager:
152    """Manages multiple agents and their execution"""
153
154    def __init__(self):
155        self.agents: Dict[str, Agent] = {}
156        self._initialize_agents()
157
158    def _initialize_agents(self):
159        """Initialize all agents from configuration"""
160        for agent_id, agent_config in config.AGENTS.items():
161            self.agents[agent_id] = Agent(agent_id, agent_config)
162        logger.info(f"Initialized {len(self.agents)} agents")
163
164    def get_execution_order(self) -> List[List[str]]:
165        """
166        Determine execution order based on dependencies and priority
167        Returns list of lists - each inner list can execute in parallel
168        """
169        # Build dependency graph
170        remaining = set(self.agents.keys())
171        execution_order = []
172
173        while remaining:
174            # Find agents with all dependencies satisfied
175            ready = []
176            for agent_id in remaining:
177                agent = self.agents[agent_id]
178                if all(dep not in remaining for dep in agent.depends_on):
179                    ready.append(agent_id)
180
181            if not ready:
182                raise ValueError("Circular dependency detected in agent configuration")
183
184            # Sort by priority
185            ready.sort(key=lambda x: self.agents[x].priority)
186
187            # Group by priority for parallel execution
188            if config.PARALLEL_EXECUTION:
189                priority_groups = {}
190                for agent_id in ready:
191                    priority = self.agents[agent_id].priority
192                    if priority not in priority_groups:
193                        priority_groups[priority] = []
194                    priority_groups[priority].append(agent_id)
195
196                for priority in sorted(priority_groups.keys()):
197                    execution_order.append(priority_groups[priority])
198                    remaining -= set(priority_groups[priority])
199            else:
200                for agent_id in ready:
201                    execution_order.append([agent_id])
202                    remaining.remove(agent_id)
203
204        return execution_order
205
206    def execute_agents(self, user_request: str) -> Dict[str, Any]:
207        """
208        Execute all agents in proper order
209
210        Args:
211            user_request: The user's task description
212
213        Returns:
214            Dictionary with all agent results
215        """
216        logger.info("=" * 60)
217        logger.info("🎯 Starting Multi-Agent Orchestration")
218        logger.info("=" * 60)
219
220        # Initialize context
221        context = {
222            "user_request": user_request,
223            "agent_outputs": {},
224            "shared_state": self._load_shared_state()
225        }
226
227        # Get execution order
228        execution_order = self.get_execution_order()
229        logger.info(f"\n📋 Execution Plan: {execution_order}\n")
230
231        # Execute agents
232        all_results = {}
233
234        for wave_idx, agent_wave in enumerate(execution_order, 1):
235            logger.info(f"\n🌊 Wave {wave_idx}: {', '.join(agent_wave)}")
236
237            wave_results = {}
238
239            # Execute agents in this wave (can be parallel)
240            for agent_id in agent_wave:
241                agent = self.agents[agent_id]
242                result = agent.execute(context)
243                wave_results[agent_id] = result
244                all_results[agent_id] = result
245
246                # Update context with outputs
247                if result["status"] == "success":
248                    context["agent_outputs"][agent_id] = result["outputs"]
249
250            # Update shared state after each wave
251            self._save_shared_state(context["shared_state"])
252
253        logger.info("\n" + "=" * 60)
254        logger.info("✨ Multi-Agent Orchestration Complete")
255        logger.info("=" * 60 + "\n")
256
257        return {
258            "status": "success",
259            "results": all_results,
260            "final_state": context["shared_state"]
261        }
262
263    def _load_shared_state(self) -> Dict[str, Any]:
264        """Load shared state from disk"""
265        if config.STATE_FILE.exists():
266            with open(config.STATE_FILE, 'r') as f:
267                return json.load(f)
268        return {}
269
270    def _save_shared_state(self, state: Dict[str, Any]):
271        """Save shared state to disk"""
272        config.STATE_FILE.parent.mkdir(parents=True, exist_ok=True)
273        with open(config.STATE_FILE, 'w') as f:
274            json.dump(state, indent=2, fp=f)

  1#!/usr/bin/env python3
  2"""
  3Main Orchestrator - Coordinates multi-agent execution
  4"""
  5
  6import json
  7import logging
  8import argparse
  9from pathlib import Path
 10from typing import Dict, Any
 11
 12from agent_manager import AgentManager
 13from config import config
 14
 15logging.basicConfig(
 16    level=logging.INFO,
 17    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
 18)
 19logger = logging.getLogger(__name__)
 20
 21
 22class Orchestrator:
 23    """Central orchestrator for multi-agent system"""
 24
 25    def __init__(self):
 26        self.agent_manager = AgentManager()
 27        self.execution_history = []
 28
 29    def run(self, user_request: str) -> Dict[str, Any]:
 30        """
 31        Execute multi-agent workflow
 32
 33        Args:
 34            user_request: User's task description
 35
 36        Returns:
 37            Final orchestration result
 38        """
 39        logger.info("\n" + "=" * 80)
 40        logger.info("🚀 MULTI-AGENT ORCHESTRATION SYSTEM")
 41        logger.info("=" * 80)
 42        logger.info(f"\n📝 User Request:\n{user_request}\n")
 43
 44        # Execute agents
 45        result = self.agent_manager.execute_agents(user_request)
 46
 47        # Record execution
 48        self.execution_history.append({
 49            "request": user_request,
 50            "result": result
 51        })
 52
 53        # Generate summary
 54        self._print_summary(result)
 55
 56        return result
 57
 58    def _print_summary(self, result: Dict[str, Any]):
 59        """Print execution summary"""
 60        logger.info("\n" + "=" * 80)
 61        logger.info("📊 EXECUTION SUMMARY")
 62        logger.info("=" * 80)
 63
 64        for agent_id, agent_result in result["results"].items():
 65            status_icon = "✅" if agent_result["status"] == "success" else "❌"
 66            logger.info(f"{status_icon} {agent_result['agent_name']}: {agent_result['status']}")
 67
 68            if agent_result["status"] == "success":
 69                outputs = agent_result.get("outputs", {})
 70                for file_name in outputs.keys():
 71                    logger.info(f"   📄 {file_name}")
 72            else:
 73                logger.error(f"   ⚠️  Error: {agent_result.get('error', 'Unknown error')}")
 74
 75        logger.info("\n" + "=" * 80)
 76        logger.info(f"📁 Outputs saved to: {config.SHARED_STATE_DIR}")
 77        logger.info("=" * 80 + "\n")
 78
 79    def validate_outputs(self) -> bool:
 80        """Validate that all expected outputs were generated"""
 81        if not config.ENABLE_VALIDATION:
 82            return True
 83
 84        logger.info("\n🔍 Validating outputs...")
 85
 86        all_valid = True
 87        for agent_id, agent_config in config.AGENTS.items():
 88            for output_file in agent_config["output_files"]:
 89                output_path = config.SHARED_STATE_DIR / output_file
 90                if not output_path.exists():
 91                    logger.error(f"❌ Missing output: {output_file} from {agent_id}")
 92                    all_valid = False
 93                else:
 94                    logger.info(f"✅ Found: {output_file}")
 95
 96        return all_valid
 97
 98
 99def main():
100    """Main entry point"""
101    parser = argparse.ArgumentParser(
102        description="Multi-Agent Orchestration System with Claude Code"
103    )
104    parser.add_argument(
105        "request",
106        nargs="?",
107        help="User request/task description"
108    )
109    parser.add_argument(
110        "--file",
111        "-f",
112        help="Read request from file"
113    )
114    parser.add_argument(
115        "--validate",
116        action="store_true",
117        help="Validate outputs after execution"
118    )
119
120    args = parser.parse_args()
121
122    # Get user request
123    if args.file:
124        with open(args.file, 'r') as f:
125            user_request = f.read()
126    elif args.request:
127        user_request = args.request
128    else:
129        # Interactive mode
130        print("\n🤖 Multi-Agent Orchestration System")
131        print("=" * 50)
132        print("Enter your task description (press Enter twice to finish):\n")
133
134        lines = []
135        empty_count = 0
136        while empty_count < 2:
137            line = input()
138            if line:
139                lines.append(line)
140                empty_count = 0
141            else:
142                empty_count += 1
143
144        user_request = "\n".join(lines)
145
146    if not user_request.strip():
147        print("❌ No request provided. Exiting.")
148        return
149
150    # Run orchestration
151    orchestrator = Orchestrator()
152    result = orchestrator.run(user_request)
153
154    # Validate if requested
155    if args.validate:
156        if orchestrator.validate_outputs():
157            print("\n✅ All outputs validated successfully!")
158        else:
159            print("\n⚠️  Some outputs are missing. Check logs for details.")
160
161    # Save execution log
162    log_file = config.LOGS_DIR / f"execution_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
163    log_file.parent.mkdir(parents=True, exist_ok=True)
164    with open(log_file, 'w') as f:
165        json.dump(result, f, indent=2)
166
167    print(f"\n📋 Execution log saved to: {log_file}")
168
169
170if __name__ == "__main__":
171    from datetime import datetime
172    main()

1ANTHROPIC_API_KEY=your_api_key_here

1# Interactive mode
2python orchestrator.py
3
4# Direct command
5python orchestrator.py "Build a todo list application with user authentication"
6
7# From file
8python orchestrator.py --file task_description.txt --validate

1python orchestrator.py --file task_description.txt --validate

 1# agents/validator/prompt.txt
 2You are a Validation Agent that reviews outputs from other agents.
 3
 4## Your Role
 5- Check requirements for completeness and clarity
 6- Validate API contracts against requirements
 7- Ensure design specs match requirements
 8- Verify frontend plan implements all features
 9
10## Input
11All outputs from PM, Design, Backend, and Frontend agents
12
13## Output
14validation-report.md with:
15- Issues found (missing requirements, inconsistencies, etc.)
16- Severity (critical, major, minor)
17- Recommendations for fixes
18
19If critical issues found, flag for re-execution.

 1class Orchestrator:
 2    def run_with_feedback(self, user_request: str, max_iterations: int = 3):
 3        """Execute with validation and feedback loops"""
 4
 5        for iteration in range(max_iterations):
 6            logger.info(f"\n🔄 Iteration {iteration + 1}/{max_iterations}")
 7
 8            # Execute agents
 9            result = self.agent_manager.execute_agents(user_request)
10
11            # Validate
12            validator = Validator()
13            validation_result = validator.validate(result)
14
15            if validation_result["status"] == "pass":
16                logger.info("✅ Validation passed!")
17                return result
18
19            # Provide feedback for next iteration
20            user_request = self._generate_feedback_request(
21                user_request,
22                validation_result
23            )
24
25        logger.warning("⚠️  Max iterations reached")
26        return result

 1PM_AGENT_PROMPT = """
 2You define WHAT to build, not HOW.
 3Focus on requirements, acceptance criteria, and priorities.
 4Do not write code or make technical decisions.
 5"""
 6
 7BACKEND_AGENT_PROMPT = """
 8You design HOW to implement the backend.
 9Focus on API contracts, database schemas, and architecture.
10Do not define UI or user stories.
11"""

1GENERIC_AGENT_PROMPT = """
2You help with software development.
3Do whatever is needed.
4"""

1api-contract.json must follow OpenAPI 3.0:
2{
3  "openapi": "3.0.0",
4  "info": { ... },
5  "paths": { ... },
6  "components": { ... }
7}

1AGENTS = {
2    "pm": {
3        "depends_on": []  # No dependencies
4    },
5    "frontend": {
6        "depends_on": ["pm", "design", "backend"]  # Clear dependencies
7    }
8}

1def execute(self, context):
2    try:
3        return self._execute_with_retry(context, max_retries=3)
4    except Exception as e:
5        return {
6            "status": "partial",
7            "error": str(e),
8            "fallback_outputs": self._generate_fallback()
9        }

1logging.basicConfig(
2    level=logging.INFO,
3    format='%(asctime)s - [%(name)s] - %(levelname)s - %(message)s',
4    handlers=[
5        logging.FileHandler('logs/orchestration.log'),
6        logging.StreamHandler()
7    ]
8)

 1class PerformanceTracker:
 2    def track_agent_execution(self, agent_id: str, duration: float):
 3        """Track agent performance metrics"""
 4        metrics = {
 5            "agent_id": agent_id,
 6            "duration_seconds": duration,
 7            "tokens_used": self._count_tokens(),
 8            "timestamp": datetime.now().isoformat()
 9        }
10        self._save_metrics(metrics)

1# Use Claude Code to run orchestration
2claude-code "Orchestrate my multi-agent system to build a blog platform"

1# Create an MCP tool for orchestration
2@server.call_tool()
3async def orchestrate_agents(name: str, arguments: Dict):
4    if name == "run_orchestration":
5        orchestrator = Orchestrator()
6        result = orchestrator.run(arguments["task"])
7        return [TextContent(type="text", text=json.dumps(result))]