┌─────────────────────────────────────────────────────────────┐
│                    AGENT ORCHESTRATOR                        │
│                  (Coordination Layer)                        │
└──────────────┬──────────────┬──────────────┬─────────────────┘
               │              │              │
    ┌──────────▼──┐  ┌───────▼───┐  ┌───────▼────┐
    │   Research  │  │  Analysis │  │   Action   │
    │    Agent    │  │   Agent   │  │    Agent   │
    └──────────────┘  └───────────┘  └────────────┘
           │                │                │
           └────────────────┼────────────────┘
                            │
                    ┌───────▼────┐
                    │   Memory   │
                    │   Store    │
                    └─────────────┘

// n8n workflow structure for hierarchical pattern
{
  "nodes": [
    {
      "type": "n8n-nodes-base.agent",
      "name": "Supervisor Agent",
      "parameters": {
        "options": {
          "systemMessage": "You are a supervisor agent. Analyze incoming requests and delegate to appropriate specialist agents: research, analysis, or action."
        }
      }
    },
    {
      "type": "n8n-nodes-base.agent",
      "name": "Research Agent",
      "parameters": {
        "options": {
          "systemMessage": "You are a research specialist. Gather information from available tools and provide structured findings."
        }
      }
    },
    {
      "type": "n8n-nodes-base.agent",
      "name": "Analysis Agent",
      "parameters": {
        "options": {
          "systemMessage": "You are an analysis specialist. Review data and provide insights, recommendations, and risk assessments."
        }
      }
    }
  ]
}

// Peer negotiation pattern
const agentNetwork = {
  agents: [
    {
      id: "sales-agent",
      role: "sales_specialist",
      capabilities: ["pricing", "negotiation", "crm_update"],
      priorities: ["revenue", "customer_satisfaction"]
    },
    {
      id: "support-agent", 
      role: "support_specialist",
      capabilities: ["troubleshooting", "ticket_management", "escalation"],
      priorities: ["resolution_time", "satisfaction_score"]
    },
    {
      id: "inventory-agent",
      role: "inventory_specialist", 
      capabilities: ["stock_check", "supplier_contact", "fulfillment"],
      priorities: ["availability", "cost_optimization"]
    }
  ],
  
  // Shared protocol for agent communication
  protocol: {
    messageTypes: ["request", "offer", "accept", "reject", "delegate", "report"],
    negotiationStrategy: "cooperative",
    conflictResolution: "priority_weighted_voting"
  }
};

Raw Input → [Agent A: Extract] → [Agent B: Transform] → [Agent C: Load] → Output
                ↓                      ↓                      ↓
           Data Cleaning          Enrichment          Validation

// Master orchestrator workflow
{
  "name": "Multi-Agent Orchestrator",
  "nodes": [
    {
      "type": "n8n-nodes-base.webhook",
      "name": "Agent Request Receiver",
      "parameters": {
        "httpMethod": "POST",
        "responseMode": "responseNode"
      }
    },
    {
      "type": "n8n-nodes-base.function",
      "name": "Route to Agent",
      "parameters": {
        "functionCode": "// Determine which agent(s) should handle the request\nconst request = items[0].json;\nconst intent = request.intent;\n\nconst agentRouting = {\n  'research': ['research-agent'],\n  'support': ['triage-agent', 'support-agent'],\n  'sales': ['qualification-agent', 'sales-agent'],\n  'complex': ['supervisor-agent']\n};\n\nreturn [{\n  json: {\n    targetAgents: agentRouting[intent] || ['supervisor-agent'],\n    originalRequest: request\n  }\n}];"
      }
    },
    {
      "type": "n8n-nodes-base.executeWorkflow",
      "name": "Execute Agent Workflow",
      "parameters": {
        "workflowId": "={{ $('Route to Agent').item.json.targetAgents[0] }}"
      }
    },
    {
      "type": "n8n-nodes-base.respondToWebhook",
      "name": "Return Response",
      "parameters": {
        "respondWith": "json",
        "responseBody": "={{$json}}"
      }
    }
  ]
}

from langgraph.graph import StateGraph, END
from typing import TypedDict, List

class AgentState(TypedDict):
    messages: List[dict]
    next_agent: str
    final_answer: str
    iteration_count: int

# Define agent nodes
def supervisor_agent(state):
    """Determines which agent should act next"""
    messages = state["messages"]
    
    # LLM-based routing decision
    decision = llm.invoke(f"""
    Based on these messages, which agent should handle this next?
    Options: research_agent, analysis_agent, action_agent, FINAL
    
    Messages: {messages}
    """)
    
    return {"next_agent": decision.content}

def research_agent(state):
    """Specialized research capabilities"""
    # Research logic here
    return {"messages": [research_result]}

def analysis_agent(state):
    """Analytical processing"""
    # Analysis logic here
    return {"messages": [analysis_result]}

# Build the graph
workflow = StateGraph(AgentState)

# Add nodes
workflow.add_node("supervisor", supervisor_agent)
workflow.add_node("research", research_agent)
workflow.add_node("analysis", analysis_agent)

# Add edges with conditional routing
workflow.add_edge("supervisor", conditional_router)
workflow.add_edge("research", "supervisor")
workflow.add_edge("analysis", "supervisor")

# Compile to runnable
chain = workflow.compile()

from crewai import Agent, Task, Crew
from crewai.tools import tool

