Delegation MCP

Instructions

Implementation Reference

• src/delegation_mcp/server.py:234-275 (handler)

  MCP tool handler for 'discover_agents': invokes AgentDiscovery, auto-registers new agents, formats and returns discovery summary as text.

  elif name == "discover_agents":
      # Discover available agents
      force_refresh = arguments.get("force_refresh", False)
      discovered = await self.agent_discovery.discover_agents(force_refresh=force_refresh)
  
      # Register newly discovered agents
      registered_count = 0
      for agent_name, metadata in discovered.items():
          if metadata.available and agent_name not in self.config.orchestrators:
              agent_config = OrchestratorConfig(
                  name=agent_name,
                  command=metadata.command,
                  enabled=True,
                  timeout=300,
              )
              self.config.orchestrators[agent_name] = agent_config
              self.registry.register(agent_config)
              registered_count += 1
              logger.info(f"Registered new agent: {agent_name}")
  
      # Build response
      summary = self.agent_discovery.get_discovery_summary()
      text = f"Agent Discovery Results:\n\n"
      text += f"Total agents scanned: {summary['total_agents']}\n"
      text += f"Available: {summary['available']}\n"
      text += f"Unavailable: {summary['unavailable']}\n"
      text += f"Newly registered: {registered_count}\n\n"
  
      if summary['available_agents']:
          text += "Available Agents:\n"
          for agent in summary['available_agents']:
              text += f"  ✓ {agent['name']}: {agent['version']}\n"
              text += f"    Path: {agent['path']}\n"
  
      if summary['unavailable_agents']:
          text += "\nUnavailable Agents:\n"
          for agent in summary['unavailable_agents']:
              text += f"  ✗ {agent['name']}\n"
              text += f"    {agent['error']}\n"
  
      return [TextContent(type="text", text=text)]

• src/delegation_mcp/server.py:184-197 (registration)

  Tool registration in list_tools() handler, defining name, description, and input schema for 'discover_agents'.

  Tool(
      name="discover_agents",
      description="Discover and register available CLI agents on the system",
      inputSchema={
          "type": "object",
          "properties": {
              "force_refresh": {
                  "type": "boolean",
                  "description": "Force re-discovery even if cache exists",
                  "default": False,
              },
          },
      },
  ),

• src/delegation_mcp/server.py:184-197 (schema)

  Input schema definition for 'discover_agents' tool: optional boolean force_refresh.

  Tool(
      name="discover_agents",
      description="Discover and register available CLI agents on the system",
      inputSchema={
          "type": "object",
          "properties": {
              "force_refresh": {
                  "type": "boolean",
                  "description": "Force re-discovery even if cache exists",
                  "default": False,
              },
          },
      },
  ),

• src/delegation_mcp/agent_discovery.py:296-359 (helper)

  Core implementation of agent discovery logic (AgentDiscovery.discover_agents method), performing parallel discovery, verification, caching; invoked by MCP tool handler.

  async def discover_agents(
      self,
      force_refresh: bool = False,
      agents_to_check: list[str] | None = None,
  ) -> dict[str, AgentMetadata]:
      """Discover available agents on the system.
  
      Args:
          force_refresh: Force re-discovery even if cache exists
          agents_to_check: Specific agents to check (default: all known agents)
  
      Returns:
          Dictionary of agent name to metadata
      """
      if not force_refresh and self._discovered_agents:
          logger.info("Using cached agent discovery results")
          return self._discovered_agents
  
      logger.info("Starting agent discovery...")
  
      # Determine which agents to check
      agents = agents_to_check or list(self.KNOWN_AGENTS.keys())
  
      # Use a semaphore to limit concurrency on all platforms to avoid resource spikes
      # On Windows this is critical, on others it's just good practice
      concurrency_limit = 5 if platform.system() == "Windows" else 10
      semaphore = asyncio.Semaphore(concurrency_limit)
  
      async def _bounded_discover(name: str, config: dict[str, Any]) -> AgentMetadata | Exception:
          async with semaphore:
              try:
                  return await self._discover_single_agent(name, config)
              except Exception as e:
                  logger.error(f"Discovery task failed for {name}: {e}")
                  return e
  
      logger.debug(f"Running agent discovery in parallel (limit={concurrency_limit})")
      tasks = []
      for name in agents:
          if name not in self.KNOWN_AGENTS:
              logger.warning(f"Unknown agent: {name}")
              continue
  
          config = self.KNOWN_AGENTS[name]
          tasks.append(_bounded_discover(name, config))
  
      results = await asyncio.gather(*tasks, return_exceptions=True)
  
      # Process results
      for result in results:
          if isinstance(result, AgentMetadata):
              self._discovered_agents[result.name] = result
          elif isinstance(result, Exception):
              # Already logged in _bounded_discover
              pass
  
      # Save to cache
      self._save_cache()
  
      logger.info(
          f"Discovery complete: {sum(1 for a in self._discovered_agents.values() if a.available)}/{len(self._discovered_agents)} agents available"
      )
  
      return self._discovered_agents

• src/delegation_mcp/agent_discovery.py:22-34 (helper)

  Dataclass defining AgentMetadata structure used in discovery results.

  @dataclass
  class AgentMetadata:
      """Metadata for a discovered agent."""
  
      name: str
      command: str | list[str]
      version: str | None = None
      available: bool = False
      path: str | None = None
      error_message: str | None = None
      capabilities: list[str] | None = None
      verified_at: str | None = None