Adapter Development Guide

First, you should install the standard communication library for MaiMai: maim_message

pip install maim_message

Detailed documentation for the maim_message library is available in the maim_message documentation.

Your Adapter should implement upstream message reception, processing, and downstream message construction. This means implementing message passing and processing between external programs ↔ Adapter and Adapter ↔ MaiBot Core. This guide will cover how to implement the communication between Adapter and MaiBot Core.

Note: In practice, communication between different Adapters can also use the maim_message library for message processing, but this guide does not cover that. For specific examples, refer to the implementations of maimbot_tts_adapter and MaiBot-Napcat-Adapter.

The Adapter workflow is as follows:

1. Receive upstream messages (not covered in this guide)
2. Process upstream messages and construct downstream message metadata
3. Process upstream messages and construct downstream message content
4. Construct the final message body and send it
5. Parse the returned message

Next, this guide will cover how to implement the communication between Adapter and MaiBot Core.

Processing Upstream Messages

For any MaiBot Adapter targeting version 0.7.0 or later, you should implement a chat whitelist/blacklist on the Adapter side.

Starting from version 0.7.0, MaiBot will no longer check whether messages exist in whitelists or blacklists.

Constructing Downstream Message Metadata

This section will use the MessageBaseInfo class from the maim_message library to construct downstream message metadata.

platform field: You should use this field to identify the source platform of the message, and also as an identifier for the adapter.

message_id field: This ID can be generated by you or returned from other platforms. The main purpose of this field is to tell MaiBot Core your message number, making it easier for the Adapter to locate messages when MaiBot Core sends Seg messages with reply type.

time field: To ensure message ordering, this should use a float-type Unix timestamp.

group_info and user_info fields: These two fields represent group information and user information respectively. These will be covered in detail below.

format_info field: This part is used to identify the message formats that the Adapter can send and receive.

template_info field: Optional.

additional_config field: This part is used to pass some additional configuration. For details, refer to the relevant section of the maim_message documentation.

Construction example:

message_info = BaseMessageInfo(
        platform="my_platform_instance_1",
        message_id="12345678",
        time=1234567.001,
        group_info=group_info,
        user_info=user_info,
        format_info=format_info,
        template_info=None,
        additional_config=None
    )

The following sections detail the construction of specific fields.

Constructing the group_info Field

For private messages, the group_info field of MessageBaseInfo should be set to None.

For group messages, the group_info field should be constructed as follows:

The platform field value should match the platform field in the MessageBaseInfo class.

The group_id field value should be the ID of the group where the message is located.

The group_name field value should be the name of the group where the message is located. This part is optional.

Then pass the group_info field to the MessageBaseInfo class.

Construction example:

# For group chats
group_info = GroupInfo(
    platform = "my_platform_instance_1",
    group_id = "123456",
    group_name = "Test Group"
)
# For private chats
group_info = None

Constructing the user_info Field

This should be constructed for both private and group chats.

The platform field value should match the platform field in the MessageBaseInfo class.

The user_id field value should be the ID of the message sender.

The user_nickname field value should be the nickname/username of the message sender. This part is optional.

The user_cardname field value should be the remark name/group nickname of the message sender. This part is optional.

Construction example:

user_info = UserInfo(
    platform = "my_platform_instance_1",
    user_id = "123456",
    user_nickname = "Test User",
    user_cardname = "Test User's Group Nickname"
)

Constructing the format_info Field

This part identifies the message formats that the Adapter can send and receive. For actual content, refer to the relevant section of the maim_message documentation.

Construction example:

format_info = FormatInfo(
    content_format=["text", "image"],  # Formats contained in the message
    accept_format=["text", "image"]   # Acceptable formats
)

Constructing Downstream Message Content

This is a required part, consisting of a Seg-wrapped message body.

When constructing this part, you should convert the corresponding message into a format usable by maim_message before passing parameters. For details, refer to the relevant section of the maim_message documentation.

Messages in this part can be recursively constructed using the seglist type of the Seg class. MaiBot Core will flatten them during parsing.

Construction example:

submit_seg = Seg(
    type = "seglist",
    data = [
        Seg(
            type = "text",
            data = "Test text"
        ),
        Seg(
            type = "image",
            data = "base64://some_base64_string"
        )
    ]
)

Constructing the Final Message to Send

This section uses the MessageBase class for construction. Construction example:

message = MessageBase(
    message_info=message_info,  # Message metadata constructed above
    message_segment=submit_seg,  # Message body constructed above
    raw_message=None
)

Constructing Routes for Message Sending and Receiving

This section uses the Router class from the maim_message library to construct routes and handle message sending and receiving.