# Define specialized agents
researcher = Agent(
    role='Market Research Specialist',
    goal='Gather comprehensive data on market trends and competitors',
    backstory='Expert in data analysis with 10 years of market research experience',
    tools=[search_tool, data_scraper],
    verbose=True
)

analyst = Agent(
    role='Business Intelligence Analyst', 
    goal='Transform raw data into actionable strategic insights',
    backstory='Former McKinsey consultant specializing in competitive analysis',
    tools=[visualization_tool, calculation_tool],
    verbose=True
)

writer = Agent(
    role='Strategy Content Writer',
    goal='Create compelling strategic recommendations from analysis',
    backstory='Award-winning business writer with expertise in executive communications',
    tools=[document_tool, template_tool],
    verbose=True
)

# Define tasks with dependencies
research_task = Task(
    description='Research the competitive landscape for AI automation tools in 2026',
    agent=researcher,
    expected_output='Comprehensive market research report with key players and trends'
)

analysis_task = Task(
    description='Analyze research findings and identify strategic opportunities',
    agent=analyst,
    expected_output='Strategic analysis with SWOT and recommendations',
    context=[research_task]  # Depends on research
)

report_task = Task(
    description='Write executive summary and strategic recommendations',
    agent=writer,
    expected_output='Professional strategy document ready for C-suite',
    context=[analysis_task]  # Depends on analysis
)

# Create and run the crew
crew = Crew(
    agents=[researcher, analyst, writer],
    tasks=[research_task, analysis_task, report_task],
    process='sequential',  # or 'hierarchical' for manager oversight
    memory=True,
    cache=True
)

result = crew.kickoff()

// agents.config.js
module.exports = {
  agents: [
    {
      name: 'coordinator',
      model: 'claude-3-opus-20240229',
      systemPrompt: `You are a coordinator agent. Your job is to:
      1. Receive high-level tasks
      2. Break them into subtasks
      3. Delegate to specialized agents
      4. Synthesize results into cohesive output`,
      tools: ['delegate', 'synthesize', 'memory'],
      permissions: ['all']
    },
    {
      name: 'code-specialist',
      model: 'claude-3-sonnet-20240229',
      systemPrompt: `You are a code specialist. Handle all programming, debugging, 
      and technical implementation tasks. Use best practices and provide well-documented solutions.`,
      tools: ['read_file', 'write_file', 'execute_command', 'web_search'],
      permissions: ['file_operations', 'execution']
    },
    {
      name: 'research-specialist',
      model: 'claude-3-haiku-20240307',
      systemPrompt: `You are a research specialist. Gather information from web sources, 
      documents, and databases. Provide structured, factual findings with citations.`,
      tools: ['web_search', 'web_fetch', 'memory_search'],
      permissions: ['read_only']
    }
  ],
  
  orchestration: {
    mode: 'hierarchical', // or 'collaborative'
    maxIterations: 10,
    timeoutSeconds: 300
  }
};

┌──────────────────────────────────────────────────────────────┐
│                    Support Request Ingestion                  │
└────────────────────┬─────────────────────────────────────────┘
                     │
        ┌────────────▼────────────┐
        │   Triage Agent            │
        │   - Intent classification │
        │   - Priority scoring      │
        │   - Initial routing       │
        └────────────┬──────────────┘
                     │
       ┌─────────────┼─────────────┐
       │             │             │
┌──────▼──────┐ ┌────▼──────┐ ┌────▼──────┐
│ Billing     │ │ Technical │ │ Account   │
│ Agent       │ │ Agent     │ │ Agent     │
└──────┬──────┘ └─────┬─────┘ └─────┬─────┘
       │              │             │
       └──────────────┼─────────────┘
                      │
           ┌──────────▼──────────┐
           │  Resolution Agent     │
           │  - Quality check       │
           │  - Response synthesis  │
           │  - Customer follow-up  │
           └──────────┬────────────┘
                      │
           ┌──────────▼──────────┐
           │  Feedback Loop       │
           │  - Performance data   │
           │  - Agent optimization │
           └───────────────────────┘

// Triage Agent Node
{
  "type": "n8n-nodes-base.agent",
  "parameters": {
    "name": "Support Triage Agent",
    "systemMessage": `You are a support triage specialist. Analyze incoming support requests:
    
    1. Classify the request category: billing, technical, account, or general
    2. Assign priority: urgent (response <1h), high (<4h), normal (<24h), low (<48h)
    3. Extract key entities: customer_id, product, issue_summary
    4. Route to appropriate specialist agent
    
    Output must be valid JSON with keys: category, priority, entities, routing_decision`,
    
    "options": {
      "temperature": 0.3,
      "responseFormat": "json_object"
    }
  }
}

// Conditional routing based on triage output
{
  "type": "n8n-nodes-base.if",
  "parameters": {
    "conditions": {
      "options": {
        "caseSensitive": true,
        "leftValue": "={{ $json.routing_decision }}",
        "operator": {
          "type": "string",
          "operation": "equals"
        },
        "rightValue": "billing"
      }
    }
  }
}

