ACP-MCP-Server

Instructions

Implementation Reference

• acp_mcp_server/agent_discovery.py:83-91 (handler)

  The main execution handler for the 'get_agent_info' tool. It retrieves agent capabilities via the AgentDiscoveryTool or returns a 'not found' message if the agent is unknown. The @mcp.tool() decorator registers it with FastMCP using the function name as the tool name.

  @mcp.tool()
  async def get_agent_info(agent_name: str) -> str:
      """Get detailed information about a specific ACP agent"""
      capabilities = await discovery.get_agent_capabilities(agent_name)
      if capabilities:
          return str(capabilities)
      else:
          return f"Agent '{agent_name}' not found"

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

  Invocation of register_discovery_tools during server initialization, which defines and registers the get_agent_info tool (and discover_acp_agents) to the FastMCP instance.

  register_discovery_tools(self.mcp, self.discovery)

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

  Core helper function called by get_agent_info to fetch and format detailed capabilities information for a specific agent from the discovered agents cache.

  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 {}

• acp_mcp_server/agent_discovery.py:60-91 (registration)

  The registration function that defines both discovery tools using @mcp.tool() decorators and binds them to the provided FastMCP instance and AgentDiscoveryTool instance.

  def register_discovery_tools(mcp: FastMCP, discovery: AgentDiscoveryTool):
      
      @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)
      
      @mcp.tool()
      async def get_agent_info(agent_name: str) -> str:
          """Get detailed information about a specific ACP agent"""
          capabilities = await discovery.get_agent_capabilities(agent_name)
          if capabilities:
              return str(capabilities)
          else:
              return f"Agent '{agent_name}' not found"