NcatBot 文档

外观

协议处理

约 1257 字大约 4 分钟

2026-03-19

协议处理 — OB11Protocol 请求-响应匹配、NapCatEventParser 事件解析、NapCatBotAPI 实现

OB11Protocol — 协议编解码

模块: ncatbot.adapter.napcat.connection.protocol

使用 echo + UUID + asyncio.Future 实现 OneBot v11 请求-响应匹配。同时将非响应消息（事件推送）转发给事件回调。

class OB11Protocol:
    def __init__(self, ws: NapCatWebSocket): ...
    def set_event_handler(self, handler: Callable[[dict], Awaitable[None]]) -> None: ...
    async def call(self, action: str, params: Optional[dict] = None, timeout: float = 30.0) -> dict: ...
    def cancel_all(self) -> None: ...
    async def on_message(self, data: dict) -> None: ...

方法详解

方法签名说明
set_event_handlerdef set_event_handler(self, handler: Callable[[dict], Awaitable[None]]) -> None设置事件推送回调
callasync def call(self, action: str, params: Optional[dict] = None, timeout: float = 30.0) -> dict发送 API 请求并等待响应（默认 30 秒超时）
cancel_alldef cancel_all(self) -> None取消所有挂起的请求（断连时调用）
on_messageasync def on_message(self, data: dict) -> None处理收到的消息：区分 API 响应和事件推送

请求-响应匹配流程

匹配机制：

  1. call() 为每个 API 请求生成唯一的 UUID 作为 echo 字段
  2. 同时创建一个 asyncio.Future 与该 echo 关联
  3. 请求发送后，call() 等待 Future 完成（带超时）
  4. on_message() 收到消息时，若包含 echo 字段，则匹配到对应的 Future 并设置结果
  5. 若消息不包含 echo 字段，识别为事件推送，转发给事件回调

错误处理

场景行为
API 请求超时call() 抛出 asyncio.TimeoutError（默认 30 秒）
连接断开cancel_all() 取消所有挂起的 Future
响应 retcode 非零返回完整响应 dict，由上层判断是否为错误

OneBot v11 响应格式

{
    "status": "ok",
    "retcode": 0,
    "data": { ... },
    "echo": "uuid-xxx"
}
字段类型说明
statusstr"ok" 或 "failed"
retcodeint返回码，0 表示成功
dataany响应数据
echostr请求时传入的 echo 标识

NapCatEventParser — 事件解析

模块: ncatbot.adapter.napcat.parser

将 OB11 原始 JSON 转换为 BaseEventData 数据模型。内部使用 EventParser 注册表，基于 (post_type, secondary_key) 二元组路由到对应的数据模型类。

class NapCatEventParser:
    def parse(self, raw_data: dict) -> Optional[BaseEventData]: ...
方法签名说明
parsedef parse(self, raw_data: dict) -> Optional[BaseEventData]解析原始 JSON → 数据模型，解析失败返回 None

EventParser 注册表

通过 EventParser.register(post_type, secondary_key) 装饰器注册事件数据模型：

post_typesecondary_key数据模型
messageprivatePrivateMessageEventData
messagegroupGroupMessageEventData
requestfriendFriendRequestEventData
requestgroupGroupRequestEventData
meta_eventlifecycleLifecycleMetaEventData
meta_eventheartbeatHeartbeatMetaEventData
noticegroup_uploadGroupUploadNoticeEventData
noticegroup_adminGroupAdminNoticeEventData
noticegroup_decreaseGroupDecreaseNoticeEventData
noticegroup_increaseGroupIncreaseNoticeEventData
noticegroup_banGroupBanNoticeEventData
noticefriend_addFriendAddNoticeEventData
noticegroup_recallGroupRecallNoticeEventData
noticefriend_recallFriendRecallNoticeEventData
noticegroup_msg_emoji_likeGroupMsgEmojiLikeNoticeEventData
noticepokePokeNotifyEventData
noticelucky_kingLuckyKingNotifyEventData
noticehonorHonorNotifyEventData