// Billing Agent Node
{
  "type": "n8n-nodes-base.agent",
  "parameters": {
    "name": "Billing Specialist Agent",
    "systemMessage": `You are a billing support specialist. Handle:
    - Subscription inquiries
    - Payment issues
    - Refund requests
    - Invoice generation
    
    Access customer billing data via the Stripe integration.
    Always verify account security before making changes.
    Escalate refund requests >$500 to supervisor.`,
    
    "tools": [
      {
        "name": "stripe_customer_lookup",
        "description": "Look up customer billing information"
      },
      {
        "name": "payment_history",
        "description": "Retrieve payment and invoice history"
      },
      {
        "name": "process_refund",
        "description": "Process refund requests up to $500"
      }
    ]
  }
}

from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator

class ContentState(TypedDict):
    topic: str
    research_notes: Annotated[list, operator.add]
    outline: dict
    draft: str
    edited_content: str
    seo_analysis: dict
    final_content: str
    stage: str

# Agent definitions
def research_node(state: ContentState):
    """Research agent gathers comprehensive information"""
    llm = ChatOpenAI(model="gpt-4-turbo")
    
    prompt = f"""Research the following topic thoroughly: {state['topic']}
    
    Provide:
    1. Key facts and statistics
    2. Current industry trends
    3. Expert opinions and quotes
    4. Common questions/misconceptions
    5. Competitor content analysis
    
    Format as structured research notes."""
    
    research = llm.invoke(prompt)
    
    return {
        "research_notes": [research.content],
        "stage": "outline"
    }

def outline_node(state: ContentState):
    """Structure agent creates content outline"""
    llm = ChatOpenAI(model="gpt-4-turbo")
    
    prompt = f"""Create a detailed content outline based on these research notes:
    
    Research: {state['research_notes']}
    
    Create an outline with:
    - Compelling headline options
    - Hook/introduction approach
    - Section breakdown with key points
    - Call-to-action recommendations
    - SEO keywords to include
    
    Output as structured JSON."""
    
    outline = llm.invoke(prompt)
    
    return {
        "outline": outline.content,
        "stage": "draft"
    }

def writing_node(state: ContentState):
    """Writing agent produces the content draft"""
    llm = ChatOpenAI(model="gpt-4-turbo")
    
    prompt = f"""Write a comprehensive blog post based on this outline:
    
    Outline: {state['outline']}
    Research: {state['research_notes']}
    
    Requirements:
    - 2,000+ words
    - Engaging, conversational tone
    - Include practical examples
    - Use subheadings and bullet points
    - Add relevant statistics with citations
    
    Write the complete article."""
    
    draft = llm.invoke(prompt)
    
    return {
        "draft": draft.content,
        "stage": "editing"
    }

def editing_node(state: ContentState):
    """Editing agent refines and polishes"""
    llm = ChatOpenAI(model="gpt-4-turbo")
    
    prompt = f"""Edit this content for publication:
    
    Draft: {state['draft']}
    
    Editorial checklist:
    - Fix grammar and spelling
    - Improve clarity and flow
    - Ensure consistent tone
    - Verify factual accuracy
    - Optimize readability (Flesch score >60)
    - Add transition sentences
    - Check for plagiarism/AI-detectability
    
    Provide the edited version with tracked changes summary."""
    
    edited = llm.invoke(prompt)
    
    return {
        "edited_content": edited.content,
        "stage": "seo"
    }

def seo_node(state: ContentState):
    """SEO agent optimizes for search"""
    llm = ChatOpenAI(model="gpt-4-turbo")
    
    prompt = f"""Optimize this content for SEO:
    
    Content: {state['edited_content']}
    
    SEO tasks:
    1. Craft meta title (50-60 chars)
    2. Write meta description (150-160 chars)
    3. Suggest internal linking opportunities
    4. Add schema markup recommendations
    5. Optimize header tags (H1, H2, H3)
    6. Keyword density analysis
    7. Image alt text suggestions
    8. URL slug recommendation
    
    Output JSON with all SEO elements."""
    
    seo = llm.invoke(prompt)
    
    return {
        "seo_analysis": seo.content,
        "final_content": state['edited_content'],
        "stage": "complete"
    }

# Build workflow graph
workflow = StateGraph(ContentState)

workflow.add_node("research", research_node)
workflow.add_node("outline", outline_node)
workflow.add_node("writing", writing_node)
workflow.add_node("editing", editing_node)
workflow.add_node("seo", seo_node)

# Define flow
workflow.add_edge("research", "outline")
workflow.add_edge("outline", "writing")
workflow.add_edge("writing", "editing")
workflow.add_edge("editing", "seo")
workflow.add_edge("seo", END)

workflow.set_entry_point("research")

# Compile
content_pipeline = workflow.compile()

# Execute
result = content_pipeline.invoke({
    "topic": "Multi-Agent AI Systems for Business Automation",
    "stage": "research"
})

// Sales Multi-Agent System in n8n

// Agent 1: Lead Scoring Agent
{
  "name": "Lead Scoring Agent",
  "type": "n8n-nodes-base.agent",
  "parameters": {
    "systemMessage": `You are a lead scoring specialist. Analyze incoming leads and assign scores:
    
    Scoring criteria (0-100):
    - Company size: 0-20 points
    - Job title relevance: 0-20 points  
    - Industry fit: 0-15 points
    - Engagement level: 0-25 points
    - Technology stack alignment: 0-20 points
    
    Categories:
    - 80-100: Hot (immediate outreach)
    - 60-79: Warm (nurture sequence)
    - 40-59: Cool (educational content)
    - 0-39: Cold (long-term nurture)
    
    Output JSON with score, category, and rationale.`,
    
    "options": {
      "responseFormat": "json_object",
      "temperature": 0.2
    }
  }
}

