modify: mqtt broker 适配

README.md

@@ -1,22 +1,23 @@
 # Chongming-Zenoh
 
-基于 **FastAPI + Zenoh + Vue 3** 的**配置驱动微服务网关平台**。
+基于 **FastAPI + Zenoh + MQTT + Vue 3** 的**配置驱动微服务网关平台**。
 
-核心理念：通过 TOML 配置文件定义 RPC 端点，自动生成 Pydantic 模型和 FastAPI 路由，
+核心理念：通过 TOML 配置文件定义 RPC 端点，自动生成 Pydantic 模型，同时暴露 **HTTP** 和 **MQTT** 双通道。
-底层通过 Zenoh 分布式消息协议进行 RPC 通信。Worker 插件可独立开发、热加载、水平扩展。
+底层通过 Zenoh 分布式消息协议进行 RPC 通信，Worker 插件可独立开发、热加载、水平扩展。
 
 ---
 
 ## 核心特性
 
 - **FastAPI + Uvicorn**：高性能异步 Web 框架，自动生成 OpenAPI 文档
-- **Zenoh 服务发现**：基于 Zenoh liveliness 的 Worker 注册与 RPC 调用，支持负载均衡、超时重试
+- **Zenoh 服务发现**：基于 Zenoh liveliness 的 Worker 注册与 RPC 调用，负载均衡 + 超时重试
-- **动态路由生成**：根据 `config/workers` 下的 TOML 配置自动生成 API 端点，无需手写路由代码
+- **MQTT 集成**：内嵌 amqtt Broker，MQTT 消息自动桥接到 Worker 插件，支持请求/响应与单向上报
-- **Worker 框架**：使用 `@rpc_method` 装饰器定义 RPC 方法，自动参数解析与 Pydantic 模型绑定
+- **动态路由生成**：扫描 `config/workers` TOML 配置自动生成 HTTP 路由，`[MQTT]` 节单独处理
-- **自动化模型生成**：`gen_models` 脚本根据 TOML 配置自动生成请求/响应 Pydantic 模型
+- **Worker 框架**：`@rpc_method` 装饰器定义 RPC 方法，自动参数解析与 Pydantic 模型绑定
-- **JWT 鉴权**：Rust 实现的 JWT 编码/验证，支持路由级别的 `auth_required` 控制
+- **自动化模型生成**：`gen_models` 脚本根据 TOML 配置生成请求/响应 Pydantic 模型（含 MQTT）
+- **JWT 鉴权**：Rust 实现，路由级别 `auth_required` 控制
 - **Argon2id 密码哈希**：Rust 实现的高性能密码安全方案
-- **对象存储代理**：集成 `nanio`（S3 兼容）文件上传/下载
+- **对象存储代理**：集成 nanio（S3 兼容）文件上传/下载
 - **构建打包**：支持 PyInstaller / Nuitka 一键打包为独立可执行文件
 - **Vue 3 前端**：Element Plus UI、Pinia 状态管理、自动生成类型安全的 API 客户端
 
@@ -26,41 +27,37 @@
 
 ```text
 chongming-zenoh/
-├── config.toml                   # 应用主配置
+├── config.toml                   # 应用主配置（含 MQTT、Zenoh、DB 等）
 ├── pyproject.toml                # 项目依赖与工具配置
 ├── config/                       # 配置文件目录
-│   ├── README.md                 # 配置说明
 │   ├── zenoh_config.json5        # Zenoh 会话配置
-│   └── workers/                  # Worker 路由定义
+│   └── workers/                  # Worker 路由定义（TOML）
-│       ├── config.toml           # Worker 插件注册列表
+│       ├── config.toml           # 插件注册列表
-│       ├── auth/                 # 鉴权插件 TOML
+│       ├── auth/                 # 鉴权插件（login / register / me / …）
-│       ├── doubao/               # 豆包插件 TOML
+│       ├── doubao/               # 豆包视觉理解插件
-│       └── example/              # 示例插件 TOML
+│       └── example/              # 示例插件（add / heartbeat / …）
 ├── src/
 │   ├── chongming/                # 核心应用
 │   │   ├── app/
-│   │   │   ├── core/             # WorkerManager、RouteGenerator、鉴权
+│   │   │   ├── core/             # WorkerManager, RouteGenerator, MQTTBridge, auth
 │   │   │   └── routes/           # 内置路由（存储上传等）
-│   │   └── scripts/              # 工具脚本（build, gen_models）
+│   │   └── scripts/              # build, gen_models
 │   ├── chongming-web/            # Vue 3 前端
-│   │   ├── src/
+│   │   └── src/
-│   │   │   ├── api/generated/    # 自动生成的 API 客户端（@hey-api/openapi-ts）
+│   │       ├── api/generated/    # 自动生成 API 客户端
-│   │   │   ├── views/            # 页面组件
+│   │       ├── views/            # 页面组件
-│   │   │   ├── router/           # 路由配置
+│   │       └── router/           # Vue Router 配置
-│   │   │   └── stores/           # Pinia 状态管理
-│   │   └── vite.config.ts
 │   └── plugins/                  # Worker 插件源码
-│       ├── auth/                 # 鉴权（注册/登录/用户信息/改密）
+│       ├── auth/                 # JWT 鉴权
-│       ├── doubao/               # 豆包视觉理解
+│       ├── doubao/               # 豆包 AI
-│       └── example/              # 示例（四则运算）
+│       └── example/              # 示例（含 MQTT 端点）
-├── utils/                        # 子工具包（独立 Python 包）
+├── utils/                        # 7 个独立工具包
-│   ├── config/                   # 配置加载
+│   ├── config/  database/  jwt/  lock/
-│   ├── database/                 # 异步 SQLite 封装
+│   ├── password/  request/  worker/
-│   ├── jwt/                      # JWT 编码/验证（Rust）
+├── test_mqtt_e2e.py              # MQTT 端到端测试脚本
-│   ├── password/                 # Argon2id 密码哈希（Rust）
+├── publisher.py                  # MQTT 发布示例
-│   ├── request/                  # Zenoh 客户端封装
+├── subscriber.py                 # MQTT 订阅示例
-│   └── worker/                   # Worker 框架核心
+├── test_amqtt.py                 # amqtt Broker 独立测试
-├── public/                       # 公共启动文件
 ├── data/storage/chongming/       # 本地数据存储
 ├── logs/                         # 日志目录
 └── resources/                    # 模块银行打包资源
@@ -74,32 +71,30 @@ chongming-zenoh/
 
 - Python >= 3.12
 - Rust 工具链（用于编译 `jwt` / `password` 扩展）
-- Node.js >= 22（用于前端开发）
+- Node.js >= 22（前端开发）
 - [uv](https://docs.astral.sh/uv/) 包管理器
 
 ### 安装与启动
 
 ```bash
