discover_acp_agents
Identify and register all available ACP agents as resources, enabling integration between ACP-based AI agents and MCP-compatible tools via the ACP-MCP-Server bridge.
Instructions
Discover all available ACP agents and register them as resources
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
No arguments | |||
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| result | Yes |
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) - 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] = {} - 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 [] - 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 {}