qqbot-sdk 0.1.0

现代化、人性化的腾讯 QQ 机器人 Python SDK (API v2)

qqbot-sdk

现代化、生产级的腾讯 QQ 机器人 Python 框架,基于官方 OpenAPI v2,面向 QQ 群机器人优化。

设计上博采众长:botpy(官方 SDK,长期未维护)的接口对照、NoneBot2 的规则/优先级/插件理念、mirai 的事件与会话模型,并原生支持 OneBot v12 协议输出。

亮点一览

维度	能力
接口覆盖	官方 OpenAPI v2 全量:四场景消息、富媒体、频道/成员/身份组、禁言、公告、精华、表态、权限、日程、论坛、音频、API 权限申请
连接模式	WebSocket(心跳/重连/Resume/多分片)+ Webhook(Ed25519 验签)可同时多路运行,自定义连接器可扩展
生态兼容	内置 OneBot v12 适配器(正向 WS + HTTP),NoneBot2 / Koishi 可直连
事件系统	洋葱模型中间件、声明式规则(&/|/~ 组合)、优先级 + 阻断、on_any 旁路监听
会话交互	msg.ask() 一行实现多轮对话、bot.wait_for() 任意事件等待
插件系统	目录自动发现、运行时加载/卸载/热重载、独立存储命名空间、on_load/on_unload 钩子
指令系统	类型化参数(注解自动转换)、冷却、权限检查器、用法提示、自动 /帮助
数据管理	统一异步 KV 抽象:内存 / JSON / SQLite 三后端,会话状态 bot.sessions 按群+用户隔离
定时任务	@bot.schedule.every(3600) 间隔任务 + 五段式 cron
可观测性	/healthz /stats /metrics(Prometheus)、事件/API/延迟指标、trace_id 透出
稳定性	token 自动刷新、401 重试、出站令牌桶限流、网络错误退避重试、事件异常隔离、优雅停机(等在途任务)

安装

pip install aiohttp cryptography
pip install -e .          # 本仓库根目录

Python ≥ 3.9。

三分钟上手

import qqbot

bot = qqbot.Bot("你的AppID", "你的AppSecret")

@bot.on_message                  # QQ单聊 + 群@ + 频道@ 一网打尽
async def echo(msg: qqbot.Message):
    await msg.reply(f"你说了: {msg.text}")

bot.run()                        # WebSocket 模式,本地零配置可跑

msg.reply() 统一四种场景(单聊/群/频道/频道私信),自动携带 msg_id 走被动通道、自动递增 msg_seq;reply(image=URL或路径或bytes) 时群/单聊自动先走富媒体上传。

核心概念

连接模式(可多路并存)

bot.add_websocket()                                   # 开发:零配置
bot.add_webhook(port=8443, max_timestamp_skew=300)    # 生产:官方推荐,自动验签+防重放
bot.add_connector(OneBotV12Adapter(port=6700))        # 生态:OneBot v12 输出
bot.run()

Webhook 的回调地址验证(op=13)由框架自动应答;WebSocket 的 intents 根据你注册的处理器自动推导。

规则 / 优先级 / 阻断(NoneBot2 风格)

from qqbot import rules

@bot.on_group_message(rule=rules.keyword("天气") & rules.scene("group"))
async def weather(msg): ...

@bot.on_message(rule=rules.regex(r"^签到$"), priority=1, block=True)
async def sign(msg): ...          # 数字越小越先执行;block 阻断后续处理器

内置规则:keyword / startswith / endswith / fullmatch / regex / scene / from_user / from_group / has_image,支持 & | ~ 组合,自定义任何 (data) -> bool 皆可。

中间件(洋葱模型)

@bot.middleware
async def audit(ctx: qqbot.Context, call_next):
    if ctx.message and is_spam(ctx.message.content):
        return                    # 不调 call_next => 拦截
    await call_next()

指令系统

@bot.command("骰子", usage="/骰子 [次数] [面数]", cooldown=5)
async def roll(msg, times: int = 1, sides: int = 6):    # 注解自动转换类型
    ...

from qqbot import permissions as perm
@bot.command("重启", permission=perm.owner_only)         # 权限检查器
async def restart(msg): ...

bot.enable_help()                                        # 自动生成 /帮助

参数解析规则:无额外参数→只传 msg;单个无默认值 str→整段文本;多参数→按空格分词并按注解转换(int/float/bool);keyword-only str→吃掉剩余整段;解析失败自动回复用法。

会话式交互

@bot.command("点餐")
async def order(msg):
    food = await msg.ask("想吃什么?")                    # 等同一用户下一条消息
    count = await food.ask(f"{food.text},要几份?")
    await count.reply("已下单 ✅")

