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
Tool call template
This tool declares no required parameters. Add optional parameters from the input schema as needed.
{
"id": 1,
"jsonrpc": "2.0",
"method": "tools/call",
"params": {
"arguments": {},
"name": "discover_acp_agents"
}
}Input Schema
| Name | Required | Description | Default |
|---|---|---|---|
No arguments | |||
Output 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 {}