installer.py•10.4 kB
"""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()