<plugin_name>/
├── __init__.py        # 入口：Plugin 子类 + create_workers/cleanup 绑定
├── global_var.py      # 模块级全局状态（Database、API client 等）
├── router/            # RPC 端点实现
│   ├── __init__.py    # 导入所有端点模块（`from . import *` 或显式导入）
│   └── <操作名>.py    # 每个端点一个文件，使用 @rpc_method 装饰器
├── models/            # 数据模型
│   ├── __init__.py    # 模型导出（由 gen_models 自动生成，勿手动修改）
│   ├── pydantic.py    # Pydantic 请求/响应模型（由 gen_models 自动生成）
│   └── <表名>.py      # SQLModel ORM 模型（如需要数据库）
├── routes.py          # 自定义 FastAPI 路由（可选，用于特殊端点）
└── README.md          # 插件说明文档（可选）

已有插件概览

插件	特点	适合参考
example	最简示例，Plugin 基类 + Database + 生命周期	入门学习
auth	SQLModel ORM + 管理员初始化 + 多进程幂等	数据库操作
doubao	外部 API 客户端（Ark/OpenAI）+ AICC 安全通道	外部集成
aibox_sophon_adapter	自定义 `APIRouter` + MQTT 客户端组件（`mqtt_clients.py`）+ 后台任务	高级模式

开发新插件：完整流程

第一步：创建目录结构

推荐：直接用脚手架生成统一风格的插件骨架，无需手动创建目录：
uv run gen_plugin <name> [operation ...]
以下为手动创建方式（了解结构时参考）。

mkdir -p src/plugins/<name>/router
mkdir -p src/plugins/<name>/models
touch src/plugins/<name>/__init__.py
touch src/plugins/<name>/global_var.py
touch src/plugins/<name>/router/__init__.py

第二步：注册插件

在 config/workers/<name>/config.toml 中通过 [workers] 节注册（网关启动时自动扫描发现）：

# config/workers/<name>/config.toml
[workers]
type = "<name>"              # 插件类型（worker type）
module = "plugins.<name>"    # Python 模块路径
factory = "create_workers"   # 工厂函数名（Plugin 基类绑定的方法）
cleanup = "cleanup"          # 清理函数名
count = 3                    # Worker 副本数

# 运行时配置（展平后传给 create_workers 的 config 参数）
[workers.config]
data_dir = "./data"

旧式全局注册文件 config/workers/config.toml（[[workers]] 数组）仅作兜底兼容，新插件请使用各插件自己的 config.toml。

第三步：编写端点 TOML 配置

创建 config/workers/<name>/config.toml（分组配置）：

[worker]
defaults = { timeout = 3.0, auth_required = true, cacheable = false, internal = false }

[worker.registration]
router_prefix = "/<prefix>"
tags = ["<name>"]

创建每个操作的 .toml 文件，如 config/workers/<name>/my_action.toml：

[POST]
summary = "我的操作"
auth_required = true

[POST.response_model]
result = { type = "str", required = true }
message = { type = "str", default = "" }

第四步：生成 Pydantic 模型

uv run python -m chongming.scripts.gen_models

自动在 src/plugins/<name>/models/ 下生成 pydantic.py 和 __init__.py。

第五步：实现 RPC 端点

src/plugins/<name>/router/my_action.py：

from chongming_worker.decorators import rpc_method
from ..models.pydantic import MyActionPostRequest, MyActionPostResponse

@rpc_method(
    subject="<name>.my_action",
    method="POST",
    response_model=MyActionPostResponse,
)
async def my_action(request: MyActionPostRequest) -> MyActionPostResponse:
    """业务逻辑"""
    result = await do_something(request)
    return MyActionPostResponse(result="success", message="done")

第六步：更新 router 导出

src/plugins/<name>/router/__init__.py：

from . import my_action
__all__ = ["my_action"]

第七步：实现插件入口（推荐 Plugin 基类）

src/plugins/<name>/__init__.py（由 gen_plugin 脚手架生成）：

import logging

from chongming_worker.plugin import Plugin

from . import router  # noqa: F401  触发 @rpc_method 注册
# from .routes import router  # 可选：自定义 FastAPI 路由（网关自动发现）

logger = logging.getLogger(__name__)