// Agent 2: Research Agent (runs in parallel for hot leads)
{
  "name": "Lead Research Agent",
  "type": "n8n-nodes-base.agent", 
  "parameters": {
    "systemMessage": `Research this lead and their company:
    
    Gather:
    1. Company overview and recent news
    2. Decision maker background
    3. Current technology stack clues
    4. Competitor relationships
    5. Trigger events (funding, expansion, etc.)
    6. Mutual connections
    
    Output structured intelligence report.`,
    
    "tools": [
      {
        "name": "linkedin_lookup",
        "description": "Search LinkedIn for company and person details"
      },
      {
        "name": "company_research",
        "description": "Search web for company news and information"
      },
      {
        "name": "tech_stack_detection",
        "description": "Analyze website for technology clues"
      }
    ]
  }
}

// Agent 3: Outreach Personalization Agent
{
  "name": "Outreach Personalization Agent",
  "type": "n8n-nodes-base.agent",
  "parameters": {
    "systemMessage": `Create personalized outreach messages:
    
    Input: Lead data + Research report
    
    Create:
    1. LinkedIn connection request (300 chars)
    2. Cold email sequence (3 emails)
    3. Call script for SDR
    4. Value proposition tailored to their situation
    
    Tone: Professional but warm
    Avoid: Generic templates, over-familiarity
    Include: Specific insights from research
    
    Output all materials in JSON format.`,
    
    "options": {
      "responseFormat": "json_object",
      "temperature": 0.7
    }
  }
}

// Agent 4: Follow-up Scheduling Agent
{
  "name": "Follow-up Scheduler",
  "type": "n8n-nodes-base.agent",
  "parameters": {
    "systemMessage": `Schedule optimal follow-up sequence:
    
    Based on:
    - Lead category (hot/warm/cool/cold)
    - Industry norms
    - Historical response data
    - Time zones
    
    Create schedule:
    - Touch 1: [Timing + Channel]
    - Touch 2: [Timing + Channel]
    - Touch 3: [Timing + Channel]
    - Touch 4: [Timing + Channel]
    
    Add to calendar and CRM.
    Set reminders for sales rep.`,
    
    "tools": [
      {
        "name": "calendar_schedule",
        "description": "Add tasks to sales team calendar"
      },
      {
        "name": "crm_update",
        "description": "Update CRM with sequence details"
      }
    ]
  }
}

interface AgentMessage {
  messageId: string;
  timestamp: Date;
  from: {
    agentId: string;
    role: string;
  };
  to: {
    agentId: string | 'broadcast';
    role?: string;
  };
  messageType: 'request' | 'response' | 'delegate' | 'notify' | 'query';
  payload: {
    taskId: string;
    context: any;
    priority: 'critical' | 'high' | 'normal' | 'low';
    deadline?: Date;
  };
  metadata: {
    conversationId: string;
    parentMessageId?: string;
    requiresResponse: boolean;
    estimatedEffort: 'quick' | 'medium' | 'complex';
  };
}

// Implementation in n8n
const messageBroker = {
  async sendMessage(message: AgentMessage): Promise<void> {
    // Store in shared state (Redis, database, etc.)
    await storeMessage(message);
    
    // Trigger webhook on recipient agent
    if (message.to.agentId !== 'broadcast') {
      await triggerAgentWebhook(message.to.agentId, message);
    } else {
      // Broadcast to all agents of specified role
      const recipients = await getAgentsByRole(message.to.role);
      await Promise.all(
        recipients.map(agent => triggerAgentWebhook(agent.id, message))
      );
    }
  },
  
  async waitForResponse(
    messageId: string, 
    timeoutMs: number = 30000
  ): Promise<AgentMessage> {
    return new Promise((resolve, reject) => {
      const timer = setTimeout(() => {
        reject(new Error('Response timeout'));
      }, timeoutMs);
      
      subscribeToResponse(messageId, (response) => {
        clearTimeout(timer);
        resolve(response);
      });
    });
  }
};

from typing import Dict, Any, Optional
import redis
from datetime import datetime, timedelta

class AgentSharedMemory:
    """Redis-backed shared memory for agent coordination"""
    
    def __init__(self, redis_client: redis.Redis):
        self.redis = redis_client
        self.ttl = timedelta(hours=24)
    
    def store_context(
        self, 
        conversation_id: str, 
        agent_id: str, 
        context: Dict[str, Any]
    ):
        """Store agent-specific context"""
        key = f"context:{conversation_id}:{agent_id}"
        self.redis.setex(
            key,
            self.ttl,
            json.dumps(context)
        )
    
    def get_shared_context(
        self, 
        conversation_id: str
    ) -> Dict[str, Any]:
        """Retrieve all agents' context for a conversation"""
        pattern = f"context:{conversation_id}:*"
        contexts = {}
        
        for key in self.redis.scan_iter(match=pattern):
            agent_id = key.decode().split(':')[-1]
            value = self.redis.get(key)
            if value:
                contexts[agent_id] = json.loads(value)
        
        return contexts
    
    def append_event(
        self,
        conversation_id: str,
        event: Dict[str, Any]
    ):
        """Append event to conversation log"""
        key = f"events:{conversation_id}"
        event['timestamp'] = datetime.utcnow().isoformat()
        self.redis.rpush(key, json.dumps(event))
        self.redis.expire(key, self.ttl)
    
    def get_event_history(
        self,
        conversation_id: str,
        limit: int = 100
    ) -> list:
        """Retrieve event history"""
        key = f"events:{conversation_id}"
        events = self.redis.lrange(key, -limit, -1)
        return [json.loads(e) for e in events]

