Multi-Agent Collaboration Systems: The Core Architecture Paradigm for Enterprise AI Applications in 2026

# Python Example: Building an API Gateway for Multi-Agent Systems Using FastAPI
from fastapi import FastAPI, HTTPException, Depends
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from pydantic import BaseModel
from typing import Optional, List, Dict, Any
import asyncio
from datetime import datetime
import hashlib

app = FastAPI(title="Multi-Agent Collaboration System API")
security = HTTPBearer()

# Request Models
class AgentRequest(BaseModel):
    session_id: str
    message: str
    context: Optional[Dict[str, Any]] = None
    agent_types: Optional[List[str]] = None

class AgentResponse(BaseModel):
    session_id: str
    message: str
    agent_id: str
    timestamp: str
    metadata: Optional[Dict[str, Any]] = None

# In-memory session storage (use Redis in production)
sessions: Dict[str, Dict[str, Any]] = {}

# Authentication dependency
async def verify_token(credentials: HTTPAuthorizationCredentials = Depends(security)):
    token = credentials.credentials
    if not token.startswith("Bearer "):
        raise HTTPException(status_code=401, detail="Invalid token format")
    return {"user_id": hashlib.md5(token.encode()).hexdigest()[:8]}

@app.post("/api/v1/agent/invoke", response_model=AgentResponse)
async def invoke_agent(
    request: AgentRequest,
    auth: dict = Depends(verify_token)
):
    """Invoke the multi-Agent system to process user requests"""
    session = sessions.get(request.session_id)
    if session is None:
        session = {
            "created_at": datetime.now().isoformat(),
            "history": [],
            "agent_states": {}
        }
        sessions[request.session_id] = session
    
    session["history"].append({
        "role": "user",
        "content": request.message,
        "timestamp": datetime.now().isoformat()
    })
    
    response = AgentResponse(
        session_id=request.session_id,
        message=f"Processed by Multi-Agent System: {request.message}",
        agent_id="orchestrator",
        timestamp=datetime.now().isoformat(),
        metadata={
            "agent_count": 3,
            "processing_time_ms": 150
        }
    )
    
    session["history"].append({
        "role": "assistant",
        "content": response.message,
        "timestamp": response.timestamp
    })
    
    return response

@app.get("/api/v1/session/{session_id}/history")
async def get_session_history(
    session_id: str,
    auth: dict = Depends(verify_token)
):
    """Get session history"""
    session = sessions.get(session_id)
    if session is None:
        raise HTTPException(status_code=404, detail="Session not found")
    return session["history"]

@app.delete("/api/v1/session/{session_id}")
async def delete_session(
    session_id: str,
    auth: dict = Depends(verify_token)
):
    """Delete session"""
    if session_id in sessions:
        del sessions[session_id]
    return {"status": "deleted", "session_id": session_id}

# Python Example: Task Planner Implementation
from dataclasses import dataclass, field
from typing import List, Dict, Any, Optional, Set
from enum import Enum
import json

class TaskStatus(Enum):
    PENDING = "pending"
    IN_PROGRESS = "in_progress"
    COMPLETED = "completed"
    FAILED = "failed"
    BLOCKED = "blocked"

@dataclass
class Task:
    id: str
    name: str
    description: str
    required_skills: List[str]
    estimated_duration: int  # seconds
    dependencies: List[str] = field(default_factory=list)
    status: TaskStatus = TaskStatus.PENDING
    assigned_agent: Optional[str] = None
    result: Optional[Any] = None
    error: Optional[str] = None

@dataclass
class ExecutionPlan:
    tasks: List[Task]
    parallel_groups: List[List[str]]
    total_duration: int
    critical_path: List[str]

class TaskPlanner:
    """Task Planner: Responsible for decomposing complex tasks into executable subtasks"""
    
    def __init__(self):
        self.task_templates = {
            "code_development": [
                Task(id="req_analysis", name="Requirements Analysis",
                     description="Analyze user requirements, output requirements specification",
                     required_skills=["analysis", "documentation"], estimated_duration=300),
                Task(id="code_design", name="Architecture Design",
                     description="Design system architecture and data models",
                     required_skills=["architecture_design", "UML"], estimated_duration=600,
                     dependencies=["req_analysis"]),
                Task(id="code_write", name="Code Writing",
                     description="Write code according to design documents",
                     required_skills=["programming"], estimated_duration=1800,
                     dependencies=["code_design"]),
                Task(id="code_review", name="Code Review",
                     description="Review code quality and security",
                     required_skills=["code_review"], estimated_duration=600,
                     dependencies=["code_write"]),
                Task(id="unit_test", name="Unit Testing",
                     description="Write and execute unit tests",
                     required_skills=["testing"], estimated_duration=900,
                     dependencies=["code_write"]),
                Task(id="integration_test", name="Integration Testing",
                     description="Execute integration tests",
                     required_skills=["testing"], estimated_duration=600,
                     dependencies=["code_review", "unit_test"]),
            ],
            "data_analysis": [
                Task(id="data_collect", name="Data Collection",
                     description="Collect data from multiple sources",
                     required_skills=["data_collection"], estimated_duration=300),
                Task(id="data_clean", name="Data Cleaning",
                     description="Clean and preprocess data",
                     required_skills=["data_processing"], estimated_duration=600,
                     dependencies=["data_collect"]),
                Task(id="data_explore", name="Exploratory Analysis",
                     description="Perform exploratory data analysis",
                     required_skills=["analysis"], estimated_duration=600,
                     dependencies=["data_clean"]),
                Task(id="model_build", name="Modeling Analysis",
                     description="Build analysis models",
                     required_skills=["modeling"], estimated_duration=1200,
                     dependencies=["data_explore"]),
                Task(id="report_gen", name="Report Generation",
                     description="Generate analysis reports and visualizations",
                     required_skills=["visualization", "documentation"], estimated_duration=600,
                     dependencies=["model_build"]),
            ]
        }
    
    def parse_user_request(self, message: str) -> str:
        """Infer task type based on user message"""
        message_lower = message.lower()
        
        if any(keyword in message_lower for keyword in ["develop", "write", "create", "implement", "code"]):
            return "code_development"
        elif any(keyword in message_lower for keyword in ["analyze", "data", "report", "statistics", "mining"]):
            return "data_analysis"
        elif any(keyword in message_lower for keyword in ["search", "find", "query", "research"]):
            return "research"
        else:
            return "general"
    
    def create_execution_plan(self, task_type: str, context: Optional[Dict] = None) -> ExecutionPlan:
        """Create execution plan based on task type"""
        if task_type not in self.task_templates:
            tasks = [Task(
                id="general_task",
                name="General Task",
                description="Process user request",
                required_skills=["general"],
                estimated_duration=300
            )]
            return ExecutionPlan(
                tasks=tasks,
                parallel_groups=[["general_task"]],
                total_duration=300,
                critical_path=["general_task"]
            )
        
        tasks = self.task_templates[task_type]
        
        if context:
            for task in tasks:
                pass
        
        parallel_groups = self._compute_parallel_groups(tasks)
        critical_path = self._compute_critical_path(tasks)
        total_duration = sum(t.estimated_duration for t in critical_path)
        
        return ExecutionPlan(
            tasks=tasks,
            parallel_groups=parallel_groups,
            total_duration=total_duration,
            critical_path=critical_path
        )
    
    def _compute_parallel_groups(self, tasks: List[Task]) -> List[List[str]]:
        """Compute groups of tasks that can be executed in parallel"""
        groups = []
        remaining = set(t.id for t in tasks)
        completed = set()
        
        while remaining:
            current_group = []
            for task_id in list(remaining):
                task = next(t for t in tasks if t.id == task_id)
                if all(dep in completed for dep in task.dependencies):
                    current_group.append(task_id)
            
            if not current_group:
                current_group = [min(remaining, key=lambda x: len(
                    next(t for t in tasks if t.id == x).dependencies
                ))]
            
            groups.append(current_group)
            completed.update(current_group)
            remaining -= set(current_group)
        
        return groups
    
    def _compute_critical_path(self, tasks: List[Task]) -> List[str]:
        """Compute critical path (longest dependency chain)"""
        critical = []
        remaining = set(t.id for t in tasks)
        completed = {}
        
        while remaining:
            for task_id in list(remaining):
                task = next(t for t in tasks if t.id == task_id)
                if all(dep in completed for dep in task.dependencies):
                    if not critical or task.dependencies:
                        critical.append(task_id)
                    completed[task_id] = task
                    remaining.remove(task_id)
                    break
        
        return critical

