1. 什么是 JSON-RPC？ #

JSON-RPC 的全称是 "JSON Remote Procedure Call"，即「用 JSON 格式进行的远程过程调用」。

通俗地说：你在一台电脑上写代码，想调用另一台电脑（服务器）上的某个函数，JSON-RPC 就是规定「请求怎么写、响应怎么回」的一套规则。客户端按规则发一个 JSON 请求，服务器按规则执行并返回 JSON 响应。

打个比方：就像点外卖——你（客户端）按固定格式下单（发 JSON 请求），餐厅（服务器）按格式做好菜并送回（返回 JSON 响应）。JSON-RPC 就是这套「下单格式」的规范。

2. 前置知识：JSON 是什么？ #

在学 JSON-RPC 之前，需要先了解 JSON，因为整个协议的数据都是用 JSON 表示的。

2.1 JSON 简介 #

JSON（JavaScript Object Notation）是一种轻量级的数据交换格式，用文本表示结构化数据，人类可读，机器易解析。

常见数据类型对应关系：

JSON 类型	示例	说明
字符串	"hello"	双引号包裹
数字	42、3.14	整数或小数
布尔	true、false	真或假
空值	null	空
数组	[1, 2, 3]	方括号，元素用逗号分隔
对象	{"name": "张三", "age": 18}	花括号，键值对用逗号分隔

2.2 Python 中的 JSON 操作 #

在 Python 中，用 json 模块处理 JSON：

# 导入 json 模块
import json

# 将 Python 字典转换为 JSON 字符串
data = {"name": "张三", "age": 18}
json_str = json.dumps(data)
print(json_str)  # 输出: {"name": "\u5f20\u4e09", "age": 18}

# 将 JSON 字符串解析为 Python 字典
parsed = json.loads(json_str)
print(parsed["name"])  # 输出: 张三

3. JSON-RPC 的核心概念 #

3.1 一句话理解 #

客户端发送「要调用哪个方法、传什么参数」的 JSON，服务器执行后返回「结果或错误」的 JSON。

3.2 两个重要特点 #

• 无状态：每次请求独立，服务器不记住你上次做了什么。
• 简单：只用 JSON，不依赖复杂协议，任何语言都能实现。

4. 请求格式（客户端发给服务器） #

客户端发送的必须是一个 JSON 对象，包含以下字段：

字段	类型	是否必需	说明
jsonrpc	字符串	是	固定写 "2.0"，表示协议版本
method	字符串	是	要调用的方法名，如 "add"
params	数组或对象	否	传给方法的参数
id	数字/字符串/null	视情况	请求标识，用于匹配响应；为 null 时表示「通知」，服务器不返回

参数两种写法：

• 数组：[42, 23]，按位置对应参数，如 subtract(42, 23)
• 对象：{"a": 42, "b": 23}，按名称对应参数，如 subtract(a=42, b=23)

示例：调用减法，参数用数组

{
  "jsonrpc": "2.0",
  "method": "subtract",
  "params": [42, 23],
  "id": 1
}

示例：调用加法，参数用对象

{
  "jsonrpc": "2.0",
  "method": "add",
  "params": {"a": 2, "b": 3},
  "id": 2
}

5. 响应格式（服务器返回给客户端） #

服务器返回的也是一个 JSON 对象，成功时带 result，失败时带 error。

字段	类型	说明
jsonrpc	字符串	固定 "2.0"
result	任意类型	成功时的返回值，与 error 二选一
error	对象	失败时的错误信息，与 result 二选一
id	与请求相同	对应请求的 id

成功响应示例：

{
  "jsonrpc": "2.0",
  "result": 19,
  "id": 1
}

错误响应示例：

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

6. 错误对象结构 #

当发生错误时，error 字段是一个对象：

字段	类型	说明
code	整数	错误码，负数表示协议错误
message	字符串	错误描述
data	任意	可选，额外调试信息

常用预定义错误码：

错误码	含义
-32700	解析错误（JSON 无效）
-32600	无效请求（格式不对）
-32601	方法不存在
-32602	参数无效
-32603	内部错误

7. 完整工作流程 #

