Metadata-Version: 2.4
Name: qbotsdk
Version: 0.7.0
Summary: Async SDK for the QQ bot open platform
Keywords: qq,qqbot,bot,sdk,websocket,webhook,asyncio
Author-Email: Half_nothing <Half_nothing@163.com>
Maintainer-Email: Half_nothing <Half_nothing@163.com>
License-Expression: Apache-2.0
License-File: LICENSE
Classifier: Development Status :: 4 - Beta
Classifier: Framework :: AsyncIO
Classifier: Intended Audience :: Developers
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3 :: Only
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Topic :: Internet
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Typing :: Typed
Project-URL: Homepage, https://github.com/half-nothing/qbotsdk
Project-URL: Repository, https://github.com/half-nothing/qbotsdk
Project-URL: Issues, https://github.com/half-nothing/qbotsdk/issues
Project-URL: Changelog, https://github.com/half-nothing/qbotsdk/blob/main/CHANGELOG.md
Requires-Python: >=3.12
Requires-Dist: aiohttp>=3.14.3
Requires-Dist: pydantic>=2.13.4
Requires-Dist: yarl>=1.24.5
Requires-Dist: loguru>=0.7.3
Requires-Dist: cryptography>=50.0.2
Requires-Dist: async-event-bus>=1.0.0
Provides-Extra: redis
Requires-Dist: redis>=8.1.0; extra == "redis"
Provides-Extra: sqlite
Requires-Dist: sqlalchemy[asyncio]>=2.0; extra == "sqlite"
Requires-Dist: aiosqlite>=0.20; extra == "sqlite"
Provides-Extra: mysql
Requires-Dist: sqlalchemy[asyncio]>=2.0; extra == "mysql"
Requires-Dist: asyncmy>=0.2.9; extra == "mysql"
Provides-Extra: postgres
Requires-Dist: sqlalchemy[asyncio]>=2.0; extra == "postgres"
Requires-Dist: asyncpg>=0.29; extra == "postgres"
Description-Content-Type: text/markdown

# qbotsdk

非官方 QQ 机器人 SDK