# Usage Example
planner = TaskPlanner()
plan = planner.create_execution_plan("code_development")
print(f"Execution Plan:")
print(f"  Total Duration: {plan.total_duration}s")
print(f"  Task Count: {len(plan.tasks)}")
print(f"  Critical Path: {' -> '.join(plan.critical_path)}")
print(f"  Parallel Groups: {plan.parallel_groups}")

// Go Example: Multi-Agent System Agent Definition and Scheduler
package main

import (
	"context"
	"encoding/json"
	"fmt"
	"sync"
	"time"
)

// AgentType defines the type of agent
type AgentType string

const (
	AgentTypePlanner  AgentType = "planner"
	AgentTypeCoder    AgentType = "coder"
	AgentTypeSearch   AgentType = "search"
	AgentTypeData     AgentType = "data"
	AgentTypeWriter   AgentType = "writer"
	AgentTypeReviewer AgentType = "reviewer"
)

// Task represents a task to be executed
type Task struct {
	ID              string                 `json:"id"`
	Name            string                 `json:"name"`
	RequiredAgent   AgentType              `json:"required_agent"`
	Input           map[string]interface{} `json:"input"`
	Dependencies    []string               `json:"dependencies,omitempty"`
	Status          string                 `json:"status"`
	Result          interface{}            `json:"result,omitempty"`
	Error           string                 `json:"error,omitempty"`
	AssignedAgentID string                 `json:"assigned_agent_id,omitempty"`
}

// AgentResult represents the result of agent execution
type AgentResult struct {
	TaskID   string        `json:"task_id"`
	AgentID  string        `json:"agent_id"`
	Success  bool          `json:"success"`
	Output   interface{}   `json:"output,omitempty"`
	Error    string        `json:"error,omitempty"`
	Duration time.Duration `json:"duration"`
}

// Agent interface for all agents
type Agent interface {
	GetType() AgentType
	GetCapabilities() []string
	Execute(ctx context.Context, task *Task) (*AgentResult, error)
}

// BaseAgent provides common functionality for all agents
type BaseAgent struct {
	ID           string
	AgentType    AgentType
	Capabilities []string
}

func (b *BaseAgent) GetType() AgentType {
	return b.AgentType
}

func (b *BaseAgent) GetCapabilities() []string {
	return b.Capabilities
}

// PlannerAgent handles task planning and decomposition
type PlannerAgent struct {
	BaseAgent
}

func NewPlannerAgent() *PlannerAgent {
	return &PlannerAgent{
		BaseAgent: BaseAgent{
			ID:           "planner-001",
			AgentType:    AgentTypePlanner,
			Capabilities: []string{"task_decomposition", "dependency_analysis", "resource_planning"},
		},
	}
}

func (a *PlannerAgent) Execute(ctx context.Context, task *Task) (*AgentResult, error) {
	start := time.Now()
	
	inputJSON, _ := json.Marshal(task.Input)
	fmt.Printf("[PlannerAgent] Planning task: %s, input: %s\n", task.Name, string(inputJSON))
	
	subTasks := []map[string]interface{}{
		{"id": "sub_1", "name": "Data Collection", "agent": "search"},
		{"id": "sub_2", "name": "Data Analysis", "agent": "data", "depends_on": []string{"sub_1"}},
		{"id": "sub_3", "name": "Report Writing", "agent": "writer", "depends_on": []string{"sub_2"}},
	}
	
	return &AgentResult{
		TaskID:   task.ID,
		AgentID:  a.ID,
		Success:  true,
		Output:   map[string]interface{}{"sub_tasks": subTasks, "execution_order": []string{"sub_1", "sub_2", "sub_3"}},
		Duration: time.Since(start),
	}, nil
}

// CoderAgent handles code generation
type CoderAgent struct {
	BaseAgent
	languages map[string]bool
}

func NewCoderAgent() *CoderAgent {
	return &CoderAgent{
		BaseAgent: BaseAgent{
			ID:           "coder-001",
			AgentType:    AgentTypeCoder,
			Capabilities: []string{"code_generation", "code_review", "refactoring"},
		},
		languages: map[string]bool{
			"python": true,
			"go":     true,
			"java":   true,
			"rust":   true,
		},
	}
}

func (a *CoderAgent) Execute(ctx context.Context, task *Task) (*AgentResult, error) {
	start := time.Now()
	
	language, _ := task.Input["language"].(string)
	if !a.languages[language] {
		return &AgentResult{
			TaskID:   task.ID,
			AgentID:  a.ID,
			Success:  false,
			Error:    fmt.Sprintf("Language %s not supported", language),
			Duration: time.Since(start),
		}, nil
	}
	
	code := fmt.Sprintf("// Generated %s code for: %s\nfunc main() {\n    // TODO: implement %s\n}\n",
		language, task.Name, task.Name)
	
	fmt.Printf("[CoderAgent] Generated %s code for task: %s\n", language, task.Name)
	
	return &AgentResult{
		TaskID:   task.ID,
		AgentID:  a.ID,
		Success:  true,
		Output:   map[string]interface{}{"language": language, "code": code, "lines": 5},
		Duration: time.Since(start),
	}, nil
}

// SearchAgent handles web and document search
type SearchAgent struct {
	BaseAgent
}

func NewSearchAgent() *SearchAgent {
	return &SearchAgent{
		BaseAgent: BaseAgent{
			ID:           "search-001",
			AgentType:    AgentTypeSearch,
			Capabilities: []string{"web_search", "document_search", "knowledge_retrieval"},
		},
	}
}

