安装
pip install moemoe[uvicorn]挂进已有的 ASGI 应用就不装 [uvicorn],把 Moemoe.asgi 交给你的服务器。
需要 Python 3.11+。
三分钟上手
from moemoe import Moemoe
app = Moemoe(access_token='') # 和 robot 配的一致,留空表示不校验
@app.command('ping')
async def _(event, args):
return 'pong' # 返回非 None 会当消息发出去
app.run(host='127.0.0.1', port=8000)robot 那侧把 backend.url 指到 ws://127.0.0.1:8000/robot 就通了。
多机器人
几十个 robot 连进来是常态。Event.bot 指回收到这条消息的那个,所以「回哪条、从哪个机器人回」永远是确定的:
@app.on_message()
async def _(event):
await event.send(f'我是 {event.bot.name}')
# 主动推送要指名机器人
await app.bot('4013755093').send(group_openid, '早安')
# 只连了一个时可以省略
await app.bot().send(group_openid, '早安')app.bots 是 {bot_id: Bot},app.bot(id) 按机器人账号查找。想广播就自己遍历。
event.bot.id # 机器人账号,对应 bot_id
event.bot.app_id # 开放平台应用 ID,对应 app_id
event.bot.openid # 机器人在群内的 openid
事件帧和 ready 帧都会更新 bot.app_id,消息分发前即可读取;
刚连接、还没收到这两种帧时为空字符串。0.0.5 需配套更新 robot,使事件信封和
ready.data 都携带 app_id,缺失时报告连接错误。
三个发消息的方法
QQ 有两套互不相干的机制:msg_id 决定这条免不免费,message_reference
决定界面上出不出引用气泡。
await event.send('收到') # 回应这次事件。免费,无气泡
await event.reply('收到') # 同上,加引用气泡
await event.reply('看这条', target='REFIDX_x') # 引用别的消息| 方法 | msg_id | message_reference | 额度 |
|---|---|---|---|
event.send | ✓ | — | 免费 |
event.reply | ✓ | ✓ | 免费 |
bot.send | — | — | 烧主动消息额度,走审核 |
Event 只管回应手上这条,发给「某个目标」是 Bot 的事。
两边都叫 send,对象本身说明了是哪一种——Event 手上有那条消息所以免费,
Bot 手上什么都没有所以只能主动发:
await event.bot.send(event.group_openid, '这条不蹭凭据')
await app.bot().send(group_openid, '早安') # 跟事件无关的推送也是同一个方法处理函数跑太久、send 的凭据过期了,就是上面第一种写法。
平台的规矩(moemoe 不代管,超了会拿到 QQ 的错误码):
- 凭据群聊 5 分钟、单聊 60 分钟内有效
- 同一条
msg_id最多回 5 次(单聊 4 次) - 引用的值是引用索引
REFIDX_xxx,不是message_id,两者不通用 - QQ 不是每条消息都给引用索引,拿不到时
reply的效果等同send
免费还是烧额度,robot 自己定
await app.bot().send(gid, '早安')
不用你判断。每条消息都在信封上写 passive: "auto",robot 按手上的凭据三步退让:
- 你给了
msg_id(event.send就是这种)→ 还能用就用它 - 那条过期了、五次回满了 → 换这个会话最新那条
- 一条能用的都没有 → 摘掉凭据走主动消息,要额度也要过审
第三步是有意的:攥着一张必定被拒的凭据发过去只会换回一个错误码,消息就没了。所以
哪怕你用的是 event.send,robot 也可能把它改成主动消息发出去——回执里的
passive.used 说的是真话,想知道烧没烧额度看它。
msg_seq 也归 robot 盖,你不用数。那串号跟着 msg_id 走,推送和回复取的是同一个池子,
两边各数各的会撞出重复消息。
先问清楚再决定:
out = await app.bot().call('GET', '/robot/passive', {'scene': gid})
if out['left']:
...命令
@app.command('weather', alias=('天气', 'w'), patterns=r'今天.*天气')
async def _(event, args):
"""查天气""" # docstring 进 command.doc
city = args.get(0, '北京')
return f'{city}今天晴'
Moemoe(prefixes=('/', '')) 表示带不带斜杠都认,nicknames=('萌萌',)
让「萌萌,天气」也能触发。
正则触发时命名组直接从 args.match 取:
@app.command('开关', patterns=r'^(?P<act>启用|停用)(?P<what>\w+)$')
async def _(event, args):
return f'{args.match["act"]}了{args.match["what"]}'按名字触发的命令 args.match 是 None。
args 认引号:/say "a b" c 拆成 ['a b', 'c']。
args.rest() 拿原文,args.int(0) 按整数取。
权限
权限谓词可以用 & | ~ 组合:
from moemoe import GROUP_ADMIN, SUPERUSER
@app.command('kick', permission=GROUP_ADMIN | SUPERUSER, denied='你不行')
async def _(event, args): ...SUPERUSER 读的是 Moemoe(superusers=[...])。
to_me:群里默认要求 @ 机器人
命令默认 to_me=True,群里必须 @ 机器人才触发。订了全量群消息时,别人聊天里打个
/ping 不会让机器人抢答。整个应用改掉用 Moemoe(to_me=False),
单条命令写 @app.command('ping', to_me=False),命令自己写的说了算。
关掉它之前要先订上 GROUP_MESSAGE_CREATE——没订的话根本收不到没 @ 的消息,改了也没用。
to_me=True 就是严格要求 @,昵称不算。 两者的关系取决于你在 robot 那侧订了哪些事件:
| robot 订的事件 | to_me=True(默认) | to_me=False |
|---|---|---|
只订 GROUP_AT_MESSAGE_CREATE |
每条都是 @ 来的,命令照常触发,昵称也能用 | 一样 |
订了 GROUP_MESSAGE_CREATE |
必须 @,昵称触发不了 | 昵称、裸命令都能触发 |
也就是说,订了全量群消息又想用昵称触发的话,那几条命令要写 to_me=False:
@app.command('ping', to_me=False) # 「萌萌,ping」和裸的「/ping」都能触发
async def _(event, args):
return 'pong'单聊一律算冲机器人来的,不受影响。
多轮问答
invoke(event, name, text) 与普通命令共用权限和 to_me 检查。spawn(handler, *args, bot=bot) 返回可等待或取消的 Task。
处理函数可以停下来等后续消息,状态留在局部变量里,不用维护一张「谁在玩」的表。
from moemoe import SessionTimeout
@app.command('改名')
async def _(event, args):
try:
answer = await event.prompt('叫什么?', timeout=30)
except SessionTimeout:
return '等太久了'
return f'好的,以后叫你{answer.text}'群级抢答用 capture,它既是异步迭代器也是异步上下文管理器:
@app.command('猜数')
async def _(event, args):
answer = random.randint(1, 10)
async for guess in event.capture(scope='scene', timeout=20,
match=lambda one: one.text.isdigit()):
if int(guess.text) == answer:
return f'{guess.username} 答对了'
await guess.send('不对,再猜')
return f'没人猜出来,答案是 {answer}'要点:
scope='user'(默认)只等这个人,scope='scene'等这个群/单聊里的任何人timeout是每条消息的等待上限,收到一条就重新计时async for在超时或关闭时正常结束,不用自己接异常;prompt超时抛SessionTimeoutmatch挡掉不相干的消息,被挡的照常走命令和事件处理函数- 被会话接住的消息不再走命令,否则一句「42」会既当答案又当命令
- 会话按机器人连接、群聊/私聊场景和发送者隔离,重连后的消息不会进入旧连接的会话
- 挂着的会话数有上限,
Moemoe(sessions=1000),超了抛TooManySessions
消息段
from moemoe import At, File, Image, Keyboard, Markdown, Text, Voice
await event.send(Text('看图') + Image('https://...'))
await event.send(Voice('https://example.com/voice.mp3'))
await event.send(File('https://example.com/song.mp3', '歌曲.mp3'))
await event.send(Markdown('## 标题\n正文 **加粗**') + Keyboard(rows=[...]))
await event.send(At('U9') + ' 在吗')段帮你兜住 QQ 那三条互斥规矩:markdown 和纯文本互斥(文本段会并进 markdown 正文)、 一条消息只挂一个富媒体(图片、语音和文件逐条发送)、按钮挂在 markdown 上(只给按钮时正文包一层)。
Voice(url) 支持群聊和私聊,地址需能被 QQ 访问。框架上传后发送,不下载或转码。
File(url, name) 按文件类型上传,显示为文件卡片,name 包含扩展名。
图片、语音和文件保持原有顺序;第一段媒体是图片时,普通文字附在该图片上,否则文字单独先发。
Markdown 和按钮仍单独发送。
Keyboard(rows=...) 原样透传给 QQ,字段照官方消息按钮文档填。
不想用糖就直接给 dict 当请求体:
await event.send({'msg_type': 2, 'markdown': {'custom_template_id': 'tpl'}})直接调官方接口
moemoe 不重新建模 QQ 的载荷。路径和字段照腾讯的文档写:
await event.call('GET', f'/v2/groups/{gid}/members')
await event.bot.call('POST', f'/v2/groups/{gid}/batch_remove_members',
{'member_openids': ['xxx']})事件也一样,类型化访问只是取值方便,event.raw 是 QQ 的原文:
event.group_openid # 好写
event.raw # QQ 给什么这里就有什么4xx / 5xx 抛 CallError,里面带着 QQ 的原始错误体和 trace:
from moemoe import CallError
try:
await event.call('GET', f'/v2/groups/{gid}/members')
except CallError as error:
error.status # HTTP 状态码,robot 自己出错时是 0
error.body # QQ 的响应体,err_code 在里面
error.trace # 报障要附的链路号钩子
@app.on_startup
async def _():
... # 在事件循环里跑
@app.on_shutdown
async def _():
... # 关服前收尾,会话会先被关掉
@app.on_bot_connect
async def _(bot):
logger.info('{} 上线', bot) # 钩子里可以调接口
@app.on_bot_disconnect
async def _(bot):
...
@app.on_error
async def _(failure):
failure.error # 抛出来的异常
failure.event # 出错时手上那条事件,没有就是 None
failure.handler # 哪个处理函数启动钩子失败会中止启动并执行关闭流程;关闭钩子和事件处理函数隔离异常并交给 on_error。应用与机器人连接退出时会取消并等待各自持有的后台任务。
HTTP 路由
跟 WebSocket 那条路共用端口和进程:
from starlette.responses import JSONResponse
@app.route('/healthz')
async def _(request):
return JSONResponse({'bots': len(app.bots)})不用装额外的东西,starlette 本来就是核心依赖。
定时任务
可选模块,先装:
pip install "moemoe[scheduler]"from moemoe.scheduler import Scheduler
scheduler = Scheduler(app) # 自动挂 startup / shutdown
@scheduler.scheduled_job('cron', hour=7, minute=15, id='daily')
async def _():
await app.bot().send(group_openid, '早上好')
调度器不是 moemoe 写的——cron 解析、时区、夏令时、错过补发,APScheduler 已经写对十几年了。
这个模块只解决接线:什么时候 start()、什么时候 shutdown()、没装时报一句人话。
认不出的属性转给底层的 AsyncIOScheduler,add_job / get_jobs /
pause 照 APScheduler 的文档写。
只支持 APScheduler 3.x。
插件
应用对象单独放一个模块,插件从那儿 import:
myrobot/
app.py app = Moemoe(...)
main.py from app import app; load_plugins('plugins'); app.run()
plugins/
__init__.py
ping.py from app import app → @app.command('ping')from moemoe import load_plugins
load_plugins('plugins')日志
from moemoe import logger 拿到的就是 loguru 的全局 logger 原样再导出。
logger.disable('moemoe') 让库彻底闭嘴。
跑测试
不需要装任何东西:
python -m unittest discover -s tests -t .装了 pytest 的话 pytest 也直接认。
0.0.7 更新
最低支持 Python 3.11。
- 新增
File(url, name)文件消息,支持群聊、私聊上传与发送,保留文件名。 - 新增
Voice(url)语音消息;文件、语音、图片和文字组合时按原顺序发送。 Bot.upload()支持file_type和file_name;上传失败直接抛出异常。- 修复 robot 断开或重连时重复关闭 WebSocket 引发的异常。
拼布蓝图发送需要本版新增的 File,线上安装完成后需重启机器人业务服务。
0.0.6 更新
最低支持 Python 3.11。
- 普通分发与
invoke()统一检查权限和to_me,权限函数异常按原样抛出。 - 命令注册冲突不再留下残余名称,参数支持制表符和换行分隔。
- 启动失败会中止启动并清理资源;关闭、断线与重连时回收所属任务和会话。
- 调用超时覆盖发送与等待,断开的机器人拒绝新调用和新任务。
- 多段 Markdown 按组合顺序保留全部正文,重复按钮和未知消息段明确报错。
spawn()返回 Task;补齐参数说明,整理排版、示例和验包资源释放。
升级后,群消息触发的内部命令若不要求 @,需设置 to_me=False,或沿用应用默认设置。
Keyboard 的 rows 与 id 只提供一个。
状态
0.0.7,早期版本,API 还会动。 本版通过 Python 3.11 单元测试、构建验包和 seasun 插件加载验证。 历史版本已实连验证 QQ 收消息到机器人回复的完整链路;本轮未重新实连 QQ。
没做的:命令中间件。
铁律见 RULES.md, 设计理由见 docs/design.md。