anp.invokeOpenRPC
Execute JSON-RPC 2.0 method calls on OpenRPC endpoints to interact with ANP agents, handling structured communication with required endpoint URLs and method names.
Instructions
使用 JSON-RPC 2.0 协议调用 OpenRPC 端点上的方法。此工具处理与暴露 OpenRPC 接口的 ANP 智能体的结构化交互。
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| endpoint | Yes | OpenRPC 端点 URL | |
| id | No | 用于跟踪的可选请求 ID | |
| method | Yes | 要调用的 RPC 方法名称 | |
| params | No | 传递给方法的参数 |
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, )