func (a *SearchAgent) Execute(ctx context.Context, task *Task) (*AgentResult, error) {
	start := time.Now()
	
	query, _ := task.Input["query"].(string)
	fmt.Printf("[SearchAgent] Searching for: %s\n", query)
	
	results := []map[string]string{
		{"title": "Result 1", "url": "https://example.com/1", "snippet": "Relevant information..."},
		{"title": "Result 2", "url": "https://example.com/2", "snippet": "More information..."},
	}
	
	return &AgentResult{
		TaskID:   task.ID,
		AgentID:  a.ID,
		Success:  true,
		Output:   map[string]interface{}{"query": query, "results": results, "total": len(results)},
		Duration: time.Since(start),
	}, nil
}

// DataAgent handles data analysis
type DataAgent struct {
	BaseAgent
}

func NewDataAgent() *DataAgent {
	return &DataAgent{
		BaseAgent: BaseAgent{
			ID:           "data-001",
			AgentType:    AgentTypeData,
			Capabilities: []string{"data_processing", "statistical_analysis", "visualization"},
		},
	}
}

func (a *DataAgent) Execute(ctx context.Context, task *Task) (*AgentResult, error) {
	start := time.Now()
	
	data, _ := json.Marshal(task.Input)
	fmt.Printf("[DataAgent] Processing data: %s\n", string(data))
	
	analysis := map[string]interface{}{
		"record_count":     1000,
		"summary":          "Data shows an upward trend",
		"anomalies":        []string{},
		"recommendations":  []string{"Recommend optimizing metric A", "Monitor fluctuation in category B data"},
	}
	
	return &AgentResult{
		TaskID:   task.ID,
		AgentID:  a.ID,
		Success:  true,
		Output:   analysis,
		Duration: time.Since(start),
	}, nil
}

// WriterAgent handles document writing
type WriterAgent struct {
	BaseAgent
}

func NewWriterAgent() *WriterAgent {
	return &WriterAgent{
		BaseAgent: BaseAgent{
			ID:           "writer-001",
			AgentType:    AgentTypeWriter,
			Capabilities: []string{"document_writing", "report_generation", "content_creation"},
		},
	}
}

func (a *WriterAgent) Execute(ctx context.Context, task *Task) (*AgentResult, error) {
	start := time.Now()
	
	content, _ := task.Input["content"].(string)
	style, _ := task.Input["style"].(string)
	
	fmt.Printf("[WriterAgent] Writing content in %s style\n", style)
	
	document := fmt.Sprintf("# %s\n\n%s\n\n## Conclusion\n\nBased on the above analysis, generate a professional report.",
		task.Name, content)
	
	return &AgentResult{
		TaskID:   task.ID,
		AgentID:  a.ID,
		Success:  true,
		Output:   map[string]interface{}{"document": document, "word_count": len(document)},
		Duration: time.Since(start),
	}, nil
}

// ReviewerAgent handles quality review
type ReviewerAgent struct {
	BaseAgent
}

func NewReviewerAgent() *ReviewerAgent {
	return &ReviewerAgent{
		BaseAgent: BaseAgent{
			ID:           "reviewer-001",
			AgentType:    AgentTypeReviewer,
			Capabilities: []string{"code_review", "quality_check", "compliance_verification"},
		},
	}
}

func (a *ReviewerAgent) Execute(ctx context.Context, task *Task) (*AgentResult, error) {
	start := time.Now()
	
	content, _ := json.Marshal(task.Input)
	fmt.Printf("[ReviewerAgent] Reviewing content: %s\n", string(content))
	
	review := map[string]interface{}{
		"quality_score":  85,
		"issues":         []string{},
		"suggestions":    []string{"Performance can be further optimized"},
		"approved":       true,
	}
	
	return &AgentResult{
		TaskID:   task.ID,
		AgentID:  a.ID,
		Success:  true,
		Output:   review,
		Duration: time.Since(start),
	}, nil
}

// AgentRegistry manages available agents
type AgentRegistry struct {
	mu     sync.RWMutex
	agents map[AgentType]Agent
}

func NewAgentRegistry() *AgentRegistry {
	registry := &AgentRegistry{
		agents: make(map[AgentType]Agent),
	}
	
	registry.Register(NewPlannerAgent())
	registry.Register(NewCoderAgent())
	registry.Register(NewSearchAgent())
	registry.Register(NewDataAgent())
	registry.Register(NewWriterAgent())
	registry.Register(NewReviewerAgent())
	
	return registry
}

func (r *AgentRegistry) Register(agent Agent) {
	r.mu.Lock()
	defer r.mu.Unlock()
	r.agents[agent.GetType()] = agent
}

func (r *AgentRegistry) Get(agentType AgentType) (Agent, bool) {
	r.mu.RLock()
	defer r.mu.RUnlock()
	agent, ok := r.agents[agentType]
	return agent, ok
}

func (r *AgentRegistry) List() []Agent {
	r.mu.RLock()
	defer r.mu.RUnlock()
	
	agents := make([]Agent, 0, len(r.agents))
	for _, agent := range r.agents {
		agents = append(agents, agent)
	}
	return agents
}

// AgentScheduler handles task scheduling
type AgentScheduler struct {
	registry *AgentRegistry
	mu       sync.Mutex
}

func NewAgentScheduler(registry *AgentRegistry) *AgentScheduler {
	return &AgentScheduler{
		registry: registry,
	}
}

func (s *AgentScheduler) Schedule(task *Task) (Agent, error) {
	s.mu.Lock()
	defer s.mu.Unlock()
	
	agent, ok := s.registry.Get(task.RequiredAgent)
	if !ok {
		return nil, fmt.Errorf("no agent available for type: %s", task.RequiredAgent)
	}
	
	return agent, nil
}

// MultiAgentOrchestrator orchestrates multiple agents
type MultiAgentOrchestrator struct {
	registry  *AgentRegistry
	scheduler *AgentScheduler
}

func NewMultiAgentOrchestrator() *MultiAgentOrchestrator {
	registry := NewAgentRegistry()
	return &MultiAgentOrchestrator{
		registry:  registry,
		scheduler: NewAgentScheduler(registry),
	}
}

func (o *MultiAgentOrchestrator) ExecuteTask(ctx context.Context, task *Task) (*AgentResult, error) {
	agent, err := o.scheduler.Schedule(task)
	if err != nil {
		return nil, err
	}
	
	return agent.Execute(ctx, task)
}

func (o *MultiAgentOrchestrator) ExecuteParallelTasks(ctx context.Context, tasks []*Task) ([]*AgentResult, error) {
	var wg sync.WaitGroup
	results := make([]*AgentResult, len(tasks))
	errors := make([]error, len(tasks))
	
	for i, task := range tasks {
		wg.Add(1)
		go func(idx int, t *Task) {
			defer wg.Done()
			
			result, err := o.ExecuteTask(ctx, t)
			results[idx] = result
			errors[idx] = err
		}(i, task)
	}
	
	wg.Wait()
	
	for _, err := range errors {
		if err != nil {
			return results, err
		}
	}
	
	return results, nil
}

