1. 引言

WorkBuddy 是一款面向团队协作与自动化任务管理的智能助手平台，支持通过消息渠道将任务提醒、审批通知、数据报表等能力推送到用户日常使用的聊天工具中。QQ 作为国内用户基数最大的即时通讯软件之一，是 WorkBuddy 触达用户的重要渠道。本文将从账号准备、协议选型、消息收发、富文本适配到生产部署，完整演示如何将 WorkBuddy 接入 QQ，并提供可直接运行的代码示例。

2. 前置准备

在开始编码之前，需要完成以下准备工作：

• 一个可用的 QQ 账号，建议使用企业专用账号，避免影响个人使用。
• 一个可公网访问的服务器，用于接收 QQ 消息回调。
• WorkBuddy 平台账号，并创建好对应的应用与 API Token。
• Python 3.9 及以上运行环境，或 Node.js 16 及以上运行环境。

本文以 Python 为例进行演示，Node.js 版本的实现思路完全一致。

3. 整体架构设计

WorkBuddy 接入 QQ 的整体架构如下图所示：

flowchart TD
    A[QQ 用户] -- 发送消息 --> B[QQ 机器人服务]
    B -- 消息回调 --> C[WorkBuddy 网关]
    C -- 解析意图 --> D[WorkBuddy 任务引擎]
    D -- 执行结果 --> C
    C -- 回复消息 --> B
    B -- 推送给用户 --> A
    E[WorkBuddy 管理后台] -- 配置渠道 --> C

整个链路中，QQ 机器人服务负责与 QQ 服务器保持长连接，WorkBuddy 网关负责消息的解析与任务分发，两者通过 HTTP Webhook 进行通信。

4. 创建 QQ 机器人

首先需要在 QQ 开放平台注册一个机器人应用。注册完成后，你会获得 app_id 和 app_secret，这两个凭证用于获取访问令牌。

下面是一个获取 QQ 机器人访问令牌的示例代码：

import requests
APP_ID = "your_app_id"
APP_SECRET = "your_app_secret"
def get_access_token():
url = "https://bots.qq.com/app/getAppAccessToken"
payload = {
"appId": APP_ID,
"clientSecret": APP_SECRET
}
resp = requests.post(url, json=payload, timeout=10)
resp.raise_for_status()
data = resp.json()
return data["access_token"]
if name == "main":
token = get_access_token()
print(f"获取到访问令牌: {token[:16]}...")

访问令牌的有效期通常为 7200 秒，建议在服务中缓存并定时刷新，避免频繁请求。

5. 搭建消息接收服务

QQ 机器人通过 WebSocket 长连接接收消息事件。下面使用官方 SDK 搭建一个最简单的消息监听服务：

import asyncio
from qqbot.core.util.yaml_util import load_yaml
from qqbot.bot import Bot
from qqbot.model.message import Message
async def on_message(event: Message):
"""处理收到的 QQ 消息"""
content = event.content
print(f"收到消息: {content}")
# 这里将消息转发给 WorkBuddy 网关
await forward_to_workbuddy(event)
async def forward_to_workbuddy(event: Message):
"""将消息转发到 WorkBuddy 网关"""
import httpx
webhook_url = "https://your-workbuddy-gateway.com/webhook/qq"
payload = {
"msg_id": event.id,
"user_id": event.author.id,
"content": event.content,
"channel": "qq"
}
async with httpx.AsyncClient() as client:
resp = await client.post(webhook_url, json=payload, timeout=10)
print(f"转发结果: {resp.status_code}")
def main():
yaml_config = load_yaml("config.yaml")
bot = Bot(yaml_config)
bot.register_message_handler(on_message)
bot.run()
if name == "main":
main()

对应的 config.yaml 配置文件内容如下：

appid: "your_app_id"
secret: "your_app_secret"
sandbox: false

6. WorkBuddy 网关对接

WorkBuddy 网关负责接收 QQ 消息并转换为内部任务。下面是一个基于 FastAPI 的网关接收端示例：

from fastapi import FastAPI, Request
from pydantic import BaseModel
import uvicorn
app = FastAPI()
class QQMessage(BaseModel):
msg_id: str
user_id: str
content: str
channel: str
@app.post("/webhook/qq")
async def receive_qq_message(msg: QQMessage):
"""接收来自 QQ 机器人的消息"""
# 解析用户指令
intent = parse_intent(msg.content)
# 调用 WorkBuddy 任务引擎
result = await workbuddy_execute(intent, msg.user_id)
# 返回结果给 QQ 机器人
return {
"code": 0,
"reply": result
}
def parse_intent(content: str) -> dict:
"""简单的意图解析，实际可接入 NLP 服务"""
if "日报" in content:
return {"action": "daily_report", "params": {}}
if "审批" in content:
return {"action": "approval", "params": {}}
return {"action": "unknown", "params": {}}
async def workbuddy_execute(intent: dict, user_id: str) -> str:
"""调用 WorkBuddy 执行任务（示例）"""
# 这里替换为真实的 WorkBuddy API 调用
return f"已收到指令: {intent['action']}，正在处理中..."
if name == "main":
uvicorn.run(app, host="0.0.0.0", port=8000)

