编写一个适配器
接一个新平台只需要写三个方法。 其余部分——连接 MaiBot、断线重连、认证、消息格式——都由官方 maim_message 库和一份百来行的骨架代码包办。这一页带你从零把它跑起来。
前置条件：
- Python 3.10+，能
pip install maim_message； - 一个能收发消息的平台 SDK（平台官方库、OneBot 实现或你自己的协议栈）；
- 一个已经跑起来的 MaiBot，并且知道它的消息服务器地址与令牌。
先确认你要接哪套服务
本页用经典消息服务器（默认开启，ws://127.0.0.1:8000/ws）。如果你要接的是 API 服务器（enable_api_server），连接参数换成 x-apikey / x-platform，报文要在外层多包一层信封——差别见消息协议参考。
骨架的分工
PlatformBridge 只有三个方法：发文本、发图片、收事件。骨架负责把事件拼成 MaiBot 的统一格式、把回复拆成分段、把平台真实消息 ID 回执给 MaiBot。
动手
装依赖。
bashpip install maim_messageMaiBot 侧对版本有下限要求：低于
0.6.2会直接拒绝启动消息服务，当前配套版本是0.6.8。建目录。
my-adapter/ ├── adapter.py # 骨架：连接、路由、回执 ├── bridge.py # 你的平台收发实现 └── requirements.txt实现平台侧的三个方法。只关心平台 SDK 的调用，不要在这里碰 MaiBot 的格式。
pyclass PlatformBridge: """平台侧收发封装。接一个新平台，只需要实现这三个方法。""" async def send_text(self, target: str, text: str) -> str: """给 target 发一条文本，返回平台侧的消息 ID。""" raise NotImplementedError async def send_image(self, target: str, image_base64: str) -> str: """给 target 发一张图片（base64），返回平台侧的消息 ID。""" raise NotImplementedError async def poll_events(self) -> list[dict[str, Any]]: """阻塞收取平台事件，返回归一化后的事件列表。 每个事件的字段约定见 build_inbound()。 """ raise NotImplementedErrorsend_text— 发送后返回平台侧消息 ID；拿不到就返回空串，回执会自动跳过。send_image— 入站图片是 base64，多数平台需要先落盘或上传再发。poll_events— 平台事件收成字典列表，字段约定见下一步。
拼入站报文。这是全篇最容易踩坑的地方：MaiBot 对必填字段是断言而不是友好校验，缺一个这条消息就没了。
pyclass Adapter: def __init__(self, bridge: PlatformBridge) -> None: self.bridge = bridge self.client = MessageClient() self.client.register_message_handler(self.on_outbound) self.client.register_custom_message_handler("message_id_echo", self.on_echo) # ---- 入站：平台 → MaiBot --------------------------------------------- def build_inbound(self, event: dict[str, Any]) -> dict[str, Any]: """把平台事件拼成 MaiBot 的统一消息格式。""" is_group = event.get("group_id") is not None message_info: dict[str, Any] = { "platform": PLATFORM, "message_id": str(event["message_id"]), "time": float(event.get("time") or time.time()), "user_info": { "platform": PLATFORM, "user_id": str(event["user_id"]), "user_nickname": str(event.get("user_name") or event["user_id"]), }, # 这三个字段让出站回复能找到回来的路 "additional_config": { "platform_io_account_id": SELF_ID, "platform_io_target_user_id": str(event["user_id"]), }, } if is_group: message_info["group_info"] = { "platform": PLATFORM, "group_id": str(event["group_id"]), "group_name": str(event.get("group_name") or event["group_id"]), } return { "message_info": message_info, "message_segment": {"type": "text", "data": str(event["text"])}, } async def poll_loop(self) -> None: while True: try: events = await self.bridge.poll_events() except Exception: logger.exception("收取平台事件失败，2 秒后重试") await asyncio.sleep(2) continue for event in events: await self.client.send_message(self.build_inbound(event))三条硬性要求：
user_info.user_id与user_info.user_nickname必须是非空字符串；群消息还要group_info.group_id与group_info.group_name；message_id、time必须存在，time是 float 秒；- 文本分段的
data必须是字符串，数字 ID 要先str()。
顺手多填两个字段，出站回复才找得到路：
additional_config.platform_io_account_id— 本适配器实例的账号标识；同平台多账号时靠它区分。additional_config.platform_io_target_user_id— 私聊回复的收件人。
处理出站分段。MaiBot 推过来的正文多半是
seglist，按type逐个翻译；不认识的分段类型记一条日志跳过，不要让整个适配器崩掉。py# ---- 出站：MaiBot → 平台 --------------------------------------------- async def on_outbound(self, message: dict[str, Any]) -> None: info = message["message_info"] segment = message["message_segment"] target = self.resolve_target(info) message_id = str(info["message_id"]) for seg in self.iter_segments(segment): actual_id = "" if seg["type"] == "text": actual_id = await self.bridge.send_text(target, seg["data"]) elif seg["type"] == "image": actual_id = await self.bridge.send_image(target, seg["data"]) elif seg["type"] == "at": actual_id = await self.bridge.send_text(target, f"@{seg['data']} ") else: logger.warning("暂不支持的分段类型：%s", seg["type"]) continue if actual_id: # 把平台真实消息 ID 回填给 MaiBot await self.client.send_custom_message( "message_id_echo", {"type": "echo", "echo": message_id, "actual_id": actual_id}, ) @staticmethod def iter_segments(segment: dict[str, Any]): """出站正文通常是 seglist，这里统一成可迭代的分段。""" if segment["type"] == "seglist": yield from segment["data"] else: yield segment @staticmethod def resolve_target(info: dict[str, Any]) -> str: """群消息发到群，私聊发到人。""" group_info = info.get("group_info") if group_info: return str(group_info["group_id"]) return str(info["additional_config"]["platform_io_target_user_id"])回传真实消息 ID。发送成功后把平台返回的 ID 回执给 MaiBot，之后它才知道"这条回复在平台上是哪条"，撤回、引用、回复链才挂得上。
pythonawait client.send_custom_message( "message_id_echo", {"type": "echo", "echo": mmc_message_id, "actual_id": platform_message_id}, )这里的
client是maim_message.MessageClient：send_custom_message(message_type_name, message)只有 2 个参数，platform由连接身份决定，不用传。3 参数的send_custom_message(platform, message_type_name, message)是服务端MessageServer的方法，适配器用不到。经典服务的类型名就是message_id_echo；API 服务器要写成custom_message_id_echo（API 客户端会自动补custom_前缀）。启动并连起来。
pythonasync def run(self) -> None: await self.client.connect(url=MAIBOT_WS_URL, platform=PLATFORM, token=MAIBOT_TOKEN or None) # run() 阻塞并自带指数退避重连，和收事件循环并发跑 await asyncio.gather(self.poll_loop(), self.client.run())connect()只是存下连接参数，真正建连在run()里；别忘了并发跑你自己的收事件循环。
完整代码
下面就是上面每一步拼起来的成品。复制出去，把 PlatformBridge 三个方法填上，改掉顶部的四个常量即可运行。
py
"""最小可用的 MaiBot 适配器骨架。
用法：把 PlatformBridge 里三个方法的 TODO 换成你平台的 SDK 调用，
其余代码不需要改动。运行前先 `pip install maim_message`。
python adapter.py
"""
from __future__ import annotations
import asyncio
import logging
import time
from typing import Any
from maim_message import MessageClient
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
logger = logging.getLogger("my-adapter")
# ---- 只改这一段 ----------------------------------------------------------
PLATFORM = "myplatform" # 平台名：全局唯一，不要用 webui / bot_console
MAIBOT_WS_URL = "ws://127.0.0.1:8000/ws" # 对应 maim_message.ws_server_host / port
MAIBOT_TOKEN = "" # 与 maim_message.auth_token 中的某一项完全一致
SELF_ID = "bot_1" # 本适配器实例的账号标识，用于多账号路由
# --------------------------------------------------------------------------
# region bridge
class PlatformBridge:
"""平台侧收发封装。接一个新平台，只需要实现这三个方法。"""
async def send_text(self, target: str, text: str) -> str:
"""给 target 发一条文本，返回平台侧的消息 ID。"""
raise NotImplementedError
async def send_image(self, target: str, image_base64: str) -> str:
"""给 target 发一张图片（base64），返回平台侧的消息 ID。"""
raise NotImplementedError
async def poll_events(self) -> list[dict[str, Any]]:
"""阻塞收取平台事件，返回归一化后的事件列表。
每个事件的字段约定见 build_inbound()。
"""
raise NotImplementedError
# endregion bridge
# region inbound
class Adapter:
def __init__(self, bridge: PlatformBridge) -> None:
self.bridge = bridge
self.client = MessageClient()
self.client.register_message_handler(self.on_outbound)
self.client.register_custom_message_handler("message_id_echo", self.on_echo)
# ---- 入站：平台 → MaiBot ---------------------------------------------
def build_inbound(self, event: dict[str, Any]) -> dict[str, Any]:
"""把平台事件拼成 MaiBot 的统一消息格式。"""
is_group = event.get("group_id") is not None
message_info: dict[str, Any] = {
"platform": PLATFORM,
"message_id": str(event["message_id"]),
"time": float(event.get("time") or time.time()),
"user_info": {
"platform": PLATFORM,
"user_id": str(event["user_id"]),
"user_nickname": str(event.get("user_name") or event["user_id"]),
},
# 这三个字段让出站回复能找到回来的路
"additional_config": {
"platform_io_account_id": SELF_ID,
"platform_io_target_user_id": str(event["user_id"]),
},
}
if is_group:
message_info["group_info"] = {
"platform": PLATFORM,
"group_id": str(event["group_id"]),
"group_name": str(event.get("group_name") or event["group_id"]),
}
return {
"message_info": message_info,
"message_segment": {"type": "text", "data": str(event["text"])},
}
async def poll_loop(self) -> None:
while True:
try:
events = await self.bridge.poll_events()
except Exception:
logger.exception("收取平台事件失败，2 秒后重试")
await asyncio.sleep(2)
continue
for event in events:
await self.client.send_message(self.build_inbound(event))
# endregion inbound
# region outbound
# ---- 出站：MaiBot → 平台 ---------------------------------------------
async def on_outbound(self, message: dict[str, Any]) -> None:
info = message["message_info"]
segment = message["message_segment"]
target = self.resolve_target(info)
message_id = str(info["message_id"])
for seg in self.iter_segments(segment):
actual_id = ""
if seg["type"] == "text":
actual_id = await self.bridge.send_text(target, seg["data"])
elif seg["type"] == "image":
actual_id = await self.bridge.send_image(target, seg["data"])
elif seg["type"] == "at":
actual_id = await self.bridge.send_text(target, f"@{seg['data']} ")
else:
logger.warning("暂不支持的分段类型：%s", seg["type"])
continue
if actual_id:
# 把平台真实消息 ID 回填给 MaiBot
await self.client.send_custom_message(
"message_id_echo",
{"type": "echo", "echo": message_id, "actual_id": actual_id},
)
@staticmethod
def iter_segments(segment: dict[str, Any]):
"""出站正文通常是 seglist，这里统一成可迭代的分段。"""
if segment["type"] == "seglist":
yield from segment["data"]
else:
yield segment
@staticmethod
def resolve_target(info: dict[str, Any]) -> str:
"""群消息发到群，私聊发到人。"""
group_info = info.get("group_info")
if group_info:
return str(group_info["group_id"])
return str(info["additional_config"]["platform_io_target_user_id"])
# endregion outbound
async def on_echo(self, payload: dict[str, Any]) -> None:
logger.debug("回执：%s -> %s", payload.get("echo"), payload.get("actual_id"))
# ---- 生命周期 --------------------------------------------------------
async def run(self) -> None:
await self.client.connect(
url=MAIBOT_WS_URL, platform=PLATFORM, token=MAIBOT_TOKEN or None
)
logger.info("已连接 MaiBot：%s（platform=%s）", MAIBOT_WS_URL, PLATFORM)
await asyncio.gather(self.poll_loop(), self.client.run())
if __name__ == "__main__":
asyncio.run(Adapter(PlatformBridge()).run())MaiBot 侧要改什么
默认配置只监听本机。适配器和 MaiBot 不在同一台机器（或不在同一个容器网络）时，改 config/bot_config.toml：
toml
[maim_message]
ws_server_host = "0.0.0.0" # 从 127.0.0.1 改成 0.0.0.0 才能被外部连上
ws_server_port = 8000
auth_token = ["换成一段足够长的随机串"] # 非空即开启认证；适配器侧要填一模一样的值认证值是原文比对
auth_token 走 authorization 请求头做整串比较，没有 Bearer 前缀解析。适配器侧多打一个 Bearer 就会被以关闭码 1008 断开。
多账号与访问范围
- 同平台多账号：经典服务下一个
platform只能有一条连接，第二条会把第一条顶掉。多账号要么把account_id填进additional_config并用不同platform，要么改用 API 服务器的x-uuid区分，或走插件网关的account_id/scope。 - 谁能用麦麦：适配器不自带黑白名单，入站放行统一由
config/adapter_policy.toml控制，见访问策略与账户路由。
部署建议
- 单独进程 — 适配器崩了不要拖垮麦麦；用 systemd、supervisor 或一个独立的 Docker 容器托管。
- 别自己写重连 —
maim_message客户端自带指数退避重连（2 秒起，上限 10 秒），重复实现只会打架。 - 日志带平台名与账号 — 多账号排查时，没有这两个字段的日志基本没用。
- 大图先压缩 — 单帧上限 100 MB，图片与语音在协议上是 base64（体积约涨三分之一）；平台侧能压缩就先压缩。
- 保留一个测试群 — 上线前先在固定测试群跑通收发，再放开正式群。
验证与排错
验收动作 — 在平台上发一条"你好"：MaiBot 控制台出现入站日志，平台侧收到回复，且 MaiBot 日志里能看到 收到回送消息ID 的调试信息。
- 启动报"连接失败"或立刻断开 — 九成是地址问题：MaiBot 仍在监听
127.0.0.1，或端口没放行；容器部署还要确认端口映射。 - 关闭码 1008 — 令牌不一致，注意值要完全一致、不要加前缀。
- 连上了但麦麦没反应 — 先看平台名是否与配置一致，再查
adapter_policy.toml是否放行该群 / 该用户。 - 报断言错误 / 消息直接消失 — 回到第 4 步逐项核对必填字段与类型，尤其是
user_nickname和group_name。 - 回复发不出去 — 私聊检查
platform_io_target_user_id是否填了；群聊确认group_info.group_id与平台侧一致。 - 反复重连 — 同一个
platform有两条连接在互相顶替，或者心跳超时（服务端 30 秒一次 PING、10 秒无响应即断）。