-# 1. 克隆项目
+uv sync                          # 安装依赖 + 编译 Rust 扩展
-git clone <repo-url>
+uv run serve                     # 启动服务（开发模式，热重载）
-cd chongming-zenoh
-
-# 2. 安装 Python 依赖（自动编译 Rust 扩展）
-uv sync
-
-# 3. 启动服务（开发模式，含热重载）
-uv run serve
 ```
 
-访问 API 文档：`http://localhost:8000/docs`
+服务启动后可用：
+
+| 通道    | 地址                         | 说明                     |
+| ------- | ---------------------------- | ------------------------ |
+| HTTP    | `http://localhost:8000`      | REST API + Swagger 文档  |
+| MQTT    | `mqtt://localhost:1883`      | 内嵌 Broker，Worker 桥接 |
+| Nanio   | `http://localhost:9000`      | S3 兼容对象存储          |
+| Swagger | `http://localhost:8000/docs` | OpenAPI 文档             |
 
 ### 前端开发
 
 ```bash
 cd src/chongming-web
-npm install
+npm install && npm run dev        # http://localhost:5173
-npm run dev          # 开发服务器 http://localhost:5173
-npm run build        # 生产构建
 ```
 
 ---
@@ -107,42 +102,122 @@ npm run build        # 生产构建
 ## 架构与数据流
 
 ```mermaid
-graph LR
+graph TB
-    A[用户请求] --> B[FastAPI 路由<br/>自动生成]
+    HTTP[HTTP 客户端] -->|REST API| FastAPI[FastAPI 路由<br/>自动生成]
-    B --> C{鉴权检查}
+    MQTT[MQTT 客户端] -->|chongming/+/+| Bridge[MQTT Bridge<br/>内嵌 Broker]
-    C -->|通过| D[WorkerManager.call]
+
-    D --> E[Zenoh Query]
+    FastAPI --> WM[WorkerManager]
-    E --> F[Worker Queryable]
+    Bridge --> WM
-    F --> G[插件业务逻辑]
+
-    G --> H[响应]
+    WM -->|Zenoh Query| W1[Worker 副本 1]
+    WM -->|负载均衡| W2[Worker 副本 2]
+    WM -->|超时重试| W3[Worker 副本 N]
+
+    W1 & W2 & W3 -->|插件业务逻辑| DB[(SQLite)]
+    W1 & W2 & W3 -->|AI 调用| AI[豆包 API]
 ```
 
 启动流程：
 
 ```
-serve() → FastAPI lifespan:
+uv run serve → FastAPI lifespan:
   1. setup_nanio()           — 本地 S3 子进程
   2. setup_s3()              — boto3 S3 客户端
   3. setup_zenoh()           — Zenoh peer 会话
-  4. setup_worker_manager()  — WorkerManager + liveliness 发现
+  4. setup_worker_manager()  — WorkerManager + liveliness 刷新
-  5. load_workers()          — 动态 import 插件
+  5. load_workers()          — 动态 import 插件 → 启动 Worker 协程
-  6. register_routes()       — 扫描 config/workers/ 生成路由
+  6. register_routes()       — 扫描 config/workers/ 生成 HTTP 路由
+  7. setup_mqtt()            — 内嵌 Broker + MQTT→Worker 桥接
+```
+
+---
+
+## MQTT 集成
+
+### 协议约定
+
+| 方向 | Topic                            | Payload                                    |
+| ---- | -------------------------------- | ------------------------------------------ |
+| 请求 | `chongming/<type>/<op>`          | `{"_method":"MQTT","param1":"value1",...}` |
+| 响应 | `chongming/<type>/<op>/response` | 由 Worker 返回（`None` 则跳过）            |
+
+### 快速测试
+
+```bash
+# 终端 1: 启动服务
+uv run serve
+
+# 终端 2: 端到端测试
+python test_mqtt_e2e.py
+
+# 或手动测试
+python subscriber.py                    # 监听所有响应
+python publisher.py example add '{"a":3,"b":5,"_method":"MQTT"}'       # 有响应
+python publisher.py example heartbeat '{"worker_id":"node-1","_method":"MQTT"}'  # 无响应
+```
+
+### 无响应模式（心跳上报等）
+
+Worker 返回 `None` 时，MQTT Bridge 自动跳过响应发布：
+
+```python
+@rpc_method(subject="example.heartbeat", method="MQTT")
+async def heartbeat(worker_id: str = "") -> None:
+    return None  # Bridge 跳过发布
+```
+
+### 配置
+
+```toml
+# config.toml — 按环境区分
+[development.mqtt]
+enabled = true          # false 禁用 MQTT
+host = "localhost"
+port = 1883
+
+[development.mqtt.broker]
+# 可选：amqtt Broker 高级参数
 ```
 
 ---
 
 ## 开发指南
 
