Mesh Agent MCP Server

Instructions

Implementation Reference

• mesh_mcp_server/tool_server.py:392-410 (handler)

  The MCP call_tool handler that executes any registered tool, including 'firecrawlsearchagent_firecrawl_scrape_url', by proxying to the remote Mesh API after looking up the agent and tool details.

  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:264-301 (registration)

  Dynamic registration of tools from agent metadata into tool_registry, constructing tool names like 'firecrawlsearchagent_firecrawl_scrape_url' via f\"{agent_id.lower()}_{tool_name}\". Enabled based on config.json per 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}")

• mesh_mcp_server/tool_server.py:379-390 (schema)

  Provides the input schema for all tools, including the target tool, from the loaded agent metadata parameters.

  @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/tool_server.py:315-353 (helper)

  Helper function that actually calls the remote Mesh API to execute the tool on the agent, used by the handler.

  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

• mesh_mcp_server/tool_server.py:120-153 (helper)

  Low-level helper to make HTTP calls to the Mesh API endpoint, used by execute_tool.

  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