"""Main installer logic.""" import asyncio import logging import platform import sys from pathlib import Path from rich.console import Console from rich.panel import Panel from rich.prompt import Prompt from ..agent_discovery import AgentDiscovery, AgentMetadata from .agent_selector import AgentSelector from .config_generator import ConfigGenerator from .mcp_configurator import MCPConfigurator from .task_mapper import TASK_CATEGORIES, TaskMapper logger = logging.getLogger(__name__) console = Console() class DelegationInstaller: """Automated installer for delegation-mcp.""" def __init__(self): self.project_dir = Path.cwd() self.agent_discovery = AgentDiscovery() self.mcp_configurator = MCPConfigurator() self.agent_selector = AgentSelector() self.task_mapper = TaskMapper() self.config_generator = ConfigGenerator() self.discovered_agents: dict[str, AgentMetadata] = {} self.selected_agents: list[str] = [] self.task_mappings: dict[str, str] = {} def install(self) -> bool: """Run full installation.""" try: self._welcome() if not self._check_system(): return False asyncio.run(self._discover_agents()) if not self.discovered_agents: self._no_agents_guide() return False # Select which agents to enable self.selected_agents = self._select_agents() if len(self.selected_agents) < 2: console.print("[red]Need at least 2 agents for delegation.[/red]") return False # Select primary orchestrator self.selected_orchestrator = "claude" # Check if Claude is available if "claude" not in self.discovered_agents: console.print("[yellow]Claude Code not detected. Please install it first: npm install -g @anthropic/claude-code[/yellow]") return False # Map tasks to agents if len(self.selected_agents) >= 2: self.task_mappings = self._map_tasks() # Ask about installation scope install_project_instructions, install_user_instructions, mcp_scope = self._ask_user_level_instructions() # Get the actual path to Claude executable orchestrator_path = None if "claude" in self.discovered_agents: orchestrator_path = self.discovered_agents["claude"].path if not self._configure_mcp(install_project_instructions, install_user_instructions, orchestrator_path, mcp_scope): return False self._print_success() return True except KeyboardInterrupt: console.print("\n[yellow]Installation cancelled[/yellow]") return False except Exception as e: console.print(f"\n[red]Installation failed: {e}[/red]") logger.error(f"Installation error: {e}", exc_info=True) return False def _welcome(self): """Print welcome message.""" console.print(Panel.fit( "[bold blue]Delegation MCP Installer[/bold blue]\n" "Automated multi-agent orchestration setup", border_style="blue" )) def _check_system(self) -> bool: """Check system requirements.""" console.print("\n[bold]Checking system requirements...[/bold]") # Check Python version if sys.version_info < (3, 10): console.print(f"[red]X Python 3.10+ required (found {sys.version_info.major}.{sys.version_info.minor})[/red]") return False console.print(f"[green]OK Python {sys.version_info.major}.{sys.version_info.minor}[/green]") # Check OS system = platform.system() console.print(f"[green]OK {system}[/green]") return True async def _discover_agents(self): """Discover available agents.""" console.print("\n[bold]Discovering agents...[/bold]") discovered = await self.agent_discovery.discover_agents(force_refresh=True) self.discovered_agents = {k: v for k, v in discovered.items() if v.available} if self.discovered_agents: console.print(f"[green]Found {len(self.discovered_agents)} agents:[/green]") for name, meta in self.discovered_agents.items(): console.print(f" [green]OK[/green] {name} ({meta.version})") else: console.print("[yellow]No agents detected[/yellow]") def _no_agents_guide(self): """Guide user to install agents.""" console.print("\n[bold yellow]No agents detected![/bold yellow]\n") console.print("Install Claude Code:") console.print(" • Claude Code: npm install -g @anthropic/claude-code\n") console.print("Then run: [bold]python install.py[/bold]") def _select_agents(self) -> list[str]: """Let user select which agents to enable.""" return self.agent_selector.prompt_selection(self.discovered_agents) def _map_tasks(self) -> dict[str, str]: """Interactive task-to-agent mapping.""" console.print("\n[bold]Task Assignment Configuration[/bold]") console.print("Configure which agents should handle which types of tasks.\n") return self.task_mapper.map_tasks(self.selected_agents) def _ask_user_level_instructions(self) -> tuple[bool, bool, str]: """ Ask where to install system instructions and MCP server. Returns: Tuple of (install_project_level, install_user_level, mcp_scope) """ console.print("\n[bold]Installation Scope:[/bold]") console.print(" 1. Project-level only (.claude/CLAUDE.md + local MCP) - Only for this project") console.print(" 2. User-level only (~/.claude/CLAUDE.md + global MCP) - For ALL Claude sessions") console.print(" 3. Both project and user level (project instructions + global MCP)") console.print(f"\n[dim]Recommended: Option 1 for project-specific setup, or 2 for global access.[/dim]") choice = Prompt.ask( "Choose installation scope", choices=["1", "2", "3"], default="1" ) if choice == "1": return (True, False, "local") # Project only elif choice == "2": return (False, True, "user") # User only else: return (True, True, "user") # Both (use user scope for MCP) def _configure_mcp(self, install_project_instructions: bool = True, install_user_instructions: bool = False, orchestrator_path: Path = None, scope: str = "local") -> bool: """Configure MCP client for Claude (the only orchestrator).""" console.print(f"\n[bold]Configuring MCP client and delegation rules for Claude...[/bold]") # Generate delegation configuration files if self.task_mappings and len(self.selected_agents) >= 2: console.print("\n[cyan]Generating delegation configuration files...[/cyan]") self.config_generator.generate_configs( selected_agents=self.selected_agents, task_mappings=self.task_mappings, agent_metadata=self.discovered_agents, primary_orchestrator="claude", project_dir=self.project_dir, scope=scope, ) console.print("[green]✓ Configuration files generated[/green]") # Configure MCP server results = self.mcp_configurator.inject_delegation_mcp( self.project_dir, "claude", # Claude is the only orchestrator install_project_instructions, install_user_instructions, orchestrator_path, scope, task_mappings=self.task_mappings, selected_agents=self.selected_agents, ) configured_any = False for client, success in results.get("clients", {}).items(): if success: console.print(f"[green]OK {client}: delegation-mcp registered[/green]") configured_any = True else: console.print(f"[yellow]! {client} not configured automatically[/yellow]") for message in results.get("messages", []): console.print(message) manual_steps = results.get("manual_instructions", []) if manual_steps: console.print("\n[bold]Manual setup required:[/bold]") for step in manual_steps: console.print(f" [cyan]{step}[/cyan]") # Claude supports auto-injection, so no manual instructions needed if not configured_any and not results.get("allow_continue", False): console.print("[red]X Failed to configure MCP client automatically.[/red]") return False return True def _print_success(self): """Print success message.""" console.print(f"\n[bold cyan]For Claude:[/bold cyan]") console.print(" 1. Restart Claude Desktop app") console.print(" 2. Open a chat and try: [cyan]'scan for security vulnerabilities'[/cyan]") console.print("\n[bold]The system will automatically:[/bold]") # Dynamic task routing summary grouped by agent if self.task_mappings: # Create a lookup for task names task_names = {cat["key"]: cat["name"] for cat in TASK_CATEGORIES} # Group tasks by agent agent_tasks = {} for task_key, agent in self.task_mappings.items(): task_name = task_names.get(task_key, task_key) if agent not in agent_tasks: agent_tasks[agent] = [] agent_tasks[agent].append(task_name) # Display grouped by agent for agent, tasks in sorted(agent_tasks.items()): tasks_str = ", ".join(tasks) console.print(f" • Route {tasks_str} to {agent}") console.print(" • Fall back if agent fails") def main(): """CLI entry point.""" logging.basicConfig(level=logging.INFO) installer = DelegationInstaller() success = installer.install() sys.exit(0 if success else 1) if __name__ == "__main__": main()