Delegation MCP

Instructions

Implementation Reference

• src/delegation_mcp/server.py:215-232 (handler)

  The main tool handler in call_tool that processes the query, calls classification and delegation logic from the engine, and returns the recommended agent.
  if name == "get_routing_guidance": # Get routing guidance without executing the task query = arguments["query"] # Classify the task to determine routing task_info = self.engine._classify_task(query) task_type = task_info[0] if isinstance(task_info, tuple) else task_info timeout = task_info[1] if isinstance(task_info, tuple) and len(task_info) > 1 else 300 # Determine which agent should handle it agent, _ = self.engine._determine_delegation(query, None) # KISS: Just return the agent name # - "gemini" / "aider" / "copilot" → delegate to that agent # - "claude" → orchestrator handles directly (since Claude is the orchestrator) response = agent if agent else "claude" return [TextContent(type="text", text=response)]
• src/delegation_mcp/server.py:170-183 (registration)

  Tool registration in list_tools(), including name, description, and input schema.
  Tool( name="get_routing_guidance", description="Get routing guidance for a task - returns which agent should handle it and the exact CLI command to run (guidance only, no execution)", inputSchema={ "type": "object", "properties": { "query": { "type": "string", "description": "The task query to get routing guidance for", }, }, "required": ["query"], }, ),
• src/delegation_mcp/delegation.py:323-363 (helper)

  Helper method _classify_task used by the handler to determine task type and recommended timeout based on keyword matching.
  def _classify_task(self, query: str) -> tuple[str, int]: """Classify task type and return recommended timeout.""" query_lower = query.lower() keywords = { "security_audit": ["security", "vulnerability", "audit", "cve", "exploit", "penetration"], "vulnerability_scan": ["scan", "vulnerability", "vuln", "security issue"], "code_review": ["review", "code quality", "best practice", "lint"], "architecture": ["architecture", "design", "system design", "structure"], "refactoring": ["refactor", "restructure", "clean up", "improve code"], "quick_fix": ["fix", "bug", "error", "issue", "broken"], "documentation": ["document", "docs", "readme", "guide", "explain"], "testing": ["test", "unittest", "integration test", "e2e"], "performance": ["performance", "optimize", "speed", "latency", "benchmark"], "git_workflow": ["commit", "push", "rebase", "merge", "cherry-pick", "squash", "git history"], "github_operations": ["pull request", "pr create", "pr review", "issue create", "release"], } # Timeout presets based on task complexity TIMEOUT_PRESETS = { "quick_fix": 60, # 1 min - simple bug fixes "refactoring": 300, # 5 min - code refactoring "security_audit": 600, # 10 min - comprehensive security review "code_review": 600, # 10 min - full code review "performance": 900, # 15 min - profiling/optimization "testing": 300, # 5 min - test generation "documentation": 180, # 3 min - documentation writing "architecture": 300, # 5 min - design work "vulnerability_scan": 300, # 5 min - automated scanning "git_workflow": 180, # 3 min - git operations "github_operations": 240, # 4 min - GitHub API operations "general": 300, # 5 min - default } for task_type, terms in keywords.items(): if any(term in query_lower for term in terms): timeout = TIMEOUT_PRESETS.get(task_type, 300) return task_type, timeout return "general", 300
• src/delegation_mcp/delegation.py:392-435 (helper)

  Helper method _determine_delegation used by the handler to decide the target agent based on complexity, rules, and capability ranking.
  def _determine_delegation( self, query: str, force_delegate: str | None, ) -> tuple[str, DelegationRule | None]: """ Determine which orchestrator should handle the query using capability-based routing. Returns: tuple: (target_orchestrator, matching_rule) """ # Force delegation overrides everything if force_delegate: logger.info(f"Routing: FORCED → {force_delegate}") return force_delegate, None # Check task complexity first - simple tasks handled directly by Claude complexity = self._estimate_task_complexity(query) if complexity == "simple": logger.info(f"Routing: SIMPLE task → claude (delegation overhead not worth it)") return "claude", None # Check explicit delegation rules rule = self.config.find_delegation_rule(query) if rule: logger.info(f"Routing: {rule.pattern} → {rule.delegate_to} (rule-based)") return rule.delegate_to, rule # Use capability-based routing for medium/complex tasks if self.config.routing_strategy in ["capability", "hybrid"]: ranked = self._rank_by_capabilities(query) if ranked: task_type, _ = self._classify_task(query) # Unpack tuple # If top ranked agent is Claude, check if delegation is still worth it if ranked[0] == "claude" and complexity == "medium": logger.info(f"Routing: {task_type} → claude (best match, medium complexity)") return "claude", None logger.info(f"Routing: {task_type} [{complexity}] → {ranked[0]} (capability-based)") return ranked[0], None # Fallback to primary orchestrator logger.info(f"Routing: DEFAULT → claude") return "claude", None