class MyPlugin(Plugin):
    type = "<name>"
    db_name = "<name>.db"   # 自动在 data_dir 下建 SQLite 库（不需要则置 None）
    auto_bind = True        # 首次初始化时自动 auto_bind_models("<name>")

    async def on_init(self):
        """首次 create_workers 时调用一次：初始化资源、启动后台任务。"""
        self.set_var("db", self.db)   # 发布到 global_var 供业务代码读取
        # self.spawn_task(self._background_loop())

    async def on_cleanup(self):
        """cleanup 时调用：关闭自定义客户端等（后台任务与数据库由基类处理）。"""


# 暴露给 app lifespan 的模块级入口（bound method，签名与旧写法完全一致）
_plugin = MyPlugin()
create_workers = _plugin.create_workers
cleanup = _plugin.cleanup

基类自动处理：幂等（初始化锁）、统一日志、数据库建库/释放、后台任务取消、清理钩子、模型自动绑定、cleanup 时重置已发布的 global_var。详见 utils/worker/README.md 的「Plugin 基类」章节。

旧写法已废弃：不依赖基类的传统 create_workers / cleanup 模块级函数写法仅框架层面保留兼容，现有插件（auth / doubao / aibox_sophon_adapter / example）均已迁移到 Plugin 基类，新插件请勿采用旧写法。

自定义 FastAPI 路由（高级）

当端点无法用 TOML 描述时（如文件上传），插件可暴露自定义 APIRouter。

src/plugins/<name>/routes.py：

from fastapi import APIRouter, UploadFile, File, Form

router = APIRouter(prefix="/<prefix>", tags=["<name>"])

@router.post("/upload")
async def upload(
    file: UploadFile = File(...),
    description: str = Form(...),
):
    # 处理文件上传
    return {"status": "ok"}

在 __init__.py 中暴露 router：

from .routes import router  # 网关通过 getattr 自动发现

网关在 load_workers() 阶段自动检测并注册该 router，无需修改网关代码。

为现有插件添加新端点

创建 config/workers/<name>/<new_op>.toml
运行 uv run python -m chongming.scripts.gen_models
创建 router/<new_op>.py，引用生成的模型
更新 router/__init__.py
重启服务

共享对象（global_var）在两种路由中的使用

数据流

共享对象（Database、API client 等）在 Plugin.on_init 中初始化，通过 set_var 发布到 global_var 模块；两种路由的业务代码（模块级函数）在函数体内 from .global_var import ... 读取。

flowchart LR
    P["Plugin.on_init<br/>self.db = ..."] -->|"self.set_var('db', self.db)"| GV["global_var 模块<br/>db / require()"]
    R1["router/*.py<br/>@rpc_method 函数"] -->|"from ..global_var import require"| GV
    R2["routes.py<br/>APIRouter 处理器"] -->|"from .global_var import db"| GV

为什么需要 global_var.py：@rpc_method 函数和 APIRouter 处理器都是模块级函数，拿不到 Plugin 实例，只能通过模块级容器读取；global_var.py 独立成模块，可避免与 __init__.py 循环导入。set_var() 内部会调用 global_var.set_global_var()（白名单校验）完成发布。

注入：on_init → global_var

src/plugins/<name>/__init__.py：

class MyPlugin(Plugin):
    type = "<name>"
    db_name = "<name>.db"

    async def on_init(self):
        self.set_var("db", self.db)   # 发布到 global_var 模块

路由一：TOML 驱动标准路由（@rpc_method）

router/<op>.py（请求经 Zenoh → Worker Queryable → 调用函数）：

from chongming_worker.decorators import rpc_method
from ..models.pydantic import OpGetRequest, OpGetResponse

@rpc_method(subject="<name>.op", method="GET", response_model=OpGetResponse)
async def op(request: OpGetRequest) -> OpGetResponse:
    from ..global_var import require
    db = require("db")   # 未初始化立即抛清晰错误
    async with db.get_session() as session:
        ...
    return OpGetResponse(...)

路由二：插件自定义路由（APIRouter）

routes.py（FastAPI 直接调用，适合 multipart 上传等 TOML 无法描述的端点）：

from fastapi import APIRouter, UploadFile, File

router = APIRouter(prefix="/<name>", tags=["<name>"])

