什么是JSON-RPC 2.0,在项目中应该怎么使用
它是什么
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