MaiBot Docs

English

简体中文

English

简体中文

Appearance

Writing an Adapter ​

Updated: 10/10/2026Word count: 1068 wordsReading time: 4 min

Connecting a new platform takes only three methods. Everything else — connecting to MaiBot, reconnecting after a drop, authentication, message formatting — is handled by the official maim_message library and a skeleton of about a hundred lines. This page gets it running from scratch.

Prerequisites:

  • Python 3.10+, with pip install maim_message working;
  • A platform SDK that can send and receive messages (the platform's official library, an OneBot implementation, or your own protocol stack);
  • A MaiBot instance that is already running, plus its message server address and token.

Decide which service you are connecting to first

This page uses the legacy message server (enabled by default, ws://127.0.0.1:8000/ws). If you are connecting to the API server (enable_api_server) instead, the connection parameters become x-apikey / x-platform, and every packet needs an extra envelope on the outside — for the differences, see Message Protocol Reference.

What the Skeleton Handles ​

PlatformBridge has only three methods: send text, send image, receive events. The skeleton assembles events into MaiBot's unified format, splits replies into segments, and echoes the platform's real message ID back to MaiBot.

Build It ​

  1. Install the dependency.

    bash
    pip install maim_message

    MaiBot enforces a minimum version: anything below 0.6.2 makes it refuse to start the message service. The version it currently pairs with is 0.6.8.

  2. Create the directory.

    my-adapter/
├── adapter.py       # Skeleton: connection, routing, echo
├── bridge.py        # Your platform send/receive implementation
└── requirements.txt

  3. Implement the three platform-side methods. Only care about calls into the platform SDK — do not touch MaiBot's format here.

    py
    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
    • send_text — return the platform-side message ID after sending; if you cannot get one, return an empty string and the echo is skipped automatically.
    • send_image — inbound images are base64; most platforms need them written to disk or uploaded before they can be sent.
    • poll_events — collect platform events into a list of dictionaries; the field conventions are in the next step.

  4. Assemble the inbound packet. This is the easiest place in the whole page to trip up: MaiBot uses assertions rather than friendly validation for required fields, and one missing field loses the whole message.

    py
    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))

    Three hard requirements:

    • user_info.user_id and user_info.user_nickname must be non-empty strings; group messages also need group_info.group_id and group_info.group_name;
    • message_id and time must be present, and time is a float in seconds;
    • the data of a text segment must be a string — cast numeric IDs with str() first.

    Fill in two more fields while you are at it, so outbound replies can find their way:

    • additional_config.platform_io_account_id — the account identifier of this adapter instance; it is what tells multiple accounts on the same platform apart.
    • additional_config.platform_io_target_user_id — the recipient of private-chat replies.

  5. Handle outbound segments. Most of the body MaiBot pushes over is a seglist; translate it one type at a time. Log and skip segment types you do not recognize — do not let the whole adapter crash.

    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"])

  6. Echo back the real message ID. After a successful send, echo the ID the platform returned to MaiBot; only then does it know "which message on the platform this reply is", so that recalls, quotes, and reply chains can attach to it.

    python
    await client.send_custom_message(
    "message_id_echo",
    {"type": "echo", "echo": mmc_message_id, "actual_id": platform_message_id},
)

    Here client is maim_message.MessageClient: send_custom_message(message_type_name, message) takes 2 arguments, and platform comes from the connection identity, so it is not passed. The 3-argument send_custom_message(platform, message_type_name, message) is a method of the server-side MessageServer and is not used by adapters. For the legacy service the type name is exactly message_id_echo; on the API server it must be custom_message_id_echo (the API client adds the custom_ prefix automatically).

  7. Start it and connect.

    python
    async def run(self) -> None:
    await self.client.connect(url=MAIBOT_WS_URL, platform=PLATFORM, token=MAIBOT_TOKEN or None)
    # run() blocks and brings its own exponential-backoff reconnection; run it concurrently with the event loop
    await asyncio.gather(self.poll_loop(), self.client.run())

    connect() only stores the connection parameters — the connection is actually established inside run(); do not forget to run your own event-receiving loop concurrently.

Full Code ​

Below is the finished result, assembled from every step above. Copy it out, fill in the three PlatformBridge methods, and change the four constants at the top to run it.

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())

What to Change on the MaiBot Side ​

The default configuration listens on the local machine only. When the adapter and MaiBot are not on the same machine (or not in the same container network), edit config/bot_config.toml:

toml
[maim_message]
ws_server_host = "0.0.0.0"   # Change 127.0.0.1 to 0.0.0.0 so external clients can connect
ws_server_port = 8000
auth_token = ["换成一段足够长的随机串"]   # Non-empty enables authentication; the adapter must send exactly the same value

The auth value is compared verbatim

auth_token is compared as a whole string against the authorization header, with no Bearer prefix parsing. One extra Bearer on the adapter side and you are disconnected with close code 1008.

Multiple Accounts and Access Scope ​

  • Multiple accounts on one platform: on the legacy service a platform can have only one connection, and a second one kicks the first out. For multiple accounts, either put account_id into additional_config and use a different platform for each, or switch to the API server's x-uuid, or use the plugin gateway's account_id / scope.
  • Who can use MaiBot: adapters do not carry their own allow/deny lists; inbound admission is controlled uniformly by config/adapter_policy.toml, see Access Policy and Account Routing.

Deployment Advice ​

  • Run it as a separate process — a crashing adapter should not drag MaiBot down with it; host it with systemd, supervisor, or a standalone Docker container.
  • Do not write your own reconnect logic — the maim_message client ships exponential-backoff reconnection (starting at 2 seconds, capped at 10 seconds); a duplicate implementation only fights with it.
  • Log the platform name and account — when troubleshooting multiple accounts, logs without these two fields are close to useless.
  • Compress large images first — the per-frame limit is 100 MB, and images and voice are base64 on the wire (about a third larger); compress on the platform side whenever you can.
  • Keep one test group — get send and receive working in a fixed test group before you open up your production groups.

Verification & Troubleshooting ​

Acceptance check — send "hello" on the platform: an inbound log appears in the MaiBot console, the platform side receives a reply, and the MaiBot log shows the 收到回送消息ID (echoed message ID received) debug line.

  • Startup reports "connection failed", or it disconnects immediately — nine times out of ten it is an address problem: MaiBot is still listening on 127.0.0.1, or the port is not open; with a container deployment, confirm the port mapping as well.
  • Close code 1008 — the tokens do not match; the values must be exactly identical, with no prefix added.
  • Connected but MaiBot does not respond — first check whether the platform name matches the configuration, then whether adapter_policy.toml allows that group / user.
  • Assertion errors / messages just vanish — go back to step 4 and check the required fields and their types one by one, especially user_nickname and group_name.
  • Replies will not go out — for private chats, check that platform_io_target_user_id is filled in; for group chats, confirm group_info.group_id matches the platform side.
  • Repeated reconnects — two connections share the same platform and keep replacing each other, or the heartbeat timed out (the server PINGs every 30 seconds and drops the connection after 10 seconds without a response).