当前位置：首页 > news >正文

什么是JSON-RPC 2.0，在项目中应该怎么使用

news 2025/8/26 6:06:21

它是什么

JSON-RPC 2.0 是一种超轻量、与传输无关的远程调用协议：用 JSON 表达“方法名 + 参数 → 结果/错误”。可跑在 HTTP、WebSocket、Unix 管道，甚至 stdio 上（很多开发协议如 LSP 就用它）。

报文长这样
• 请求：

{"jsonrpc":"2.0","method":"sum","params":[1,2,3],"id":"42"}

params 可是数组（按位置）或对象（按名），id 用来对应响应（字符串/数字均可）。
• 通知（不需要回应）：去掉 id

{"jsonrpc":"2.0","method":"log","params":{"msg":"hi"}}

•	响应（二选一：要么 result，要么 error）：

{"jsonrpc":"2.0","result":6,"id":"42"}

{"jsonrpc":"2.0","error":{"code":-32601,"message":"Method not found"},"id":"42"}

•	批量：请求或响应都是数组。

常用错误码：
-32700 解析错误、
-32600 非法请求、
-32601 方法不存在、
-32602 参数错误、
-32603 服务器内部错；
-32000~-32099 留给服务端自定义。

⸻

什么时候用（和 REST/gRPC 对比）
• 你想要**“方法调用式”接口、低 ceremony、浏览器/脚本语言友好、支持批量/通知**时 → JSON-RPC 很合适。
• 资源/对象语义清晰、需要缓存/幂等/超媒体 → REST 更自然。
• 强类型/高性能/流式、多语言代码生成 → gRPC 更强。

⸻

在 FastAPI 中落地（最小实现，支持批量/通知）

pip install fastapi uvicorn httpx

from typing import Any, Dict, Callable, List, Optional
from fastapi import FastAPI, Request, Response
from fastapi.responses import JSONResponseapp = FastAPI()

1) 方法注册表

methods: Dict[str, Callable[..., Any]] = {}def rpc_method(name: str):def deco(fn: Callable[..., Any]):methods[name] = fnreturn fnreturn deco@rpc_method("sum")
def sum_(numbers: List[float]) -> float:return float(sum(numbers))@rpc_method("echo")
def echo(msg: str) -> str:return msg

2) 协议工具

def error(code: int, message: str, _id: Any = None, data: Any = None):e = {"code": code, "message": message}if data is not None:e["data"] = datareturn {"jsonrpc": "2.0", "error": e, "id": _id}def success(result: Any, _id: Any):return {"jsonrpc": "2.0", "result": result, "id": _id}def handle_one(req: Dict[str, Any]) -> Optional[Dict[str, Any]]:# 通知：没有 id -> 不返回_id = req.get("id", None)is_notification = "id" not in reqif req.get("jsonrpc") != "2.0" or "method" not in req:return None if is_notification else error(-32600, "Invalid Request", _id)method = req["method"]fn = methods.get(method)if not fn:return None if is_notification else error(-32601, "Method not found", _id)params = req.get("params", [])try:if isinstance(params, list):result = fn(*params)elif isinstance(params, dict):result = fn(**params)else:return None if is_notification else error(-32602, "Invalid params", _id)return None if is_notification else success(result, _id)except TypeError as te:return None if is_notification else error(-32602, "Invalid params", _id, str(te))except Exception as ex:return None if is_notification else error(-32603, "Internal error", _id, str(ex))@app.post("/rpc")
async def rpc_entry(request: Request):try:payload = await request.json()except Exception:# 解析失败：id 必须为 nullreturn JSONResponse(error(-32700, "Parse error", None), status_code=200)# 批量if isinstance(payload, list):responses = []for item in payload:if not isinstance(item, dict):responses.append(error(-32600, "Invalid Request", None))continuer = handle_one(item)if r is not None:responses.append(r)# 纯通知批：不返回 body（204）if not responses:return Response(status_code=204)return JSONResponse(responses, status_code=200)# 单个if not isinstance(payload, dict):return JSONResponse(error(-32600, "Invalid Request", None), status_code=200)r = handle_one(payload)if r is None:  # 通知return Response(status_code=204)return JSONResponse(r, status_code=200)

调用示例（客户端）

import httpxresp = httpx.post("http://localhost:8000/rpc", json={"jsonrpc":"2.0","method":"sum","params":{"numbers":[1,2,3]},"id":"42"
})
print(resp.json())  # -> {"jsonrpc":"2.0","result":6,"id":"42"}

批量：一个请求 + 一个通知

batch = [{"jsonrpc":"2.0","method":"echo","params":{"msg":"Hi"},"id":1},{"jsonrpc":"2.0","method":"echo","params":{"msg":"No reply"}}   # 通知
]
print(httpx.post("http://localhost:8000/rpc", json=batch).json())

说明：JSON-RPC 对 HTTP 状态码不做规定。上面示例把应用层错误都塞在 200 的响应体里；纯通知返回 204。

⸻

最佳实践清单
• 鉴权：用 HTTP 头（如 Authorization: Bearer ）或在传输层做认证（mTLS/WebSocket 子协议）。不要把密钥放进 params。
• 类型与校验：为方法参数建 Pydantic 模型；服务端在进入方法前先校验，定位错误更清晰。
• 日志与追踪：记录 method 与 id；链路里加超时与重试（注意幂等）。
• 批量与通知：通知无响应，适合非关键且幂等的操作（如埋点）。批量要考虑部分成功。
• ID 生成：客户端生成可追踪的字符串 ID（UUID/雪花）；不要用自增数字以免冲突。
• 传输选择：
• HTTP：简单、易部署。
• WebSocket：天然双向，适合实时推送/订阅。
• stdio/管道：本地进程间协议（很多“工具/智能体”生态走这条）。

⸻

现成库（按语言）
• Python：fastapi-jsonrpc、jsonrpcserver、msgspec（自实现时做编解码）
• Node.js：jayson、json-rpc-2.0
• Go：github.com/sourcegraph/jsonrpc2
• Rust：jsonrpsee
• Java：jsonrpc4j

查看全文

http://www.dtcms.com/a/349354.html