介绍

aiocqhttpOneBot (原 酷QCQHTTP 插件 ) 的 Python SDK,采用异步 I/O,封装了 web 服务器相关的代码,支持 OneBot 的 HTTP 和反向 WebSocket 两种通信方式,让使用 Python 的开发者能方便地开发插件。
本 SDK 要求使用 Python 3.7 或更高版本,以及建议搭配支持 OneBot v11 的 OneBot 实现。

特点

  • 基于 asyncio,异步 I/O 使程序运行效率更高
  • 支持反向 WebSocket 通信方式,允许同时作为多个机器人账号的后端
  • 接口类似 Flask,直观易懂,上手成本低
  • 封装了消息类,便于进行消息处理和构造

开始使用

安装

首先安装 aiocqhttp 包:
pip install aiocqhttp
note:
可能需要将 pip 换成 pip3
默认情况下将安装必要依赖 Quartaiocqhttp。若要安装可选依赖,可使用:
pip install aiocqhttp[all]
这将会额外安装 ujson

最小实例

新建 Python 文件(这里假设名为 bot.py),内容如下:
# bot.py from aiocqhttp import CQHttp, Event bot = CQHttp() @bot.on_message('private') async def _(event: Event): await bot.send(event, '你发了:') return {'reply': event.message} bot.run(host='127.0.0.1', port=8080)
note:
如果需要在不同主机上运行此 bot 后端和 酷Q(CQHTTP),则此处 host 应该为 bot 后端所在主机的 IP 或 0.0.0.0
如果你在使用 VPS,应确保在安全组中开放 8080 端口。
运行该文件,看到如下输出即为启动成功:
Running on <http://127.0.0.1:8080> (CTRL + C to quit) [2020-01-29 19:27:57,133] Running on 127.0.0.1:8080 over http (CTRL + C to quit)
note:
如果没有启动成功,建议检查 Python 版本是否过旧、端口是否被占用等。

配置 CQHTTP

使用反向 WebSocket

在 CQHTTP 配置文件中,填写 ws_reverse_url 值为 ws://127.0.0.1:8080/ws/,这里 127.0.0.1:8080 应根据情况改为 bot.py 中传给 bot.runhostport 参数。
然后,如果有的话,删掉 ws_reverse_event_urlws_reverse_api_url 这两个配置项。
接着设置 use_ws_reversetrue
最后重启 CQHTTP。

使用 HTTP

修改 bot.py 中创建 bot 对象部分的代码为:
bot = CQHttp(api_root='<http://127.0.0.1:5700>')
这里 127.0.0.1:5700 应根据情况改为 CQHTTP 所监听的 IP 和端口(由 CQHTTP 配置中的 hostport 指定)。
然后在 CQHTTP 配置文件中,填写 post_urlhttp://127.0.0.1:8080/,这里 127.0.0.1:8080 应根据情况改为 bot.py 中传给 bot.runhostport 参数。
接着设置 use_httptrue(默认就是 true)。
最后重启 CQHTTP。

测试对话

给你的机器人发一段私聊消息,如果一切正常,ta 应该会回复你。
note:
如果没有回复,请检查 bot.py 运行是否报错、酷Q 日志是否报错。如果都没有报错,则可能是机器人账号被腾讯风控,需要在同一环境中多登录一段时间。

发生了什么

本节围绕 开始使用 中的最小实例,来解释它如何工作。
先贴出代码:
# bot.py from aiocqhttp import CQHttp, Event bot = CQHttp() # M1 @bot.on_message('private') # M2 async def _(event: Event): # M3 await bot.send(event, '你发了:') # M4 return {'reply': event.message} # M5 bot.run(host='127.0.0.1', port=8080) # M6

CQHttp

# bot.py bot = CQHttp() # M1
M1 处首先创建了 aiocqhttp.CQHttp 类的对象 bot
该类是本 SDK 的主体,内部封装了 Quart 对象作为 web 服务器。其中添加了 //ws/ 等路由,从而使 CQHTTP 能够通过 HTTP 或 WebSocket 协议连接 bot
使用 HTTP 通信时,需要传入 api_root 参数是因为 bot.send 需要主动调用 CQHTTP API,它需要知道 CQHTTP「在哪」。
使用反向 WebSocket 通信时,可以有多个 CQHTTP 同时连接到一个 botbot.send 会自动选择对应的账号发送。