+### 端点类型说明
+
+TOML 文件中的节类型决定端点暴露方式：
+
+| 节类型   | HTTP 路由 | OpenAPI | MQTT | 请求模型命名              | 备注                |
+| -------- | --------- | ------- | ---- | ------------------------- | ------------------- |
+| `[GET]`  | ✅        | ✅      | ✅   | `XxxGetRequest/Response`  |                     |
+| `[POST]` | ✅        | ✅      | ✅   | `XxxPostRequest/Response` |                     |
+| `[MQTT]` | ❌        | ❌      | ✅   | `XxxMqttRequest`          | 无 `response_model` |
+
+```toml
+# config/workers/example/add.toml
+[GET]
+summary = "Add via HTTP GET"
+params = ["a: float", "b: float"]
+
+[POST]
+summary = "Add via HTTP POST"
+params = ["a: float", "b: float"]
+
+[MQTT]
+summary = "Add via MQTT"
+params = ["a: float", "b: float"]
+```
+
 ### 添加新 Worker 插件
 
-1. 创建 `src/plugins/<name>/` 目录结构
+1. 创建 `src/plugins/<name>/` — `__init__.py`, `global_var.py`, `router/`, `models/`
-2. 在 `config/workers/config.toml` 注册插件
+2. 注册到 `config/workers/config.toml`
 3. 创建 `config/workers/<name>/*.toml` 端点定义
-4. 运行 `uv run python -m chongming.scripts.gen_models` 生成模型
+4. `uv run gen_models config/workers/<name>/ src/plugins/<name>/`
-5. 实现 `router/*.py` 业务逻辑
+5. 实现 `router/<op>.py`，引用生成的模型
-6. 实现 `__init__.py` 工厂函数
+6. 更新 `router/__init__.py` 导入
 
-详见 [插件开发指南](src/plugins/README.md) 和 [配置说明](config/README.md)。
+详见 [插件开发指南](src/plugins/README.md)。
 
 ### 常用命令
 
@@ -157,26 +232,26 @@ uv pip install -e utils/<name>            # 安装/更新工具包子包
 ### TOML 配置字段速查
 
 | 字段                 | 位置                  | 说明                                    |
-| -------------------- | --------------------- | --------------------------------------------------- |
+| -------------------- | --------------------- | --------------------------------------- |
 | `count`              | `workers/config.toml` | Worker 副本数                           |
 | `router_prefix`      | `<插件>/config.toml`  | API 路由前缀                            |
 | `defaults.timeout`   | `<插件>/config.toml`  | 默认超时（秒）                          |
 | `defaults.cacheable` | `<插件>/config.toml`  | 是否缓存 RPC 结果                       |
-| `params`             | `<操作>.toml` `[GET]` | GET 参数列表（`"name: type"`）                      |
+| `params`             | `<操作>.toml`         | 参数列表（`"name: type"` 格式）         |
 | `auth_required`      | `<操作>.toml`         | 是否需要 JWT 鉴权                       |
-| `internal`           | `<操作>.toml`         | 设为 `true` 则仅允许内部 RPC 调用，不注册 HTTP 路由 |
+| `internal`           | `<操作>.toml`         | `true` 则仅限内部 RPC，不注册 HTTP 路由 |
 | `cacheable`          | `<操作>.toml`         | 端点是否可缓存                          |
-| `summary`            | `<操作>.toml`         | 端点描述（OpenAPI docs）                            |
+| `summary`            | `<操作>.toml`         | 端点描述（用于 OpenAPI / 日志）         |
 
 ---
 
 ## 内置插件
 
-| 插件        | 路由前缀  | 说明                                         |
+| 插件        | HTTP 前缀 | MQTT 主题前缀 | 说明                                 |
-| ----------- | --------- | -------------------------------------------- |
+| ----------- | --------- | ------------- | ------------------------------------ |
-| **auth**    | `/auth`   | 用户注册、登录、JWT 鉴权、个人信息、密码修改 |
+| **auth**    | `/auth`   | `auth.*`      | 注册、登录、JWT 鉴权、个人信息、改密 |
-| **doubao**  | `/doubao` | 火山引擎豆包视觉理解模型集成                 |
+| **doubao**  | `/doubao` | `doubao.*`    | 火山引擎豆包视觉理解模型             |
-| **example** | `/calc`   | 四则运算示例（加/减/乘/除）+ 访问计数器      |
+| **example** | `/calc`   | `example.*`   | 四则运算 + 访问计数 + MQTT 心跳示例  |
 
 ---
 
@@ -185,10 +260,10 @@ uv pip install -e utils/<name>            # 安装/更新工具包子包
 | 包名                 | 说明                                          | 技术栈      |
 | -------------------- | --------------------------------------------- | ----------- |
 | `chongming-config`   | TOML 配置加载、TypedDict 类型定义             | Python      |
