MCP2ANP Bridge Server

Instructions

Implementation Reference

• mcp2anp/core/handlers.py:86-150 (handler)

  Core handler implementation for the 'anp.invokeOpenRPC' tool. Validates input using InvokeOpenRPCRequest model, processes parameters, invokes ANPCrawler.execute_tool_call, and returns structured result or error.

  async def handle_invoke_openrpc(self, arguments: dict[str, Any]) -> dict[str, Any]:
      """处理 invokeOpenRPC 工具调用。"""
      try:
          # 验证参数
          request = models.InvokeOpenRPCRequest(**arguments)
  
          logger.info(
              "Invoking OpenRPC method",
              endpoint=request.endpoint,
              method=request.method,
              params=request.params,
          )
  
          # 检查 ANPCrawler 是否已初始化
          if self.anp_crawler is None:
              logger.error("ANPCrawler not initialized. Please check DID credentials.")
              return {
                  "ok": False,
                  "error": {
                      "code": "ANP_NOT_INITIALIZED",
                      "message": "ANPCrawler not initialized. Please check DID credentials.",
                  },
              }
  
          # 构建工具名称（ANPCrawler 需要这种格式）
          tool_name = f"{request.method}"
  
          # 调用工具
          if request.params is None:
              tool_params = {}
          elif isinstance(request.params, dict):
              tool_params = request.params
          elif isinstance(request.params, list):
              # 如果是列表，转换为字典
              tool_params = {"args": request.params}
          else:
              tool_params = {"value": request.params}
  
          # 因为execute_json_rpc在调用的时候，会有一定的概率失败。主要是Endpoint错误，所以这里暂时用execute_tool_call来调用。
          # execute_json_rpc的通用性最好，后面在根据模型能力调整
          # result = await self.anp_crawler.execute_json_rpc(request.endpoint, request.method, tool_params)
          result = await self.anp_crawler.execute_tool_call(tool_name, tool_params)
  
          logger.info("OpenRPC method invoked successfully", method=request.method)
          return {
              "ok": True,
              "result": result,
              "raw": result,  # ANPCrawler 已经返回结构化结果
          }
  
      except Exception as e:
          logger.error(
              "Failed to invoke OpenRPC method",
              endpoint=arguments.get("endpoint"),
              method=arguments.get("method"),
              error=str(e),
          )
          return {
              "ok": False,
              "error": {
                  "code": "ANP_RPC_ERROR",
                  "message": str(e),
              },
          }

• mcp2anp/utils/models.py:37-47 (schema)

  Pydantic model used for input validation in the handler for 'anp.invokeOpenRPC'.

  class InvokeOpenRPCRequest(ConfigMixin, BaseModel):
      """invokeOpenRPC 工具请求模型。"""
  
      endpoint: str = Field(..., description="OpenRPC 端点 URL")
      method: str = Field(..., description="要调用的方法名")
      params: dict[str, Any] | list[Any] | None = Field(
          None,
          description="方法参数",
      )
      id: str | None = Field(None, description="请求 ID")

• mcp2anp/server.py:64-92 (registration)

  MCP tool registration in the stdio server, including name, description, and input schema.

  Tool(
      name="anp.invokeOpenRPC",
      description=(
          "使用 JSON-RPC 2.0 协议调用 OpenRPC 端点上的方法。"
          "此工具处理与暴露 OpenRPC 接口的 ANP 智能体的结构化交互。"
      ),
      inputSchema={
          "type": "object",
          "properties": {
              "endpoint": {
                  "type": "string",
                  "description": "OpenRPC 端点 URL",
                  "format": "uri",
              },
              "method": {
                  "type": "string",
                  "description": "要调用的 RPC 方法名称",
              },
              "params": {
                  "description": "传递给方法的参数",
              },
              "id": {
                  "type": "string",
                  "description": "用于跟踪的可选请求 ID",
              },
          },
          "required": ["endpoint", "method"],
      },
  ),

• mcp2anp/server_remote.py:277-334 (registration)

  FastMCP tool registration and wrapper function that delegates to the core handler.

  @mcp.tool()
  async def anp_invokeOpenRPC(
      endpoint: str,
      method: str,
      ctx: Context,
      params: Any = None,
      request_id: str | None = None,
  ) -> str:
      """使用 ANP的 JSON-RPC 2.0 协议调用 OpenRPC 端点上的方法。
  
      此工具处理与暴露 OpenRPC 接口的 ANP 智能体的结构化交互。
  
      Args:
          endpoint: OpenRPC 端点 URL
          method: 要调用的 RPC 方法名称
          ctx: FastMCP 上下文对象
          params: 传递给方法的参数（可选）
          request_id: 用于跟踪的可选请求 ID
  
      Returns:
          JSON 格式的结果字符串
      """
      arguments: dict[str, Any] = {"endpoint": endpoint, "method": method}
      if params is not None:
          arguments["params"] = params
      if request_id is not None:
          arguments["id"] = request_id
  
      logger.info("Tool called", tool_name="anp.invokeOpenRPC", args=arguments)
  
      try:
          state = ensure_session_initialized(ctx)
          if state is None:
              return json.dumps(
                  {
                      "ok": False,
                      "error": {
                          "code": "AUTHENTICATION_FAILED",
                          "message": "Authentication failed. Please provide valid credentials.",
                      },
                  },
                  indent=2,
                  ensure_ascii=False,
              )
  
          anp_handler: ANPHandler = state["anp_handler"]
          result = await anp_handler.handle_invoke_openrpc(arguments)
          return json.dumps(result, indent=2, ensure_ascii=False)
      except Exception as e:
          logger.error(
              "Tool execution failed", tool_name="anp.invokeOpenRPC", error=str(e)
          )
          return json.dumps(
              {"ok": False, "error": {"code": "EXECUTION_ERROR", "message": str(e)}},
              indent=2,
              ensure_ascii=False,
          )