# Usage in agents
memory = AgentSharedMemory(redis_client)

# Agent A stores its findings
memory.store_context(
    conversation_id="conv_123",
    agent_id="research_agent",
    context={"findings": [...], "confidence": 0.92}
)

# Agent B retrieves all contexts
all_contexts = memory.get_shared_context("conv_123")
# Use research_agent's findings in analysis

from typing import List, Dict, Any
from collections import defaultdict

class ConsensusManager:
    """Manage multi-agent consensus and conflict resolution"""
    
    def __init__(self, min_agents: int = 3, consensus_threshold: float = 0.6):
        self.min_agents = min_agents
        self.consensus_threshold = consensus_threshold
    
    async def reach_consensus(
        self,
        agents: List[Agent],
        decision_prompt: str,
        options: List[str]
    ) -> Dict[str, Any]:
        """
        Have multiple agents vote on a decision and reach consensus
        """
        votes = defaultdict(list)
        
        # Parallel agent voting
        vote_tasks = [
            agent.vote(decision_prompt, options)
            for agent in agents
        ]
        
        results = await asyncio.gather(*vote_tasks)
        
        # Collect votes
        for agent_id, vote, confidence in results:
            votes[vote].append({
                'agent_id': agent_id,
                'confidence': confidence
            })
        
        # Calculate consensus
        total_votes = len(results)
        vote_counts = {option: len(votes[option]) for option in options}
        
        # Find winning option
        winner = max(vote_counts.items(), key=lambda x: x[1])
        winner_option, winner_count = winner
        consensus_ratio = winner_count / total_votes
        
        if consensus_ratio >= self.consensus_threshold:
            return {
                'decision': winner_option,
                'consensus_reached': True,
                'confidence': consensus_ratio,
                'vote_breakdown': vote_counts,
                'dissenting_agents': [
                    v['agent_id'] 
                    for option, voters in votes.items()
                    for v in voters
                    if option != winner_option
                ]
            }
        else:
            # No consensus - escalate or use weighted decision
            return await self._handle_no_consensus(
                votes, vote_counts, total_votes
            )
    
    async def _handle_no_consensus(
        self,
        votes: Dict,
        vote_counts: Dict,
        total_votes: int
    ) -> Dict[str, Any]:
        """Handle cases where consensus wasn't reached"""
        
        # Option 1: Weight by confidence scores
        weighted_scores = defaultdict(float)
        for option, voters in votes.items():
            for voter in voters:
                weighted_scores[option] += voter['confidence']
        
        winner = max(weighted_scores.items(), key=lambda x: x[1])
        
        # Option 2: Escalate to supervisor agent
        return {
            'decision': winner[0],
            'consensus_reached': False,
            'requires_escalation': True,
            'reason': f'No clear consensus (best option: {winner[1]:.2f} weighted score)',
            'vote_breakdown': vote_counts,
            'weighted_scores': dict(weighted_scores)
        }

# Usage example
consensus = ConsensusManager(min_agents=3)

result = await consensus.reach_consensus(
    agents=[research_agent, analyst_agent, domain_expert],
    decision_prompt="What is the best pricing strategy for our new product?",
    options=['premium', 'competitive', 'penetration', 'freemium']
)

if result['consensus_reached']:
    execute_strategy(result['decision'])
else:
    escalate_to_supervisor(result)

# agent-security-config.yaml

security:
  # 1. Principle of Least Privilege
  permissions:
    default: read_only
    escalate_by_role:
      - role: researcher
        permissions: [read, search, analyze]
      - role: executor
        permissions: [read, write, execute, search]
      - role: supervisor
        permissions: [all]
    
  # 2. Mandatory Approval Gates
  approval_required:
    - action: delete_file
      approvers: [human, supervisor]
    - action: execute_command
      approvers: [supervisor]
      whitelist: ["git", "npm", "pip"]
    - action: external_api_call
      approvers: [supervisor]
      blacklist_domains: ["internal.corp", "localhost"]
    - action: database_write
      approvers: [human]
    
  # 3. Audit and Monitoring
  audit:
    log_all_actions: true
    log_level: INFO
    retention_days: 90
    alert_on:
      - unauthorized_access_attempt
      - permission_escalation
      - unusual_execution_pattern
      - external_communication
      
  # 4. Sandbox and Isolation
  isolation:
    file_system: chroot_jail
    network: restricted_egress
    process: containerized
    resources:
      max_cpu_percent: 50
      max_memory_mb: 2048
      max_execution_time_seconds: 300
      
  # 5. Inter-Agent Verification
  verification:
    require_signature: true
    cross_validate_critical_decisions: true
    consensus_threshold: 0.67
    timeout_seconds: 60

