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