-| `chongming-database` | 异步 SQLite 封装（aiosqlite + SQLModel）      | Python      |
+| `chongming-database` | 异步 SQLite（aiosqlite + SQLModel）           | Python      |
 | `chongming-jwt`      | JWT 编码/验证（HS256/384/512, RS256/384/512） | Rust + PyO3 |
 | `chongming-password` | Argon2id 密码哈希与验证                       | Rust + PyO3 |
-| `chongming-request`  | Zenoh Worker 直连客户端（负载均衡、超时重试） | Python      |
+| `chongming-request`  | Zenoh Worker 客户端（负载均衡、超时重试）     | Python      |
 | `chongming-worker`   | Worker 框架核心（`@rpc_method`、运行时）      | Python      |
 
 ---

config.toml

@@ -50,6 +50,15 @@ allow_headers = ["*"]
 allow_methods = ["*"]
 allow_origins = ["*"]
 
+[development.mqtt]
+enabled = true
+host = "localhost"
+port = 1883
+
+[development.mqtt.broker]
+# amqtt Broker 可选配置，默认允许匿名连接
+# 详细参数参考: https://github.com/Yakifo/amqtt
+
 [plugins_worker]
 config_path = "./config/workers/config.toml"
 

publisher.py

@@ -0,0 +1,47 @@
+"""
+MQTT Publisher — 向 chongming Worker 发送 MQTT 请求
+
+用法:
+  python publisher.py <worker_type> <operation> [json_payload]
+  
+示例:
+  python publisher.py example add '{"a": 3, "b": 5, "_method": "GET"}'
+  python publisher.py doubao analyze '{"text": "hello", "_method": "POST"}'
+"""
+
+import asyncio
+import sys
+import json
+import logging
+from amqtt.client import MQTTClient
+from amqtt.mqtt.constants import QOS_1
+
+logging.basicConfig(level=logging.INFO)
+
+# MQTT 主题前缀（与 mqtt_bridge.py 保持一致）
+PREFIX = "chongming"
+
+# 默认参数
+DEFAULT_WORKER = "example"
+DEFAULT_OP = "add"
+DEFAULT_PAYLOAD = {"a": 3, "b": 5, "_method": "GET"}
+
+async def main():
+    worker_type = sys.argv[1] if len(sys.argv) > 1 else DEFAULT_WORKER
+    operation = sys.argv[2] if len(sys.argv) > 2 else DEFAULT_OP
+    payload_str = sys.argv[3] if len(sys.argv) > 3 else json.dumps(DEFAULT_PAYLOAD)
+
+    topic = f"{PREFIX}/{worker_type}/{operation}"
+    payload = json.loads(payload_str)
+
+    client = MQTTClient()
+    await client.connect("mqtt://localhost:1883/")
+
+    await client.publish(topic, json.dumps(payload).encode("utf-8"), QOS_1)
+    print(f"[Publisher] Sent to '{topic}': {payload}")
+
+    await client.disconnect()
+    print("[Publisher] Disconnected")
+
+if __name__ == "__main__":
+    asyncio.run(main())

src/chongming/app/__init__.py

@@ -52,7 +52,10 @@ async def lifespan(app: FastAPI):
     # 6. 注册路由
     await lifespan_setup.register_routes(app, manager)
 
-    # 7. 短暂等待，确保 worker 状态同步
+    # 7. 启动 MQTT Bridge（内嵌 Broker + Worker 桥接）
+    await lifespan_setup.setup_mqtt(app, manager)
+
+    # 8. 短暂等待，确保 worker 状态同步
     await asyncio.sleep(0.5)
 
     # 打印 worker 就绪状态
@@ -68,10 +71,13 @@ async def lifespan(app: FastAPI):
     # ---- 关闭阶段 ----
     logger.info("正在关闭应用...")
 
-    # 1. 清理 workers（调用 cleanup）
+    # 1. 停止 MQTT Bridge
+    await lifespan_setup.shutdown_mqtt(app)
+
+    # 2. 清理 workers（调用 cleanup）
     await lifespan_setup.cleanup_workers(app)
 
-    # 2. 取消刷新任务
+    # 3. 取消刷新任务
     refresh_task = getattr(app.state, "refresh_task", None)
     if refresh_task:
         refresh_task.cancel()
@@ -80,17 +86,17 @@ async def lifespan(app: FastAPI):
         except asyncio.CancelledError:
             pass
 
-    # 3. 取消所有 worker 任务
+    # 4. 取消所有 worker 任务
     for task in worker_tasks:
         task.cancel()
     if worker_tasks:
         await asyncio.gather(*worker_tasks, return_exceptions=True)
 
-    # 4. 关闭 Zenoh 会话
+    # 5. 关闭 Zenoh 会话
     session.close()
     logger.info("Zenoh 会话已关闭")
 
-    # 5. 终止 nanio
+    # 6. 终止 nanio
     await lifespan_setup.shutdown_nanio(app)
 
     logger.info("应用已关闭")

src/chongming/app/core/config_.py

@@ -102,6 +102,15 @@ class Cors(TypedDict, total=False):
     allow_headers: List[str]
 
 
+class MQTTConfig(TypedDict, total=False):
+    """MQTT Bridge 配置（按环境区分）"""
+
+    enabled: bool
+    host: str
+    port: int
+    broker: dict
+
+
 class ModuleSystem(TypedDict, total=False):
     """模块系统配置，定义模块加载类型和路径"""
 