func main() {
	orchestrator := NewMultiAgentOrchestrator()
	ctx := context.Background()
	
	tasks := []*Task{
		{
			ID:            "task-001",
			Name:          "Search Information",
			RequiredAgent: AgentTypeSearch,
			Input:         map[string]interface{}{"query": "Latest AI technology developments"},
		},
		{
			ID:            "task-002",
			Name:          "Data Analysis",
			RequiredAgent: AgentTypeData,
			Input:         map[string]interface{}{"data": []float64{1.1, 2.2, 3.3}},
		},
		{
			ID:            "task-003",
			Name:          "Generate Code",
			RequiredAgent: AgentTypeCoder,
			Input:         map[string]interface{}{"language": "python", "task": "Sorting algorithm"},
		},
	}
	
	fmt.Println("Starting parallel task execution...")
	results, err := orchestrator.ExecuteParallelTasks(ctx, tasks)
	if err != nil {
		fmt.Printf("Execution error: %v\n", err)
		return
	}
	
	fmt.Println("\nExecution Results:")
	for _, result := range results {
		if result.Success {
			outputJSON, _ := json.MarshalIndent(result.Output, "", "  ")
			fmt.Printf("  Task %s: Success (Duration: %v)\n  Output: %s\n",
				result.TaskID, result.Duration, string(outputJSON))
		} else {
			fmt.Printf("  Task %s: Failed - %s\n", result.TaskID, result.Error)
		}
	}
}

# Python Example: MCP Client Implementation
import json
import asyncio
from typing import Any, Dict, List, Optional
from dataclasses import dataclass, field
from enum import Enum

class MCPErrorCode(Enum):
    INVALID_REQUEST = -32600
    METHOD_NOT_FOUND = -32601
    INVALID_PARAMS = -32602
    INTERNAL_ERROR = -32603

@dataclass
class MCPRequest:
    jsonrpc: str = "2.0"
    id: Optional[int] = None
    method: str = ""
    params: Dict[str, Any] = field(default_factory=dict)

@dataclass
class MCPResponse:
    jsonrpc: str = "2.0"
    id: Optional[int] = None
    result: Optional[Any] = None
    error: Optional[Dict[str, Any]] = None

@dataclass
class MCPTool:
    name: str
    description: str
    inputSchema: Dict[str, Any]

@dataclass
class MCPToolResult:
    content: List[Dict[str, Any]]
    isError: bool = False

class MCPClient:
    """MCP Client Implementation"""
    
    def __init__(self, server_url: str, api_key: Optional[str] = None):
        self.server_url = server_url
        self.api_key = api_key
        self.tools: Dict[str, MCPTool] = {}
        self._request_id = 0
    
    async def initialize(self) -> Dict[str, Any]:
        """Initialize connection and get server capabilities"""
        request = MCPRequest(
            id=self._next_id(),
            method="initialize",
            params={
                "protocolVersion": "2024-11-05",
                "capabilities": {
                    "roots": {"listChanged": True},
                    "sampling": {}
                },
                "clientInfo": {
                    "name": "multi-agent-client",
                    "version": "1.0.0"
                }
            }
        )
        
        response = {
            "protocolVersion": "2024-11-05",
            "serverInfo": {
                "name": "example-mcp-server",
                "version": "1.0.0"
            },
            "capabilities": {
                "tools": {"listChanged": True}
            },
            "instructions": "Example MCP Server for demonstration"
        }
        
        await self.list_tools()
        
        return response
    
    async def list_tools(self) -> List[MCPTool]:
        """List all available tools"""
        request = MCPRequest(
            id=self._next_id(),
            method="tools/list"
        )
        
        tool_list = [
            MCPTool(
                name="search_web",
                description="Search the web for information",
                inputSchema={
                    "type": "object",
                    "properties": {
                        "query": {"type": "string", "description": "Search query"},
                        "limit": {"type": "integer", "description": "Max results"}
                    },
                    "required": ["query"]
                }
            ),
            MCPTool(
                name="read_file",
                description="Read contents of a file",
                inputSchema={
                    "type": "object",
                    "properties": {
                        "path": {"type": "string", "description": "File path"},
                        "encoding": {"type": "string", "default": "utf-8"}
                    },
                    "required": ["path"]
                }
            ),
            MCPTool(
                name="execute_code",
                description="Execute code in a sandboxed environment",
                inputSchema={
                    "type": "object",
                    "properties": {
                        "language": {"type": "string", "enum": ["python", "javascript", "bash"]},
                        "code": {"type": "string"}
                    },
                    "required": ["language", "code"]
                }
            ),
        ]
        
        self.tools = {tool.name: tool for tool in tool_list}
        return tool_list
    
    async def call_tool(self, tool_name: str, arguments: Dict[str, Any]) -> MCPToolResult:
        """Call the specified tool"""
        if tool_name not in self.tools:
            return MCPToolResult(
                content=[{"type": "text", "text": f"Tool '{tool_name}' not found"}],
                isError=True
            )
        
        request = MCPRequest(
            id=self._next_id(),
            method="tools/call",
            params={
                "name": tool_name,
                "arguments": arguments
            }
        )
        
        if tool_name == "search_web":
            results = [
                {"type": "text", "text": f"Search results for '{arguments.get('query', '')}':"},
                {"type": "text", "text": "1. Result A - relevant information about the topic"},
                {"type": "text", "text": "2. Result B - another relevant source"},
            ]
            return MCPToolResult(content=results)
        
        elif tool_name == "read_file":
            return MCPToolResult(
                content=[{"type": "text", "text": f"File content from {arguments.get('path', '')}:\n[simulated content]"}]
            )
        
        elif tool_name == "execute_code":
            return MCPToolResult(
                content=[{"type": "text", "text": f"Executed {arguments.get('language')} code successfully"}]
            )
        
        return MCPToolResult(content=[{"type": "text", "text": "Unknown tool"}], isError=True)
    
    def _next_id(self) -> int:
        self._request_id += 1
        return self._request_id
    
    async def close(self):
        """Close connection"""
        self.tools.clear()


async def main():
    client = MCPClient("https://mcp-server.example.com", api_key="demo-key")
    
    info = await client.initialize()
    print(f"Connected to {info['serverInfo']['name']} v{info['serverInfo']['version']}")
    
    tools = await client.list_tools()
    print(f"\nAvailable tools ({len(tools)}):")
    for tool in tools:
        print(f"  - {tool.name}: {tool.description}")
    
    result = await client.call_tool("search_web", {"query": "multi-agent AI systems"})
    print(f"\nTool result: {result.content}")
    
    await client.close()

if __name__ == "__main__":
    asyncio.run(main())

# Python Example: A2A Protocol Implementation
import asyncio
import json
import uuid
from typing import Dict, List, Callable, Any, Optional
from dataclasses import dataclass, field, asdict
from enum import Enum
from datetime import datetime
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


class MessageType(Enum):
    TASK_REQUEST = "task_request"
    TASK_RESPONSE = "task_response"
    TASK_UPDATE = "task_update"
    TASK_COMPLETE = "task_complete"
    TASK_FAILED = "task_failed"
    HEARTBEAT = "heartbeat"
    AGENT_REGISTER = "agent_register"
    AGENT_DISCOVER = "agent_discover"


@dataclass
class A2AMessage:
    id: str
    type: MessageType
    sender_id: str
    receiver_id: Optional[str]
    task_id: Optional[str]
    payload: Dict[str, Any]
    timestamp: str = field(default_factory=lambda: datetime.now().isoformat())
    correlation_id: Optional[str] = None


