OpenClaw Agent 与 QQ 深度集成实战：基于 NapCat 的全双工本地化智能体架构
本指南聚焦于在 Linux 服务器上构建一个完全脱离云端 API 调用、复用 OpenClaw 原生执行链路的 QQ 接入方案。核心目标是实现消息路由统一、上下文记忆共享、工具调用一致的多端协同智能体，避免因重复初始化导致的人设断裂或状态丢失。
整体架构设计
系统采用容器化分层结构，所有组件运行于同一 Docker 自定义网络（如 claw-net），确保服务间低延迟互通：
- NapCatQQ 容器：以 iPad 模式登录轻量级 QQ 小号，暴露 OneBot v11 WebSocket 接口（
ws://napcat:3001）； - OpenClaw Kernel 容器：预装完整 OpenClaw CLI 环境，不暴露 HTTP 服务，仅作为本地命令执行中枢；
- Bridge Service：独立 Python 脚本，部署于 OpenClaw 容器内，负责监听 NapCat 消息、触发
openclaw agent子进程，并将响应回传至 QQ。
关键问题与解决方案
1. 容器文件同步失效
常见现象：修改宿主机脚本后容器内仍运行旧逻辑。根本原因在于 Docker 镜像层只读，挂载卷未配置或路径映射错误。
推荐修复方式（非暴力覆盖）：
# 创建绑定挂载卷，使代码实时生效
docker run -d \
--name openclaw_kernel \
--network claw-net \
-v /root/claw-bridge:/home/node/bridge \
openclaw/kernel:latest
# 启动时直接运行挂载目录下的脚本
docker exec -it openclaw_kernel python3 /home/node/bridge/qq_gateway.py
2. CLI 参数顺序引发的命令解析失败
当使用 subprocess 调用 openclaw agent 时，若将 --no-color 放在 agent 子命令之后，会导致参数解析器误判为 agent 的专属选项而报错。
正确调用模式（必须严格遵循）：
await asyncio.create_subprocess_exec(
"openclaw", "--no-color", "agent",
"--to", "13*******75",
"--message", user_input,
...
)
该写法确保全局标志被主程序识别，子命令接收其专属参数，避免 unknown option '--no-color' 类错误。
3. NapCat 连接假活跃状态
现象：日志显示"Login Success"，但无消息收发能力。本质是腾讯服务端已撤销 Token 权限，而 NapCat 本地缓存未刷新。
标准化恢复流程：
- 手机端进入「设置 → 账号安全 → 登录设备管理」，强制下线除当前设备外的所有会话；
- 重启 NapCat 容器：
docker restart napcat； - 实时跟踪日志：
docker logs -f napcat | grep -i "qrcode\|success"； - 扫码完成认证后，再启动 bridge 脚本。
桥接服务核心实现（qq_gateway.py）
以下为精简、健壮、生产就绪的桥接逻辑，省略异常重试与日志分级，突出关键路径：
import asyncio
import json
import websockets
import os
WS_ENDPOINT = "ws://napcat:3001"
MASTER_QQ = os.getenv("MASTER_QQ_ID", "13*******75")
async def invoke_agent(message: str) -> str:
proc = await asyncio.create_subprocess_exec(
"openclaw", "--no-color", "agent",
"--to", MASTER_QQ,
"--message", message,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE
)
stdout, stderr = await proc.communicate()
return stdout.decode().strip() if proc.returncode == 0 else f"[ERR] {stderr.decode().strip()}"
async def handle_message(ws, data: dict):
if data.get("post_type") != "message":
return
if str(data.get("user_id")) != MASTER_QQ:
return
text = data.get("raw_message", "").strip()
if not text:
return
response = await invoke_agent(text)
await ws.send(json.dumps({
"action": "send_private_msg",
"params": {"user_id": int(MASTER_QQ), "message": response}
}))
async def main():
async with websockets.connect(WS_ENDPOINT, ping_interval=30) as ws:
print("[INFO] Bridge connected to NapCat")
while True:
try:
raw = await ws.recv()
payload = json.loads(raw)
await handle_message(ws, payload)
except websockets.exceptions.ConnectionClosed:
print("[WARN] WebSocket disconnected, reconnecting...")
break
except Exception as e:
print(f"[ERROR] Handler failed: {e}")
if __name__ == "__main__":
asyncio.run(main())
