Delegation MCP

"""MCP client auto-configuration.""" import json import logging import os import platform import shutil from pathlib import Path logger = logging.getLogger(__name__) class MCPConfigurator: """Auto-configure MCP clients (Claude Code and Claude Desktop).""" def __init__(self, client_type: str = "auto"): """ Initialize configurator. Args: client_type: "code", "desktop", or "auto" to detect """ self.client_type = client_type self.config_path = None if client_type in ("desktop", "auto"): self.desktop_config_path = self._detect_desktop_config_path() else: self.desktop_config_path = None def _detect_desktop_config_path(self) -> Path: """Detect Claude Desktop config location (platform-aware).""" system = platform.system() if system == "Windows": locations = [ Path(os.environ.get("APPDATA", "")) / "Claude" / "claude_desktop_config.json", Path.home() / "AppData" / "Roaming" / "Claude" / "claude_desktop_config.json", ] elif system == "Darwin": # macOS locations = [ Path.home() / "Library" / "Application Support" / "Claude" / "claude_desktop_config.json", ] else: # Linux locations = [ Path.home() / ".config" / "Claude" / "claude_desktop_config.json", ] for loc in locations: if loc.exists(): logger.info(f"Found Claude Desktop config: {loc}") return loc # Default fallback fallback = locations[0] logger.info(f"Will create new Desktop config: {fallback}") return fallback def configure_claude_code( self, project_dir: Path, system_instructions: str = None, install_project_instructions: bool = True, install_user_instructions: bool = False, claude_path: Path = None, scope: str = "local" ) -> tuple[bool, bool]: """ Configure MCP for Claude Code using 'claude mcp add' command and inject system instructions. Args: project_dir: Project directory path system_instructions: System instructions text to inject install_project_instructions: If True, install CLAUDE.md at project level install_user_instructions: If True, install CLAUDE.md at user level (~/.claude/) claude_path: Path to claude executable (if None, uses 'claude' from PATH) scope: Configuration scope - "local" (default), "project", or "user" Returns: Tuple of (mcp_configured, instructions_configured) """ import subprocess mcp_configured = False instructions_configured = False # 1. Configure MCP server try: # Determine claude command to use claude_cmd = str(claude_path) if claude_path else "claude" # First, try to remove any existing delegation-mcp server to ensure clean install try: remove_result = subprocess.run( [claude_cmd, "mcp", "remove", "delegation-mcp"], cwd=str(project_dir), capture_output=True, text=True, timeout=10 ) if remove_result.returncode == 0: logger.info("Removed existing delegation-mcp server before reinstalling") except Exception as e: logger.debug(f"No existing delegation-mcp to remove (or removal failed): {e}") # Use the official claude mcp add command with scope cmd = [ claude_cmd, "mcp", "add", "--scope", scope, "delegation-mcp", "delegation-mcp" ] result = subprocess.run( cmd, cwd=str(project_dir), capture_output=True, text=True, timeout=30 ) if result.returncode == 0: logger.info(f"Configured delegation-mcp for Claude Code using 'claude mcp add'") # Verify the server was actually added try: verify_result = subprocess.run( [claude_cmd, "mcp", "list"], cwd=str(project_dir), capture_output=True, text=True, timeout=10 ) if verify_result.returncode == 0 and "delegation-mcp" in verify_result.stdout: logger.info("Verified delegation-mcp is registered with Claude Code") mcp_configured = True else: logger.warning(f"delegation-mcp not found in 'claude mcp list' output") mcp_configured = False except Exception as e: logger.warning(f"Could not verify MCP server registration: {e}") # Still mark as configured if 'claude mcp add' succeeded mcp_configured = True else: logger.warning(f"'claude mcp add' failed (exit code {result.returncode}): {result.stderr}") except FileNotFoundError: logger.warning(f"Claude executable not found at: {claude_cmd}") except subprocess.TimeoutExpired: logger.error("'claude mcp add' command timed out") except Exception as e: logger.error(f"Failed to configure Claude Code MCP: {e}") # 2. Inject system instructions if provided if system_instructions: # Helper function to safely add instructions to a file def add_instructions_to_file(file_path: Path, backup: bool = True): """Add or update delegation instructions using marker tags.""" import re try: # Define markers BEGIN_MARKER = "<!-- BEGIN DELEGATION-MCP INSTRUCTIONS -->" END_MARKER = "<!-- END DELEGATION-MCP INSTRUCTIONS -->" # Wrap instructions with markers wrapped_instructions = f"{BEGIN_MARKER}\n{system_instructions}\n{END_MARKER}" existing_content = "" if file_path.exists(): with open(file_path, "r", encoding="utf-8") as f: existing_content = f.read() # Backup existing file if backup: backup_path = file_path.with_suffix(".md.backup") shutil.copy2(file_path, backup_path) logger.info(f"Backed up existing file to: {backup_path}") # Check if markers exist (update case) marker_pattern = re.compile( r"<!-- BEGIN DELEGATION-MCP INSTRUCTIONS.*?-->\n.*?\n<!-- END DELEGATION-MCP INSTRUCTIONS -->", re.DOTALL ) if existing_content and marker_pattern.search(existing_content): # Replace existing wrapped section new_content = marker_pattern.sub(wrapped_instructions, existing_content) logger.info(f"Replaced existing delegation instructions in {file_path}") elif existing_content.strip(): # Prepend wrapped instructions to existing content new_content = wrapped_instructions + "\n\n" + "="*80 + "\n" new_content += "# Existing Instructions\n" new_content += "="*80 + "\n\n" new_content += existing_content logger.info(f"Prepended delegation instructions to {file_path}") else: # New file, just write wrapped instructions new_content = wrapped_instructions logger.info(f"Created new delegation instructions in {file_path}") # Create parent directory if needed file_path.parent.mkdir(parents=True, exist_ok=True) # Write the file with open(file_path, "w", encoding="utf-8") as f: f.write(new_content) return True except Exception as e: logger.warning(f"Failed to add instructions to {file_path}: {e}") return False # Project-level CLAUDE.md if install_project_instructions: project_claude_dir = project_dir / ".claude" project_claude_md = project_claude_dir / "CLAUDE.md" if add_instructions_to_file(project_claude_md, backup=True): instructions_configured = True # User-level CLAUDE.md if install_user_instructions: user_claude_dir = Path.home() / ".claude" user_claude_md = user_claude_dir / "CLAUDE.md" if add_instructions_to_file(user_claude_md, backup=True): instructions_configured = True return (mcp_configured, instructions_configured) def configure_claude_desktop(self, project_dir: Path) -> bool: """Configure MCP for Claude Desktop using global config.""" try: # Backup existing config if self.desktop_config_path.exists(): backup_path = self.desktop_config_path.with_suffix(".json.backup") shutil.copy2(self.desktop_config_path, backup_path) logger.info(f"Backed up Desktop config to: {backup_path}") # Load existing config if self.desktop_config_path.exists(): with open(self.desktop_config_path) as f: config = json.load(f) else: config = {} # Ensure mcpServers exists if "mcpServers" not in config: config["mcpServers"] = {} # Add delegation-mcp config["mcpServers"]["delegation-mcp"] = { "command": "uv", "args": [ "--directory", str(project_dir), "run", "delegation-mcp" ] } # Save config self.desktop_config_path.parent.mkdir(parents=True, exist_ok=True) with open(self.desktop_config_path, "w") as f: json.dump(config, f, indent=2) self.config_path = self.desktop_config_path logger.info(f"Configured delegation-mcp for Claude Desktop: {self.desktop_config_path}") return True except Exception as e: logger.error(f"Failed to configure Claude Desktop: {e}") return False def inject_delegation_mcp( self, project_dir: Path, orchestrator: str, install_project_instructions: bool = True, install_user_instructions: bool = False, orchestrator_path: Path = None, scope: str = "local", task_mappings: dict[str, str] | None = None, selected_agents: list[str] | None = None, ) -> dict[str, object]: """ Inject delegation-mcp server into config(s). Args: project_dir: Project directory path orchestrator: Name of the orchestrator (claude, gemini, etc.) install_project_instructions: If True, install system instructions at project level install_user_instructions: If True, install system instructions at user level orchestrator_path: Path to orchestrator executable scope: Configuration scope for Claude Code - "local" (default), "project", or "user" task_mappings: Dictionary mapping task categories to agent names selected_agents: List of selected agent names Returns: Dict containing client results, status, and manual instructions """ from .system_instructions import get_system_instructions orchestrator = (orchestrator or "").lower() outcome: dict[str, object] = { "orchestrator": orchestrator, "clients": {}, "messages": [], "manual_instructions": [], "system_instructions": get_system_instructions( orchestrator, task_mappings=task_mappings, selected_agents=selected_agents ), "allow_continue": False, } if orchestrator == "claude": system_instructions = str(outcome["system_instructions"]) if self.client_type in ("code", "auto"): mcp_configured, instructions_configured = self.configure_claude_code( project_dir, system_instructions, install_project_instructions, install_user_instructions, orchestrator_path, scope ) code_success = mcp_configured or instructions_configured outcome["clients"]["Claude Code"] = code_success if code_success: # Determine what was actually configured configured_items = [] if mcp_configured: configured_items.append("MCP server") if instructions_configured: configured_items.append("system instructions") outcome["messages"].append( f"[green]✓ Claude Code: {', '.join(configured_items)} configured![/green]" ) if mcp_configured: outcome["messages"].append( "[dim] • MCP server verified with 'claude mcp list'[/dim]" ) if install_project_instructions: outcome["messages"].append( f"[dim] • System instructions (project): {project_dir / '.claude' / 'CLAUDE.md'}[/dim]" ) if install_user_instructions: user_claude_md = Path.home() / ".claude" / "CLAUDE.md" outcome["messages"].append( f"[dim] • System instructions (user): {user_claude_md}[/dim]" ) # Only show manual instructions if MCP server wasn't configured if not mcp_configured: outcome["messages"].append( "[yellow]! MCP server not added - manual setup required:[/yellow]" ) outcome["manual_instructions"].append("claude mcp add delegation-mcp delegation-mcp") outcome["manual_instructions"].append( "Then verify with: claude mcp list" ) else: outcome["manual_instructions"].append("claude mcp add delegation-mcp delegation-mcp") outcome["manual_instructions"].append( "Add system instructions from delegation_instructions.txt to .claude/CLAUDE.md" ) if self.client_type in ("desktop", "auto"): desktop_success = self.configure_claude_desktop(project_dir) outcome["clients"]["Claude Desktop"] = desktop_success if desktop_success and not self.config_path: self.config_path = self.desktop_config_path outcome["allow_continue"] = True return outcome outcome["allow_continue"] = True return outcome outcome["messages"].append( f"[yellow]! Automatic MCP configuration for '{orchestrator}' is not supported yet.[/yellow]" ) outcome["manual_instructions"].append( "Please register the delegation-mcp server manually in your MCP client." ) outcome["allow_continue"] = True return outcome def verify_config(self) -> bool: """Verify config is valid JSON and has delegation-mcp.""" if not self.config_path or not self.config_path.exists(): return False try: with open(self.config_path) as f: config = json.load(f) return "mcpServers" in config and "delegation-mcp" in config["mcpServers"] except Exception as e: logger.error(f"Config verification failed: {e}") return False