事件处理

# bot.py @bot.on_message('private') # M2
M2 处通过 @bot.on_message('private') 装饰器注册了 私聊消息事件 的处理函数。
除了 bot.on_message,还有类似的 bot.on_noticebot.on_requestbot.on_meta_event,它们分别用于注册消息、通知、请求、元事件这四种事件类型(对应 CQHTTP 事件post_type 字段)的处理函数。
这些装饰器可以带参数,也可以不带参数,参数可以有多个,对应 CQHTTP 事件的 ?_type 字段,这里 ?_type 根据事件类型的不同,分别为 message_typenotice_typerequest_typemeta_event_type
async def _(event: Event): # M3
M3 处定义了事件处理函数,它必须接受一个 Event 对象作为唯一的参数。Event 对象是对 CQHTTP 事件数据的简单封装,提供了属性以方便获取其中的字段,例如 event.messageevent.user_id 等。

API 调用

await bot.send(event, '你发了:') # M4
M4 处调用了 bot.send 方法,该方法是对 CQHTTP APIsend_msg 的简单封装,它会向 event 对应的主体发送消息(由第二个参数指定),本例中这个主体是「发私聊消息来的人」。
除此之外,你可以在 bot 对象上直接调用任何 CQHTTP API,见 API 列表 ,所需参数通过命名参数传递,例如:
friends = await bot.get_friend_list() await bot.set_group_ban(group_id=10010, user_id=10001000) credentials = await bot.get_credentials(domain='qun.qq.com')
note:
如果有多个 CQHTTP 连接,可能需要在调用 API 时增加 self_id 参数以指定要调用的机器人账号,例如:
await bot.get_friend_list(self_id=event.self_id)
调用 API 时需向 CQHTTP 发出请求,这一步可能出错:
  • 如果 API 当前不可用(例如没有任何连接了的 CQHTTP、或未配置 api_root),抛出 aiocqhttp.ApiNotAvailable
  • 如果 API 可用,但网络无法连接或连接出现错误,抛出 aiocqhttp.NetworkError
一旦请求成功,SDK 会判断 HTTP 响应状态码是否为 2xx:
  • 如果不是,抛出 aiocqhttp.HttpFailed,在这个异常中可通过 status_code 获取 HTTP 响应状态码
  • 如果是,则进一步查看响应 JSON 的 status 字段,如果 status 字段为 failed,抛出 aiocqhttp.ActionFailed,在这个异常中可通过 retcode 获取 API 调用的返回码
以上异常全都继承自 aiocqhttp.Error
如果 statusokasync,则不抛出异常,函数返回 CQHTTP API 响应数据的 data 字段(有可能为 None)。
HTTP 响应状态码和 retcode 的具体含义,见 响应说明

快速操作

return {'reply': event.message} # M5
M5 处事件处理函数返回了一个字典,这会被 SDK 序列化为 JSON 并返回给 CQHTTP,作为 CQHTTP 事件上报的响应 (通过 HTTP 响应正文或 WebSocket 传送)。这称为「快速操作」,可用于对事件进行一些简单的操作,本例中对事件进行了「回复」操作,对于群聊等事件,快速操作还包括「禁言」「撤回」等,具体请见 事件列表 的「响应数据」。
快速操作不是必须的,事件处理函数可以不返回任何值。

运行

bot.run(host='127.0.0.1', port=8080) # M6
M6 处调用 bot.run 运行了 bot 后端,该方法是 Quart 对象的 run 方法的简单封装,可直接传入更多参数,参数会直接进入 Quart.run

更多

更丰富的例子见 demo.py
本节所提到的类、方法、函数等,均可在右上角「模块 API」中找到更详细的说明。

常见主题

消息处理

