BotClient 组件介绍
约 1027 字大约 3 分钟
2025-09-26
BotClient:事件回调注册与运行方式
BotClient 是 NcatBot 的对外门面,一个进程只允许出现一个 BotClient 实例,负责:
- 持有 Adapter 与 BotAPI,并把 Adapter 的底层能力包装成易用的 API(
self.api = BotAPI(self.adapter.send)
)。 - 为「官方事件」建立回调分发组,并提供两种注册方式(函数式与装饰器)。
- 自动把官方事件转发到插件系统的 EventBus,让插件按需订阅处理。
本文重点介绍回调的注册方法与使用建议。
回调函数机制
NcatBot 采用回调函数机制来上报事件. 当对应事件发生时, 由 BotClient 调用事件绑定的回调函数, 并将事件相关信息作为参数传递。
NcatBot 的回调函数只有一个参数, 用于传递所发生事件的信息。
官方事件分组
NcatBot 会上报以下事件:
OFFICIAL_GROUP_MESSAGE_EVENT = "ncatbot.group_message_event"
OFFICIAL_PRIVATE_MESSAGE_EVENT = "ncatbot.private_message_event"
OFFICIAL_REQUEST_EVENT = "ncatbot.request_event"
OFFICIAL_NOTICE_EVENT = "ncatbot.notice_event"
OFFICIAL_STARTUP_EVENT = "ncatbot.startup_event"
OFFICIAL_SHUTDOWN_EVENT = "ncatbot.shutdown_event"
OFFICIAL_HEARTBEAT_EVENT = "ncatbot.heartbeat_event"
分为四大类:
- 元事件:
- Bot 启动事件 (STARTUP_EVENT)
- Bot 心跳事件 (HEARTBEAT_EVENT)
- Bot 关闭事件 (SHUTDOWN_EVENT)
- 消息事件:
- 群消息事件 (GROUP_MESSAGE_EVENT)
- 私聊消息事件 (PRIVATE_MESSAGE_EVENT)
- 请求事件 (REQUEST_EVENT)
- 通知事件 (NOTICE_EVENT)
上报事件时传递的数据结构。
BotClient 初始化时会为每个事件创建一个处理器列表,并把 Adapter 的 event_callback[event_name]
指向 BotClient
内部的异步分发函数。
发生事件后,异步分发函数被调用,以确保消息到达后能分发到你注册的处理器。
注册回调的两种方式
BotClient 同时提供「函数式 API」与「装饰器 API」。两种方式等效,可自由选择。
1) 函数式 API
- 群消息:
add_group_message_handler(handler, filter=None)
- 私聊:
add_private_message_handler(handler, filter=None)
- 通知:
add_notice_handler(handler, filter=None)
- 请求:
add_request_handler(handler, filter=Literal['group','friend'])
- 启动:
add_startup_handler(handler)
- 关闭:
add_shutdown_handler(handler)
- 心跳:
add_heartbeat_handler(handler)
要点:
- handler 可以是同步或异步函数,内部会自动判断并在合适的上下文中执行。
- 「消息类事件」支持
filter
参数,用于对event.message
做类型过滤。filter
必须是MessageSegment
的子类;若过滤后没有消息片段,处理器会被跳过。 - 请求事件的
filter
用于筛选「好友请求」或「群请求」。
示例(函数式):
from ncatbot.core.client import BotClient
from ncatbot.core.event.message_segment import Image
from ncatbot.core.event import GroupMessageEvent
bot = BotClient()
def on_group(e: GroupMessageEvent):
# 仅当消息中包含图片片段时才会进入
print("收到一条含图片的群消息")
# 只处理包含图片的消息
bot.add_group_message_handler(on_group, filter=Image)
2) 装饰器 API(推荐)
- 群消息:
@bot.on_group_message(filter=...)
- 私聊:
@bot.on_private_message(filter=...)
- 通知:
@bot.on_notice()
- 请求:
@bot.on_request(filter='group'|'friend')
- 启动:
@bot.on_startup()
- 关闭:
@bot.on_shutdown()
- 心跳:
@bot.on_heartbeat()
示例(装饰器):
from ncatbot.core.client import BotClient
bot = BotClient()
@bot.on_request(filter="group")
def on_group_request(e):
# 仅处理加群请求
pass
兼容别名(3xx 版本):
group_event/private_event/notice_event/request_event/startup_event/shutdown_event/heartbeat_event
- 以及
add_*_event_handler
等旧接口均已兼容映射。
执行模型与注意事项
- BotClient 会把 Adapter 输入的事件分发到内部线程池(
ThreadPool
),再调用你注册的处理器,避免阻塞收包循环。 - 处理器内部可以是同步或异步;若为异步函数会被正确
await
。 - 请尽量避免在处理器内执行长时间阻塞操作;若有耗时 I/O,请自行创建任务或使用异步 API。
- 消息过滤:
filter
需为MessageSegment
子类,否则会抛出TypeError
。
运行 Bot 与插件系统对接
- 前台阻塞运行:
bot.run_frontend(**kwargs)
- 后台非阻塞:
bot.run_backend(**kwargs)
(返回BotAPI
以便在外部直接调用接口)
start()
过程会:
- 校验并应用配置;
- 创建
EventBus
与PluginLoader
并加载插件; - 非 mock 模式下启动 NapCat 与 WebSocket;
- 通过
Adapter
接收事件,触发上述回调与插件总线。
插件访问:
get_registered_plugins() -> List[BasePlugin]
get_plugin(Type[T]) -> T
(按类型检索实例)
小贴士
- 想让插件收到官方事件?无需手动桥接。BotClient 已在内部注册了一个内置处理器:把官方事件封装为
NcatBotEvent
并发布到EventBus
,插件只需订阅相应事件类型即可。
版权所有
版权归属:huan-yp