@@ -123,6 +132,7 @@ class Env(TypedDict, total=False):
     server: Required[Server]
     logging: Required[Logging]
     cors: Cors
+    mqtt: MQTTConfig
     module_system: ModuleSystem
     file_system: FileSystem
 

src/chongming/app/core/lifespan_setup.py

@@ -17,6 +17,7 @@ from chongming_config import get_config, get_workers_config
 from ..core.logger import get_logger
 from ..core.route_generator import register_dynamic_routes
 from ..core.worker_manager import WorkerManager
+from ..core.mqtt_bridge import MQTTBridge
 
 config = get_config()
 logger = get_logger("app")
@@ -182,6 +183,46 @@ async def register_routes(app: FastAPI, manager: WorkerManager):
     register_dynamic_routes(app, manager, prefix=prefix)
     # 可选：等待路由注册生效（已在 register_dynamic_routes 内部完成日志）
 
+async def setup_mqtt(app: FastAPI, worker_manager: WorkerManager) -> MQTTBridge | None:
+    """
+    启动 MQTT Bridge（内嵌 Broker + 客户端桥接）。
+
+    从 config.toml 的环境段（如 [development.mqtt]）读取配置，
+    默认监听 localhost:1883。若 enabled 为 false，则跳过启动。
+    """
+    env_config = config.get(env, {})
+    mqtt_config = env_config.get("mqtt", {})
+    if not mqtt_config.get("enabled", True):
+        logger.info("MQTT Bridge 已禁用（mqtt.enabled = false）")
+        return None
+
+    host = mqtt_config.get("host", "localhost")
+    port = mqtt_config.get("port", 1883)
+    broker_config = mqtt_config.get("broker", {})
+
+    bridge = MQTTBridge(
+        worker_manager=worker_manager,
+        host=host,
+        port=port,
+        broker_config=broker_config,
+    )
+
+    app.state.mqtt_bridge = bridge
+    await bridge.start()
+    logger.info(f"MQTT Bridge 已启动 (host={host}, port={port})")
+    return bridge
+
+
+async def shutdown_mqtt(app: FastAPI):
+    """停止 MQTT Bridge。"""
+    bridge: MQTTBridge | None = getattr(app.state, "mqtt_bridge", None)
+    if bridge is None:
+        return
+
+    if bridge.is_running:
+        await bridge.stop()
+
+
 async def shutdown_nanio(app: FastAPI):
     """终止 nanio 进程。"""
     proc = getattr(app.state, "nanio_process", None)

src/chongming/app/core/mqtt_bridge.py