插件

# plugins/greet.py
from qqbot import Plugin
plugin = Plugin("greet", version="1.0")

@plugin.command("hello")
async def hello(msg): await msg.reply("你好!")

# main.py
bot.load_plugins("plugins")          # 自动发现
await bot.reload_plugin("greet")     # 热重载,改代码不重启

状态与数据

bot = qqbot.Bot(appid, secret, storage=qqbot.SQLiteStore("bot.db"))

await bot.storage.set("global_key", {"any": "json"})
store = bot.storage.namespace("my_plugin")               # 插件隔离

async with bot.sessions.open(msg) as sess:               # 群+用户 维度会话状态
    sess["count"] = sess.get("count", 0) + 1             # 退出 with 自动保存

定时任务

@bot.schedule.every(3600)            # 每小时
async def hourly(): ...

@bot.schedule.cron("0 8 * * *")      # 每天 8:00(分 时 日 月 周;周: 0=周一)
async def morning(): ...

OneBot v12 桥接

from qqbot.ext.onebot import OneBotV12Adapter
bot.add_websocket()
bot.add_connector(OneBotV12Adapter(port=6700, access_token="token"))
# NoneBot2 等以 ws://127.0.0.1:6700/onebot/v12 正向连接

选择 v12 而非 v11:v12 的 ID 为字符串,与官方 openid 无损兼容。已实现 send_message / delete_message / get_self_info / get_status / get_version / get_supported_actions,消息段 text / image / mention / reply,群管理事件以 qq.* 扩展 notice 下发。

可观测性 & 生产化

bot = qqbot.Bot(appid, secret,
                rate_limit=20,        # 出站令牌桶限流(req/s)
                api_retries=2)        # 网络错误退避重试
bot.enable_observability(port=9100)   # /healthz /stats /metrics(Prometheus)

@bot.on_startup
async def up(b): ...
@bot.on_shutdown
async def down(b): ...                # SIGTERM/Ctrl+C 触发优雅停机

多账号:qqbot.run_bots(bot1, bot2)。

示例索引

文件	内容
examples/01_echo_websocket.py	最小复读机
examples/02_echo_webhook.py	Webhook 生产部署
examples/03_commands.py	指令系统基础
examples/04_rich_content.py	图片/按钮/Markdown/Ark/按钮回调
examples/05_api_only.py	纯 API 主动推送
examples/06_middleware_rules.py	中间件 + 规则 + 优先级
examples/07_conversation.py	多轮对话 + 会话状态
examples/08_plugins/	插件化项目结构 + 热重载
examples/09_scheduler_production.py	定时任务 + 生产级配置
examples/10_onebot_bridge.py	OneBot v12 桥接 NoneBot2

测试

pip install pytest pytest-asyncio
pytest tests            # 49 项:签名/构建器/模型/分发/规则/中间件/会话/存储/调度/插件/OneBot

项目结构

qqbot/
├── bot.py            # 总装:事件/中间件/指令/插件/会话/连接器/生命周期
├── api.py            # OpenAPI 全量封装
├── connection.py     # 连接器抽象(WebSocket / Webhook / 自定义)
├── gateway.py        # WebSocket 网关(心跳/重连/Resume/分片)
├── webhook.py        # Webhook 服务器(Ed25519 验签 + 防重放)
├── middleware.py     # 洋葱模型中间件
├── rules.py          # 声明式规则组合子
├── commands.py       # 增强指令(类型化参数/冷却/权限)
├── permissions.py    # 权限检查器
├── plugin.py         # 插件系统(热重载)
├── storage.py        # KV 存储抽象 + 会话管理(内存/JSON/SQLite)
├── scheduler.py      # 定时任务(interval + cron)
├── observability.py  # 指标 + 健康检查/Prometheus
├── limiter.py        # 令牌桶限流 / 冷却
├── models.py         # 统一消息模型 + 事件模型
├── event.py          # 事件定义、intents 映射、payload 解析
├── builders.py       # Markdown/Keyboard/Ark/Embed 构建器
├── http.py           # 鉴权 HTTP(401 重试/限流/重试/指标)
├── token.py          # access_token 自动管理
└── ext/
    └── onebot.py     # OneBot v12 适配器

已知边界

• 主动消息受平台配额限制(每目标每月 4 条),被动回复不受限,优先 reply() / event.reply()
• 私域专属能力(频道全量消息、论坛事件)公域机器人订阅会鉴权失败,框架已在自动推导时规避
• OneBot 适配器覆盖常用动作;未覆盖的平台能力可在同进程内用原生 API 补齐

License

MIT