Delegation MCP

"""Configuration models for delegation MCP server.""" from typing import Any, Literal from pathlib import Path from pydantic import BaseModel, Field import yaml import re from collections import defaultdict class ConfigValidationError(Exception): """Exception raised when configuration validation fails.""" def __init__(self, errors: list[str]): self.errors = errors super().__init__(f"Configuration validation failed: {'; '.join(errors)}") class AgentCapabilities(BaseModel): """Capability scores for an agent.""" security_audit: float = 0.5 vulnerability_scan: float = 0.5 code_review: float = 0.5 architecture: float = 0.5 refactoring: float = 0.5 quick_fix: float = 0.5 documentation: float = 0.5 testing: float = 0.5 performance: float = 0.5 git_workflow: float = 0.5 # Git operations (commit, push, merge, rebase) github_operations: float = 0.5 # GitHub operations (PR, issues, releases) general: float = 0.5 # Fallback capability class OrchestratorConfig(BaseModel): """Configuration for a single orchestrator/CLI.""" name: str command: str | list[str] args: list[str] = Field(default_factory=list) enabled: bool = True env: dict[str, str] = Field(default_factory=dict) timeout: int = 300 # seconds max_retries: int = 3 capabilities: AgentCapabilities = Field(default_factory=AgentCapabilities) cost_per_1k_tokens: float = 0.001 # Cost estimate for routing decisions class DelegationRule(BaseModel): """Rule for delegating tasks to specific orchestrators.""" pattern: str # Regex pattern to match delegate_to: str # Target orchestrator name priority: int = 0 # Higher priority wins requires_approval: bool = False description: str = "" class DelegationConfig(BaseModel): """Main configuration for delegation MCP server.""" orchestrator: Literal["claude"] = "claude" # Primary orchestrator orchestrators: dict[str, OrchestratorConfig] = Field(default_factory=dict) rules: list[DelegationRule] = Field(default_factory=list) auto_approve: bool = False log_delegations: bool = True routing_strategy: Literal["capability", "pattern", "hybrid"] = "capability" # Routing approach def to_yaml(self, path: Path) -> None: """Save configuration to YAML file.""" with open(path, "w", encoding="utf-8") as f: yaml.dump(self.model_dump(), f, default_flow_style=False) def get_orchestrator(self, name: str) -> OrchestratorConfig | None: """Get orchestrator configuration by name.""" return self.orchestrators.get(name) def find_delegation_rule(self, query: str) -> DelegationRule | None: """Find matching delegation rule for query.""" matching_rules = [ rule for rule in self.rules if re.search(rule.pattern, query, re.IGNORECASE) ] if not matching_rules: return None # Return highest priority rule return max(matching_rules, key=lambda r: r.priority) def _validate_minimum_agents(self) -> list[str]: """Validate that at least 2 agents are enabled.""" errors = [] enabled_count = sum( 1 for orch in self.orchestrators.values() if orch.enabled ) if enabled_count < 2: errors.append( f"At least 2 agents must be enabled, but only {enabled_count} " f"{'is' if enabled_count == 1 else 'are'} enabled. " f"Enable more agents in orchestrators configuration." ) return errors def _validate_regex_patterns(self) -> list[str]: """Validate YAML regex syntax in routing rules.""" errors = [] for i, rule in enumerate(self.rules): try: # Attempt to compile the regex pattern re.compile(rule.pattern) except re.error as e: errors.append( f"Rule #{i + 1} (delegate_to: {rule.delegate_to}): " f"Invalid regex pattern '{rule.pattern}': {str(e)}" ) return errors def _validate_agent_references(self) -> list[str]: """Validate that all referenced agents exist.""" errors = [] # Check primary orchestrator exists if self.orchestrator not in self.orchestrators: errors.append( f"Primary orchestrator '{self.orchestrator}' is not defined " f"in orchestrators configuration. Available orchestrators: " f"{', '.join(self.orchestrators.keys())}" ) # Check all delegation rule targets exist for i, rule in enumerate(self.rules): if rule.delegate_to not in self.orchestrators: errors.append( f"Rule #{i + 1} (pattern: '{rule.pattern}'): " f"Target orchestrator '{rule.delegate_to}' is not defined. " f"Available orchestrators: {', '.join(self.orchestrators.keys())}" ) return errors def _validate_no_circular_delegation(self) -> list[str]: """ Validate that there are no obvious circular delegation patterns. Note: In the current delegation system, queries are processed once and returned to the user - there is no automatic re-delegation. Therefore, true circular delegation is not possible. This check looks for potential issues like: - Duplicate patterns delegating to different orchestrators (ambiguous routing) - Rules with very similar patterns that might cause confusion Since circular delegation is not a real concern in this architecture, this validation performs only basic sanity checks. """ errors = [] if not self.rules: return errors # Check for duplicate or highly similar patterns # This could cause ambiguous delegation behavior pattern_targets: dict[str, list[str]] = defaultdict(list) for rule in self.rules: # Normalize pattern for comparison (case-insensitive) normalized_pattern = rule.pattern.lower().strip() pattern_targets[normalized_pattern].append(rule.delegate_to) # Check for exact duplicate patterns with different targets for pattern, targets in pattern_targets.items(): unique_targets = set(targets) if len(unique_targets) > 1: # Same pattern delegates to multiple different orchestrators # This is actually fine - highest priority wins # But if priorities are the same, it could be ambiguous # Find rules with this pattern matching_rules = [ r for r in self.rules if r.pattern.lower().strip() == pattern ] # Check if they have the same priority priorities = {r.priority for r in matching_rules} if len(priorities) == 1: # All have same priority - ambiguous! targets_str = ", ".join(sorted(unique_targets)) errors.append( f"Ambiguous delegation: pattern '{pattern}' delegates to " f"multiple orchestrators ({targets_str}) with the same priority. " f"This could cause unpredictable behavior. Consider using " f"different priorities or combining into a single rule." ) return errors def validate(self) -> None: """ Validate the entire configuration. Performs the following validations: 1. At least 2 agents are enabled 2. All regex patterns in routing rules are valid 3. All referenced agents exist in orchestrators 4. No circular delegation rules exist Raises: ConfigValidationError: If any validation fails, with detailed error messages. """ all_errors: list[str] = [] # Run all validation checks all_errors.extend(self._validate_minimum_agents()) all_errors.extend(self._validate_regex_patterns()) all_errors.extend(self._validate_agent_references()) all_errors.extend(self._validate_no_circular_delegation()) # Raise exception if any errors found if all_errors: raise ConfigValidationError(all_errors) @classmethod def from_yaml(cls, path: Path, validate: bool = True) -> "DelegationConfig": """ Load configuration from YAML file. Args: path: Path to YAML configuration file validate: Whether to validate the configuration after loading (default: True) Returns: DelegationConfig instance Raises: ConfigValidationError: If validation is enabled and fails """ with open(path, encoding="utf-8") as f: data = yaml.safe_load(f) config = cls(**data) if validate: config.validate() return config