@dataclass
class AgentCapability:
    name: str
    description: str
    input_schema: Dict[str, Any]
    output_schema: Dict[str, Any]


@dataclass
class AgentInfo:
    id: str
    name: str
    capabilities: List[AgentCapability]
    status: str = "online"
    endpoint: str = ""


class A2AMessageBus:
    """A2A Message Bus - Core infrastructure for Agent communication"""
    
    def __init__(self):
        self.agents: Dict[str, AgentInfo] = {}
        self.subscribers: Dict[MessageType, List[Callable]] = {}
        self.message_queue: asyncio.Queue = asyncio.Queue()
        self.running = False
        self._message_history: List[A2AMessage] = []
    
    def register_agent(self, agent: AgentInfo):
        """Register an Agent to the message bus"""
        self.agents[agent.id] = agent
        logger.info(f"Agent registered: {agent.id} ({agent.name})")
        
        self._broadcast(A2AMessage(
            id=str(uuid.uuid4()),
            type=MessageType.AGENT_REGISTER,
            sender_id="message_bus",
            receiver_id=None,
            task_id=None,
            payload={"agent": asdict(agent)}
        ))
    
    def discover_agents(self, capability: Optional[str] = None) -> List[AgentInfo]:
        """Discover Agents with specific capabilities"""
        if capability is None:
            return list(self.agents.values())
        
        return [
            agent for agent in self.agents.values()
            if any(cap.name == capability for cap in agent.capabilities)
        ]
    
    def subscribe(self, message_type: MessageType, callback: Callable):
        """Subscribe to specific message types"""
        if message_type not in self.subscribers:
            self.subscribers[message_type] = []
        self.subscribers[message_type].append(callback)
    
    def unsubscribe(self, message_type: MessageType, callback: Callable):
        """Unsubscribe"""
        if message_type in self.subscribers:
            self.subscribers[message_type].remove(callback)
    
    async def send_message(self, message: A2AMessage) -> bool:
        """Send a message"""
        self._message_history.append(message)
        
        if message.receiver_id and message.receiver_id in self.agents:
            await self._deliver_to_agent(message.receiver_id, message)
            return True
        
        self._broadcast(message)
        return True
    
    def _broadcast(self, message: A2AMessage):
        """Broadcast message to all subscribers"""
        if message.type in self.subscribers:
            for callback in self.subscribers[message.type]:
                try:
                    callback(message)
                except Exception as e:
                    logger.error(f"Error in subscriber callback: {e}")
    
    async def _deliver_to_agent(self, agent_id: str, message: A2AMessage):
        """Deliver message to specific Agent"""
        logger.info(f"Delivering message {message.id} to agent {agent_id}")
    
    async def get_task_status(self, task_id: str) -> List[A2AMessage]:
        """Get all messages for specific task"""
        return [msg for msg in self._message_history if msg.task_id == task_id]
    
    async def start(self):
        """Start message bus"""
        self.running = True
        logger.info("A2A Message Bus started")
        
        while self.running:
            try:
                message = await self.message_queue.get()
                await self._process_message(message)
            except asyncio.CancelledError:
                break
            except Exception as e:
                logger.error(f"Error processing message: {e}")
    
    async def stop(self):
        """Stop message bus"""
        self.running = False
        logger.info("A2A Message Bus stopped")
    
    async def _process_message(self, message: A2AMessage):
        """Process message"""
        self._broadcast(message)


class BaseA2AAgent:
    """Base class for A2A-enabled Agents"""
    
    def __init__(self, agent_id: str, name: str, capabilities: List[AgentCapability], message_bus: A2AMessageBus):
        self.agent_id = agent_id
        self.name = name
        self.capabilities = capabilities
        self.message_bus = message_bus
        self.inbox: asyncio.Queue = asyncio.Queue()
        self.task_results: Dict[str, Any] = {}
        self._running = False
    
    @property
    def agent_info(self) -> AgentInfo:
        return AgentInfo(
            id=self.agent_id,
            name=self.name,
            capabilities=self.capabilities,
            status="online",
            endpoint=f"agent://{self.agent_id}"
        )
    
    async def register(self):
        """Register to message bus"""
        self.message_bus.register_agent(self.agent_info)
    
    async def send_task_request(self, receiver_id: str, task_id: str, task_data: Dict[str, Any]) -> str:
        """Send task request"""
        message = A2AMessage(
            id=str(uuid.uuid4()),
            type=MessageType.TASK_REQUEST,
            sender_id=self.agent_id,
            receiver_id=receiver_id,
            task_id=task_id,
            payload=task_data,
            correlation_id=task_id
        )
        await self.message_bus.send_message(message)
        return message.id
    
    async def send_task_response(self, receiver_id: str, task_id: str, result: Any) -> str:
        """Send task response"""
        message = A2AMessage(
            id=str(uuid.uuid4()),
            type=MessageType.TASK_RESPONSE,
            sender_id=self.agent_id,
            receiver_id=receiver_id,
            task_id=task_id,
            payload={"result": result}
        )
        await self.message_bus.send_message(message)
        return message.id
    
    async def send_task_update(self, receiver_id: str, task_id: str, progress: int, status: str) -> str:
        """Send task update"""
        message = A2AMessage(
            id=str(uuid.uuid4()),
            type=MessageType.TASK_UPDATE,
            sender_id=self.agent_id,
            receiver_id=receiver_id,
            task_id=task_id,
            payload={"progress": progress, "status": status}
        )
        await self.message_bus.send_message(message)
        return message.id
    
    async def handle_message(self, message: A2AMessage):
        """Handle received message"""
        if message.type == MessageType.TASK_REQUEST:
            result = await self.execute_task(message)
            await self.send_task_response(message.sender_id, message.task_id, result)
        
        elif message.type == MessageType.TASK_COMPLETE:
            self.task_results[message.task_id] = message.payload.get("result")
        
        elif message.type == MessageType.AGENT_DISCOVER:
            response = A2AMessage(
                id=str(uuid.uuid4()),
                type=MessageType.AGENT_REGISTER,
                sender_id=self.agent_id,
                receiver_id=message.sender_id,
                task_id=None,
                payload={"agent": asdict(self.agent_info)}
            )
            await self.message_bus.send_message(response)
    
    async def execute_task(self, message: A2AMessage) -> Any:
        """Execute task - subclasses must implement"""
        raise NotImplementedError("Subclasses must implement execute_task")
    
    async def start(self):
        """Start Agent"""
        self._running = True
        await self.register()
        logger.info(f"Agent {self.agent_id} started")
    
    async def stop(self):
        """Stop Agent"""
        self._running = False
        logger.info(f"Agent {self.agent_id} stopped")


# Usage Example: Concrete Agent Implementation
class DataAnalysisA2AAgent(BaseA2AAgent):
    """A2A implementation of Data Analysis Agent"""
    
    def __init__(self, message_bus: A2AMessageBus):
        super().__init__(
            agent_id="data-agent-001",
            name="Data Analysis Agent",
            capabilities=[
                AgentCapability(
                    name="analyze",
                    description="Perform data analysis",
                    input_schema={"type": "object", "properties": {"data": {}}},
                    output_schema={"type": "object"}
                )
            ],
            message_bus=message_bus
        )
    
    async def execute_task(self, message: A2AMessage) -> Any:
        data = message.payload.get("data", {})
        logger.info(f"DataAgent analyzing data: {data}")
        
        await asyncio.sleep(0.5)
        
        return {
            "summary": "Analysis complete",
            "record_count": 1000,
            "key_findings": ["Finding A", "Finding B"]
        }


