MCP Router

Instructions

Execute a tool on a target MCP server.

Args:
    target_server_name: Name of the target server
    target_tool_name: Name of the tool to execute
    parameters: Parameters for the tool

Implementation Reference

• mcp_router.py:114-140 (handler)

  The primary handler function for the 'exec_mcp_tool' MCP tool. It retrieves the target server's endpoint from the discovery service and dispatches to either SSE or HTTP execution helpers based on the endpoint type.

  @mcp.tool()
  async def exec_mcp_tool(
      target_server_name: str, 
      target_tool_name: str, 
      parameters: Dict[str, Any]
  ) -> str:
      """
      Execute a tool on a target MCP server.
      
      Args:
          target_server_name: Name of the target server
          target_tool_name: Name of the tool to execute
          parameters: Parameters for the tool
      """
      try:
          # Get target server endpoint
          target_endpoint = discovery_service.get_server_endpoint(target_server_name)
          
          # Check if this is an SSE service (like AMap)
          if "sse" in target_endpoint.lower() and HAVE_SSE_SUPPORT:
              # Use SSE connection for services that require it
              return await _execute_sse_tool(target_endpoint, target_tool_name, parameters)
          else:
              # Use HTTP POST for standard MCP services
              return await _execute_http_tool(target_endpoint, target_tool_name, parameters)
      except Exception as e:
          return json.dumps({"error": str(e)}, ensure_ascii=False)

• mcp_router.py:53-57 (schema)

  Pydantic BaseModel defining the input schema structure for the exec_mcp_tool, matching its parameter types.

  class ExecToolRequest(BaseModel):
      target_server_name: str
      target_tool_name: str
      parameters: Dict[str, Any]

• mcp_router.py:142-179 (helper)

  Helper function that handles execution of tools on SSE-based MCP servers using MCP ClientSession and proper resource cleanup.

  async def _execute_sse_tool(endpoint: str, tool_name: str, parameters: Dict[str, Any]) -> str:
      """
      Execute a tool on a target MCP server using SSE connection.
      
      Args:
          endpoint: The SSE endpoint URL
          tool_name: Name of the tool to execute
          parameters: Parameters for the tool
      """
      exit_stack = AsyncExitStack()
      
      try:
          # Create SSE client
          sse_cm = sse_client(endpoint)
          streams = await exit_stack.enter_async_context(sse_cm)
          
          # Create session
          session_cm = ClientSession(streams[0], streams[1])
          session = await exit_stack.enter_async_context(session_cm)
          
          # Initialize session
          await session.initialize()
          
          # Execute tool
          result = await session.call_tool(tool_name, parameters)
          
          # Convert result to dict if it's a CallToolResult object
          if hasattr(result, '_asdict'):
              result = result._asdict()
          elif hasattr(result, '__dict__'):
              result = result.__dict__
          
          # Return result as JSON
          return json.dumps(result, ensure_ascii=False, default=str)
      finally:
          # Clean up
          await exit_stack.aclose()

• mcp_router.py:180-211 (helper)

  Helper function that executes tools on standard HTTP-based MCP servers by sending JSON-RPC 'tools/call' requests.

  async def _execute_http_tool(endpoint: str, tool_name: str, parameters: Dict[str, Any]) -> str:
      """
      Execute a tool on a target MCP server using HTTP POST.
      
      Args:
          endpoint: The HTTP endpoint URL
          tool_name: Name of the tool to execute
          parameters: Parameters for the tool
      """
      # Construct JSON-RPC request
      json_rpc_request = {
          "jsonrpc": "2.0",
          "method": "tools/call",
          "params": {
              "name": tool_name,
              "arguments": parameters
          },
          "id": 1
      }
      
      # Send request to target server
      async with httpx.AsyncClient() as client:
          response = await client.post(
              endpoint,
              json=json_rpc_request,
              headers={"Content-Type": "application/json"}
          )
          
          # Return the response from the target server
          response_data = response.json()
          return json.dumps(response_data, ensure_ascii=False)

• mcp_router.py:114-114 (registration)

  The @mcp.tool() decorator registers the exec_mcp_tool function as an MCP tool with FastMCP.

  @mcp.tool()