// Security middleware node
{
  "type": "n8n-nodes-base.function",
  "name": "Security Validator",
  "parameters": {
    "functionCode": `
      const action = items[0].json.action;
      const agentId = items[0].json.agent_id;
      const requestedPermissions = items[0].json.permissions;
      
      // Check agent's authorized permissions
      const agentProfile = getAgentProfile(agentId);
      const authorized = agentProfile.permissions;
      
      // Validate requested vs authorized
      const unauthorized = requestedPermissions.filter(
        p => !authorized.includes(p) && !authorized.includes('all')
      );
      
      if (unauthorized.length > 0) {
        // Log security event
        await logSecurityEvent({
          type: 'unauthorized_permission_request',
          agentId,
          requested: unauthorized,
          authorized,
          timestamp: new Date().toISOString()
        });
        
        // Block and alert
        return [{
          json: {
            approved: false,
            reason: 'Unauthorized permissions requested',
            blocked_permissions: unauthorized,
            requires_escalation: true
          }
        }];
      }
      
      // Check approval gates
      const requiresApproval = checkApprovalGates(action, requestedPermissions);
      
      if (requiresApproval) {
        return [{
          json: {
            approved: false,
            reason: 'Requires approval',
            pending_approval: true,
            approvers_required: requiresApproval.approvers
          }
        }];
      }
      
      return [{
        json: {
          approved: true,
          agent_id: agentId,
          action: action
        }
      }];
    `
  }
}

from enum import Enum
from typing import List, Set
from dataclasses import dataclass

class Permission(Enum):
    READ = "read"
    WRITE = "write"
    EXECUTE = "execute"
    DELETE = "delete"
    ADMIN = "admin"
    IMPERSONATE = "impersonate"

@dataclass
class AgentRole:
    name: str
    permissions: Set[Permission]
    max_concurrent_tasks: int
    can_delegate: bool
    can_access_sensitive: bool
    approval_required_for: List[str]

# Define standard roles
ROLES = {
    "observer": AgentRole(
        name="Observer",
        permissions={Permission.READ},
        max_concurrent_tasks=10,
        can_delegate=False,
        can_access_sensitive=False,
        approval_required_for=[]
    ),
    
    "researcher": AgentRole(
        name="Researcher",
        permissions={Permission.READ, Permission.EXECUTE},
        max_concurrent_tasks=5,
        can_delegate=False,
        can_access_sensitive=False,
        approval_required_for=["external_search"]
    ),
    
    "executor": AgentRole(
        name="Executor",
        permissions={Permission.READ, Permission.WRITE, Permission.EXECUTE},
        max_concurrent_tasks=3,
        can_delegate=True,
        can_access_sensitive=False,
        approval_required_for=["file_write", "database_write", "api_call"]
    ),
    
    "administrator": AgentRole(
        name="Administrator",
        permissions={Permission.READ, Permission.WRITE, Permission.EXECUTE, 
                     Permission.DELETE, Permission.ADMIN},
        max_concurrent_tasks=1,
        can_delegate=True,
        can_access_sensitive=True,
        approval_required_for=["permission_grant", "system_config"]
    )
}

class RBACManager:
    def __init__(self):
        self.agent_roles = {}
        self.permission_cache = {}
    
    def assign_role(self, agent_id: str, role_name: str):
        """Assign a role to an agent"""
        if role_name not in ROLES:
            raise ValueError(f"Unknown role: {role_name}")
        
        self.agent_roles[agent_id] = ROLES[role_name]
        self.permission_cache[agent_id] = ROLES[role_name].permissions
    
    def check_permission(self, agent_id: str, permission: Permission) -> bool:
        """Check if agent has a specific permission"""
        if agent_id not in self.permission_cache:
            return False
        return permission in self.permission_cache[agent_id]
    
    def can_execute_action(
        self, 
        agent_id: str, 
        action: str,
        context: dict = None
    ) -> dict:
        """Check if agent can execute specific action"""
        role = self.agent_roles.get(agent_id)
        
        if not role:
            return {
                "allowed": False,
                "reason": "Agent has no assigned role"
            }
        
        # Check approval requirements
        if action in role.approval_required_for:
            return {
                "allowed": False,
                "reason": "Action requires approval",
                "requires_approval_from": ["supervisor"]
            }
        
        # Check concurrent task limit
        current_tasks = get_agent_active_tasks(agent_id)
        if len(current_tasks) >= role.max_concurrent_tasks:
            return {
                "allowed": False,
                "reason": "Agent at concurrent task limit",
                "current_tasks": len(current_tasks),
                "max_tasks": role.max_concurrent_tasks
            }
        
        return {
            "allowed": True,
            "role": role.name,
            "permissions": list(role.permissions)
        }

# enterprise-lead-pipeline.yaml

system:
  name: Enterprise Lead Processing System
  version: 2.0
  description: Multi-agent system for processing and qualifying sales leads

