Process 插件开发指南

Process 插件作为独立进程运行，通过 HTTP API 与主应用通信。适合全功能应用、AI Agent、复杂业务场景。

---

架构概览

┌──────────────────────────┐       HTTP        ┌─────────────────────────┐
│     UnityAI 主应用        │ ◄──────────────► │    Process Plugin        │
│  (下发 PORT / JWT Token)  │                   │  (独立进程 / 独立服务器)   │
└──────────────────────────┘                   └─────────────────────────┘

• 主应用为每个插件分配 PORT 和 MAIN_APP_URL 环境变量
• 插件启动后监听 0.0.0.0:{PORT}
• 所有请求携带 JWT Token 进行认证

---

支持语言

Python、Node.js、Go、Rust 或任何能起 HTTP 服务的语言。

---

项目结构

my-plugin/
├── plugin.json          # 插件元数据（必需）
├── main.py              # 入口文件
├── requirements.txt     # 依赖声明
├── icon.png             # 图标 (256x256 PNG)
└── README.md

---

plugin.json 配置

{
  "id": "com.example.my-plugin",
  "name": "My Plugin",
  "version": "1.0.0",
  "description": "插件简要描述",
  "author": "Your Name",
  "type": "process",
  "entry_point": "main.py",
  "runtime": "python3",
  "permissions": ["network", "storage"],
  "icon": "icon.png",
  "min_platform_version": "2.0.0"
}

字段	类型	必需	说明
id	string	是	唯一标识，推荐反向域名
name	string	是	显示名称
version	string	是	semver 版本号
type	string	是	必须为 "process"
entry_point	string	是	入口文件路径
runtime	string	是	python3 或 node
permissions	array	否	所需权限列表

---

认证机制

Token 流程

1. 主应用签发短期 JWT Token（5 分钟有效）
2. Token 通过请求 Header 传递：Authorization: Bearer <token>
3. 插件调用主应用验证接口确认身份

import os
import requests

MAIN_APP_URL = os.environ["MAIN_APP_URL"]

def verify_token(token: str) -> dict:
    """向主应用验证 Token 并获取用户信息"""
    resp = requests.get(
        f"{MAIN_APP_URL}/api/plugin/model-config",
        headers={"Authorization": f"Bearer {token}"},
        timeout=5
    )
    resp.raise_for_status()
    return resp.json()  # 包含 user_id、模型配置等

用户隔离

从 JWT payload 中提取 user_id，所有数据操作必须按用户隔离：

import base64, json

def extract_user_id(token: str) -> str:
    """从 JWT payload 解析 user_id"""
    payload = token.split(".")[1]
    # 补齐 base64 padding
    payload += "=" * (4 - len(payload) % 4)
    data = json.loads(base64.urlsafe_b64decode(payload))
    return data["user_id"]

---

HTTP API 规范

插件必须实现以下端点：

健康检查

GET /health
Response: { "status": "ok" }

执行入口

POST /execute
Headers: Authorization: Bearer <token>
Body: { "input": "用户输入内容", "context": {} }
Response: { "output": "处理结果", "type": "text" }

流式响应（可选）

POST /execute/stream
Headers: Authorization: Bearer <token>
Body: { "input": "...", "stream": true }
Response: text/event-stream (SSE)

data: {"chunk": "第一段"}
data: {"chunk": "第二段"}
data: [DONE]

---

完整示例：Python 插件

main.py

#!/usr/bin/env python3
import os
import json
import base64
from http.server import HTTPServer, BaseHTTPRequestHandler
import requests

PORT = int(os.environ.get("PORT", 8000))
MAIN_APP_URL = os.environ.get("MAIN_APP_URL", "http://localhost:3000")


def extract_user_id(token: str) -> str:
    payload = token.split(".")[1]
    payload += "=" * (4 - len(payload) % 4)
    data = json.loads(base64.urlsafe_b64decode(payload))
    return data["user_id"]


def verify_token(token: str) -> dict:
    resp = requests.get(
        f"{MAIN_APP_URL}/api/plugin/model-config",
        headers={"Authorization": f"Bearer {token}"},
        timeout=5,
    )
    resp.raise_for_status()
    return resp.json()


class PluginHandler(BaseHTTPRequestHandler):
    def do_GET(self):
        if self.path == "/health":
            self._respond(200, {"status": "ok"})
            return
        self._respond(404, {"error": "Not Found"})

    def do_POST(self):
        if self.path == "/execute":
            token = self.headers.get("Authorization", "").replace("Bearer ", "")
            if not token:
                self._respond(401, {"error": "Missing token"})
                return

            try:
                config = verify_token(token)
                user_id = extract_user_id(token)
            except Exception as e:
                self._respond(401, {"error": f"Auth failed: {e}"})
                return

            body = self._read_body()
            result = self.process(body, user_id, config)
            self._respond(200, result)
            return

        self._respond(404, {"error": "Not Found"})

    def process(self, body: dict, user_id: str, config: dict) -> dict:
        """插件核心业务逻辑"""
        user_input = body.get("input", "")
        return {"output": f"Hello {user_id}, processed: {user_input}", "type": "text"}

    def _read_body(self) -> dict:
        length = int(self.headers.get("Content-Length", 0))
        raw = self.rfile.read(length)
        return json.loads(raw) if raw else {}

    def _respond(self, code: int, data: dict):
        self.send_response(code)
        self.send_header("Content-Type", "application/json")
        self.end_headers()
        self.wfile.write(json.dumps(data).encode())


if __name__ == "__main__":
    server = HTTPServer(("0.0.0.0", PORT), PluginHandler)
    print(f"Plugin running on port {PORT}")
    server.serve_forever()

---

平台 API 调用

插件可调用主应用提供的接口：

接口	方法	用途
/api/plugin/model-config	GET	获取用户模型配置 + 验证 Token
/api/plugin/data/{key}	GET/PUT	读写插件数据（按用户隔离）
/api/plugin/notify	POST	向用户发送通知

所有调用需携带 Authorization: Bearer <token> 请求头。

---

部署方式

Process 插件支持两种部署模式：

托管模式（推荐新手）

打包为 .tar.gz 上传到插件商店，平台自动管理：

• 自动分配端口、启动进程
• 健康检查 + 崩溃自动重启（最多 3 次）
• 资源限制：256MB 内存、1hr CPU

自托管模式

自行部署到服务器，提供 HTTPS URL：

1. 将插件代码部署到你的服务器
2. 确保 GET /health 返回 200
3. 提交时 hosting_mode 设为 "self_hosted"，填写 self_hosted_url
4. 平台每 5 分钟做可用性检查

详见 托管与部署指南

---

发布到插件商店

参见 第三方插件开发指南 - 发布流程

---

安全要求

• 所有数据按 user_id 隔离，不得跨用户访问
• Token 仅在内存中使用，不得持久化存储
• 不得将 Token 转发给第三方
• 插件崩溃不得泄露用户数据
• 使用 HTTPS 进行生产环境通信

---

下一步

• 第三方插件开发指南 — 社区精选与官方上架
• JSON Schema 规范 — 配置校验