async def demonstrate_a2a():
    """Demonstrate A2A protocol workflow"""
    message_bus = A2AMessageBus()
    
    data_agent = DataAnalysisA2AAgent(message_bus)
    
    await message_bus.start()
    await data_agent.start()
    
    task_id = "task-123"
    print(f"Sending task request: {task_id}")
    
    await data_agent.send_task_request(
        receiver_id="data-agent-001",
        task_id=task_id,
        task_data={"data": {"source": "database", "query": "SELECT *"}}
    )
    
    await asyncio.sleep(1)
    
    result = data_agent.task_results.get(task_id)
    print(f"Task result: {result}")
    
    await data_agent.stop()
    await message_bus.stop()


if __name__ == "__main__":
    asyncio.run(demonstrate_a2a())

// Go Example: Code Development Multi-Agent Workflow
package main

import (
	"context"
	"fmt"
	"time"
)

// SDLCPhase represents phases in software development lifecycle
type SDLCPhase string

const (
	PhaseRequirements SDLCPhase = "requirements"
	PhaseDesign      SDLCPhase = "design"
	PhaseCode        SDLCPhase = "code"
	PhaseTest        SDLCPhase = "test"
	PhaseReview      SDLCPhase = "review"
	PhaseDeploy      SDLCPhase = "deploy"
)

// CodeTask represents a code development task
type CodeTask struct {
	ID            string
	Description   string
	Language      string
	Framework     string
	CurrentPhase  SDLCPhase
	Requirements  string
	Design        string
	Code          string
	Tests         string
	ReviewNotes   string
}

// RequirementsAgent analyzes requirements
type RequirementsAgent struct{}

func (a *RequirementsAgent) Analyze(task *CodeTask) (string, error) {
	fmt.Printf("[RequirementsAgent] Analyzing requirements for: %s\n", task.Description)
	time.Sleep(200 * time.Millisecond)
	
	task.Requirements = fmt.Sprintf("Functional Requirements:\n1. %s\n2. Error handling\n3. Logging\n\nNon-Functional Requirements:\n1. Performance\n2. Security\n3. Scalability", task.Description)
	
	return task.Requirements, nil
}

// DesignAgent designs architecture
type DesignAgent struct{}

func (a *DesignAgent) Design(task *CodeTask) (string, error) {
	fmt.Printf("[DesignAgent] Designing architecture for: %s\n", task.Description)
	time.Sleep(300 * time.Millisecond)
	
	task.Design = fmt.Sprintf(`Architecture Design for %s (%s/%s):

1. Project Structure:
   - cmd/: Entry points
   - internal/: Core business logic
   - pkg/: Reusable packages
   - api/: API definitions

2. Key Components:
   - Handler layer
   - Service layer
   - Repository layer

3. Data Models:
   - Define domain entities
   - Create database schemas

4. API Design:
   - RESTful endpoints
   - Request/Response formats`,
		task.Description, task.Language, task.Framework)
	
	return task.Design, nil
}

// CodeGenerationAgent generates code
type CodeGenerationAgent struct{}

func (a *CodeGenerationAgent) Generate(task *CodeTask) (string, error) {
	fmt.Printf("[CodeGenerationAgent] Generating %s code for: %s\n", task.Language, task.Description)
	time.Sleep(500 * time.Millisecond)
	
	task.Code = fmt.Sprintf(`// main.go - Generated %s code
package main

import (
    "fmt"
    "log"
)

func main() {
    fmt.Println("Starting: %s")
    // TODO: Implement %s
}`, task.Description, task.Description)
	
	return task.Code, nil
}

// TestGenerationAgent generates tests
type TestGenerationAgent struct{}

func (a *TestGenerationAgent) GenerateTests(task *CodeTask) (string, error) {
	fmt.Printf("[TestGenerationAgent] Generating tests for: %s\n", task.Description)
	time.Sleep(300 * time.Millisecond)
	
	task.Tests = fmt.Sprintf(`// main_test.go - Generated tests
package main

import "testing"

func TestMain(t *testing.T) {
    // Test case for: %s
    expected := true
    result := true // TODO: Implement actual test
    if result != expected {
        t.Errorf("Test failed: expected %%v, got %%v", expected, result)
    }
}`, task.Description)
	
	return task.Tests, nil
}

// CodeReviewAgent reviews code
type CodeReviewAgent struct{}

func (a *CodeReviewAgent) Review(task *CodeTask) (string, error) {
	fmt.Printf("[CodeReviewAgent] Reviewing code for: %s\n", task.Description)
	time.Sleep(200 * time.Millisecond)
	
	task.ReviewNotes = `Code Review Report:
- Code Quality: Good
- Security: No major issues found
- Performance: Acceptable
- Suggestions: Consider adding more error handling`
	
	return task.ReviewNotes, nil
}

// CodeDevelopmentWorkflow orchestrates the development process
type CodeDevelopmentWorkflow struct {
	requirementsAgent *RequirementsAgent
	designAgent       *DesignAgent
	codeAgent         *CodeGenerationAgent
	testAgent         *TestGenerationAgent
	reviewAgent       *CodeReviewAgent
}

func NewCodeDevelopmentWorkflow() *CodeDevelopmentWorkflow {
	return &CodeDevelopmentWorkflow{
		requirementsAgent: &RequirementsAgent{},
		designAgent:       &DesignAgent{},
		codeAgent:         &CodeGenerationAgent{},
		testAgent:         &TestGenerationAgent{},
		reviewAgent:       &CodeReviewAgent{},
	}
}

func (w *CodeDevelopmentWorkflow) Execute(ctx context.Context, task *CodeTask) error {
	fmt.Printf("\n=== Starting Code Development Workflow ===\n")
	fmt.Printf("Task: %s\n\n", task.Description)
	
	// Phase 1: Requirements Analysis
	fmt.Println("--- Phase 1: Requirements Analysis ---")
	if _, err := w.requirementsAgent.Analyze(task); err != nil {
		return fmt.Errorf("requirements analysis failed: %w", err)
	}
	fmt.Printf("Requirements:\n%s\n\n", task.Requirements)
	
	// Phase 2: Architecture Design
	fmt.Println("--- Phase 2: Architecture Design ---")
	if _, err := w.designAgent.Design(task); err != nil {
		return fmt.Errorf("design failed: %w", err)
	}
	fmt.Printf("Design:\n%s\n\n", task.Design)
	
	// Phase 3: Code Generation
	fmt.Println("--- Phase 3: Code Generation ---")
	if _, err := w.codeAgent.Generate(task); err != nil {
		return fmt.Errorf("code generation failed: %w", err)
	}
	fmt.Printf("Generated Code:\n%s\n\n", task.Code)
	
	// Phase 4: Test Generation
	fmt.Println("--- Phase 4: Test Generation ---")
	if _, err := w.testAgent.GenerateTests(task); err != nil {
		return fmt.Errorf("test generation failed: %w", err)
	}
	fmt.Printf("Generated Tests:\n%s\n\n", task.Tests)
	
	// Phase 5: Code Review
	fmt.Println("--- Phase 5: Code Review ---")
	if _, err := w.reviewAgent.Review(task); err != nil {
		return fmt.Errorf("code review failed: %w", err)
	}
	fmt.Printf("Review Notes:\n%s\n\n", task.ReviewNotes)
	
	fmt.Println("=== Code Development Workflow Complete ===")
	return nil
}

