后端即时通讯【免费下载链接】nonebot2跨平台 Python 异步聊天机器人框架 / Asynchronous multi-platform chatbot framework written in Python项目地址https://gitcode.com/gh_mirrors/no/nonebot2点击查看免费下载适配器Adapter是 NoneBot2 中机器人与平台交互的核心桥梁。本文以官方适配器开发文档为骨架结合当前仓库nonebot2的源码实现系统讲解如何从零编写一个可运行的适配器项目项目组织结构、脚手架创建、Adapter/Bot/Event/Message四大组成部分的实现要点、客户端与服务端两种通信方式、事件转换与平台 API 调用、适配器测试与商店发布流程。读完本文你将掌握把任意聊天平台接入 NoneBot2 的完整开发套路。一、适配器的功能与组成在编写适配器之前需要先理解适配器在 NoneBot 中的职责。根据 使用适配器 文档的说明适配器通常具备两种功能接收事件将驱动器Driver收到的事件数据转换为 NoneBot 定义的事件模型然后交由机器人插件处理调用平台接口将机器人插件调用平台接口的数据转换为平台指定的格式交由驱动器发送并接收接口返回数据。为了实现这两种功能适配器通常由四个部分组成组成部分职责Adapter负责转换事件和调用接口正确创建 Bot 对象并注册到 NoneBot 中Bot负责存储平台机器人相关信息并提供回复事件、调用 API 的上层方法Event负责定义事件内容以及事件主体对象Message负责正确序列化消息以便机器人插件处理在 NoneBot 中这四部分的基类统一从nonebot.adapters模块导出见 nonebot/adapters/init.pyfrom nonebot.internal.adapter import Adapter as Adapter from nonebot.internal.adapter import Bot as Bot from nonebot.internal.adapter import Event as Event from nonebot.internal.adapter import Message as Message from nonebot.internal.adapter import MessageSegment as MessageSegment from nonebot.internal.adapter import MessageTemplate as MessageTemplate编写适配器时我们需要继承这些基类并根据实际平台逐一实现每个部分的功能。二、项目组织结构与脚手架创建NoneBot 适配器项目通常以nonebot-adapter-{adapter-name}作为项目名并以命名空间包的形式编写即在nonebot/adapters/{adapter-name}目录中编写实际代码。推荐的项目结构如下 nonebot-adapter-{adapter-name} ├── nonebot │ ├── adapters │ │ ├── {adapter-name} │ │ │ ├── __init__.py │ │ │ ├── adapter.py │ │ │ ├── bot.py │ │ │ ├── config.py │ │ │ ├── event.py │ │ │ └── message.py ├── pyproject.toml └── README.md:::tip 提示 上述项目结构仅作推荐不做强制要求保证实际可用性即可。命名空间包的形式使适配器代码可以无缝并入 NoneBot 的nonebot.adapters包命名空间插件开发者可以直接通过from nonebot.adapters.{adapter-name} import ...导入。 :::使用 NB-CLI 创建项目可以使用 NoneBot 官方脚手架快速创建带基本结构的适配器项目nb adapter create按照指引输入适配器名称以及存储位置即可得到一个包含基本结构的适配器项目。三、Log为适配器定制日志前缀适配器在处理时通常需要打印日志但直接使用 NoneBot 的默认logger不方便区分适配器输出和其他日志。NoneBot 在 nonebot/utils.py 中提供了logger_wrapper方法其实现如下def logger_wrapper(logger_name: str): def log(level: str, message: str, exception: Exception | None None): logger.opt(colorsTrue, exceptionexception).log( level, fm{escape_tag(logger_name)}/m | {message} ) return log因此适配器项目可以自定义一个log函数用于快捷打印带适配器名称前缀的日志建议放在独立的log.py中from nonebot.utils import logger_wrapper log logger_wrapper(your_adapter_name)这个log函数接收三个参数日志等级、日志内容以及可选的异常具体用法如下from .log import log log(DEBUG, A DEBUG log.) log(INFO, A INFO log.) try: ... except Exception as e: log(ERROR, something error., e)从源码可以看到logger_wrapper使用logger.opt(colorsTrue, exceptionexception)开启了 loguru 的彩色标签支持因此日志内容中可以安全使用y.../y、r.../r等 loguru 颜色标签适配器名称会以m洋红色显示并通过escape_tag转义避免适配器名称中的特殊字符破坏颜色标签解析。四、Config读取适配器专属配置通常适配器需要一些配置项例如平台连接密钥等。适配器的配置方法与 插件配置 类似只需定义一个继承自pydantic.BaseModel的配置模型from pydantic import BaseModel class Config(BaseModel): xxx_id: str xxx_token: str配置项的读取在Adapter中通过get_plugin_config完成。从 nonebot/plugin/init.py 的源码看get_plugin_config会从全局配置对象中提取当前适配器或插件需要的配置项并兼容环境变量文件env_file、嵌套分隔符env_nested_delimiter等全局配置参数def get_plugin_config(config: type[C]) - C: 从全局配置获取当前插件需要的配置项。 global_config get_driver().config return type_validate_python( config, BaseSettings._settings_build_values( config, model_dump(global_config), env_fileglobal_config._env_file, env_file_encodingglobal_config._env_file_encoding, env_nested_delimiterglobal_config._env_nested_delimiter, ), )也就是说适配器配置既可以写在.env文件中也可以写在pyproject.toml的[tool.nonebot]插件配置段中NoneBot 会自动完成解析与校验。五、Adapter与平台通信的核心枢纽Adapter负责转换事件、调用接口以及正确创建 Bot 对象并注册到 NoneBot 中。在编写平台相关内容之前需要继承基类并实现适配器的基本信息。基类定义见 nonebot/internal/adapter/adapter.py其__init__保存了driver实例并初始化了一个以机器人 ID 为键的bots字典class Adapter(abc.ABC): def __init__(self, driver: Driver, **kwargs: Any): self.driver: Driver driver self.bots: dict[str, Bot] {}适配器基本骨架如下from typing import Any from typing_extensions import override from nonebot.drivers import Driver from nonebot import get_plugin_config from nonebot.adapters import Adapter as BaseAdapter from .config import Config class Adapter(BaseAdapter): override def __init__(self, driver: Driver, **kwargs: Any): super().__init__(driver, **kwargs) # 读取适配器所需的配置项 self.adapter_config: Config get_plugin_config(Config) classmethod override def get_name(cls) - str: 适配器名称 return your_adapter_name关键点说明get_name是基类声明的抽象类方法abc.abstractmethod返回的字符串将作为该适配器的唯一名称用于注册与获取也与Bot.type属性一一对应见 nonebot/internal/adapter/bot.py 中type属性的实现return self.adapter.get_name()Driver实例由 NoneBot 在注册适配器时自动传入适配器后续的所有网络通信都经由该driver完成。5.1 与平台交互客户端通信方式NoneBot 提供了多种 Driver 帮助适配器进行网络通信主要分为客户端Forward和服务端Reverse两种类型。需要根据平台文档和特性选择合适的通信方式。相关混入基类定义在 nonebot/internal/driver/abstract.pyWebSocketClientMixin继承自ForwardMixin提供websocket(setup: Request)异步上下文管理器用于发起 WebSocket 客户端连接HTTPClientMixin提供request/stream_request/get_session等方法用于发起 HTTP 客户端请求ASGIMixin继承自ReverseMixin提供setup_http_server/setup_websocket_server方法用于注册服务端路由。以下以 WebSocket 客户端为例展示完整的适配器生命周期管理import asyncio from typing_extensions import override from nonebot import get_plugin_config from nonebot.exception import WebSocketClosed from nonebot.drivers import Request, WebSocketClientMixin class Adapter(BaseAdapter): override def __init__(self, driver: Driver, **kwargs: Any): super().__init__(driver, **kwargs) self.adapter_config: Config get_plugin_config(Config) self.task: Optional[asyncio.Task] None # 存储 ws 任务 self.setup() def setup(self) - None: if not isinstance(self.driver, WebSocketClientMixin): # 判断用户配置的Driver类型是否符合适配器要求不符合时应抛出异常 raise RuntimeError( fCurrent driver {self.config.driver} doesnt support websocket client connections! f{self.get_name()} Adapter need a WebSocket Client Driver to work. ) # 在 NoneBot 启动和关闭时进行相关操作 self.driver.on_startup(self.startup) self.driver.on_shutdown(self.shutdown) async def startup(self) - None: 定义启动时的操作例如和平台建立连接 self.task asyncio.create_task(self._forward_ws()) # 建立 ws 连接 async def _forward_ws(self): request Request( methodGET, urlyour_platform_websocket_url, headers{token: ...}, # 鉴权请求头 ) while True: try: async with self.websocket(request) as ws: try: # 处理 websocket ... except WebSocketClosed as e: log( ERROR, rbg #f8bbd0WebSocket Closed/bg #f8bbd0/r, e, ) except Exception as e: log( ERROR, rbg #f8bbd0Error while process data from websocket platform_websocket_url. Trying to reconnect.../bg #f8bbd0/r, e, ) finally: # 这里要断开 Bot 连接 except Exception as e: # 尝试重连 log( ERROR, rbg #f8bbd0Error while setup websocket to platform_websocket_url. Trying to reconnect.../bg #f8bbd0/r, e, ) await asyncio.sleep(3) # 重连间隔 async def shutdown(self) - None: 定义关闭时的操作例如停止任务、断开连接 # 断开 ws 连接 if self.task is not None and not self.task.done(): self.task.cancel()要点解析self.websocket(request)是Adapter基类提供的内置方法内部会校验 driver 是否为WebSocketClientMixin并转发到driver.websocket(setup)见 nonebot/internal/adapter/adapter.py 的websocket方法async with退出时连接自动关闭Request对象是 NoneBot 统一的网络请求模型定义于 nonebot/internal/driver/model.py支持method、url、params、headers、cookies、content、data、json、files、timeout、proxy等丰富的参数WebSocketClosed是 NoneBot 在 nonebot/exception.py 中提供的异常用于区分连接被正常关闭与处理过程中发生错误从而决定是否触发重连逻辑driver.on_startup/driver.on_shutdown用于注册生命周期回调底层由 nonebot/internal/driver/_lifespan.py 的Lifespan管理。5.2 与平台交互服务端通信方式如果平台以主动推送Webhook / 反向 WebSocket的方式下发事件则需要适配器作为服务端。此时需要 driver 支持 ASGI如 FastAPI / Quart并通过HTTPServerSetup与WebSocketServerSetup注册路由from nonebot import get_plugin_config from nonebot.drivers import ( Request, ASGIMixin, WebSocket, HTTPServerSetup, WebSocketServerSetup, ) class Adapter(BaseAdapter): override def __init__(self, driver: Driver, **kwargs: Any): super().__init__(driver, **kwargs) self.adapter_config: Config get_plugin_config(Config) self.setup() def setup(self) - None: if not isinstance(self.driver, ASGIMixin): raise RuntimeError( fCurrent driver {self.config.driver} doesnt support asgi server! f{self.get_name()} Adapter need a asgi server driver to work. ) # 建立服务端路由 # HTTP Webhook 路由 http_setup HTTPServerSetup( URL(your_webhook_url), # 路由地址 POST, # 接收的方法 WEBHOOK name, # 路由名称 self._handle_http, # 处理函数 ) self.setup_http_server(http_setup) # 反向 Websocket 路由 ws_setup WebSocketServerSetup( URL(your_websocket_url), # 路由地址 WebSocket name, # 路由名称 self._handle_ws, # 处理函数 ) self.setup_websocket_server(ws_setup) async def _handle_http(self, request: Request) - Response: HTTP 路由处理函数只有一个类型为 Request 的参数且返回值类型为 Response ... return Response( status_code200, # 状态码 headers{something: something}, # 响应头 contentxxx, # 响应内容 ) async def _handle_ws(self, websocket: WebSocket) - Any: WebSocket 路由处理函数只有一个类型为 WebSocket 的参数 ...要点解析setup_http_server/setup_websocket_server同样是Adapter基类提供的内置方法见 nonebot/internal/adapter/adapter.py内部会校验 driver 是否实现了ASGIMixin再转发到driver.setup_http_server/driver.setup_websocket_serverHTTP 处理函数的签名是固定的接收一个Request参数返回一个ResponseResponse模型同样定义于 nonebot/internal/driver/model.py支持status_code、headers、content与可选的request字段WebSocket 处理函数接收WebSocket对象可以通过await ws.receive_text()/await ws.receive_bytes()/await ws.receive()接收消息通过ws.send_text(...)/ws.send_bytes(...)/ws.send(...)发送消息WebSocket.send会自动根据str或bytes类型分发。:::tip 通信方式选择参考 从 NoneBot 社区既有实现来看不同平台适配器选择的通信方式各有侧重有的平台同时支持 WebSocket 客户端、WebSocket 服务端、HTTP Webhook、HTTP POST 四种方式有的仅支持 WebSocket 服务端与 HTTP Webhook有的仅支持 HTTP Webhook。编写时请以目标平台的官方文档为准选择平台支持且特性最匹配的通信方式。 :::5.3 建立 Bot 连接在与平台建立连接后需要将Bot实例化并调用适配器基类提供的bot_connect方法告知 NoneBot 建立了 Bot 连接在与平台断开连接或出现某些异常进行重连时调用bot_disconnect方法告知 NoneBot 断开了 Bot 连接from .bot import Bot class Adapter(BaseAdapter): def _handle_connect(self): bot_id ... # 通过配置或者平台 API 等方式获取到 Bot 的 ID bot Bot(self, self_idbot_id) # 实例化 Bot self.bot_connect(bot) # 建立 Bot 连接 def _handle_disconnect(self): self.bot_disconnect(bot) # 断开 Bot 连接bot_connect/bot_disconnect的底层行为见 nonebot/internal/adapter/adapter.pydef bot_connect(self, bot: Bot) - None: self.driver._bot_connect(bot) self.bots[bot.self_id] bot def bot_disconnect(self, bot: Bot) - None: if self.bots.pop(bot.self_id, None) is None: raise RuntimeError(f{bot} not found in adapter {self.get_name()}) self.driver._bot_disconnect(bot)它们会完成两件事一是把 Bot 注册进当前适配器的bots字典与驱动器的全局 Bot 注册表driver._bots二是触发driver.on_bot_connect/driver.on_bot_disconnect注册的钩子函数见 nonebot/internal/driver/abstract.py。正因如此插件开发者才能通过nonebot.get_bots()/nonebot.get_bot(id)获取到已连接的 Bot 对象。5.4 转换 Event 事件在接收到来自平台的事件数据后需要将其转换为适配器的Event并调用Bot.handle_event方法让 NoneBot 对事件进行处理。NoneBot 提供了type_validate_python见 nonebot/compat.pypydantic v2 下等价于TypeAdapter(type_).validate_python(data)来校验并实例化 pydantic 模型import asyncio from typing import Any, Dict from nonebot.compat import type_validate_python from .bot import Bot from .event import Event from .log import log class Adapter(BaseAdapter): classmethod def payload_to_event(cls, payload: Dict[str, Any]) - Event: 根据平台事件的特性转换平台 payload 为具体 Event Event 模型继承自 pydantic.BaseModel具体请参考 pydantic 文档 # 做一层异常处理以应对平台事件数据的变更 try: return type_validate_python(your_event_class, payload) except Exception as e: # 无法正常解析为具体 Event 时给出日志提示 log( WARNING, fParse event error: {str(payload)}, ) # 也可以尝试转为基础 Event 进行处理 return type_validate_python(Event, payload) async def _forward(self, bot: Bot): payload: Dict[str, Any] # 接收到的事件数据 event self.payload_to_event(payload) # 让 bot 对事件进行处理 asyncio.create_task(bot.handle_event(event))建议为payload_to_event增加异常兜底平台事件数据结构可能随时变更直接解析失败时给出 WARNING 日志并降级尝试转换为基类Event避免单个异常事件导致整个连接崩溃。5.5 调用平台 API需要实现Adapter的_call_api抽象方法基类中声明为abc.abstractmethod使开发者能够调用平台提供的 API。如果通过 WebSocket 通信可以使用send方法发送数据如果采用 HTTP 请求则需要构造Request对象并调用driver的request方法发送请求from typing import Any from typing_extensions import override from nonebot.drivers import Request, WebSocket from .bot import Bot class Adapter(BaseAdapter): override async def _call_api(self, bot: Bot, api: str, **data: Any) - Any: log(DEBUG, fCalling API y{api}/y) # 给予日志提示 platform_data your_handle_data_method(data) # 自行将数据转为平台所需要的格式 # 采用 HTTP 请求的方式需要构造一个 Request 对象 request Request( methodGET, # 请求方法 urlapi, # 接口地址 headers..., # 请求头通常需要包含鉴权信息 paramsplatform_data, # 自行处理数据的传输形式 # jsonplatform_data, # dataplatform_data, ) # 发送请求返回结果 return await self.driver.request(request) # 采用 WebSocket 通信的方式可以直接调用 send 方法发送数据 # 通过某种方式获取到 bot 对应的 websocket 对象 ws: WebSocket your_get_websocket_method(bot.self_id) await ws.send_text(platform_data) # 发送 str 类型的数据 await ws.send_bytes(platform_data) # 发送 bytes 类型的数据 await ws.send(platform_data) # 是以上两种方式的合体 # 接收并返回结果同样的也有 str 和 bytes 的区别 return await ws.receive_text() return await ws.receive_bytes() return await ws.receive()要点解析_call_api的调用链插件中await bot.call_api(send_msg, message...)甚至await bot.send_msg(...)Bot.__getattr__会将任意属性访问包装为partial(self.call_api, name)最终都会进入Bot.call_api再调用self.adapter._call_api(self, api, **data)见 nonebot/internal/adapter/bot.py。因此_call_api是实现平台 API 调用的唯一落点Bot.call_api还内置了on_calling_api/on_called_api钩子体系支持在调用前后进行拦截、Mock、记录等操作适配器实现_call_api后即可自动获得这些能力同一个 WebSocket 连接上往往有多个 Bot多账号通常需要按bot.self_id维护Bot ID → WebSocket 对象的映射以便在_call_api中把请求发往正确的连接。六、Bot机器人开发者直接使用的核心对象Bot是机器人开发者能够直接获取并使用的核心对象负责存储平台机器人相关信息并提供回复事件、调用 API 的上层方法。需要继承基类Bot并实现相关方法from typing import TYPE_CHECKING, Any, Union from typing_extensions import override from nonebot.message import handle_event from nonebot.adapters import Bot as BaseBot from .event import Event from .message import Message, MessageSegment if TYPE_CHECKING: from .adapter import Adapter class Bot(BaseBot): your_adapter_name 协议 Bot 适配。 override def __init__(self, adapter: Adapter, self_id: str, **kwargs: Any): super().__init__(adapter, self_id) self.adapter: Adapter adapter # 一些有关 Bot 的信息也可以在此定义和存储 async def handle_event(self, event: Event): # 根据需要对事件进行某些预处理例如 # 检查事件是否和机器人有关操作去除事件消息首尾的 bot # 检查事件是否有回复消息调用平台 API 获取原始消息的消息内容 ... # 调用 handle_event 让 NoneBot 对事件进行处理 await handle_event(self, event) override async def send( self, event: Event, message: Union[str, Message, MessageSegment], **kwargs: Any, ) - Any: # 根据平台实现 Bot 回复事件的方法 # 将消息处理为平台所需的格式后调用发送消息接口进行发送例如 data message_to_platform_data(message) await self.send_message( datadata, ... )要点解析handle_event是适配器侧Adapter._forward与 NoneBot 事件分发流程之间的入口。nonebot.message.handle_event的完整实现位于 nonebot/message.py它会依次执行事件预处理、按优先级分发事件给所有事件响应器Matcher、执行运行前后处理器等完整流程。适配器的handle_event中可以按平台特性做预处理如剥离消息首尾的bot、回填被回复消息的原文处理完毕后再调用await handle_event(self, event)交给 NoneBotsend是基类声明的抽象方法见 nonebot/internal/adapter/bot.py 中abc.abstractmethod async def send(...)实现时通常需要将str/Message/MessageSegment统一转为平台要求的消息格式再调用平台发送消息接口。七、Event事件主体对象Event是 NoneBot 中的事件主体对象所有平台消息在进入处理流程前需要转换为 NoneBot 事件。基类Event同时继承abc.ABC与pydantic.BaseModel见 nonebot/internal/adapter/event.py且配置了extraallow因此适配器定义的事件模型可以包含平台特有的额外字段。需要继承基类并实现以下抽象方法from typing_extensions import override from nonebot.compat import model_dump from nonebot.adapters import Event as BaseEvent class Event(BaseEvent): override def get_event_name(self) - str: # 返回事件的名称用于日志打印 return event name override def get_event_description(self) - str: # 返回事件的描述用于日志打印请注意转义 loguru tag return escape_tag(repr(model_dump(self))) override def get_message(self): # 获取事件消息的方法根据事件具体实现如果事件非消息类型事件则抛出异常 raise ValueError(Event has no message!) override def get_user_id(self) - str: # 获取用户 ID 的方法根据事件具体实现如果事件没有用户 ID则抛出异常 raise ValueError(Event has no context!) override def get_session_id(self) - str: # 获取事件会话 ID 的方法根据事件具体实现如果事件没有相关 ID则抛出异常 raise ValueError(Event has no context!) override def is_tome(self) - bool: # 判断事件是否和机器人有关 return False要点解析get_event_description中使用escape_tag(repr(model_dump(self)))其中model_dump是 nonebot/compat.py 提供的跨 pydantic v1/v2 的模型转字典工具escape_tag见 nonebot/utils.py用于转义 loguru 的颜色标签字符防止事件内容破坏日志格式基类还提供了get_log_string()与get_plaintext()的默认实现分别基于事件名称/描述拼接与get_message().extract_plain_text()通常无需重写。然后根据平台消息的类型编写各种不同的事件并且注意要根据事件类型实现get_type方法。根据 使用适配器 文档NoneBot 约定的事件类型通常为meta_event元事件、message消息事件、notice通知事件、request请求事件。消息类型事件还应重写get_message和get_user_id等方法例如from .message import Message class HeartbeatEvent(Event): 心跳时间通常为元事件 override def get_type(self) - str: return meta_event class MessageEvent(Event): 消息事件 message_id: str user_id: str override def get_type(self) - str: return message override def get_message(self) - Message: # 返回事件消息对应的 NoneBot Message 对象 return self.message override def get_user_id(self) - str: return self.user_id class JoinRoomEvent(Event): 加入房间事件通常为通知事件 user_id: str room_id: str override def get_type(self) - str: return notice class ApplyAddFriendEvent(Event): 申请添加好友事件通常为请求事件 user_id: str override def get_type(self) - str: return request从这段示例可以看到事件模型的常见组织方式元事件meta_event如心跳、连接状态变化通常不包含消息消息事件message包含message_id、user_id等字段重写get_message/get_user_id通知事件notice如成员加入房间、群变动等通常重写get_user_id/get_session_id提供事件主体信息请求事件request如好友申请通常需要插件调用平台 API 作出同意/拒绝等响应。get_type的返回值会被 NoneBot 的事件分发机制用于匹配对应的事件类型依赖注入如Event参数的类型判断与event.get_type()过滤务必与 NoneBot 约定的四种类型保持一致。八、Message消息的正确序列化Message负责正确序列化消息以便机器人插件处理。需要同时继承MessageSegment和Message两个类并实现相关方法。基类定义见 nonebot/internal/adapter/message.pyfrom typing import Type, Iterable from typing_extensions import override from nonebot.utils import escape_tag from nonebot.adapters import Message as BaseMessage from nonebot.adapters import MessageSegment as BaseMessageSegment class MessageSegment(BaseMessageSegment[Message]): classmethod override def get_message_class(cls) - Type[Message]: # 返回适配器的 Message 类型本身 return Message override def __str__(self) - str: # 返回该消息段的纯文本表现形式通常在日志中展示 return text of MessageSegment override def is_text(self) - bool: # 判断该消息段是否为纯文本 return self.type text class Message(BaseMessage[MessageSegment]): classmethod override def get_segment_class(cls) - Type[MessageSegment]: # 返回适配器的 MessageSegment 类型本身 return MessageSegment staticmethod override def _construct(msg: str) - Iterable[MessageSegment]: # 实现从字符串中构造消息数组如无字符串嵌入格式可直接返回文本类型 MessageSegment ...要点解析MessageSegment基类是一个包含type: str与data: dict[str, Any]两个字段的 dataclass见 nonebot/internal/adapter/message.pytype表示消息段类型如text、image、at等data存放该消息段的参数。基类同时提供了get/keys/values/items/join/copy等便捷方法Message基类继承自list是消息段的序列容器自带template消息模板、extract_plain_text提取纯文本、include/exclude按类型过滤、get/count/only/index、/拼接、__getitem__支持按类型索引等完整能力。适配器只需实现get_segment_class与_construct两个抽象方法即可获得全套功能_construct负责实现从字符串构造消息数组的解析逻辑如果平台消息没有特殊的字符串嵌入格式如[CQ:image,filexxx]这类 CQ 码可以直接返回一个纯文本类型的MessageSegment。然后根据平台具体的消息类型实现各种MessageSegment消息段。通常的做法是在MessageSegment中定义classmethod工厂方法如MessageSegment.text(hello)、MessageSegment.image(url...)、MessageSegment.at(user_id...)每个工厂方法构造一个对应type的实例在__str__中把消息段渲染为平台约定的人类可读文本形式如xxx、[图片]该字符串用于日志展示与命令匹配在Message._construct中实现反向解析把平台下发的字符串消息体拆解为多个消息段。此外Message基类的template方法基于 nonebot/internal/adapter/template.py 的MessageTemplate用法与str.format大致相同也默认可用支持以消息对象作为模板并输出消息对象插件开发者可以据此实现富文本消息模板。九、适配器测试适配器测试相关内容在官方文档中不再展开开发者可以根据需要进行合适的测试。这里提供几个常见问题的解决方法测试中无法导入 editable 模式安装的适配器代码。在 pytest 的conftest.py内添加如下代码将适配器源码目录加入nonebot.adapters命名空间的搜索路径from pathlib import Path import nonebot.adapters nonebot.adapters.__path__.append( # type: ignore str((Path(__file__).parent.parent / nonebot / adapters).resolve()) )这一做法的原理是命名空间包namespace package的__path__是一个可变列表向其追加源码目录后Python 的包导入机制即可在该目录下继续查找nonebot.adapters.{adapter-name}子模块从而绕过 editable 安装的路径问题。需要计算适配器测试覆盖率。在pyproject.toml中添加 pytest 配置[tool.pytest.ini_options] addopts --cov nonebot/adapters/{adapter-name} --cov-report term-missing该配置需要项目已安装pytest-cov依赖运行 pytest 时会输出适配器代码的覆盖率统计并列出未覆盖的行号term-missing。十、后续工作发布适配器在完成适配器代码的编写后如果想要将适配器发布到 NoneBot 商店需要将适配器项目发布到PyPI确保pyproject.toml中的包名、版本号、依赖声明完整准确且nonebot/adapters/{adapter-name}目录被打包在内前往 商店适配器页面切换到适配器页签点击发布适配器按钮填写适配器相关信息并提交审核。另外建议为适配器编写文档或者提供一些插件开发示例以便其他开发者能够快速上手使用。一个易于接入的适配器通常还应在 README 中说明支持的 NoneBot 版本与驱动类型Forward / Reverse、环境变量配置项、注册方式示例、支持的事件类型清单等。小结编写一个 NoneBot2 适配器本质上是围绕Adapter、Bot、Event、Message四个基类把平台协议翻译成 NoneBot 的统一抽象Adapter负责网络通信、事件转换、Bot 注册与平台 API 调用_call_api是适配器的引擎Bot是插件开发者直接操作的对象提供handle_event与send入口Event将平台数据映射为 NoneBot 事件模型并通过get_type/get_user_id/get_message等抽象方法提供通用信息Message与MessageSegment负责消息的序列化与解析。结合本文给出的源码级原理bot_connect与驱动器的 Bot 注册表、Bot.call_api的钩子体系、type_validate_python的 pydantic 校验、logger_wrapper的日志前缀机制等你可以对照 nonebot/internal/adapter/ 目录下的基类实现逐项落实最终将一个全新的平台接入 NoneBot2 生态。赞分享后端即时通讯【免费下载链接】nonebot2跨平台 Python 异步聊天机器人框架 / Asynchronous multi-platform chatbot framework written in Python项目地址https://gitcode.com/gh_mirrors/no/nonebot2点击查看免费下载相关推荐NoneBot2 适配器开发实战从零编写 Adapter 对接任意聊天平台NoneBot2 适配器开发实战从零编写 Adapter 对接任意聊天平台 本篇技术指南以 NoneBot2 框架的适配器开发文档为核心完整讲解如何编写一个后端即时通讯Plano 概览AI 原生代理服务器与数据平面架构详解OverviewPlano 概览AI 原生代理服务器与数据平面架构详解Overview Plano 是面向 agentic 应用智能体应用的“交付基础设施”一个 A后端即时通讯Triton 双引擎时代CUDA Tile IR 后端与最优软件流水线/Warp Specialization 研究速览Triton 双引擎时代CUDA Tile IR 后端与最优软件流水线/Warp Specialization 研究速览 本文基于 Triton 项目社区例会后端即时通讯创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考