后端 SDK 文档

一、概述

小蜗插件后端支持多种语言：

类型	说明
python	推荐，提供现成 SDK，小蜗内置 Python 运行时
node	需自行实现通信协议
exe	任意语言编译的可执行文件，需自行实现通信协议

通信方式：前后端通过 stdin/stdout 进行 JSON-RPC 协议通信。

二、Python SDK（推荐）

Python 后端提供现成的 SDK，封装了通信协议，开发者只需专注业务逻辑。

2.1 文件结构

backend/
├── main.py              # 入口文件
├── handlers.py          # 接口函数
└── utils/
    ├── __init__.py
    └── xiaowo_sdk.py    # SDK 核心

2.2 入口文件 main.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-

from handlers import *

if __name__ == "__main__":
    sdk.run()

2.3 接口函数 handlers.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-

from utils.xiaowo_sdk import sdk

@sdk.handler("hello")
def handle_hello(params):
    name = params.get("name", "World")
    return {"message": f"Hello, {name}!"}

三、SDK API 参考

3.1 属性

属性	类型	说明
sdk.plugin_id	str	插件唯一标识
sdk.plugin_dir	str	插件目录路径
sdk.plugin_data	dict	插件扩展配置（从环境变量 PLUGIN_DATA 解析）

3.2 方法

@sdk.handler(method)

注册接口处理函数的装饰器。

@sdk.handler("get_status")
def handle_get_status(params):
    return {"status": "running"}

sdk.send_event(event_name, data)

主动推送事件到前端。

sdk.send_event("progress", {"percent": 50})

sdk.logerr(message)

输出日志到 stderr。

sdk.logerr("发生错误")

sdk.run()

启动主循环，监听前端请求。仅在 main.py 中调用。

四、完整示例

4.1 长任务进度推送

@sdk.handler("long_task")
def handle_long_task(params):
    total = params.get("total", 5)

    for i in range(total):
        time.sleep(1)
        percent = int((i + 1) / total * 100)
        sdk.send_event("progress", {
            "current": i + 1,
            "total": total,
            "percent": percent
        })

    return {"success": True}

4.2 获取插件信息

@sdk.handler("get_info")
def handle_get_info(params):
    return {
        "plugin_id": sdk.plugin_id,
        "plugin_dir": sdk.plugin_dir
    }

五、注意事项（Python）

1. 编码问题：建议在 manifest.json 中设置环境变量 PYTHONIOENCODING: utf-8
2. 返回值：handler 函数必须返回可 JSON 序列化的对象

六、Python 后端开发规范

6.1 第三方包依赖管理

如果需要使用第三方包（非 Python 标准库），必须先在插件的 manifest.json 中声明依赖：

{
  "backend": {
    "pip": {
      "requests": null,
      "numpy": "1.23.5"
    }
  }
}

• null 表示不限制版本
• 字符串值表示指定版本
• 插件启动时会自动检查并安装缺失的依赖
• 限配置小包，大型包 例如：torch等，请参考 第八节 extmodels 扩展模型包配置

6.2 代码组织规范

handlers.py 文件职责：只存放 @sdk.handler 装饰的接口函数，保持文件简洁。

# handlers.py - 只放接口函数
from utils.xiaowo_sdk import sdk
from utils.common import process_data

@sdk.handler("process")
def handle_process(params):
    result = process_data(params.get("data"))
    return {"result": result}

utils 包职责：

• 具体功能实现放在 utils/ 目录下新建的文件中
• 常用函数、公共函数统一放到 utils/common.py 中

backend/
├── main.py
├── handlers.py          # 只放 @sdk.handler 装饰的函数
└── utils/
    ├── __init__.py
    ├── xiaowo_sdk.py    # SDK 核心
    ├── common.py        # 常用函数、公共函数
    └── video_utils.py   # 视频处理相关功能（示例）

七、通信协议（其他语言）

如果使用 Node.js、Go、Rust 等其他语言开发后端，需要自行实现 JSON-RPC 通信协议。

7.1 协议概述

• 输入：从 stdin 读取 JSON 请求（每行一个）
• 输出：向 stdout 写入 JSON 响应（需包裹特殊标记）
• 日志：调试信息输出到 stderr（不影响通信）

7.2 消息格式

请求格式（从 stdin 读取）：

{"id": 1, "method": "hello", "params": {"name": "World"}}

字段	说明
id	请求 ID，响应时需原样返回
method	方法名
params	参数对象

响应格式（写入 stdout）：

<<PLUGIN_MSG_START>>{"id": 1, "result": {"message": "Hello!"}}<<PLUGIN_MSG_END>>

重要：响应必须用 <<PLUGIN_MSG_START>> 和 <<PLUGIN_MSG_END>> 包裹。

事件格式（主动推送，id 为 0）：

<<PLUGIN_MSG_START>>{"id": 0, "result": {"event": "progress", "data": {"percent": 50}}}<<PLUGIN_MSG_END>>

7.3 后端就绪信号

后端启动后必须发送就绪信号，否则前端调用会失败：

<<PLUGIN_MSG_START>>{"id": 0, "result": {"event": "__backend_ready__", "data": {"status": "ready"}}}<<PLUGIN_MSG_END>>

7.4 Node.js 示例

const readline = require('readline')

const MSG_START = '<<PLUGIN_MSG_START>>'
const MSG_END = '<<PLUGIN_MSG_END>>'

// 发送响应
function sendResponse(id, result) {
  const msg = JSON.stringify({ id, result })
  process.stdout.write(`${MSG_START}${msg}${MSG_END}`)
}

// 发送事件
function sendEvent(event, data) {
  sendResponse(0, { event, data })
}

// 处理请求
function handleRequest(request) {
  const { id, method, params } = request

  if (method === 'hello') {
    const name = params.name || 'World'
    sendResponse(id, { message: `Hello, ${name}!` })
  }
}

// 主循环
const rl = readline.createInterface({ input: process.stdin })

rl.on('line', (line) => {
  try {
    const request = JSON.parse(line)
    handleRequest(request)
  } catch (e) {
    console.error('解析错误:', e.message)
  }
})

// 发送就绪信号
sendEvent('__backend_ready__', { status: 'ready' })