func main() {
	ctx := context.Background()
	
	workflow := NewCodeDevelopmentWorkflow()
	
	task := &CodeTask{
		ID:          "task-001",
		Description: "User Authentication Module",
		Language:    "Go",
		Framework:   "Gin",
	}
	
	if err := workflow.Execute(ctx, task); err != nil {
		fmt.Printf("Workflow failed: %v\n", err)
	}
}

# Python Example: Multi-Agent System Memory Implementation
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional
from datetime import datetime
import numpy as np
from collections import deque

@dataclass
class MemoryItem:
    id: str
    content: str
    embedding: Optional[List[float]] = None
    created_at: datetime = field(default_factory=datetime.now)
    access_count: int = 0
    last_accessed: datetime = field(default_factory=datetime.now)
    memory_type: str = "generic"
    metadata: Dict[str, Any] = field(default_factory=dict)

class WorkingMemory:
    """Working Memory - Fixed-size circular buffer"""
    
    def __init__(self, max_size: int = 100):
        self.max_size = max_size
        self.buffer: deque = deque(maxlen=max_size)
    
    def add(self, item: MemoryItem):
        self.buffer.append(item)
    
    def get_recent(self, n: int = 10) -> List[MemoryItem]:
        items = list(self.buffer)[-n:]
        for item in items:
            item.access_count += 1
            item.last_accessed = datetime.now()
        return items
    
    def clear(self):
        self.buffer.clear()
    
    def __len__(self):
        return len(self.buffer)


class VectorStore:
    """Vector Store - For semantic retrieval in long-term memory"""
    
    def __init__(self):
        self.items: List[MemoryItem] = []
    
    def add(self, item: MemoryItem):
        if item.embedding is None:
            item.embedding = self._generate_embedding(item.content)
        self.items.append(item)
    
    def search(self, query: str, top_k: int = 5) -> List[MemoryItem]:
        query_embedding = self._generate_embedding(query)
        
        scores = []
        for item in self.items:
            if item.embedding:
                score = self._cosine_similarity(query_embedding, item.embedding)
                scores.append((score, item))
        
        scores.sort(key=lambda x: x[0], reverse=True)
        return [item for _, item in scores[:top_k]]
    
    def _generate_embedding(self, text: str) -> List[float]:
        np.random.seed(hash(text) % (2**32))
        return np.random.randn(384).tolist()
    
    def _cosine_similarity(self, a: List[float], b: List[float]) -> float:
        a = np.array(a)
        b = np.array(b)
        return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))
    
    def __len__(self):
        return len(self.items)


class EpisodicMemory:
    """Episodic Memory - Records complete task execution traces"""
    
    def __init__(self):
        self.episodes: Dict[str, List[Dict]] = {}
    
    def start_episode(self, task_id: str, context: Dict[str, Any]):
        self.episodes[task_id] = [{
            "type": "start",
            "timestamp": datetime.now().isoformat(),
            "context": context
        }]
    
    def record_step(self, task_id: str, step_type: str, data: Dict[str, Any]):
        if task_id not in self.episodes:
            self.episodes[task_id] = []
        
        self.episodes[task_id].append({
            "type": step_type,
            "timestamp": datetime.now().isoformat(),
            "data": data
        })
    
    def end_episode(self, task_id: str, outcome: str):
        if task_id in self.episodes:
            self.episodes[task_id].append({
                "type": "end",
                "timestamp": datetime.now().isoformat(),
                "outcome": outcome
            })
    
    def get_episode(self, task_id: str) -> Optional[List[Dict]]:
        return self.episodes.get(task_id)
    
    def find_similar_episodes(self, current_task: str, top_k: int = 3) -> List[str]:
        return list(self.episodes.keys())[-top_k:]


class MultiAgentMemory:
    """Unified Memory System for Multi-Agent Systems"""
    
    def __init__(self, agent_id: str):
        self.agent_id = agent_id
        self.working_memory = WorkingMemory(max_size=100)
        self.long_term_memory = VectorStore()
        self.episodic_memory = EpisodicMemory()
        self.current_task_id: Optional[str] = None
    
    def remember(self, content: str, memory_type: str = "generic", metadata: Optional[Dict] = None):
        item = MemoryItem(
            id=f"{self.agent_id}_{datetime.now().timestamp()}",
            content=content,
            memory_type=memory_type,
            metadata=metadata or {}
        )
        
        if memory_type == "short_term":
            self.working_memory.add(item)
        else:
            self.long_term_memory.add(item)
    
    def recall(self, query: str, search_type: str = "all") -> List[MemoryItem]:
        results = []
        
        if search_type in ["all", "recent"]:
            recent = self.working_memory.get_recent(10)
            results.extend([r for r in recent if query.lower() in r.content.lower()])
        
        if search_type in ["all", "semantic"]:
            semantic_results = self.long_term_memory.search(query)
            results.extend(semantic_results)
        
        return results
    
    def start_task(self, task_id: str, context: Dict[str, Any]):
        self.current_task_id = task_id
        self.episodic_memory.start_episode(task_id, context)
    
    def record_task_step(self, step_type: str, data: Dict[str, Any]):
        if self.current_task_id:
            self.episodic_memory.record_step(self.current_task_id, step_type, data)
    
    def complete_task(self, outcome: str):
        if self.current_task_id:
            recent = self.working_memory.get_recent(5)
            for item in recent:
                item.memory_type = "long_term"
                self.long_term_memory.add(item)
            
            self.episodic_memory.end_episode(self.current_task_id, outcome)
            self.current_task_id = None
    
    def learn_from_past(self, current_task: str) -> List[str]:
        similar_tasks = self.episodic_memory.find_similar_episodes(current_task)
        lessons = []
        
        for task_id in similar_tasks:
            episode = self.episodic_memory.get_episode(task_id)
            if episode:
                for step in episode:
                    if step["type"] == "success" or step["type"] == "error":
                        lessons.append(step.get("data", {}).get("lesson", ""))
        
        return lessons


def demonstrate_memory_system():
    agent_memory = MultiAgentMemory("data-agent-001")
    
    agent_memory.start_task("task-001", {"description": "Data analysis task"})
    agent_memory.record_task_step("agent_call", {
        "agent": "search",
        "input": "Query sales data",
        "lesson": "Search keywords should be more precise"
    })
    
    agent_memory.remember(
        content="The key to data cleaning is handling missing values and outliers",
        memory_type="short_term",
        metadata={"topic": "data_processing"}
    )
    
    results = agent_memory.recall("data cleaning")
    print(f"Retrieved {len(results)} memories about 'data cleaning'")
    
    agent_memory.complete_task("success")
    lessons = agent_memory.learn_from_past("data analysis")
    print(f"Learned {len(lessons)} lessons from past tasks")