为了方便地处理消息,本 SDK 提供了 MessageMessageSegment 类,用于解析和构造消息中的 CQ 码,例如:
from aiocqhttp import Message, MessageSegment @bot.on_message async def handle_msg(event): msg = Message(event.message) for seg in msg: if seg == MessageSegment.at(event.self_id): await bot.send(event, 'at 我干啥') break img = MessageSegment.image('<http://example.com/somepic.png>') await bot.send(event, img + '\\n上面这是一张图')
如果觉得手动从 event.message 构造 Message 对象不够方便,可以在 bot 对象初始化时传入 message_class 参数,例如:
from aiocqhttp import CQHttp, Message bot = CQHttp(message_class=Message)
这会使 SDK 在收到消息事件后,使用形如 event.message = Message(event.message) 的方式构造 Message 对象。
当然,如果内置的 Message 类不符合你的需求,你也可以自己编写消息类,同样可以传入 message_class

默认 bot 对象

如果你只是开发一些简单的功能,或临时做测试等,可以不用自己创建 bot 对象,而直接使用 SDK 内置的默认 bot 对象,例如:
from aiocqhttp.default import on_message, send, api, run @on_message async def handle_msg(event): await send(event, event.message) await api.send_private_msg(user_id=event.user_id, message='。。。') run(host='127.0.0.1', port=8080)
如需修改 CQHttp 初始化的参数,可使用 reconfigure_default_bot 函数,例如:
from aiocqhttp.default import reconfigure_default_bot reconfigure_default_bot(api_root='<http://127.0.0.1:8080>')

同步和异步

通常情况下,建议在 bot 中全部使用异步操作,例如使用 aiofiles 读写文件、使用 aiohttphttpx.AsyncClient 等进行网络请求、使用 aiomysqlasyncpgaioredisMotor 等访问数据库。
但如果出于某些原因,你更偏好或不得不使用同步函数,SDK 也提供了原生支持,例如:
@bot.on_message def sync_handle_msg(event): time.sleep(5) # 模拟耗时 I/O # 使用 bot.sync 进行 API 调用 bot.sync.send_private_msg(user_id=event.user_id, message='处理完毕')
sync_handle_msg 会在 asyncio loop 的默认 executor(多线程,需注意线程安全)里运行,可通过 loop.set_default_executor 修改。

日志

本 SDK 直接使用了内部 Quart 对象的日志器,是一个 logging.Logger 对象,可通过 bot.logger 获得,例如:
bot.logger.info('初始化成功')
如果你需要对 logger 进行配置,直接修改它即可(但不可给它赋值),但更建议的方式是在你的项目中自己创建新的 logger。

部署

默认情况下,bot.run 以 debug 模式运行,可通过 bot.run(..., debug=False) 关闭。
在实际部署环境中,不建议使用 bot.run,而应该使用专业的 ASGI 服务器,例如 UvicornDaphneHypercorn 等。这里给出使用 Uvicorn 部署的例子:
# main.py from aiocqhttp import CQHttp bot = CQHttp()
uvicorn --host 127.0.0.1 --port 8080 main:bot.asgi

添加路由

SDK 注册了 //ws/ws/api/ws/event 这几个路由,以便向 CQHTTP 提供服务。有时你可能想要注册其它自定义的路由,例如接收 webhook 推送、展示机器人状态、提供管理控制台等,可以直接操作 Quart 对象来做到,例如:
@bot.server_app.route('/webhook') async def webhook(): pass
使用 Quart 时,有时需要用到当前模块的相对路径,例如加载当前模块下的 templates 中的模板文件等,如果使用这些功能时遇到问题,可以尝试给 CQHttp 类传入 import_name 参数,例如:
bot = CQHttp(__name__) @bot.server_app.route('/admin') async def admin(): return await render_template('admin.html')

在已有事件循环中运行

通过 CQHttp.run_task 方法可以将 bot 运行在已有的事件循环中,参数同 CQHttp.run,例如:
bot = CQHttp() loop = asyncio.new_event_loop() loop.create_task(bot.run_task(host='127.0.0.1', port=8080)) loop.run_forever()
badge