# workflow_rpc_python_cli

**Repository Path**: FriendOfJustic/workflow_rpc_python_cli

## Basic Information

- **Project Name**: workflow_rpc_python_cli
- **Description**: No description available
- **Primary Language**: Python
- **License**: Not specified
- **Default Branch**: master
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 0
- **Forks**: 0
- **Created**: 2026-05-15
- **Last Updated**: 2026-05-17

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

# Workflow RPC Python Client

Workflow RPC 系统的 Python 客户端 SDK。

## 功能特性

- WebSocket 连接与认证
- 服务注册与发现
- 自动重连（指数退避）
- 方法路由（按 method 分发）
- 并发请求处理
- 调用超时保护
- 事件回调（连接/断开/注册/错误）
- 心跳保活
- 文件参数自动下载（递归解析嵌套结构）
- 文件结果返回（`$file_content` base64 编码）
- 多种认证方式（Bearer / Apikey / Query 参数）
- 同步和异步方法处理函数

## 安装

### 1. 激活虚拟环境

Windows:
```bash
.venv\Scripts\activate
```

Linux/Mac:
```bash
source .venv/bin/activate
```

### 2. 安装依赖

```bash
pip install -e .
```

## 快速开始

### 1. 获取 API Key

1. 登录系统
2. 进入「系统配置」→「API Key 管理」
3. 点击「创建 API Key」
4. **保存好生成的 Key（只显示一次！）**

### 2. 创建服务

```python
import asyncio
from rpc_client import RPCClient, ServiceConfig

config = ServiceConfig(
    name="my-python-service",
    node_id="worker-001",
    desc="我的 Python RPC 服务",
    api_key="你的 API Key",
    server_url="ws://localhost:8080/api/v1/rpc/ws",
)

client = RPCClient(config)

# 注册方法（支持同步和异步函数）
def my_method(params: dict) -> dict:
    return {"message": f"Hello {params.get('name', 'World')}!"}

async def my_async_method(params: dict) -> dict:
    await asyncio.sleep(1)
    return {"message": "Async done"}

client.register_method("my_method", my_method)
client.register_method("my_async_method", my_async_method)

# 事件回调
@client.on_registered
def on_registered():
    print("服务注册成功！")

async def main():
    try:
        await client.start()
    except KeyboardInterrupt:
        await client.stop()

asyncio.run(main())
```

### 3. 运行示例

```bash
# 先修改 examples/simple_service.py 中的 API_KEY
python examples/simple_service.py
```

## API 文档

### ServiceConfig

服务配置类：

| 参数 | 类型 | 默认值 | 说明 |
|------|------|--------|------|
| `name` | str | 必填 | 服务名称（唯一标识） |
| `node_id` | str | 必填 | 节点 ID（区分同一服务的不同实例） |
| `desc` | str | `""` | 服务描述 |
| `api_key` | str | `""` | API Key（用于认证） |
| `server_url` | str | `"ws://localhost:8080/api/v1/rpc/ws"` | WebSocket 服务地址 |
| `http_url` | str | `"http://localhost:8080/api/v1/rpc"` | HTTP API 地址 |
| `auth_mode` | str | `"bearer"` | 认证方式：`"bearer"` / `"apikey"` / `"query"` |
| `reconnect_max_attempts` | int | `10` | 最大重连次数（0 表示不限制） |
| `reconnect_initial_delay` | float | `1.0` | 初始重连延迟（秒） |
| `reconnect_max_delay` | float | `30.0` | 最大重连延迟（秒） |
| `call_timeout` | float | `55.0` | RPC 调用超时（秒） |
| `auto_download_files` | bool | `True` | 是否自动下载文件参数内容 |

### 认证方式

支持三种 API Key 认证方式（通过 `auth_mode` 配置）：

| auth_mode | Header / 参数 | 格式 |
|-----------|--------------|------|
| `"bearer"` | `Authorization` Header | `Bearer <api_key>` |
| `"apikey"` | `Authorization` Header | `Apikey <api_key>` |
| `"query"` | URL query 参数 | `?apikey=<api_key>` |

### RPCClient

客户端主类：

#### 方法

- `register_method(method: str, handler: Callable)` - 注册 RPC 方法（支持同步和异步函数）
- `unregister_method(method: str)` - 注销 RPC 方法
- `connect()` - 连接服务端
- `start()` - 启动客户端（带自动重连，阻塞）
- `stop()` - 停止客户端
- `is_connected` - 是否已连接
- `is_registered` - 是否已注册

#### 事件回调