@@ -0,0 +1,292 @@
+"""
+MQTT Bridge — 将 MQTT 消息桥接到 Worker 插件
+
+架构:
+  MQTT 客户端订阅 chongming/+/+ 主题
+  → 解析 topic: chongming/<worker_type>/<operation>
+  → 调用 WorkerManager.call(worker_type, subject, payload)
+  → 将响应发布到 chongming/<worker_type>/<operation>/response
+
+同时内嵌一个 amqtt Broker，无需外部 MQTT 服务器。
+"""
+
+import asyncio
+import json
+from typing import Any, Dict, Optional
+
+from amqtt.broker import Broker
+from amqtt.client import MQTTClient
+from amqtt.mqtt.constants import QOS_1
+
+from .logger import get_logger
+
+logger = get_logger("mqtt")
+
+# MQTT 主题前缀
+MQTT_TOPIC_PREFIX = "chongming"
+# 订阅通配符：匹配 chongming/<worker_type>/<operation>
+MQTT_SUBSCRIBE_TOPIC = f"{MQTT_TOPIC_PREFIX}/+/+"
+
+
+class MQTTBridge:
+    """
+    MQTT 桥接器
+
+    内嵌 amqtt Broker + 客户端，将 MQTT 消息转发到 WorkerManager。
+
+    用法:
+        bridge = MQTTBridge(worker_manager, host="localhost", port=1883)
+        await bridge.start()
+        # ... 应用运行 ...
+        await bridge.stop()
+    """
+
+    def __init__(
+        self,
+        worker_manager,
+        host: str = "localhost",
+        port: int = 1883,
+        broker_config: Optional[Dict[str, Any]] = None,
+    ):
+        self._worker_manager = worker_manager
+        self._host = host
+        self._port = port
+        self._broker_config = broker_config or {}
+
+        self._broker: Optional[Broker] = None
+        self._client: Optional[MQTTClient] = None
+        self._broker_task: Optional[asyncio.Task] = None
+        self._listen_task: Optional[asyncio.Task] = None
+        self._running = False
+
+    # ============================================================
+    # 生命周期
+    # ============================================================
+
+    async def start(self) -> None:
+        """启动 MQTT Broker 和客户端桥接"""
+        if self._running:
+            logger.warning("MQTT Bridge 已在运行中")
+            return
+
+        # 1. 启动内嵌 Broker
+        await self._start_broker()
+
+        # 2. 启动客户端桥接
+        await self._start_client()
+
+        self._running = True
+        logger.info(f"MQTT Bridge 已启动 (broker: {self._host}:{self._port})")
+
+    async def stop(self) -> None:
+        """停止客户端和 Broker"""
+        if not self._running:
+            return
+
+        logger.info("正在关闭 MQTT Bridge...")
+
+        # 1. 停止客户端
+        await self._stop_client()
+
+        # 2. 停止 Broker
+        await self._stop_broker()
+
+        self._running = False
+        logger.info("MQTT Bridge 已关闭")
+
+    # ============================================================
+    # Broker（内嵌 MQTT 服务器）
+    # ============================================================
+
+    async def _start_broker(self) -> None:
+        """启动内嵌 amqtt Broker"""
+        config = {
+            "listeners": {
+                "default": {
+                    "type": "tcp",
+                    "bind": f"{self._host}:{self._port}",
+                    "max_connections": 500,
+                },
+            },
+            "timeout_disconnect_delay": 2,
+            "sys_interval": 0,  # 禁用 sys 主题周期性发布，避免 None > 0 异常
+            "auth": {
+                "allow-anonymous": True,
+            },
+        }
+        # 合并用户自定义配置
+        config.update(self._broker_config)
+
+        self._broker = Broker(config)
+
+        async def _run_broker():
+            try:
+                await self._broker.start() # type: ignore
+                logger.info(f"amqtt Broker 已启动 @ {self._host}:{self._port}")
+                # 保持 broker 运行直到被取消
+                while True:
+                    await asyncio.sleep(1)
+            except asyncio.CancelledError:
+                pass
+            except Exception as e:
+                logger.error(f"Broker 运行异常: {e}")
+
+        self._broker_task = asyncio.create_task(_run_broker())
+        # 等待 broker 就绪
+        await asyncio.sleep(0.5)
+
+    async def _stop_broker(self) -> None:
+        """停止 Broker"""
+        if self._broker_task and not self._broker_task.done():
+            self._broker_task.cancel()
+            try:
+                await self._broker_task
+            except asyncio.CancelledError:
+                pass
+
+        if self._broker:
+            try:
+                await self._broker.shutdown()
+            except Exception as e:
+                logger.error(f"Broker 关闭异常: {e}")
+            self._broker = None
+
+        logger.info("amqtt Broker 已停止")
+
+    # ============================================================
+    # 客户端桥接
+    # ============================================================
+
+    async def _start_client(self) -> None:
+        """启动 MQTT 客户端并订阅 worker 主题"""
+        self._client = MQTTClient()
+        await self._client.connect(f"mqtt://{self._host}:{self._port}/")
+        await self._client.subscribe([(MQTT_SUBSCRIBE_TOPIC, QOS_1)])
+        logger.info(f"MQTT 客户端已订阅: {MQTT_SUBSCRIBE_TOPIC}")
+
+        self._listen_task = asyncio.create_task(self._message_loop())
+
+    async def _stop_client(self) -> None:
+        """断开 MQTT 客户端"""
+        if self._listen_task and not self._listen_task.done():
+            self._listen_task.cancel()
+            try:
+                await self._listen_task
+            except asyncio.CancelledError:
+                pass
+
+        if self._client:
+            try:
+                await self._client.unsubscribe([MQTT_SUBSCRIBE_TOPIC])
+                await self._client.disconnect()
+            except Exception as e:
+                logger.error(f"MQTT 客户端断开异常: {e}")
+            self._client = None
+
+        logger.info("MQTT 客户端已断开")
+
+    # ============================================================
+    # 消息处理
+    # ============================================================
+
+    async def _message_loop(self) -> None:
+        """消息循环：接收 MQTT 消息并转发到 WorkerManager"""
+        while True:
+            try:
+                message = await self._client.deliver_message() # type: ignore
+                await self._handle_message(message)
+            except asyncio.CancelledError:
+                break
+            except Exception as e:
+                logger.error(f"MQTT 消息处理异常: {e}")
+
+    async def _handle_message(self, message) -> None:
+        """
+        处理收到的 MQTT 消息
+
+        解析 topic: chongming/<worker_type>/<operation>
+        转发到 WorkerManager.call()
+        发布响应到 chongming/<worker_type>/<operation>/response
+        """
+        topic: str = message.topic  # type: ignore
+        payload_bytes: bytes = message.data  # type: ignore
+
+        # 解析主题
+        parts = topic.split("/")
+        if len(parts) < 3:
+            logger.warning(f"无效的 MQTT 主题: {topic}")
+            return
+
+        worker_type = parts[1]
+        operation = parts[2]
+        subject = f"{worker_type}.{operation}"
+        response_topic = f"{MQTT_TOPIC_PREFIX}/{worker_type}/{operation}/response"
+
+        # 解析载荷
+        try:
+            payload_str = payload_bytes.decode("utf-8") if payload_bytes else "{}"
+            payload = json.loads(payload_str)
+        except json.JSONDecodeError:
+            logger.warning(f"无效的 JSON 载荷: {payload_bytes}")
+            await self._publish_response(
+                response_topic,
+                {"error": "Invalid JSON payload"},
+            )
+            return
+
+        # 确保 _method 字段存在
+        if "_method" not in payload:
+            payload["_method"] = "GET"
+
+        logger.info(
+            f"MQTT → Worker: {topic} [{payload.get('_method', 'GET')}] "
+            f"-> {worker_type}/{subject}"
+        )
+
+        # 调用 WorkerManager
+        try:
+            result = await self._worker_manager.call(
+                worker_type=worker_type,
+                subject=subject,
+                payload=payload,
+                timeout=float(payload.pop("_timeout", 30.0)),
+                cacheable=bool(payload.pop("_cacheable", False)),
+            )
+            # result 为 None 时不发布响应（如心跳上报等无需返回的场景）
+            if result is not None:
+                await self._publish_response(response_topic, result)
+                logger.debug(f"Worker → MQTT: {response_topic} success")
+            else:
+                logger.debug(f"Worker → MQTT: {topic} (无响应，已跳过)")
+        except RuntimeError as e:
+            error_resp = {"error": str(e), "worker_type": worker_type, "subject": subject}
+            await self._publish_response(response_topic, error_resp)
+            logger.warning(f"Worker 调用失败: {e}")
+        except Exception as e:
+            error_resp = {"error": str(e), "worker_type": worker_type, "subject": subject}
+            await self._publish_response(response_topic, error_resp)
+            logger.error(f"MQTT 桥接异常: {e}")
+
+    async def _publish_response(self, topic: str, data: Any) -> None:
+        """将响应发布到 MQTT 主题"""
+        try:
+            payload = json.dumps(data, ensure_ascii=False, default=str)
+            await self._client.publish(topic, payload.encode("utf-8"), QOS_1) # type: ignore
+        except Exception as e:
+            logger.error(f"MQTT 发布失败 [{topic}]: {e}")
+
+    # ============================================================
+    # 属性
+    # ============================================================
+
+    @property
+    def is_running(self) -> bool:
+        return self._running
+
+    @property
+    def broker_host(self) -> str:
+        return self._host
+
+    @property
+    def broker_port(self) -> int:
+        return self._port

