qqbotsdk

QQ 官方机器人 Python SDK（Python 3.13 / asyncio / aiohttp），支持 WebSocket 与 Webhook 两种接入方式。

使用指南 · API 参考 · 架构 · 示例 · 更新日志

架构

flowchart TB
    subgraph 接入层
        WS["WebsocketConnecter（ws 长连接 / 重连 / 看门狗）"]
        WH["WebhookConnecter（http 回调 / 验签 / 去重）"]
    end
    subgraph 分发层
        Q[("EventQueue")]
        EE["EventEmitter（路由 / DI 注入 / 异常隔离）"]
    end
    subgraph 协议层
        WSP["WebsocketProtocol（鉴权 / 心跳 / 序列号）"]
        WHP["WebhookProtocol（op=13 验证应答）"]
    end
    subgraph 业务层
        API["BotApi（REST + ApiError）"]
        USER["用户 handler @ee.on(...)"]
    end
    QQ["QQ 开放平台"]

    WS --> Q
    WH --> Q
    Q --> EE
    EE <--> WSP
    EE --> USER
    USER --> API
    API --> QQ
    WS --> QQ

组件生命周期由 dishka 容器统一装配（APP scope，无全局单例），依赖关系、数据流与设计决策详见 docs/ARCHITECTURE.md。

特性

• 内置协议处理器（ws_protocol.py / webhook_protocol.py）：HELLO 鉴权（IDENTIFY/RESUME 自动切换）、apscheduler 定时心跳、READY 会话捕获、INVALID_SESSION 会话失效处理
• 断线自动重连（指数退避 + 心跳看门狗），重连时自动 RESUME 续传
• Webhook 接入自带 Ed25519 验签、op=13 验证应答与 60s TTL 事件去重
• 事件订阅由项目根目录的 intents.toml 控制，将对应事件改为 true 即可
• Markdown 与内嵌按钮一等支持：post_group_message(..., markdown=..., keyboard=...) 自动置 msg_type=2，respond_interaction() 回应按钮回调
• 组件生命周期由 dishka DI 容器统一管理，全 SDK 共享一个 HTTP 连接池

配置

.env（APPID/APPSECRET 必填，缺失启动即报错）：

APPID=你的AppID
APPSECRET=你的AppSecret
BASE_URL=https://api.bot.qq.com   # 可选，默认官方地址
TIMEOUT=5000                      # 可选，毫秒
CONNECTER=websocket               # websocket 或 webhook

# 仅 webhook 接入时
WEBHOOK_HOST=0.0.0.0
WEBHOOK_PORT=8080
WEBHOOK_PATH=/qqbot/webhook

使用

编写入口（如 main.py），把自己构造的 EventEmitter 传给 main()：

from qqbotsdk import EventEmitter, EventQueue, main

ee = EventEmitter(EventQueue())

# 方式一：裸参数（d 为原始事件数据）
@ee.on("GROUP_AT_MESSAGE_CREATE")
async def on_group_message(d):
    print("收到群消息:", d)

# 方式二：类型化 payload（dataclass，见 qqbotsdk.payloads）
from qqbotsdk.payloads import GroupAtMessage

@ee.on("GROUP_AT_MESSAGE_CREATE")
async def on_group_message(msg: GroupAtMessage):
    print(msg.group_openid, msg.user_openid, msg.content)

# 方式三：依赖注入（dishka）
from qqbotsdk import Inject
from qqbotsdk.api import BotApi
from qqbotsdk.events import Event

@ee.on("GROUP_AT_MESSAGE_CREATE")
async def handle(event: Event, api: Inject[BotApi]):
    print(event.user_id, event.group_id, event.content)
    # 被动回复：带 msg_id（群聊 5 分钟/单聊 60 分钟内有效，最多分别回复 5/4 次）；
    # 同一条消息多次回复递增 msg_seq
    await api.post_group_message(
        event.group_id,
        content=f"你说：{event.content}",
        msg_id=event.message_id,
    )   # 失败抛 ApiError；富媒体先 upload_group_file 拿 file_info

# 业务事件（op=0）处理器的返回值不会回流；
# 协议层（op≠0）由 ws_protocol / webhook_protocol 的类处理，经 reply 队列发回。
# 单个 handler 抛异常不会影响其他 handler 和主循环，异常会被记录。

if __name__ == "__main__":
    main(ee)

运行：

uv run python main.py

Webhook 接入：将 .env 中 CONNECTER 改为 webhook，运行后把 http://<你的域名>:8080/qqbot/webhook 配置到 QQ 开放平台回调地址（仅支持 80/443/8080/8443 端口，需公网 HTTPS）。

完整可运行示例见 examples/：群机器人（文本/图片回复、入群欢迎、单聊回复）与 Webhook 最小接入。

开发

uv run pytest    # 单元测试
uv run ruff check src tests examples && uv run pyright

待完善

• 更多 OpenAPI 封装（按需在 api.py 的 BotApi 上加薄封装即可）