1. 客户端：按规范构造 JSON 请求（含 method、params、id 等）
2. 发送：通过 HTTP POST 把 JSON 发给服务器（如 http://服务器地址/rpc）
3. 服务器：解析 JSON，检查格式，找到对应方法并执行
4. 返回：把结果或错误封装成 JSON 响应
5. 客户端：根据 id 匹配请求，解析 result 或 error

8. JSON-RPC 服务器 #

下面是一个完整的、可独立运行的 JSON-RPC 2.0 服务器示例。使用前请安装：pip install flask。

# 导入 Flask 和请求/响应相关功能
from flask import Flask, request, jsonify

# 创建 Flask 应用实例
app = Flask(__name__)


# 定义两个简单的业务方法，供 JSON-RPC 调用
def add(a, b):
    """加法"""
    return a + b


def subtract(a, b):
    """减法"""
    return a - b


# 方法名到函数的映射，方便根据 method 字符串找到对应函数
METHODS = {
    "add": add,
    "subtract": subtract,
}


# 定义 /rpc 路由，只接受 POST 请求
@app.route("/rpc", methods=["POST"])
def handle_rpc():
    # 解析请求体中的 JSON，得到字典
    req = request.get_json(silent=True)

    # 如果解析失败或没有 jsonrpc 字段，返回无效请求错误
    if req is None or req.get("jsonrpc") != "2.0":
        return jsonify({
            "jsonrpc": "2.0",
            "error": {"code": -32600, "message": "Invalid Request"},
            "id": req.get("id") if req else None,
        })

    # 取出方法名、参数和请求 id
    method = req.get("method")
    params = req.get("params") or []
    req_id = req.get("id")

    # 如果方法不存在，返回方法未找到错误
    if method not in METHODS:
        return jsonify({
            "jsonrpc": "2.0",
            "error": {"code": -32601, "message": "Method not found"},
            "id": req_id,
        })

    # 尝试执行方法
    try:
        if isinstance(params, list):
            # 参数是数组，按位置传参：add(3, 5)
            result = METHODS[method](*params)
        else:
            # 参数是对象，按名称传参：add(a=3, b=5)
            result = METHODS[method](**params)
    except Exception as e:
        # 执行出错，返回内部错误
        return jsonify({
            "jsonrpc": "2.0",
            "error": {"code": -32603, "message": str(e)},
            "id": req_id,
        })

    # 成功，返回结果
    return jsonify({
        "jsonrpc": "2.0",
        "result": result,
        "id": req_id,
    })


# 程序入口：直接运行此文件时启动服务器
if __name__ == "__main__":
    app.run(host="0.0.0.0", port=5000)

运行方式：保存为 server.py，在终端执行 python server.py，服务器会在 http://localhost:5000 启动。

9. JSON-RPC 客户端 #

下面是一个完整的客户端示例，用于调用上面的服务器。使用前请安装：pip install requests。请先启动服务器，再运行客户端。

# 导入 requests 库，用于发送 HTTP 请求
import requests

# 服务器地址
SERVER_URL = "http://localhost:5000/rpc"


# 封装一个通用的 JSON-RPC 调用函数
def call(method, params=None, req_id=1):
    """
    发送 JSON-RPC 请求并返回解析后的响应。
    :param method: 要调用的方法名
    :param params: 参数，可以是列表或字典，默认空列表
    :param req_id: 请求 id，用于匹配响应
    :return: 解析后的响应字典
    """
    # 构造符合 JSON-RPC 2.0 规范的请求体
    payload = {
        "jsonrpc": "2.0",
        "method": method,
        "params": params if params is not None else [],
        "id": req_id,
    }

    # 发送 POST 请求，json= 会自动设置 Content-Type 并序列化
    response = requests.post(SERVER_URL, json=payload)

    # 将响应体解析为字典并返回
    return response.json()


# 主程序：演示多次调用
if __name__ == "__main__":
    # 调用 add 方法，参数用数组 [3, 5]
    result1 = call("add", [3, 5], req_id=1)
    print("add(3, 5) =", result1.get("result"))

    # 调用 subtract 方法，参数用对象
    result2 = call("subtract", {"a": 42, "b": 23}, req_id=2)
    print("subtract(42, 23) =", result2.get("result"))

    # 故意调用不存在的方法，观察错误响应
    result3 = call("unknown_method", [], req_id=3)
    if "error" in result3:
        print("错误:", result3["error"]["message"])