agents:
  # Ingestion Layer
  - name: lead-ingestion-agent
    role: observer
    triggers:
      - webhook: /webhooks/new-lead
      - schedule: "*/5 * * * *"  # Check every 5 minutes
    actions:
      - validate_input
      - enrich_data
      - store_raw
    output: raw_lead_queue
    
  # Qualification Layer
  - name: qualification-agent
    role: researcher
    triggers:
      - queue: raw_lead_queue
    actions:
      - score_lead
      - categorize_lead
      - enrich_firmographic
    tools:
      - clearbit_enrichment
      - linkedin_lookup
      - company_research
    output: scored_lead_queue
    
  # Routing Layer
  - name: routing-agent
    role: executor
    triggers:
      - queue: scored_lead_queue
    actions:
      - determine_sales_rep
      - check_capacity
      - assign_lead
    tools:
      - salesforce_api
      - slack_notify
      - calendar_check
    output: assigned_lead_queue
    
  # Engagement Layer
  - name: engagement-agent
    role: executor
    triggers:
      - queue: assigned_lead_queue
      - schedule: "0 9 * * *"  # Daily at 9 AM
    actions:
      - personalize_outreach
      - schedule_sequence
      - monitor_response
    tools:
      - outreach_api
      - salesloft_api
      - email_templates
    output: engaged_lead_queue
    
  # Follow-up Layer
  - name: followup-agent
    role: executor
    triggers:
      - queue: engaged_lead_queue
      - event: no_response_48h
    actions:
      - adjust_messaging
      - try_alternative_channel
      - update_crm
    tools:
      - multi_channel_sender
      - sentiment_analyzer
    output: nurture_lead_queue
    
  # Supervision Layer
  - name: supervisor-agent
    role: administrator
    triggers:
      - event: lead_escalated
      - event: conflict_detected
      - schedule: "0 */6 * * *"  # Every 6 hours
    actions:
      - review_queue_health
      - reassign_overloaded
      - approve_high_value
      - resolve_conflicts
    tools:
      - all_system_tools

workflows:
  hot_lead_fast_track:
    trigger: lead_score > 80
    steps:
      - parallel:
          - research-agent: deep_research
          - routing-agent: immediate_assignment
      - engagement-agent: priority_sequence
      - notify: sales_manager
      
  standard_lead_process:
    trigger: lead_score 40-79
    steps:
      - enrichment-agent: standard_enrichment
      - routing-agent: round_robin_assignment
      - engagement-agent: standard_sequence
      - followup-agent: monitor_and_nurture
      
  cold_lead_nurture:
    trigger: lead_score < 40
    steps:
      - tagging-agent: long_term_nurture
      - engagement-agent: educational_sequence
      - schedule: re_evaluate_in_90_days

monitoring:
  metrics:
    - lead_processing_time
    - conversion_rate_by_agent
    - agent_response_time
    - conflict_count
    - escalation_rate
    
  alerts:
    - queue_depth > 100
    - processing_time > 5_minutes
    - agent_failure_rate > 5%
    - security_event_detected

// Main orchestrator workflow
{
  "name": "Enterprise Lead Processing",
  "nodes": [
    // Webhook trigger
    {
      "type": "n8n-nodes-base.webhook",
      "name": "Lead Ingestion",
      "parameters": {
        "httpMethod": "POST",
        "path": "lead-ingestion"
      }
    },
    
    // Initial validation
    {
      "type": "n8n-nodes-base.function",
      "name": "Validate Lead",
      "parameters": {
        "functionCode": "// Validation logic"
      }
    },
    
    // Parallel enrichment
    {
      "type": "n8n-nodes-base.executeWorkflow",
      "name": "Run Enrichment Agents",
      "parameters": {
        "workflowId": "enrichment-agents"
      }
    },
    
    // Qualification agent
    {
      "type": "n8n-nodes-base.agent",
      "name": "Qualification Agent",
      "parameters": {
        "systemMessage": "Qualify and score this lead..."
      }
    },
    
    // Conditional routing
    {
      "type": "n8n-nodes-base.switch",
      "name": "Route by Score",
      "parameters": {
        "rules": {
          "rules": [
            {
              "value": "hot",
              "conditions": [{
                "leftValue": "={{ $json.score }}",
                "operator": {
                  "type": "number",
                  "operation": "gt"
                },
                "rightValue": 80
              }]
            },
            {
              "value": "standard",
              "conditions": [{
                "leftValue": "={{ $json.score }}",
                "operator": {
                  "type": "number", 
                  "operation": "between"
                },
                "rightValue": "40,79"
              }]
            }
          ]
        }
      }
    },
    
    // Hot lead fast track
    {
      "type": "n8n-nodes-base.executeWorkflow",
      "name": "Hot Lead Workflow",
      "parameters": {
        "workflowId": "hot-lead-processing"
      }
    },
    
    // Standard processing
    {
      "type": "n8n-nodes-base.executeWorkflow", 
      "name": "Standard Lead Workflow",
      "parameters": {
        "workflowId": "standard-lead-processing"
      }
    }
  ]
}

import functools
import hashlib
import json
from typing import Callable, Any
import redis

