ACP-MCP-Server

Instructions

Implementation Reference

• acp_mcp_server/agent_discovery.py:62-81 (handler)

  The primary handler for the 'discover_acp_agents' tool. Decorated with @mcp.tool() for automatic registration. Discovers ACP agents via HTTP API, enriches with MCP resource URIs and capabilities, and returns formatted JSON string.

  @mcp.tool()
  async def discover_acp_agents() -> str:
      """Discover all available ACP agents and register them as resources"""
      agents = await discovery.discover_agents()
      
      result = {
          "discovered_count": len(agents),
          "agents": []
      }
      
      for agent in agents:
          agent_info = {
              "name": agent.name,
              "description": agent.description,
              "resource_uri": discovery.get_mcp_resource_uri(agent.name),
              "capabilities": await discovery.get_agent_capabilities(agent.name)
          }
          result["agents"].append(agent_info)
      
      return str(result)

• acp_mcp_server/server.py:85-85 (registration)

  Invocation of register_discovery_tools in the ACPMCPServer._register_all_tools method, which defines and registers the discover_acp_agents tool with the FastMCP server instance.

  register_discovery_tools(self.mcp, self.discovery)

• acp_mcp_server/agent_discovery.py:8-11 (schema)

  Pydantic BaseModel schema used to parse and type discovered ACP agent data from the API response.

  class ACPAgent(BaseModel):
      name: str
      description: str
      metadata: Dict[str, Any] = {}

• acp_mcp_server/agent_discovery.py:18-37 (helper)

  Key helper method in AgentDiscoveryTool that performs the actual HTTP discovery of ACP agents and caches them.

  async def discover_agents(self) -> List[ACPAgent]:
      """Discover all available ACP agents"""
      async with aiohttp.ClientSession() as session:
          try:
              async with session.get(f"{self.acp_base_url}/agents") as response:
                  if response.status == 200:
                      data = await response.json()
                      agents = [ACPAgent(**agent) for agent in data.get("agents", [])]
                      
                      # Update discovered agents cache
                      for agent in agents:
                          self.discovered_agents[agent.name] = agent
                      
                      return agents
                  else:
                      print(f"Failed to discover agents: {response.status}")
                      return []
          except Exception as e:
              print(f"Error discovering agents: {e}")
              return []

• acp_mcp_server/agent_discovery.py:43-57 (helper)

  Helper method that generates MCP-compatible capabilities information for discovered agents.

  async def get_agent_capabilities(self, agent_name: str) -> Dict[str, Any]:
      """Get detailed capabilities of a specific agent"""
      # This could be extended to call a capabilities endpoint
      # For now, return basic info from discovery
      agent = self.discovered_agents.get(agent_name)
      if agent:
          return {
              "name": agent.name,
              "description": agent.description,
              "metadata": agent.metadata,
              "supports_streaming": True,  # ACP supports streaming
              "supports_multimodal": True,  # ACP supports multi-modal
              "interaction_modes": ["sync", "async", "stream"]
          }
      return {}