运行方式：保存为 client.py，在终端执行 python client.py。输出示例：

add(3, 5) = 8
subtract(42, 23) = 19
错误: Method not found

10. 一个文件同时演示服务器和客户端（可选） #

如果希望在一个脚本里先启动服务器，再发请求，可以用多线程。下面是一个完整可运行的示例，适合快速体验：

# 导入所需模块
import json
import threading
import time
from http.server import HTTPServer, BaseHTTPRequestHandler

# 若未安装 requests，可改用标准库 urllib（本示例使用 urllib 以零依赖运行）
from urllib.request import urlopen, Request
from urllib.error import URLError


# 定义业务方法
def add(a, b):
    return a + b


def subtract(a, b):
    return a - b


METHODS = {"add": add, "subtract": subtract}


# 自定义 HTTP 请求处理器，用于处理 JSON-RPC 请求
class RpcHandler(BaseHTTPRequestHandler):
    def do_POST(self):
        # 读取请求体长度
        length = int(self.headers.get("Content-Length", 0))
        body = self.rfile.read(length)

        # 解析 JSON
        try:
            req = json.loads(body.decode("utf-8"))
        except json.JSONDecodeError:
            req = None

        # 校验请求格式
        if req is None or req.get("jsonrpc") != "2.0":
            resp = {
                "jsonrpc": "2.0",
                "error": {"code": -32600, "message": "Invalid Request"},
                "id": req.get("id") if req else None,
            }
        else:
            method = req.get("method")
            params = req.get("params") or []
            req_id = req.get("id")

            if method not in METHODS:
                resp = {
                    "jsonrpc": "2.0",
                    "error": {"code": -32601, "message": "Method not found"},
                    "id": req_id,
                }
            else:
                try:
                    if isinstance(params, list):
                        result = METHODS[method](*params)
                    else:
                        result = METHODS[method](**params)
                    resp = {"jsonrpc": "2.0", "result": result, "id": req_id}
                except Exception as e:
                    resp = {
                        "jsonrpc": "2.0",
                        "error": {"code": -32603, "message": str(e)},
                        "id": req_id,
                    }

        # 返回 JSON 响应
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.end_headers()
        self.wfile.write(json.dumps(resp).encode("utf-8"))


# 在后台线程中启动服务器
def run_server():
    server = HTTPServer(("127.0.0.1", 8765), RpcHandler)
    server.serve_forever()


# 客户端：发送 JSON-RPC 请求
def call(method, params=None, req_id=1):
    payload = {
        "jsonrpc": "2.0",
        "method": method,
        "params": params if params is not None else [],
        "id": req_id,
    }
    req = Request(
        "http://127.0.0.1:8765",
        data=json.dumps(payload).encode("utf-8"),
        headers={"Content-Type": "application/json"},
        method="POST",
    )
    with urlopen(req) as f:
        return json.loads(f.read().decode("utf-8"))


# 主程序
if __name__ == "__main__":
    # 启动服务器线程
    thread = threading.Thread(target=run_server, daemon=True)
    thread.start()

    # 等待服务器就绪
    time.sleep(0.5)

    # 发送请求并打印结果
    r1 = call("add", [3, 5])
    print("add(3, 5) =", r1.get("result"))

    r2 = call("subtract", {"a": 42, "b": 23})
    print("subtract(42, 23) =", r2.get("result"))

运行方式：保存为 demo.py，执行 python demo.py。无需安装 Flask 或 requests，仅用 Python 标准库即可运行。 doc

11. 总结 #

• JSON-RPC 是用 JSON 格式约定「远程调用」的协议，客户端发请求、服务器返回结果或错误。
• 请求必须包含：jsonrpc、method，可选 params、id。
• 响应包含：jsonrpc、id，以及 result（成功）或 error（失败）。
• 实际开发中，常用 HTTP + JSON 实现，Python 可用 Flask 做服务端，用 requests 做客户端。
• 本教程的示例代码均可独立运行，建议按顺序：先跑服务器和客户端，再尝试单文件 demo，以加深理解。