class AgentCache:
    """Intelligent caching for agent operations"""
    
    def __init__(self, redis_client: redis.Redis):
        self.redis = redis_client
        self.default_ttl = 3600  # 1 hour
    
    def cache_agent_response(
        self,
        ttl: int = None,
        key_prefix: str = "agent_response"
    ):
        """Decorator for caching agent responses"""
        def decorator(func: Callable) -> Callable:
            @functools.wraps(func)
            async def wrapper(*args, **kwargs):
                # Generate cache key from function signature
                cache_key = self._generate_cache_key(
                    func.__name__, args, kwargs, key_prefix
                )
                
                # Try to get from cache
                cached = self.redis.get(cache_key)
                if cached:
                    return json.loads(cached)
                
                # Execute function
                result = await func(*args, **kwargs)
                
                # Store in cache
                self.redis.setex(
                    cache_key,
                    ttl or self.default_ttl,
                    json.dumps(result)
                )
                
                return result
            return wrapper
        return decorator
    
    def _generate_cache_key(
        self,
        func_name: str,
        args: tuple,
        kwargs: dict,
        prefix: str
    ) -> str:
        """Generate deterministic cache key"""
        key_data = {
            "func": func_name,
            "args": args,
            "kwargs": kwargs
        }
        key_hash = hashlib.md5(
            json.dumps(key_data, sort_keys=True).encode()
        ).hexdigest()
        return f"{prefix}:{key_hash}"
    
    def invalidate_agent_cache(self, agent_id: str):
        """Invalidate all cache entries for an agent"""
        pattern = f"agent_response:*:{agent_id}:*"
        for key in self.redis.scan_iter(match=pattern):
            self.redis.delete(key)

# Usage
@agent_cache.cache_agent_response(ttl=7200)
async def research_agent_query(query: str, agent_id: str):
    """Cached research agent query"""
    # Expensive operation
    return await perform_research(query)

import asyncio
from typing import List, Coroutine

class ParallelAgentExecutor:
    """Execute multiple agents in parallel with result aggregation"""
    
    def __init__(self, max_concurrent: int = 10):
        self.max_concurrent = max_concurrent
        self.semaphore = asyncio.Semaphore(max_concurrent)
    
    async def execute_agents_parallel(
        self,
        agents: List[Agent],
        task: dict,
        aggregation_strategy: str = "all"
    ) -> dict:
        """
        Execute multiple agents in parallel
        
        aggregation_strategy:
        - "all": Wait for all results
        - "first": Return first successful result
        - "majority": Return result agreed upon by majority
        - "best": Return result with highest confidence
        """
        
        async def execute_with_semaphore(agent):
            async with self.semaphore:
                return await agent.execute(task)
        
        # Execute all agents
        tasks = [execute_with_semaphore(agent) for agent in agents]
        results = await asyncio.gather(*tasks, return_exceptions=True)
        
        # Process results based on strategy
        if aggregation_strategy == "all":
            return {
                "results": [
                    r for r in results 
                    if not isinstance(r, Exception)
                ],
                "errors": [
                    str(r) for r in results 
                    if isinstance(r, Exception)
                ]
            }
        
        elif aggregation_strategy == "first":
            for result in results:
                if not isinstance(result, Exception):
                    return result
            raise Exception("All agents failed")
        
        elif aggregation_strategy == "best":
            valid_results = [
                r for r in results 
                if not isinstance(r, Exception) and r.get("confidence")
            ]
            if not valid_results:
                raise Exception("No valid results")
            return max(valid_results, key=lambda x: x["confidence"])
        
        elif aggregation_strategy == "majority":
            return self._majority_vote(results)
    
    def _majority_vote(self, results: List[dict]) -> dict:
        """Determine majority consensus from results"""
        # Group results by their output
        groups = {}
        for result in results:
            if isinstance(result, Exception):
                continue
            key = json.dumps(result.get("output"), sort_keys=True)
            if key not in groups:
                groups[key] = []
            groups[key].append(result)
        
        # Find majority
        majority_key = max(groups.keys(), key=lambda k: len(groups[k]))
        return {
            "output": json.loads(majority_key),
            "confidence": len(groups[majority_key]) / len(results),
            "supporting_agents": [
                r.get("agent_id") for r in groups[majority_key]
            ]
        }

# Usage
executor = ParallelAgentExecutor(max_concurrent=5)

results = await executor.execute_agents_parallel(
    agents=[research_agent_1, research_agent_2, research_agent_3],
    task={"query": "market analysis for AI automation"},
    aggregation_strategy="majority"
)

// Agent composition from marketplace
const agentSystem = await AgentComposer.create({
  agents: [
    { id: "research-agent-v2.3", source: "marketplace:anthropic" },
    { id: "analysis-agent-v1.5", source: "marketplace:openai" },
    { id: "custom-reporting-agent", source: "local:./agents/" }
  ],
  orchestration: "hierarchical",
  memory: "shared:redis://localhost:6379"
});

class SelfImprovingAgent:
    def __init__(self):
        self.performance_history = []
        self.optimization_agent = OptimizationAgent()
    
    async def execute_with_improvement(self, task):
        result = await self.execute(task)
        
        # Evaluate performance
        metrics = self.evaluate_performance(result)
        self.performance_history.append(metrics)
        
        # Trigger optimization if needed
        if metrics['score'] < 0.8:
            improvements = await self.optimization_agent.suggest_improvements(
                self.performance_history
            )
            await self.apply_improvements(improvements)
        
        return result

compliance_agents:
  - name: gdpr-compliance-agent
    responsibilities:
      - verify_data_processing_legality
      - ensure_consent_documentation
      - manage_right_to_be_forgotten
      - audit_data_retention
      
  - name: accessibility-compliance-agent
    responsibilities:
      - verify_wcag_compliance
      - check_screen_reader_compatibility
      - validate_keyboard_navigation
      - ensure_color_contrast
      
  - name: security-compliance-agent
    responsibilities:
      - verify_encryption_standards
      - check_access_controls
      - audit_permission_grants
      - monitor_anomaly_detection