@router.post("/upload")
async def upload(file: UploadFile = File(...)):
    from .global_var import db
    if db is None:                 # 容忍未初始化，优雅降级
        return {"error": "db 未初始化"}
    async with db.get_session() as session:
        ...
    return {"ok": True}

在 __init__.py 顶部暴露给网关：from .routes import router（网关通过 getattr(module, "router") 自动发现）。

两种路由对比

	路由一 `@rpc_method`	路由二 `APIRouter`
文件位置	`router/<op>.py`	`routes.py`
入口发现	装饰器注册 → Worker Queryable	`__init__.py` 暴露 `router` 属性
函数签名	`(request: Pydantic模型) -> Pydantic模型`	FastAPI 风格（`Request`/`UploadFile`/`Form`）
适用场景	标准 JSON / query 端点	multipart 上传、手动解析报文
读共享对象	`from ..global_var import require`	`from .global_var import db`
请求路径	HTTP → Zenoh → Worker 回调	FastAPI 直接调用

经验

在函数体内 import（不要放文件顶部）：模块 import 时 create_workers 尚未执行、global_var 还是空；请求到达时变量才注入完成。
两种读取风格：require("db")（必须有，未初始化立即报错）vs from .global_var import db + if db is None（允许优雅降级）。
参考实现：aibox_sophon_adapter/routes.py 的 _save_alarm_record（自定义路由中 from .global_var import db + async with db.get_session()）。

@rpc_method 装饰器

@rpc_method(
    subject="<plugin>.<operation>",   # Zenoh 主题（必需）
    method="GET",                     # HTTP 方法：GET / POST / MQTT
    response_model=ResponseCls,       # Pydantic 响应模型类
)
async def handler(request: RequestCls) -> ResponseCls:
    ...

method	参数来源	请求绑定
`GET`	URL query string	TOML `params` 定义
`POST`	JSON body	自动绑定到 Pydantic 请求模型
`MQTT`	MQTT payload JSON	同 POST

请求信封协议：网关统一把请求组装为 {"meta": {"method", "user_info"}, "data": {...}} 发给 Worker；运行时按 meta.method 分派到对应 @rpc_method，data 经请求模型校验后传给处理器。MQTT 客户端发布消息时需按信封格式（见主 README「请求负载格式」）， MQTT 专用端点要显式 "meta": {"method": "MQTT"}。

设置 internal = true 后，端点不注册 HTTP 路由，仅可通过 Zenoh RPC 或 MQTT 内部调用
MQTT 方法返回 None 时，Bridge 自动跳过响应发布

启动契约校验

服务启动时（load_workers 之后、注册路由之前），网关自动核对三方一致性：

检查项	策略
TOML HTTP 路由缺 `@rpc_method`	硬失败，启动直接终止
`[MQTT]` / `internal` 路由缺注册	警告日志
已注册方法无 TOML 路由	警告日志（反向漂移检测）
params/response_model 缺生成模型	警告日志（gen_models 过期）
生成模型未被任何路由引用	警告日志（孤儿模型）

新增端点后如果启动报错「缺少对应的 @rpc_method 注册」，检查 TOML 的 subject/方法节与 router/<op>.py 中的 @rpc_method(subject=..., method=...) 是否一致。

命名约定

概念	约定	示例
插件名	小写下划线	`aibox_sophon_adapter`
操作名	小写下划线	`board_ping`, `list_devices`
Zenoh subject	`<插件>.<操作>`	`example.add`
TOML 文件名	与操作名一致	`add.toml`
Python 文件名	与操作名一致	`add.py`
模型类名	`<操作><方法>Request/Response`	`AddGetRequest`

概念	约定	示例
插件名	小写下划线	`doubao`, `auth`, `example`
操作名	小写下划线	`analyze`, `add`, `change_password`
Zenoh subject	`<插件>.<操作>`	`example.add`, `doubao.analyze`
TOML 文件名	与操作名一致	`analyze.toml`, `add.toml`
Python 路由文件名	与操作名一致	`analyze.py`, `add.py`

现有插件

插件	说明	文档
auth	用户注册、登录、JWT 鉴权、用户信息管理	含 SQLite 数据库
doubao	火山引擎豆包视觉理解模型集成	第三方 API 调用
example	四则运算示例 + 访问计数器	入门参考

README.md Unescape Escape

Worker 插件开发指南

目录结构约定