# python_json_rpc

**Repository Path**: enzoism/python_json_rpc

## Basic Information

- **Project Name**: python_json_rpc
- **Description**: 使用Python进行JsonRPC的格式调用
- **Primary Language**: Unknown
- **License**: Not specified
- **Default Branch**: master
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 0
- **Forks**: 0
- **Created**: 2025-05-07
- **Last Updated**: 2025-05-07

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README



以下是一个使用 Python 实现 JSON-RPC 2.0 的简单示例，包含服务端和客户端代码，并说明其工作原理：

---

### 1. 服务端代码（基于 Flask）
```python
from flask import Flask, request, jsonify
import json

app = Flask(__name__)

# 注册支持的 RPC 方法到字典
rpc_methods = {}

def register_rpc(method_name):
    """装饰器：注册一个 RPC 方法"""
    def decorator(func):
        rpc_methods[method_name] = func
        return func
    return decorator

@app.route('/rpc', methods=['POST'])
def rpc_handler():
    """处理 JSON-RPC 请求"""
    data = request.get_json()
    
    # 验证请求格式
    if not data or data.get('jsonrpc') != '2.0':
        return jsonify({
            'jsonrpc': '2.0',
            'id': None,
            'error': {
                'code': -32600,
                'message': 'Invalid Request'
            }
        }), 400

    method = data.get('method')
    params = data.get('params', [])
    _id = data.get('id')

    # 调用对应方法
    if method in rpc_methods:
        try:
            # 根据参数类型确定调用方式：数组参数 vs 对象参数
            if isinstance(params, list):
                result = rpc_methods[method](*params)
            elif isinstance(params, dict):
                result = rpc_methods[method](**params)
            else:
                raise TypeError("Parameters must be list or object")
            
            return jsonify({
                'jsonrpc': '2.0',
                'id': _id,
                'result': result
            })
        except Exception as e:
            return jsonify({
                'jsonrpc': '2.0',
                'id': _id,
                'error': {
                    'code': -32603,
                    'message': 'Internal error',
                    'data': str(e)
                }
            }), 500
    else:
        return jsonify({
            'jsonrpc': '2.0',
            'id': _id,
            'error': {
                'code': -32601,
                'message': 'Method not found'
            }
        }), 404

# 注册示例 RPC 方法

@register_rpc('add')
def add(a: int, b: int):
    """两个数相加"""
    return a + b

@register_rpc('subtract')
def subtract(minuend: int, subtrahend: int):
    """两个数相减"""
    return minuend - subtrahahend  # 故意写错拼写，演示错误处理

if __name__ == '__main__':
    app.run(port=5000)
```

---

### 2. 客户端代码（基于 requests 库）
```python
import requests
import json

base_url = 'http://localhost:5000'

# 定义请求模板
def send_rpc_request(method, params):
    payload = {
        "jsonrpc": "2.0",
        "method": method,
        "params": params,
        "id": 1  # 可以自动生成唯一ID
    }
    
    response = requests.post(f"{base_url}/rpc", json=payload)
    return response.json()

# 测试正确方法
response = send_rpc_request('add', [1, 2])
print("Test 'add':", response)

# 测试错误方法
response = send_rpc_request('multiply', [3, 4])
print("Test 'multiply':", response)

# 测试存在语法错误的方法
response = send_rpc_request('subtract', {"minuend": 5, "subtrahend": 3})
print("Test 'subtract':", response)
```

---

### 运行步骤
1. **安装依赖**：
   ```bash
   pip install flask requests
   ```

2. **启动服务端**：
   - 在 Python 文件中运行服务端代码（保存为 `server.py`）
   - Flask 会在本地启动一个 HTTP 服务，默认可用端口 5000

3. **运行客户端**：
   - 在 Python 文件中运行客户端代码（保存为 `client.py`）

---

### 示例输出
#### 成功调用 `add` 的输出：
```json
{'jsonrpc': '2.0', 'id': 1, 'result': 3}
```

#### 调用 `subtract` 方法时触发的代码错误：
```json
{
    'jsonrpc': '2.0',
    'id': 1,
    'error': {
        'code': -32603,
        'message': 'Internal error',
        'data': "name 'subtrahahend' is not defined"  # 服务端代码拼写错误导致
    }
}
```

#### 调用不存在的 `multiply` 方法：
```json
{
    'jsonrpc': '2.0',
    'id': 1,
    'error': {
        'code': -32601,
        'message': 'Method not found'
    }
}
```

---

### 主要代码解析
#### **服务端核心逻辑**
1. **可用方法注册**：
   - 通过装饰器 `@register_rpc('method_name')` 将函数注册为可用的 RPC 方法。
2. **请求处理**：
   - 检查必需字段（`jsonrpc`、`method`、`id`）。
   - 根据参数类型（列表或字典）分别解析参数：
     ```python
     if isinstance(params, list):
         result = method(*params)
     elif isinstance(params, dict):
         result = method(**params)
     ```
3. **错误处理**：
   - 内部错误返回 `code=-32603`
   - 方法未找到返回 `code=-32601`
   - 不符合请求格式时返回 `code=-32600`

#### **客户端核心逻辑**
- 构造标准的 JSON-RPC 请求对象
- 发送 POST 请求并接收响应
- 根据返回的 `id` 和结果/错误信息进行处理

---

### 扩展说明
1. **更复杂的错误处理**：
   服务端可以在错误对象中添加更多信息（如 `data` 字段）:
   ```json
   "error": {
       "code": -32000,
       "message": "Division by zero",
       "data": "Denominator is zero"
   }
   ```

2. **支持变量请求ID**：
   可以让 `id` 动态生成：
   ```python
   import uuid
   payload['id'] = str(uuid.uuid4())
   ```

3. **支持批量请求**：
   服务端可以处理包含多个 RPC 请求的数组，客户端修改请求格式：
   ```python
   payloads = [
       {"jsonrpc": "2.0", "method": "add", "params": [1,2], "id": 1},
       {"jsonrpc": "2.0", "method": "subtract", "params": {"minuend": 5, "subtrahend":3 }, "id": 2}
   ]
   
   response = requests.post(f"{base_url}/rpc", json=payloads)
   ```

---

### 使用场景说明
- **为什么选择 Flask？**
  - 易于搭建简单服务（3分钟内可启动服务端）
  - 适合快速开发原型系统
  - 支持 HTTP/HTTPS，默认兼容 JSON 格式
- **为什么选择 requests？**
  - Python 标准库函数支持构建 JSON 请求
  - 轻量级，不需要额外配置即可发送 HTTP 请求

---

### 其他实现方式
如果需要一个更完整的 JSON-RPC 库，可以使用：
- **jsonrpclib**（Python2和Python3都支持，基于HTTP的简单库）
- **aiohttp_json_rpc**（支持异步的 Python3 高级库）

但推荐从简单示例开始，理解其核心机制后再使用第三方库。

通过这个例子可以观察到：JSON-RPC 的实现核心是客户端和服务端之间按照严格的 JSON 结构进行编解码和参数映射，错误的处理异常是关键。