Mesh Agent MCP Server

Instructions

Implementation Reference

• mesh_mcp_server/server.py:301-319 (handler)

  MCP call_tool handler that executes the tool by proxying to the Mesh API after parsing the prefixed tool name to extract agent ID and original tool name.
  async def call_tool(name: str, arguments: dict) -> List[types.TextContent]: """Call the specified tool with the given arguments.""" try: if name not in self.tool_registry: raise ValueError(f"Unknown tool: {name}") tool_info = self.tool_registry[name] result = await self.execute_tool( agent_id=tool_info["agent_id"], tool_name=tool_info["tool_name"], tool_arguments=arguments, ) # Convert result to TextContent return [types.TextContent(type="text", text=str(result))] except Exception as e: logger.error(f"Error calling tool {name}: {e}") raise ValueError(f"Failed to call tool {name}: {str(e)}") from e
• mesh_mcp_server/server.py:163-223 (registration)

  Dynamically registers tools from agent metadata, constructing tool names as '{agent_id.lower()}_{tool_name}', e.g., 'elfatwitterintelligenceagent_get_trending_tokens' for agent 'ElfaTwitterIntelligenceAgent'.
  async def process_tool_metadata(self) -> Dict[str, Dict[str, Any]]: """Process agent metadata and extract tool information. Returns: Dictionary mapping tool IDs to tool information """ agents_metadata = await self.fetch_agent_metadata() tool_registry = {} # Log filtering status if self.supported_agents is not None: logger.info( f"Filtering tools using supported agent list ({len(self.supported_agents)} agents specified)" ) else: logger.info("Loading tools from all available agents (no filter applied)") for agent_id, agent_data in agents_metadata.items(): # Skip agents not in our supported list (if a list is specified) if ( self.supported_agents is not None and agent_id not in self.supported_agents ): continue # Process tools for this agent for tool in agent_data.get("tools", []): if tool.get("type") == "function": function_data = tool.get("function", {}) tool_name = function_data.get("name") if not tool_name: continue # Create a unique tool ID tool_id = f"{agent_id.lower()}_{tool_name}" # Get parameters or create default schema parameters = function_data.get("parameters", {}) if not parameters: parameters = { "type": "object", "properties": {}, "required": [], } # Store tool info tool_registry[tool_id] = { "agent_id": agent_id, "tool_name": tool_name, "description": function_data.get("description", ""), "parameters": parameters, } # Log which agents contributed tools agents_with_tools = set(info["agent_id"] for info in tool_registry.values()) logger.info(f"Loaded tools from agents: {', '.join(sorted(agents_with_tools))}") logger.info(f"Successfully loaded {len(tool_registry)} tools") return tool_registry
• mesh_mcp_server/server.py:40-53 (registration)

  Configuration listing supported agents, including 'ElfaTwitterIntelligenceAgent' whose tools are registered with the 'elfatwitterintelligenceagent_' prefix.
  DEFAULT_AGENTS = [ "CoinGeckoTokenInfoAgent", "DexScreenerTokenInfoAgent", "ElfaTwitterIntelligenceAgent", "ExaSearchAgent", "TwitterInfoAgent", "AIXBTProjectInfoAgent", "EtherscanAgent", "EvmTokenInfoAgent", "FundingRateAgent", "UnifaiTokenAnalysisAgent", "YahooFinanceAgent", "ZerionWalletAnalysisAgent" ]
• mesh_mcp_server/server.py:288-298 (registration)

  MCP list_tools decorator that exposes all registered tools, including the target tool with its schema from metadata.
  @app.list_tools() async def list_tools() -> List[types.Tool]: """List all available tools.""" return [ types.Tool( name=tool_id, description=tool_info["description"], inputSchema=tool_info["parameters"], ) for tool_id, tool_info in self.tool_registry.items() ]
• mesh_mcp_server/server.py:224-262 (helper)

  Helper that sends the tool execution request to the Mesh API endpoint.
  async def execute_tool( self, agent_id: str, tool_name: str, tool_arguments: Dict[str, Any] ) -> Dict[str, Any]: """Execute a tool on a mesh agent. Args: agent_id: ID of the agent to execute the tool on tool_name: Name of the tool to execute tool_arguments: Arguments to pass to the tool Returns: Tool execution result Raises: ToolExecutionError: If there's an error executing the tool """ request_data = { "agent_id": agent_id, "input": {"tool": tool_name, "tool_arguments": tool_arguments}, } # Add API key if available if Config.HEURIST_API_KEY: request_data["api_key"] = Config.HEURIST_API_KEY try: result = await call_mesh_api( "mesh_request", method="POST", json=request_data ) return result.get("data", result) # Prefer the 'data' field if it exists except MeshApiError as e: # Re-raise API errors with clearer context raise ToolExecutionError(str(e)) from e except Exception as e: logger.error(f"Error calling {agent_id} tool {tool_name}: {e}") raise ToolExecutionError( f"Failed to call {agent_id} tool {tool_name}: {str(e)}" ) from e