if __name__ == "__main__":
    demonstrate_memory_system()

# Python Example: Multi-Agent System Fault Tolerance
import asyncio
import time
from functools import wraps
from typing import Callable, Any, Optional, Type
from dataclasses import dataclass
from enum import Enum
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


class RetryStrategy(Enum):
    EXPONENTIAL_BACKOFF = "exponential_backoff"
    LINEAR_BACKOFF = "linear_backoff"
    FIXED_DELAY = "fixed_delay"


@dataclass
class RetryConfig:
    max_attempts: int = 3
    initial_delay: float = 1.0
    max_delay: float = 60.0
    strategy: RetryStrategy = RetryStrategy.EXPONENTIAL_BACKOFF
    retryable_exceptions: tuple = (ConnectionError, TimeoutError, asyncio.TimeoutError)


@dataclass
class CircuitBreakerState:
    failure_count: int = 0
    success_count: int = 0
    last_failure_time: Optional[float] = None
    state: str = "closed"


def with_retry(config: RetryConfig = None):
    if config is None:
        config = RetryConfig()
    
    def decorator(func: Callable):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            last_exception = None
            
            for attempt in range(config.max_attempts):
                try:
                    return await func(*args, **kwargs)
                except config.retryable_exceptions as e:
                    last_exception = e
                    if attempt < config.max_attempts - 1:
                        delay = calculate_delay(attempt, config)
                        logger.warning(f"Attempt {attempt + 1} failed: {e}. Retrying in {delay}s...")
                        await asyncio.sleep(delay)
                    else:
                        logger.error(f"All {config.max_attempts} attempts failed")
            
            raise last_exception
        
        return wrapper
    return decorator


def calculate_delay(attempt: int, config: RetryConfig) -> float:
    if config.strategy == RetryStrategy.EXPONENTIAL_BACKOFF:
        delay = min(config.initial_delay * (2 ** attempt), config.max_delay)
    elif config.strategy == RetryStrategy.LINEAR_BACKOFF:
        delay = min(config.initial_delay * (attempt + 1), config.max_delay)
    else:
        delay = config.initial_delay
    
    import random
    delay *= (0.5 + random.random())
    return delay


class CircuitBreaker:
    """Circuit Breaker - Prevents cascading failures"""
    
    def __init__(self, failure_threshold: int = 5, recovery_timeout: float = 60.0, recovery_ratio: float = 0.5):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.recovery_ratio = recovery_ratio
        self.state = CircuitBreakerState()
    
    @property
    def is_open(self) -> bool:
        if self.state.state == "open":
            if self.state.last_failure_time:
                elapsed = time.time() - self.state.last_failure_time
                if elapsed > self.recovery_timeout:
                    self.state.state = "half_open"
                    logger.info("Circuit breaker: transitioning to half_open")
            return self.state.state == "open"
        return False
    
    def record_success(self):
        self.state.success_count += 1
        self.state.failure_count = 0
        
        if self.state.state == "half_open":
            if self.state.success_count >= 2:
                self.state.state = "closed"
                self.state.success_count = 0
                logger.info("Circuit breaker: recovered")
    
    def record_failure(self):
        self.state.failure_count += 1
        self.state.last_failure_time = time.time()
        
        if self.state.failure_count >= self.failure_threshold:
            self.state.state = "open"
            logger.warning(f"Circuit breaker: opened after {self.state.failure_count} failures")


class AgentFaultTolerantClient:
    """Agent Fault-Tolerant Client"""
    
    def __init__(self, agent_id: str):
        self.agent_id = agent_id
        self.circuit_breaker = CircuitBreaker()
        self.retry_config = RetryConfig()
    
    async def execute_with_fallback(
        self, 
        primary_func: Callable,
        fallback_func: Optional[Callable] = None,
        *args, **kwargs
    ) -> Any:
        if self.circuit_breaker.is_open:
            logger.warning(f"Circuit breaker is open for {self.agent_id}, using fallback")
            if fallback_func:
                return await fallback_func(*args, **kwargs)
            raise Exception(f"Agent {self.agent_id} unavailable and no fallback provided")
        
        try:
            @with_retry(self.retry_config)
            async def execute():
                return await primary_func(*args, **kwargs)
            
            result = await execute()
            self.circuit_breaker.record_success()
            return result
            
        except Exception as e:
            self.circuit_breaker.record_failure()
            logger.error(f"Agent {self.agent_id} execution failed: {e}")
            
            if fallback_func:
                logger.info(f"Executing fallback for {self.agent_id}")
                return await fallback_func(*args, **kwargs)
            raise


class SandboxExecutor:
    """Sandbox Executor - Isolates Agent execution"""
    
    def __init__(self, timeout: float = 30.0, memory_limit: int = 512 * 1024 * 1024):
        self.timeout = timeout
        self.memory_limit = memory_limit
    
    async def execute(self, code: str, language: str) -> dict:
        logger.info(f"Executing code in sandbox: language={language}")
        
        try:
            await asyncio.sleep(0.5)
            
            return {
                "success": True,
                "output": f"Executed {language} code",
                "execution_time": 0.5
            }
            
        except asyncio.TimeoutError:
            return {
                "success": False,
                "error": "Execution timeout"
            }
        except Exception as e:
            return {
                "success": False,
                "error": str(e)
            }


class TaskRollbackManager:
    """Task Rollback Manager"""
    
    def __init__(self):
        self.checkpoints: dict = {}
    
    def create_checkpoint(self, task_id: str, state: dict):
        self.checkpoints[task_id] = {
            "timestamp": time.time(),
            "state": state.copy()
        }
        logger.info(f"Checkpoint created for task {task_id}")
    
    def rollback(self, task_id: str) -> Optional[dict]:
        if task_id in self.checkpoints:
            checkpoint = self.checkpoints[task_id]
            logger.info(f"Rolling back task {task_id} to checkpoint at {checkpoint['timestamp']}")
            return checkpoint["state"]
        return None
    
    def clear_checkpoint(self, task_id: str):
        if task_id in self.checkpoints:
            del self.checkpoints[task_id]


async def demonstrate_fault_tolerance():
    client = AgentFaultTolerantClient("data-agent")
    sandbox = SandboxExecutor(timeout=10.0)
    rollback_manager = TaskRollbackManager()
    
    async def primary_execute(data):
        raise ConnectionError("Primary service unavailable")
    
    async def fallback_execute(data):
        logger.info("Executing fallback")
        return {"result": "from fallback", "data": data}
    
    result = await client.execute_with_fallback(
        primary_execute,
        fallback_execute,
        {"test": "data"}
    )
    print(f"Result: {result}")
    
    rollback_manager.create_checkpoint("task-001", {"step": 1, "result": "partial"})
    rollback_manager.create_checkpoint("task-001", {"step": 2, "result": "partial2"})
    
    state = rollback_manager.rollback("task-001")
    print(f"Rolled back state: {state}")


if __name__ == "__main__":
    asyncio.run(demonstrate_fault_tolerance())