[![Python](https://img.shields.io/badge/Python-3.12%2B-blue)](https://www.python.org/)
[![License](https://img.shields.io/github/license/half-nothing/qbotsdk)](./LICENSE)
[![PyPI](https://img.shields.io/pypi/v/qbotsdk)](https://pypi.org/project/qbotsdk/)

## 特性

- **全异步**：基于 `asyncio` 与 `aiohttp`，所有接口均为异步接口。
- **登录态自动维护**：`access_token` 自动换取、缓存、到期前刷新。
- **连接与重连**：WebSocket 适配器自动心跳、断线重连与 Resume。
- **接入方式可插拔**：默认走 WebSocket 长连接，也可以换成 Webhook 回调，或自己实现 `Adapter`。
- **消息即对象**：收到的事件可以直接 `await event.reply(...)`。
- **命令与权限**：`Router` 用前缀树匹配命令，支持模板参数、`--` 选项与子路由 `include`；权限键按操作拆分，内置 `/permission` 管理命令，存储可选内存、文件、Redis 或关系库。
- **按钮回调**：`@router.button` 把按钮点击路由到带 pydantic 载荷的处理器，`ack` 策略自动回应客户端，不必自己调 `respond`。
- **容错内置**：429 按 `Retry-After` 重试，幂等请求遇 5xx 与网络异常退避重试，本地令牌桶限流，错误统一成 `QQBotError` 体系。
- **原始报文兜底**：每个模型都带 `raw`，官方新增字段时可以直接取用，不必等 SDK 更新。
- **类型完整**：pydantic v2 模型、全量类型标注、随包分发 `py.typed`。

## 安装

包名与导入名均为 `qbotsdk`，要求 Python 版本 >= 3.12。

```bash
pip install qbotsdk
uv add qbotsdk           # uv 项目里加依赖
uv pip install qbotsdk   # 或直接装进当前环境
```

要从源码安装

```bash
git clone https://github.com/half-nothing/qbotsdk.git
cd qbotsdk

pip install -e .         # 可编辑模式，改完立刻生效
uv pip install -e .      # uv 同理
```

## 快速开始

```python
import asyncio
import os

from qbotsdk import Bot


async def main() -> None:
    # https://q.qq.com/qqbot/dashboard/manage
    # 进入上方网站管理机器人，在开发设置中找到“AppID 接入凭证”卡片
    # 将环境变量QQ_BOT_APP_ID设为获取到的AppID
    # 将环境变量QQ_BOT_APP_SECRET设为获取到的AppSecret
    # 如果您选择直接替换对应内容，请不要将携带敏感信息的文件提交到仓库
    async with Bot(
        app_id=os.environ["QQ_BOT_APP_ID"],
        app_secret=os.environ["QQ_BOT_APP_SECRET"],
    ) as bot:
        me = await bot.api.user.me()
        print(f"登录成功：{me.username}")

        async for guild in bot.api.user.iter_guilds():
            channels = await bot.api.guild.channels(guild.id)
            print(f"{guild.name}：{len(channels)} 个子频道")


asyncio.run(main())
```

## 接收事件

默认走 WebSocket 长连接

```python
import asyncio

from async_event_bus import EventContext

from qbotsdk import Bot
from qbotsdk.model.event import C2CMessageCreate, GroupAtMessageCreate, InteractionCreate


async def main() -> None:
    # 认证信息的具体获取方式请查看“快速开始”章节
    async with Bot(app_id=..., app_secret=...) as bot:

        @bot.on(GroupAtMessageCreate)
        async def on_group_at(ctx: EventContext[GroupAtMessageCreate]) -> None:
            await ctx.event.reply("收到: " + ctx.event.content)

        @bot.on(C2CMessageCreate)
        async def on_c2c(ctx: EventContext[C2CMessageCreate]) -> None:
            await ctx.event.reply("收到: " + ctx.event.content)

        @bot.on(InteractionCreate)
        async def on_interaction(ctx: EventContext[InteractionCreate]) -> None:
            if ctx.event.needs_response:
                await ctx.event.respond()  # 不回应客户端会一直 loading

        await bot.start()
        await asyncio.sleep(3600)


asyncio.run(main())
```

bot的`adapter`参数默认为`WebSocketAdapter`，可以显式指定为`WebhookAdapter`  
也可以自行实现`Adapter`从而扩展更多接入方式，自行实现的`Adapter`需要继承`qbotsdk.adapter.Adapter`并实现两个抽象方法

`WebSocketAdapter`的会话状态默认存在内存里  
如果想关闭后自动续连，可以在构造`WebSocketAdapter`的时候传递参数`session_storage=`  
可选值为`MemorySessionStorage`, `FileSessionStorage`, `RedisSessionStorage`, 默认为`MemorySessionStorage`

```python
from qbotsdk.adapter import WebSocketAdapter
from qbotsdk.storage import FileSessionStorage

# 若不传存储路径，默认路径为软件运行目录下的.session.json文件
WebSocketAdapter(session_storage=FileSessionStorage())

# Redis 在可选依赖里, 使用前需要先安装qbotsdk[redis]
from qbotsdk.storage import RedisSessionStorage

# 给完整地址
WebSocketAdapter(session_storage=RedisSessionStorage.from_url("redis://:pwd@10.0.0.5:6379/2"))

# 或者只给零件, 地址由SDK拼
storage = RedisSessionStorage.from_url(host="10.0.0.5", port=6379, password="pwd", db=2)
WebSocketAdapter(session_storage=storage)

# Unix domain socket, 以及指定RESP协议版本
RedisSessionStorage.from_url(unix_socket="/var/run/redis/redis.sock", db=2, protocol=3)
```

也可以直接把现成的Redis客户端交给SDK

```python
# 特别注意，SDK需要的Redis实例全限定名是redis.asyncio.Redis
from redis.asyncio import Redis

from qbotsdk.storage import RedisSessionStorage

storage = RedisSessionStorage(Redis(host="10.0.0.5", port=6379))
```

### Webhook 接入

在[QQ机器人管理后台](https://q.qq.com/qqbot/dashboard/manage)填好回调地址并勾选要监听的事件后，
事件就改走 HTTP 推送，代码层面仅需把适配器换成 `WebhookAdapter`：

```python
from qbotsdk import Bot
from qbotsdk.adapter import WebhookAdapter

bot = Bot(
    # 认证信息的具体获取方式请查看“快速开始”章节
    app_id=...,
    app_secret=...,
    adapter=WebhookAdapter(
        "同app_secret",
        port=8443,
        path="/qbotsdk",
        certfile="certs/fullchain.pem",
        keyfile="certs/privkey.pem",
    ),
)
await bot.start()
```

没有公网地址时用反向代理或内网穿透把请求转到本机，务必原样透传 body 和两个签名头，任何改写都会导致验签失败。

如果事件要接进自己已有的aiohttp应用，不用起整套服务器，挂一个处理函数就行：

```python
from aiohttp import web

from qbotsdk import Bot
from qbotsdk.gateway import build_webhook_handler

bot = Bot(app_id=..., app_secret=...)
app = web.Application()
app.router.add_post("/qbotsdk", build_webhook_handler(bot.dispatcher, "机器人密钥"))
```

如果要使用flask等其他http框架，仿照该处理函数写一遍处理函数即可。

## 命令与权限

命令层是叠在事件之上的可选模块，它不订阅任何事件，也不挂在 `Bot` 上。
注册时声明每条命令响应哪些事件，再由你在自己的监听器里决定是否交给它。

### 三步接入

```python
from async_event_bus import EventContext

from qbotsdk import Bot
from qbotsdk.commands import GroupContext, Reply, Router
from qbotsdk.model.event import GroupMessageCreate

router = Router()


@router.command("ping", on=GroupMessageCreate)
async def ping(ctx: GroupContext) -> Reply:
    return Reply(text="pong")


bot = Bot(app_id=..., app_secret=...)


@bot.on(GroupMessageCreate)
async def on_group(ctx: EventContext[GroupMessageCreate]) -> None:
    await router.handle(ctx.event)  # 不是命令时返回 NOT_FOUND，什么都不发
```

就这样，群里发 `/ping` 会收到 `pong`。
注册命令必须写 `on=`，它决定这条命令响应哪种事件。

### 带权限的完整例子

```python
from async_event_bus import EventContext

from qbotsdk import Bot
from qbotsdk.commands import GroupContext, Reply, Router, Status
from qbotsdk.model.event import GroupMessageCreate
from qbotsdk.permissions import (
    MemoryPermissionStorage,
    Permission,
    PermissionChecker,
    PermissionGroup,
    install_permission_commands,
)


class Server(PermissionGroup, namespace="server"):
    START = Permission("start", "启动服务器")
    STOP = Permission("stop", "停止服务器")


store = MemoryPermissionStorage()
router = Router(checker=PermissionChecker(store))

# /permission show [target] | grant | revoke | claim | whoami | role
install_permission_commands(router, store)


@router.command("server start", on=GroupMessageCreate, permission=Server.START)
async def start(ctx: GroupContext) -> Reply:
    return Reply(text="started")


@router.command("kick <member:mention>", on=GroupMessageCreate)
async def kick(ctx: GroupContext, member: str) -> Reply:
    return Reply(text=f"kicked {member}")


bot = Bot(app_id=..., app_secret=...)


@bot.on(GroupMessageCreate)
async def on_group(ctx: EventContext[GroupMessageCreate]) -> None:
    result = await router.handle(ctx.event)
    if result.status is Status.HANDLED:
        print(result.command, result.replied)
```

### 处理器返回值

处理器返回值必须是 `Reply`，`Reply(text="...")` 当正文发出去，`Reply.ok()` 表示静默成功，
`Reply(..., markdown=True)` / `Reply(..., keyboard=...)` 带额外能力，参数错误请显式 `raise UsageError(...)`。

参数有问题就 `raise UsageError("...")`，其它异常统一回内部错误文案；
`ExecutionResult.status` 会区分 `HANDLED` / `DENIED` / `FAILED` / `NOT_FOUND`，方便在监听器里做记录或埋点。
handler 里已经 `await ctx.reply(...)` 过的话，路由器不会把返回值再发一遍。

一条命令可以同时挂在多个事件上。

### 参数与模板

模板里的占位符默认按注解转换，`str`、`int`、`float`、`mention`，末尾写 `<text:rest>` 表示吃掉剩余全部 token。
模板不支持默认值，可选参数用注册两条命令表达：

```python
@router.command("roll", on=GroupMessageCreate)
async def roll(ctx: GroupContext) -> Reply:
    return Reply(text=f"掷出了 {random.randint(1, 6)}")


@router.command("roll <sides:int>", on=GroupMessageCreate)
async def roll_sides(ctx: GroupContext, sides: int) -> Reply:
    return Reply(text=f"掷出了 {random.randint(1, sides)}")
```

想让处理器自己拿原始参数，注册时写 `bind_args=False`，此时处理器只收 `ctx` 和 `argv: list[str]`：

```python
@router.command("echo <text:rest>", on=GroupMessageCreate, bind_args=False)
async def echo(ctx: GroupContext, argv: list[str]) -> Reply:
    return Reply(text=" ".join(argv))
```

### 匹配顺序

命令匹配是一棵按事件类型分开的前缀树，字面量边优先，整棵子树走不通才回退到 `<name>` 这类可变参边，最后才是 `<text:rest>`；
token 用完却停在没有处理器的节点上时，框架会按该节点的子项回 usage，
例如 `/permission role` → `Invalid arguments. Usage: /permission role <add|*name>`。

### 选项

`--` 开头的 token 在匹配阶段被收进选项表，`--` 单独出现表示后面不再当选项。注册时声明类型即可自动转换并绑定：

```python
@router.command("roll <sides:int>", on=GroupMessageCreate, options={"loud": bool, "times": int})
async def roll(ctx: GroupContext, sides: int, *, loud: bool = False, times: int = 1) -> Reply: ...
```

`/roll 20 --loud --times=3` 会得到 `sides=20`、`loud=True`、`times=3`。
`<text:rest>` 之后的 token 原样打包，不再解析选项，交给 handler 自己处理。

### 未知命令

带前缀但没命中任何命令时默认静默；需要提示就 `Router(reply_unknown=True)`，
回复文案是 `Messages.unknown_command`，可整体或按键替换。不带前缀的普通聊天不会触发任何回复。

### 权限

权限键自己声明，注册命令时挂上去：

```python
class Demo(PermissionGroup, namespace="demo"):
    ADMIN = Permission("admin", "演示管理")
    PING = Permission("ping", "演示 ping")


@router.command("ban <member:mention>", on=GroupMessageCreate, permission=Demo.ADMIN)
async def ban(ctx: GroupContext, member: str) -> Reply:
    return Reply(text=f"banned {member}")
```

`namespace` 加键名就是完整键，`Demo.ADMIN` 即 `demo.admin`；`Demo.ADMIN.wildcard()` 是 `demo.*`，
只授 `demo.*` 就能一次覆盖 `demo.admin` 与 `demo.ping`。

判定由 `Router(checker=PermissionChecker(store))` 完成，`store` 是下面的任意一种权限存储；
没配 checker 时命令一律被拒并打 warning，不会静默放行。

#### 第一次授权

`/permission grant` 自己也要 `permission.grant`，所以第一次得在启动时给一个人：

```python
import os

from qbotsdk.permissions import Grant, install_permission_commands

owner = os.environ["QQ_BOT_OWNER"]  # 形如 group:{群 openid}:{你的 openid}
await store.grant(owner, Grant("permission.*"))
install_permission_commands(router, store)
```

主体键可以先让机器人回 `/permission whoami` 告诉你，再填进环境变量。

#### 内置命令

| 命令                                          | 作用        | 需要                  |
|---------------------------------------------|-----------|---------------------|
| `/permission whoami`                        | 打印自己的主体键  | 无                   |
| `/permission show`                          | 看自己的规则    | 无                   |
| `/permission show @某人`                      | 看别人的规则    | `permission.show`   |
| `/permission grant @某人 demo.admin`          | 授权（默认场景级） | `permission.grant`  |
| `/permission grant @某人 demo.admin --global` | 授到账号级     | `permission.grant`  |
| `/permission grant demo.admin --code`       | 生成一次性授权码  | `permission.grant`  |
| `/permission claim ABCD-2345`               | 领取授权码     | 无                   |
| `/permission revoke @某人 demo.admin`         | 撤销        | `permission.revoke` |
| `/permission role list`                     | 列出角色      | `permission.role`   |
| `/permission role save vip demo.*`          | 定义角色      | `permission.role`   |
| `/permission role assign @某人 vip`           | 把角色分给人    | `permission.role`   |

`@某人` 依赖事件里的成员标记，
`GroupMessageCreate` 下事件层把 `<@openid>` 渲染成 `@openid`，`mention` 参数直接得到裸 openid；
机器人只订阅 `GROUP_AT_MESSAGE_CREATE` 时拿不到 @ 信息，这时用 `--code` 或直接写 openid。

#### 判定规则

主体分两级：场景键与账号键，先判场景再判账号；候选链从具体往笼统走，同一级拒绝优先，默认拒绝。

### 权限存储

按需选择：`MemoryPermissionStorage`（默认）、`FilePermissionStorage`、`RedisPermissionStorage`、`SqlitePermissionStorage`、
`MysqlPermissionStorage`、`PostgresPermissionStorage`；外层套 `CachedPermissionStorage` 可以少打几次存储。

权限数据要跨重启保留时才需要换存储，本地调试用默认的内存实现即可。

### 按钮回调

按钮与快捷菜单回调用 `router.button(载荷模型)` 注册，handler 里写 `return Reply(text="已处理")`
或 `await ctx.reply("已处理")` 都可以，框架按 `ack` 策略把它变成客户端回应。

## 原始报文

SDK 给每个响应模型和事件模型都挂了一个 `raw`，内容是**解析这条数据时用的那份 dict 本身**

```python
from qbotsdk.model.event import GroupMessageCreate


@bot.on(GroupMessageCreate)
async def on_group(ctx) -> None:
    event = ctx.event
    print(event.content)  # SDK 已建模的字段
    print(event.raw["content"])  # 同一份数据，原文
    print(event.raw.get("some_new_field"))  # 官方新加、SDK 还没建模的字段
```

嵌套模型各自带一份子 dict，`event.author.raw["username"]`等都能直接读。REST 接口同理：

```python
me = await bot.api.user.me()
print(me.raw["username"])
```

注意事项：

1. `raw`字段是dict类型，生成的 `__hash__` 在调用时会抛 `TypeError`，不要把模型放进 `set` 或当作 dict 的键。
2. 内存上每个模型有一份dict引用，嵌套对象与原报文共享同一个dict对象，不是深拷贝。

## 示例

具体示例请见[examples文件夹](examples/README.md)。

## 第三方依赖

| 依赖                                                                 | 用途                      | 许可                         |
|--------------------------------------------------------------------|-------------------------|----------------------------|
| [aiohttp](https://github.com/aio-libs/aiohttp)                     | HTTP 与 WebSocket 客户端    | Apache-2.0 AND MIT         |
| [pydantic](https://github.com/pydantic/pydantic)                   | 数据模型与校验                 | MIT                        |
| [loguru](https://github.com/Delgan/loguru)                         | 日志                      | MIT                        |
| [cryptography](https://github.com/pyca/cryptography)               | Webhook 的 ed25519 签名与验签 | Apache-2.0 OR BSD-3-Clause |
| [async-event-bus](https://github.com/half-nothing/async-event-bus) | 事件总线                    | MIT                        |
| [yarl](https://github.com/aio-libs/yarl)                           | URL 拼接                  | Apache-2.0                 |

## 致谢

- [QQ 机器人官方文档](https://bot.q.qq.com/wiki/develop/api-v2)：字段、错误码与协议细节都以它为准。
- [async-event-bus](https://github.com/half-nothing/async-event-bus)：事件总线。
- [aiohttp](https://github.com/aio-libs/aiohttp)：HTTP 与 WebSocket 客户端。
- [pydantic](https://github.com/pydantic/pydantic)：数据模型与校验。
- [loguru](https://github.com/Delgan/loguru)：日志。
- [cryptography](https://github.com/pyca/cryptography)：ed25519 签名与验签。
- [yarl](https://github.com/aio-libs/yarl)：URL 拼接。
- 腾讯官方 Python SDK 文档与社区实现：心跳、Resume、关闭码语义对照过。
- 所有反馈过的使用者。

## 许可证

Apache-2.0，见 [LICENSE](LICENSE)。
