Spaces:
Sleeping
Sleeping
| """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 | |