```python
@client.on_connect
def on_connect():
    """连接成功时触发"""

@client.on_disconnect
def on_disconnect():
    """连接断开时触发"""

@client.on_registered
def on_registered():
    """服务注册成功时触发"""

@client.on_error
def on_error(e: Exception):
    """发生错误时触发"""
```

### 文件参数处理

当工作流节点的输入包含 `file` 类型参数时，SDK 会自动将文件参数（`$is_file` 标记的对象）替换为 `FileInfo` 对象，并可选地自动下载文件内容。

**服务端发来的文件参数格式：**
```json
{
  "document": {
    "$file_id": "a1b2c3d4e5f6",
    "$is_file": true,
    "$url": "http://host/api/v1/workflows/files/a1b2c3d4e5f6/content?token=xxx",
    "name": "report.pdf",
    "size": 1048576,
    "mime_type": "application/pdf"
  }
}
```

**在 handler 中使用：**
```python
def handle_analyze(params: dict) -> dict:
    document = params.get("document")
    if isinstance(document, FileInfo):
        print(document.name)       # "report.pdf"
        print(document.mime_type)  # "application/pdf"
        print(document.size)       # 1048576
        if document.is_downloaded:
            content = document.content  # bytes
    return {"status": "success"}
```

**配置：** 设置 `auto_download_files=False` 可以禁用自动下载，仅保留文件元数据和 URL。

### 返回文件结果

使用 `FileResult` 在响应中返回生成的文件：

```python
from rpc_client import FileResult

def handle_generate(params: dict) -> dict:
    content = b"file binary content"
    return {
        "status": "success",
        "result_file": FileResult(
            content=content,
            name="output.png",
            mime_type="image/png"
        )
    }
```

SDK 会自动将 `FileResult` 编码为 `{"$file_content": "base64...", "name": "...", "mime_type": "..."}` 格式，服务端会自动解码并上传到存储系统。

### ServiceRegistryClient

服务查询客户端（HTTP）：

```python
from rpc_client import ServiceRegistryClient

registry = ServiceRegistryClient(
    http_url="http://localhost:8080/api/v1/rpc",
    api_key="你的 API Key"
)

# 查询在线服务
services = await registry.list_services()

# 获取文件临时访问 URL（需要 JWT Token）
url_info = await registry.get_file_url("file_id", jwt_token, expires=3600)

# 下载文件内容
content = await registry.download_file("http://host/...?token=xxx")
```

## 协议规范

### 消息类型

**客户端 → 服务端：**
- `0` - Register（注册服务）
- `1` - Offline（下线通知）
- `2` - Online（在线通知）
- `3` - Response（响应 RPC 调用）

**服务端 → 客户端：**
- `0` - Ping（心跳，WebSocket 控制帧）
- `1` - RPCRequest（RPC 调用请求）
- `2` - Ack（确认响应）
- `3` - Error（错误响应）

### 调用流程

```
客户端                          服务端
  |                               |
  |--------- 连接 WebSocket ------>|
  |                               |
  |------- Register (type=0) ---->|
  |                               |
  |<---- Ack (type=2, code=200) --|
  |                               |
  |<------- Ping (控制帧) --------|
  |-------- Pong (控制帧) ------->|
  |                               |
  |<---- RPC Request (type=1) ----|
  |                               |
  | 处理调用...                    |
  |                               |
  |---- RPC Response (type=3) --->|
  |                               |
  | ... 重复心跳和调用 ...         |
  |                               |
  |------- Offline (type=1) ----->|
  |                               |
  |<------- 连接关闭 ------------|
```

## 常见问题

**Q: 如何调试连接问题？**

A: 启用 debug 日志：
```python
import logging
logging.basicConfig(level=logging.DEBUG)
```

**Q: 连接断开会自动重连吗？**

A: 会。默认最多重连 10 次，使用指数退避策略（1s → 2s → 4s → ... → 30s）。

**Q: 可以同时处理多个 RPC 调用吗？**

A: 可以。每个调用都会在独立的 task 中执行，WebSocket 发送操作自动序列化。

**Q: 调用超时是多少？**

A: 默认 55 秒（略小于服务端的 60 秒），可以通过 `call_timeout` 配置修改。

**Q: 如何优雅关闭？**

A: 捕获 `KeyboardInterrupt` 后调用 `await client.stop()`，会自动发送下线通知。

**Q: 文件参数的 URL 过期了怎么办？**

A: 使用 `ServiceRegistryClient.get_file_url()` 重新获取临时 URL，或在 handler 中通过 `FileInfo.url` 手动下载。

**Q: 可以注册异步方法吗？**

A: 可以。SDK 会自动检测 handler 是同步还是异步函数，异步函数直接 await，同步函数在线程池中执行。