解析路由逻辑

事件解析的路由键由两部分组成：

  1. post_type — 事件大类：message / notice / request / meta_event
  2. secondary_key — 事件子类型，具体字段因 post_type 而异：
    • message → message_type（private / group）
    • notice → notice_type（group_recall / poke 等）
    • request → request_type（friend / group）
    • meta_event → meta_event_type（lifecycle / heartbeat）

未注册的事件类型会被忽略（parse() 返回 None），同时记录警告日志。

NapCatBotAPI — IQQAPIClient 实现

模块: ncatbot.adapter.napcat.api.bot_api

NapCatBotAPI 是 IQQAPIClient 的 NapCat 平台实现，通过 Mixin 模式组合多个功能模块：

class NapCatBotAPI(
    MessageAPIMixin,      # 消息操作
    GroupAPIMixin,        # 群管理
    AccountAPIMixin,      # 账号操作
    QueryAPIMixin,        # 信息查询
    FileAPIMixin,         # 文件操作
    IQQAPIClient,
):
    def __init__(self, protocol: OB11Protocol): ...
    async def _call(self, action: str, params: Optional[dict] = None) -> dict: ...
    async def _call_data(self, action: str, params: Optional[dict] = None) -> Any: ...

Mixin 组合

Mixin模块职责
MessageAPIMixinncatbot.adapter.napcat.api.messagesend_private_msg / send_group_msg / delete_msg / send_forward_msg
GroupAPIMixinncatbot.adapter.napcat.api.group群管理（踢人、禁言、设置管理员等）
AccountAPIMixinncatbot.adapter.napcat.api.account好友/群请求处理
QueryAPIMixinncatbot.adapter.napcat.api.query信息查询（登录信息、好友列表、群成员等）
FileAPIMixinncatbot.adapter.napcat.api.file文件上传与管理

核心内部方法

方法签名说明
_callasync def _call(self, action: str, params: Optional[dict] = None) -> dict发送 API 请求，返回完整响应
_call_dataasync def _call_data(self, action: str, params: Optional[dict] = None) -> Any发送 API 请求，返回 data 字段

两者都委托给 OB11Protocol.call() 实现实际的请求-响应通信。_call_data 是 _call 的便捷封装，直接提取响应中的 data 字段。

预上传机制

NapCatBotAPI 重写了 send_group_msg、send_private_msg、send_forward_msg，在发送前自动对消息中的可下载资源执行预上传（PreUploadService），将 URL 资源转换为本地文件，提升消息发送可靠性。

async def send_group_msg(
    self, group_id: Union[str, int], message: list, **kwargs: Any,
) -> dict:
    message = await self._preupload_message(message)
    return await super().send_group_msg(group_id, message, **kwargs)

附加 API

方法签名说明
send_likeasync def send_like(self, user_id: Union[str, int], times: int = 1) -> None发送点赞
send_pokeasync def send_poke(self, group_id: Union[str, int], user_id: Union[str, int]) -> None发送戳一戳

序列化与反序列化

所有 API 请求和响应均使用 JSON 格式通过 WebSocket 传输：

请求序列化（OB11Protocol.call）：

{
    "action": "send_group_msg",
    "params": {
        "group_id": 12345,
        "message": [{"type": "text", "data": {"text": "hello"}}]
    },
    "echo": "uuid-xxx"
}

事件反序列化（NapCatEventParser.parse）：

{
    "post_type": "message",
    "message_type": "group",
    "group_id": 12345,
    "user_id": 67890,
    "message": [{"type": "text", "data": {"text": "hello"}}],
    "raw_message": "hello",
    "time": 1700000000,
    "self_id": 123456789
}

原始 JSON 经 NapCatEventParser.parse() 解析后转换为对应的 BaseEventData 子类实例（如 GroupMessageEventData），包含类型化的字段访问。

版权归属：MI

许可证：署名 4.0 国际 (CC-BY-4.0)

© 2025-2026 NcatBot