Mesh Agent MCP Server

Instructions

Implementation Reference

• mesh_mcp_server/tool_server.py:222-314 (registration)

  Dynamically processes agent metadata and registers tools by constructing tool_id = f'{agent_id.lower()}_{tool_name}', enabling the tool 'goplusanalysisagent_fetch_security_details' if configured and present in metadata.
  """Process agent metadata and extract tool information based on configuration. Returns: Dictionary mapping tool IDs to tool information """ # Load configuration self.agent_tool_config = Config.load_agent_config(self.config_path) if not self.agent_tool_config: logger.warning("No agents configured in config file") return {} agents_metadata = await self.fetch_agent_metadata() tool_registry = {} # Log configuration status logger.info( f"Processing tools for {len(self.agent_tool_config)} configured agents" ) # Track statistics agents_processed = set() tools_enabled = 0 tools_skipped = 0 for agent_id, agent_data in agents_metadata.items(): # Skip agents not in our configuration if agent_id not in self.agent_tool_config: continue agents_processed.add(agent_id) configured_tools = self.agent_tool_config[agent_id] # Log agent processing if configured_tools: logger.info( f"Processing agent {agent_id} with {len(configured_tools)} specific tools" # noqa: E501 ) else: logger.info(f"Processing agent {agent_id} with all tools enabled") # 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 # Check if this tool is enabled based on configuration if not self.is_tool_enabled(agent_id, tool_name): logger.debug( f"Skipping tool {tool_name} for agent {agent_id} (not in config)" # noqa: E501 ) tools_skipped += 1 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, } tools_enabled += 1 logger.debug(f"Enabled tool: {tool_id}") # Log summary logger.info("Configuration summary:") logger.info( f" - Agents processed: {len(agents_processed)} / {len(self.agent_tool_config)} configured" # noqa: E501 ) logger.info(f" - Tools enabled: {tools_enabled}") logger.info(f" - Tools skipped: {tools_skipped}") if agents_processed: logger.info(f" - Active agents: {', '.join(sorted(agents_processed))}") return tool_registry
• mesh_mcp_server/tool_server.py:392-410 (handler)

  The MCP tool handler that executes the tool by calling the mesh API with the agent_id and tool_name extracted from the tool_registry.
  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/tool_server.py:316-353 (helper)

  Executes the tool call to the mesh API endpoint.
  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
• mesh_mcp_server/tool_server.py:120-153 (helper)

  Helper to call the mesh API, used for tool execution and metadata fetch.
  async def call_mesh_api( path: str, method: str = "GET", json: Dict[str, Any] = None ) -> Dict[str, Any]: """Helper function to call the mesh API endpoint. Args: path: API path to call method: HTTP method to use json: Optional JSON payload Returns: API response as dictionary Raises: MeshApiError: If there's an error calling the API """ async with aiohttp.ClientSession() as session: url = f"{Config.HEURIST_API_ENDPOINT}/{path}" try: headers = {} if Config.HEURIST_API_KEY: headers["X-HEURIST-API-KEY"] = Config.HEURIST_API_KEY async with session.request( method, url, json=json, headers=headers ) as response: if response.status != 200: error_text = await response.text() raise MeshApiError(f"Mesh API error: {error_text}") return await response.json() except aiohttp.ClientError as e: logger.error(f"Error calling mesh API: {e}") raise MeshApiError(f"Failed to connect to mesh API: {str(e)}") from e
• mesh_mcp_server/tool_server.py:223-280 (registration)

  Key code block where the tool ID is constructed and registered in the tool_registry.
  Returns: Dictionary mapping tool IDs to tool information """ # Load configuration self.agent_tool_config = Config.load_agent_config(self.config_path) if not self.agent_tool_config: logger.warning("No agents configured in config file") return {} agents_metadata = await self.fetch_agent_metadata() tool_registry = {} # Log configuration status logger.info( f"Processing tools for {len(self.agent_tool_config)} configured agents" ) # Track statistics agents_processed = set() tools_enabled = 0 tools_skipped = 0 for agent_id, agent_data in agents_metadata.items(): # Skip agents not in our configuration if agent_id not in self.agent_tool_config: continue agents_processed.add(agent_id) configured_tools = self.agent_tool_config[agent_id] # Log agent processing if configured_tools: logger.info( f"Processing agent {agent_id} with {len(configured_tools)} specific tools" # noqa: E501 ) else: logger.info(f"Processing agent {agent_id} with all tools enabled") # 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 # Check if this tool is enabled based on configuration if not self.is_tool_enabled(agent_id, tool_name): logger.debug( f"Skipping tool {tool_name} for agent {agent_id} (not in config)" # noqa: E501 ) tools_skipped += 1 continue # Create a unique tool ID