Python JSON-RPC 完整教程
JSON-RPC 是一种轻量级的远程过程协议，它使用 JSON 格式进行数据编码，通过 HTTP (或其他协议) 进行通信，它非常适合构建 API 和微服务，因为它简单、易于解析，并且被广泛支持。
（图片来源网络，侵删）
本教程将涵盖以下内容：
- JSON-RPC 核心概念
- 环境准备
- 创建一个简单的 JSON-RPC 服务器
- 创建一个 JSON-RPC 客户端
- 使用成熟的库：
jsonrpcserver和requests - 进阶主题
JSON-RPC 核心概念
在开始编码前,我们需要了解 JSON-RPC 请求和响应的基本结构。
请求
客户端向服务器发送一个 JSON 对象，至少包含以下字段：
jsonrpc: 字符串，必须是"2.0"。method: 字符串，要调用的方法名。params: 数组或对象，传递给方法的参数，如果不需要参数，可以省略或设为空数组[]。id: 任意类型的值（通常是数字或字符串），用于匹配请求和响应。
示例请求：
（图片来源网络，侵删）
{
"jsonrpc": "2.0",
"method": "greet",
"params": ["Alice"],
"id": 1
}
这个请求调用 greet 方法，传入 "Alice" 作为参数，并期望服务器返回一个 id 为 1 的响应。
响应
服务器处理完请求后,返回一个 JSON 对象，包含以下字段：
jsonrpc: 字符串，必须是"2.0"。result: 任意类型的值，方法的返回结果，如果方法执行成功，此字段存在。error: 对象，如果方法执行失败，此字段存在，它包含：code: 数值，错误码。message: 字符串，错误信息。data: 可选，包含错误详情的值。
id: 与请求中id字段相同的值。
成功响应示例：
{
"jsonrpc": "2.0",
"result": "Hello, Alice!",
"id": 1
}
错误响应示例：
{
"jsonrpc": "2.0",
"error": {
"code": -32601,
"message": "Method not found"
},
"id": 1
}
环境准备
你需要安装 Python，我们将使用 Flask 作为 Web 服务器框架，requests 作为 HTTP 客户端库。
# 创建一个项目目录 mkdir jsonrpc_project cd jsonrpc_project # 创建虚拟环境 (推荐) python -m venv venv source venv/bin/activate # Linux/macOS # venv\Scripts\activate # Windows # 安装必要的库 pip install flask requests
实战一：创建一个简单的 JSON-RPC 服务器
我们将使用 Flask 来创建一个 HTTP 服务器，该服务器接收 JSON-RPC 请求并处理它们。
创建 server.py 文件：
from flask import Flask, request, jsonify
import json
app = Flask(__name__)
# 一个简单的内存存储
data_store = {"items": []}
# --- JSON-RPC 方法 ---
def greet(name):
"""返回一个问候语"""
if not isinstance(name, str):
raise ValueError("Name must be a string")
return f"Hello, {name}!"
def add_item(item):
"""向数据存储中添加一个新项目"""
data_store["items"].append(item)
return {"status": "success", "item_added": item, "total_items": len(data_store["items"])}
def get_items():
"""获取所有项目"""
return data_store["items"]
# --- Flask 路由 ---
@app.route('/api', methods=['POST'])
def handle_rpc():
"""
处理所有 JSON-RPC 请求的端点。
"""
try:
# 解析 JSON 请求
request_data = request.get_json()
if not request_data:
return jsonify({"jsonrpc": "2.0", "error": {"code": -32700, "message": "Parse error"}, "id": None}), 400
# 验证基本字段
if 'jsonrpc' not in request_data or request_data['jsonrpc'] != '2.0':
return jsonify({"jsonrpc": "2.0", "error": {"code": -32600, "message": "Invalid Request"}, "id": request_data.get('id')}), 400
method_name = request_data.get('method')
params = request_data.get('params', [])
rpc_id = request_data.get('id')
# 查找并调用方法
if method_name == 'greet':
# params 可以是数组或对象，这里我们假设是数组
result = greet(*params)
elif method_name == 'add_item':
result = add_item(*params)
elif method_name == 'get_items':
result = get_items()
else:
# 方法未找到
return jsonify({
"jsonrpc": "2.0",
"error": {"code": -32601, "message": "Method not found"},
"id": rpc_id
}), 404
# 返回成功响应
response = {
"jsonrpc": "2.0",
"result": result,
"id": rpc_id
}
return jsonify(response)
except Exception as e:
# 处理执行期间发生的错误
response = {
"jsonrpc": "2.0",
"error": {
"code": -32603, # Internal error
"message": str(e)
},
"id": request_data.get('id')
}
return jsonify(response), 500
if __name__ == '__main__':
# 运行在 0.0.0.0:5000，以便外部可以访问
app.run(host='0.0.0.0', port=5000, debug=True)
代码解释：
@app.route('/api', methods=['POST']): 我们创建了一个/api端点，它只接受 POST 请求，这是 JSON-RPC 的标准。request.get_json(): 从 HTTP 请求体中解析 JSON 数据。- 验证: 我们检查
jsonrpc版本和基本字段是否存在，确保请求格式正确。 - 方法分发: 根据
method字段的值，我们调用本地定义的 Python 函数（greet,add_item,get_items）。 - 参数处理: 我们使用
*params将参数列表解包后传递给函数。 - 错误处理: 我们捕获了所有可能的错误（解析错误、方法未找到、执行错误）并返回符合 JSON-RPC 规范的错误响应。
运行服务器：
在终端中运行：
python server.py
服务器将在 http://localhost:5000 上启动。
实战二：创建一个 JSON-RPC 客户端
现在我们来创建一个客户端,向我们的服务器发送请求。
创建 client.py 文件：
import json
import requests
# 服务器的 URL
URL = "http://localhost:5000/api"
def send_rpc_request(method, params, rpc_id):
"""
构建并发送 JSON-RPC 请求。
"""
payload = {
"jsonrpc": "2.0",
"method": method,
"params": params,
"id": rpc_id
}
print(f"Sending request: {json.dumps(payload, indent=2)}")
try:
response = requests.post(URL, json=payload)
response.raise_for_status() # 如果请求失败 (404, 500)，则抛出异常
result = response.json()
print(f"Received response: {json.dumps(result, indent=2)}")
return result
except requests.exceptions.RequestException as e:
print(f"An error occurred: {e}")
return None
if __name__ == '__main__':
# --- 测试不同的调用 ---
# 1. 调用 greet 方法
send_rpc_request("greet", ["Bob"], 1)
print("\n" + "="*40 + "\n")
# 2. 调用 add_item 方法
send_rpc_request("add_item", ["Apple"], 2)
send_rpc_request("add_item", ["Banana"], 3)
print("\n" + "="*40 + "\n")
# 3. 调用 get_items 方法
send_rpc_request("get_items", [], 4)
print("\n" + "="*40 + "\n")
# 4. 测试一个不存在的方法
send_rpc_request("non_existent_method", [], 5)
print("\n" + "="*40 + "\n")
# 5. 测试一个会引发错误的参数
send_rpc_request("greet", [123], 6) # 期望一个字符串，但传入了数字
代码解释：
requests.post(URL, json=payload): 我们使用requests库向服务器的/api端点发送一个 POST 请求。json=payload会自动将 Python 字典序列化为 JSON 并设置正确的Content-Type头。response.json(): 将服务器返回的 JSON 响应解析回 Python 字典。- 错误处理: 我们捕获了网络请求可能发生的异常。
运行客户端：
在另一个终端中运行（确保服务器仍在运行）：
python client.py
你将看到客户端发送请求并打印出服务器返回的响应,包括成功和失败的案例。
使用成熟的库：jsonrpcserver 和 requests
手动处理 JSON-RPC 的请求和响应虽然有助于理解原理，但在生产环境中，使用专门的库会更安全、更方便。
服务端：jsonrpcserver
这个库可以极大地简化服务器的编写。
安装：
pip install jsonrpcserver
创建 server_with_lib.py：
from flask import Flask, request, jsonify
from jsonrpcserver import method, Success, Error
app = Flask(__name__)
data_store = {"items": []}
# 使用 @method 装饰器注册 RPC 方法
@method
def greet(name: str) -> str:
"""Returns a greeting."""
return f"Hello, {name}!"
@method
def add_item(item: str) -> dict:
"""Adds an item to the store."""
data_store["items"].append(item)
return {"status": "success", "total_items": len(data_store["items"])}
@method
def get_items() -> list:
"""Gets all items."""
return data_store["items"]
@app.route('/api', methods=['POST'])
def handle_rpc():
# jsonrpcserver.dispatch 会自动处理请求、调用方法并格式化响应
response = dispatch(request.get_data().decode('utf-8'))
return jsonify(response)
if __name__ == '__main__':
app.run(host='0.0.0.0', port=5001, debug=True)
这个版本代码更少,更清晰，并且类型提示和错误处理都由库内部完成。
客户端：requests
客户端部分仍然使用 requests，因为它本身就是处理 HTTP 请求的优秀库，我们之前的 client.py 已经是最佳实践。
进阶主题
1 批量请求
JSON-RPC 2.0 支持在一个请求中发送多个调用。
客户端发送的批量请求：
[
{"jsonrpc": "2.0", "method": "greet", "params": ["Charlie"], "id": 101},
{"jsonrpc": "2.0", "method": "add_item", "params": ["Cherry"], "id": 102},
{"jsonrpc": "2.0", "method": "get_items", "params": [], "id": 103}
]
服务器返回的批量响应：
[
{"jsonrpc": "2.0", "result": "Hello, Charlie!", "id": 101},
{"jsonrpc": "2.0", "result": {"status": "success", "total_items": 3}, "id": 102},
{"jsonrpc": "2.0", "result": ["Apple", "Banana", "Cherry"], "id": 103}
]
注意：如果某个调用失败，响应数组中对应位置的对象将包含 error 而不是 result，服务器必须为每个请求返回一个响应，顺序与请求一致。
要实现批量请求,服务器端需要检查收到的 JSON 是一个对象（单个请求）还是一个数组（批量请求），并分别处理。
2 通知
通知是指那些不需要服务器返回响应的请求,这通过将 id 字段设置为 null 来实现。
通知请求示例：
{
"jsonrpc": "2.0",
"method": "add_item",
"params": ["Durian"]
}
服务器处理这个请求后,不会返回任何响应，这对于需要触发服务器执行某个操作但不关心结果的场景非常有用（记录日志、更新缓存等）。
| 特性 | 手动实现 | 使用 jsonrpcserver |
|---|---|---|
| 优点 | 理解原理，无额外依赖 | 代码简洁，功能强大，类型安全，自动处理错误 |
| 缺点 | 代码冗长，容易出错，需要自己处理所有边界情况 | 需要引入一个新库 |
| 适用场景 | 学习、简单脚本、对依赖有严格要求的项目 | 生产环境、任何需要稳定 JSON-RPC 服务的项目 |
推荐路径：
- 初学者：从手动实现开始，这能帮助你深刻理解 JSON-RPC 的工作原理。
- 生产环境：强烈推荐使用
jsonrpcserver这样的成熟库，它能让你专注于业务逻辑，而不是协议的细节。
希望这份教程能帮助你掌握 Python 中的 JSON-RPC 开发！