1. First, create an instance of the RouteConfig class, passing in the configuration for all platforms you need to connect to.

Construction example:

route_config = RouteConfig(
    route_config={
        # "platform_name" is a custom identifier used to distinguish different connections
        "my_platform_instance_1": TargetConfig(
            url="ws://127.0.0.1:8000/ws", # MaimCore or target server address
            token=None, # Fill this field if the server requires Token authentication
        ),
        # Multiple connections can be configured
        # "another_platform": TargetConfig(...)
    }
)

2. Then use the newly constructed route_config to create a router instance router from the Router class.

router = Router(route_config)

3. First, implement your message handler, which should be an asynchronous function. For details, refer to maim_message-messagebase. This function can convert dictionary-format messages to MessageBase objects by calling the from_dict method.
4. Then use the router.register_class_handler() method to register your message handler.

router.register_class_handler(message_handler)

5. Run the router to start receiving messages.

Parsing Returned Messages

Your message handler should be an asynchronous function that receives a dictionary as a parameter.

You can use the from_dict method of MessageBase to convert the corresponding dictionary into a MessageBase object for parsing.

After parsing, you should send the message to the corresponding platform.

Complete Example

import asyncio
import time
from maim_message import (
    BaseMessageInfo, UserInfo, GroupInfo, MessageBase, Seg,
    Router, RouteConfig, TargetConfig
)

# 1. Define connection targets (e.g., MaimCore)
route_config = RouteConfig(
    route_config={
        # "platform_name" is a custom identifier used to distinguish different connections
        "my_platform_instance_1": TargetConfig(
            url="ws://127.0.0.1:8000/ws", # MaimCore or target server address
            token=None, # If the server requires Token authentication
        ),
        # Multiple connections can be configured
        # "another_platform": TargetConfig(...)
    }
)

# 2. Create Router instance
router = Router(route_config)

# 3. Define how to handle messages received from MaimCore
async def handle_response_from_maimcore(message: MessageBase):
    """Handle messages replied from MaimCore"""
    print(f"Received reply from MaimCore ({message.message_info.platform}): {message.message_segment}")
    # Add logic here to send messages back to the original platform (e.g., QQ, Discord, etc.)
    # ...

# 4. Register message handler
# Router will automatically pass messages received from the corresponding platform to the registered handler
router.register_class_handler(handle_response_from_maimcore)

# 5. Construct messages to send to MaimCore
def construct_message_to_maimcore(platform_name: str, user_id: int, group_id: int, text_content: str) -> MessageBase:
    """Construct standard MessageBase based on platform events"""
    user_info = UserInfo(platform=platform_name, user_id=user_id)
    group_info = GroupInfo(platform=platform_name, group_id=group_id)
    message_info = BaseMessageInfo(
        platform=platform_name,
        message_id="some_unique_id_from_platform", # Original ID of the platform message
        time=time.time(), # Current timestamp
        user_info=user_info,
        group_info=group_info,
    )
    message_segment = Seg("seglist", [
        Seg("text", text_content),
        # Other Segs can be added, e.g., Seg("image", "base64data...")
    ])
    format_info = {
        "content_format": ["text"], # This sample message contains text format
        "accept_format": ["text", "image"]
    }
    message_info.format_info = format_info
    return MessageBase(message_info=message_info, message_segment=message_segment)

# 6. Run and send messages
async def run_client():
    # Start Router (it will automatically attempt to connect to all configured targets and start receiving messages)
    # run() is usually asynchronous and blocking, requiring create_task
    router_task = asyncio.create_task(router.run())
    print("Router is starting and attempting to connect...")

    # Wait for successful connection (more robust connection state checking is needed in actual applications)
    await asyncio.sleep(2)
    print("Connection should be established...")

    # Construct and send messages
    platform_id = "my_platform_instance_1"
    msg_to_send = construct_message_to_maimcore(
        platform_name=platform_id,
        user_id=12345,
        group_id=98765,
        text_content="Hello MaimCore!"
    )
    print(f"Sending message to {platform_id}...")
    await router.send_message(msg_to_send)
    print("Message sent.")

    # Let Router continue running (or stop as needed)
    # await router_task # This will block until router stops

    # Example: Stop after running for some time
    await asyncio.sleep(5)
    print("Preparing to stop Router...")
    await router.stop()
    print("Router stopped.")
    # Wait for task completion
    try:
        await router_task
    except asyncio.CancelledError:
        print("Router task has been cancelled.")

if __name__ == "__main__":
    try:
        asyncio.run(run_client())
    except KeyboardInterrupt:
        print("User interrupted.")
    # Note: In actual adapters, Router startup and message sending/receiving will be integrated into the adapter's main event loop.