7. 主动推送消息到 QQ

除了被动回复，WorkBuddy 还需要主动向用户推送通知。下面演示如何通过 QQ 机器人 API 发送主动消息：

import requests
def send_qq_message(access_token, user_openid, content):
"""向指定用户发送 QQ 消息"""
url = "https://api.sgroup.qq.com/v2/users/{openid}/messages".format(
openid=user_openid
)
headers = {
"Authorization": f"QQBot {access_token}",
"Content-Type": "application/json"
}
payload = {
"content": content,
"msg_type": 0  # 0 表示文本消息
}
resp = requests.post(url, headers=headers, json=payload, timeout=10)
resp.raise_for_status()
return resp.json()
def push_daily_report(access_token, user_openid):
"""推送日报到 QQ"""
report = "【今日工作日报】\n"
report += "1. 完成 WorkBuddy 接入 QQ 的架构设计\n"
report += "2. 实现消息接收与转发服务\n"
report += "3. 联调主动推送功能\n"
send_qq_message(access_token, user_openid, report)

8. 富文本消息适配

QQ 机器人支持多种消息类型，包括文本、图片、Markdown 等。WorkBuddy 生成的任务卡片可以通过 Markdown 消息呈现，示例代码如下：

def send_markdown_card(access_token, user_openid):
    """发送 Markdown 格式的任务卡片"""
    url = "https://api.sgroup.qq.com/v2/users/{openid}/messages".format(
        openid=user_openid
    )
    headers = {
        "Authorization": f"QQBot {access_token}",
        "Content-Type": "application/json"
    }
    markdown_content = (
        "## 任务审批通知\n\n"
        "**申请人**: 张三\n\n"
        "**事项**: 服务器采购申请\n\n"
        "**金额**: ¥ 12,000\n\n"
        "---\n\n"
        "请点击下方按钮进行审批：\n\n"
        "- [通过审批](https://workbuddy.example.com/approve/12345)\n"
        "- [拒绝审批](https://workbuddy.example.com/reject/12345)"
    )
    payload = {
        "msg_type": 2,  # 2 表示 Markdown 消息
        "markdown": {
            "content": markdown_content
        }
    }
    resp = requests.post(url, headers=headers, json=payload, timeout=10)
    resp.raise_for_status()
    return resp.json()

9. 完整联调示例

下面给出一个完整的端到端示例，演示用户发送「查日报」后，WorkBuddy 自动回复日报内容：

import asyncio
import httpx
from qqbot.bot import Bot
from qqbot.model.message import Message
WORKBUDDY_GATEWAY = "https://your-workbuddy-gateway.com"
async def handle_message(event: Message):
content = event.content.strip()
if "查日报" in content:
# 调用 WorkBuddy 获取日报
async with httpx.AsyncClient() as client:
resp = await client.get(
f"{WORKBUDDY_GATEWAY}/api/report/daily",
params={"user_id": event.author.id},
timeout=15
)
report = resp.json().get("data", "暂无日报数据")
# 回复用户
await event.reply(content=report)
else:
await event.reply(content="您好，我是 WorkBuddy 助手。发送「查日报」可获取今日工作日报。")
def main():
bot = Bot(auto_reconnect=True)
bot.register_message_handler(handle_message)
bot.run()
if name == "main":
main()

10. 生产部署注意事项

将服务部署到生产环境时，需要注意以下几点：

• 安全加固：Webhook 接口必须校验签名，防止伪造请求。
• 消息去重：QQ 可能重复推送消息，需要根据 msg_id 做幂等处理。
• 限流控制：QQ 机器人 API 有频率限制，建议在网关层做令牌桶限流。
• 日志监控：记录完整的消息链路日志，便于排查问题。
• 多实例部署：使用 Redis 等共享存储，保证多实例间的状态一致。

下面是一个简单的签名校验示例：

import hashlib
import hmac
def verify_signature(payload: bytes, signature: str, secret: str) -> bool:
"""校验 Webhook 签名"""
expected = hmac.new(
secret.encode(),
payload,
hashlib.sha256
).hexdigest()
return hmac.compare_digest(expected, signature)

11. 总结

本文从账号准备、架构设计、消息收发、富文本适配到生产部署，完整演示了 WorkBuddy 接入 QQ 的全过程。核心要点包括：通过 QQ 开放平台创建机器人并获取凭证；使用 WebSocket 长连接接收消息；通过 HTTP Webhook 与 WorkBuddy 网关通信；使用 Markdown 消息呈现丰富的任务卡片。希望本文的代码示例能帮助你快速完成接入工作，在实际项目中建议根据业务场景进一步扩展消息类型和交互能力。