src/chongming/app/core/route_generator.py

@@ -128,9 +128,11 @@ def load_worker_routes_from_dir(worker_type: str) -> List[Dict[str, Any]]:
           response_model = { ... }
           [POST]
           ...                 # 同一个接口支持多个请求方法
+          [MQTT]
+          ...                 # MQTT 专用端点（不注册 HTTP 路由）
 
     返回: List[Dict], 每个 dict 包含 flatten 后的路由配置,
-          每个方法节(如 GET) 生成一个独立的路由条目。
+          每个方法节(如 GET/POST/MQTT) 生成一个独立的路由条目。
     """
     from pathlib import Path
     import tomllib
@@ -386,6 +388,13 @@ def _register_single_route(
     """注册单个路由，生成正确的函数签名以便 OpenAPI 正确生成文档"""
     rc = RouteConfig(route_conf)
 
+    # MQTT 端点：不注册 HTTP 路由，仅通过 MQTT Bridge 调用
+    if rc.method == "MQTT":
+        logger.info(
+            f"MQTT 端点: subject={worker_type}.{rc.subject} (仅 MQTT，不注册 HTTP 路由)"
+        )
+        return
+
     # internal 端点：不注册 HTTP 路由，仅保留 Zenoh RPC 内部调用能力
     if rc.internal:
         logger.info(

src/chongming/scripts/gen_models.py

@@ -91,8 +91,9 @@ def parse_response_model(response_model: dict[str, Any]) -> list[dict[str, Any]]
 
 
 def _model_name(base: str, method: str, suffix: str) -> str:
-    """生成模型名: e.g. add + GET + Request -> AddGetRequest"""
+    """生成模型名: e.g. add + GET + Request -> AddGetRequest, add + MQTT + Request -> AddMqttRequest"""
-    return f"{base.title().replace('-', '').replace('_', '')}{method.title()}{suffix}"
+    method_title = "Mqtt" if method == "MQTT" else method.title()
+    return f"{base.title().replace('-', '').replace('_', '')}{method_title}{suffix}"
 
 
 def generate_pydantic_code(
@@ -243,7 +244,7 @@ def scan_config_dir(config_dir: Path) -> list[dict[str, Any]]:
 
         route_name = toml_file.stem
 
-        for method in ("GET", "POST", "PUT", "DELETE", "PATCH"):
+        for method in ("GET", "POST", "PUT", "DELETE", "PATCH", "MQTT"):
             if method not in data:
                 continue
 

subscriber.py

@@ -0,0 +1,64 @@
+"""
+MQTT Subscriber — 监听 chongming Worker 的响应
+
+用法:
+  python subscriber.py [worker_type] [operation]
+  
+示例:
+  python subscriber.py                    # 监听所有 chongming 响应
+  python subscriber.py example add        # 只监听 example.add 的响应
+"""
+
+import asyncio
+import json
+import sys
+import logging
+from amqtt.client import MQTTClient
+from amqtt.mqtt.constants import QOS_1
+
+logging.basicConfig(level=logging.INFO)
+
+PREFIX = "chongming"
+
+# 默认订阅所有 chongming 响应
+DEFAULT_TOPIC = f"{PREFIX}/+/+/response"
+
+async def main():
+    worker_type = sys.argv[1] if len(sys.argv) > 1 else None
+    operation = sys.argv[2] if len(sys.argv) > 2 else None
+
+    if worker_type and operation:
+        topic = f"{PREFIX}/{worker_type}/{operation}/response"
+    elif worker_type:
+        topic = f"{PREFIX}/{worker_type}/+/response"
+    else:
+        topic = DEFAULT_TOPIC
+
+    client = MQTTClient()
+    await client.connect("mqtt://localhost:1883/")
+
+    await client.subscribe([(topic, QOS_1)])
+    print(f"[Subscriber] Subscribed to '{topic}'. Waiting for messages... (Press Ctrl+C to stop)")
+
+    try:
+        while True:
+            message = await client.deliver_message()
+            payload = message.data.decode("utf-8")  # type: ignore
+            try:
+                data = json.loads(payload)
+                print(f"[Subscriber] Topic: {message.topic}")  # type: ignore
+                print(f"  Response: {json.dumps(data, indent=2, ensure_ascii=False)}")
+            except json.JSONDecodeError:
+                print(f"[Subscriber] Topic: {message.topic}, Raw: {payload}")  # type: ignore
+    except asyncio.CancelledError:
+        print("[Subscriber] Stopping...")
+    finally:
+        await client.unsubscribe([topic])
+        await client.disconnect()
+        print("[Subscriber] Disconnected")
+
+if __name__ == "__main__":
+    try:
+        asyncio.run(main())
+    except KeyboardInterrupt:
+        print("[Subscriber] Interrupted by user")

test_amqtt.py

@@ -0,0 +1,30 @@
+import asyncio
+from asyncio import CancelledError
+import logging
+
+from amqtt.broker import Broker
+
+"""
+This sample shows how run a broker
+"""
+
+formatter = "[%(asctime)s] :: %(levelname)s :: %(name)s :: %(message)s"
+logging.basicConfig(level=logging.INFO, format=formatter)
+
+async def run_server() -> None:
+    broker = Broker()
+    try:
+        await broker.start()
+        while True:
+            await asyncio.sleep(1)
+    except CancelledError:
+        await broker.shutdown()
+
+def __main__():
+    try:
+        asyncio.run(run_server())
+    except KeyboardInterrupt:
+        print("Server exiting...")
+
+if __name__ == "__main__":
+    __main__()

test_mqtt_e2e.py

@@ -0,0 +1,102 @@
+"""
+MQTT 端到端测试脚本
+
+测试流程:
+  1. 启动内嵌 amqtt Broker
+  2. 启动 subscriber（监听响应）
+  3. 启动 publisher（发送 MQTT 请求 → chongming/example/add）
+  4. 验证 subscriber 收到正确的响应
+
+注意: 此脚本独立运行，不依赖 chongming 服务端。
+      使用时需确保 chongming 服务已通过 `uv run serve` 启动。
+"""
+
+import asyncio
+import json
+import sys
+import logging
+from amqtt.client import MQTTClient
+from amqtt.mqtt.constants import QOS_1
+
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s [%(name)s] %(message)s",
+)
+
+PREFIX = "chongming"
+WORKER = "example"
+OPERATION = "add"
+REQUEST_TOPIC = f"{PREFIX}/{WORKER}/{OPERATION}"
+RESPONSE_TOPIC = f"{PREFIX}/{WORKER}/{OPERATION}/response"
+BROKER_URL = "mqtt://localhost:1883/"
+
+# 测试用例
+TEST_CASES = [
+    {"a": 3, "b": 5, "_method": "MQTT", "expected": 8},
+    {"a": -1, "b": 10, "_method": "MQTT", "expected": 9},
+    {"a": 3.14, "b": 2.86, "_method": "MQTT", "expected": 6.0},
+]
+
+PASS = 0
+FAIL = 0
+
+
+async def run_test():
+    global PASS, FAIL
+    client = MQTTClient()
+    await client.connect(BROKER_URL)
+    await client.subscribe([(RESPONSE_TOPIC, QOS_1)])
+    print(f"📡 已订阅: {RESPONSE_TOPIC}\n")
+
+    for i, case in enumerate(TEST_CASES):
+        a, b, expected = case["a"], case["b"], case["expected"]
+        payload = {"a": a, "b": b, "_method": "MQTT"}
+        payload_bytes = json.dumps(payload).encode("utf-8")
+
+        await client.publish(REQUEST_TOPIC, payload_bytes, QOS_1)
+        print(f"📤 [{i+1}/{len(TEST_CASES)}] 发送: {REQUEST_TOPIC} → {payload}")
+
+        # 等待响应（超时 5 秒）
+        try:
+            message = await asyncio.wait_for(client.deliver_message(), timeout=5.0)
+            data = json.loads(message.data.decode("utf-8"))  # type: ignore
+            actual = data.get("result")
+            status = "✅ PASS" if abs(actual - expected) < 0.01 else "❌ FAIL"
+            if status == "✅ PASS":
+                PASS += 1
+            else:
+                FAIL += 1
+            print(f"📥 [{i+1}/{len(TEST_CASES)}] 响应: {json.dumps(data)} → {status} (expect={expected}, got={actual})")
+        except asyncio.TimeoutError:
+            FAIL += 1
+            print(f"📥 [{i+1}/{len(TEST_CASES)}] ❌ FAIL: 超时无响应")
+
+    await client.unsubscribe([RESPONSE_TOPIC])
+    await client.disconnect()
+
+    print(f"\n{'='*50}")
+    print(f"测试结果: ✅ {PASS} 通过  ❌ {FAIL} 失败")
+    print(f"{'='*50}")
+    return FAIL == 0
+
+
+async def main():
+    # 先检查 Broker 是否可用
+    test_client = MQTTClient()
+    try:
+        await asyncio.wait_for(test_client.connect(BROKER_URL), timeout=3.0)
+        await test_client.disconnect()
+    except Exception:
+        print("❌ 无法连接到 MQTT Broker (localhost:1883)")
+        print("   请先启动 chongming 服务: uv run serve")
+        sys.exit(1)
+
+    success = await run_test()
+    sys.exit(0 if success else 1)
+
+
+if __name__ == "__main__":
+    try:
+        asyncio.run(main())
+    except KeyboardInterrupt:
+        print("\n⏹️  测试被中断")