How to Implement a Custom Platform Adapter for AstrBot: A Complete Developer Guide

To implement a custom platform adapter in AstrBot, create a Python class inheriting from the abstract Platform base class, implement the required meta(), run(), convert_message(), and handle_msg() methods, and register the class using the @register_platform_adapter decorator.

AstrBot is a multi-platform chatbot framework that unifies messaging services through a plugin-style architecture. Creating a custom platform adapter allows you to connect AstrBot to any messaging API—whether a private corporate chat, a niche social platform, or a custom WebSocket service—by translating native platform events into AstrBot's internal message format.

Understanding the Platform Adapter Architecture

AstrBot treats every messaging service as a platform adapter, a concrete implementation of the Platform abstract base class defined in astrbot/core/platform/platform.py. The framework discovers adapters automatically at startup through a decorator-based registration system. Once registered, the PlatformManager (located in astrbot/core/platform/manager.py) instantiates adapters based on the user's config.yaml and manages their lifecycle.

Core Components of a Platform Adapter

The Platform Abstract Base Class (astrbot/core/platform/platform.py)

The Platform ABC defines the contract all adapters must fulfill. It specifies the lifecycle methods (run, terminate), the message sending interface (send_by_session), error handling hooks, and the event-queue bridge. All adapters inherit from this class and override its abstract methods to provide platform-specific logic.

The Registration Decorator (astrbot/core/platform/register.py)

The @register_platform_adapter decorator is the entry point for adapter discovery. When applied to a class, it constructs a PlatformMetadata object and stores the class in the global platform_cls_map dictionary, keyed by the adapter name. This registration happens at import time, ensuring the framework knows about the adapter before configuration parsing begins.

Platform Metadata (astrbot/core/platform/platform_metadata.py)

PlatformMetadata is a dataclass that describes an adapter's capabilities to the Dashboard and configuration system. It includes fields for the adapter name, description, default configuration template, internationalization keys, and feature flags like support_streaming_message. The metadata generated by the decorator is what allows AstrBot's web UI to render configuration forms dynamically.

The Adapter Contract: Required Methods

meta(self) -> PlatformMetadata

This method must return a PlatformMetadata instance describing the adapter. The metadata is used by the Dashboard to display the adapter's name, description, and configuration options. It should match the information provided to the decorator.

run(self) -> Coroutine

The run method is the core lifecycle coroutine. It must initialize the platform's client (e.g., HTTP session, WebSocket connection, or SDK client), set up event listeners, and keep the coroutine alive until the bot shuts down. Typically, this is done by awaiting an asyncio.Event (e.g., self.shutdown_event).

convert_message(self, raw: Any) -> AstrBotMessage

This method translates a raw message payload from the platform's native format into an AstrBotMessage. The AstrBotMessage standardizes fields like type (group or private), session_id, sender, message_str, and raw_message. This conversion is critical for AstrBot's core to process messages uniformly regardless of origin.

handle_msg(self, message: AstrBotMessage, ...)

After conversion, handle_msg wraps the AstrBotMessage into a platform-specific AstrMessageEvent subclass and pushes it onto the central event queue via self.commit_event(event). This bridges the adapter to AstrBot's EventBus, allowing plugins to receive and respond to the message.

terminate(self) -> Coroutine (Optional)

Implement this to gracefully shut down the adapter. It should signal the shutdown_event, close HTTP sessions, disconnect WebSockets, and cancel any background tasks.

send_by_session(self, ...) (Optional)

If your platform supports session-based messaging (where plugins can send messages to a specific session without knowing the underlying platform details), implement this method. It receives a MessageSesion and a MessageChain, then dispatches the message to the platform's API.

Step-by-Step Implementation Guide

Step 1: Create the Adapter Package

Create a new directory under astrbot/core/platform/sources/<your_platform>/. This keeps your adapter organized alongside built-in adapters like Telegram and Discord.

Step 2: Implement the Adapter Class

Create a Python file (e.g., my_adapter.py) in your new directory. Define a class inheriting from Platform and implement the required methods: meta, run, convert_message, and handle_msg.

Step 3: Register the Adapter

Apply the @register_platform_adapter decorator to your class, providing a unique name, description, and optional default configuration template. This ensures AstrBot discovers your adapter at startup.

Step 4: Configure the Adapter

Add a configuration section to your config.yaml under the platforms key, using the adapter name you registered. The PlatformManager will instantiate your adapter using this configuration.

Complete Skeleton Implementation

The following example demonstrates a minimal yet complete custom platform adapter for a fictional "MyChat" service. This skeleton includes all required methods, proper registration, and event queue integration as implemented in astrbot/core/platform/sources/telegram/tg_adapter.py and astrbot/core/platform/sources/discord/discord_platform_adapter.py.

# file: astrbot/core/platform/sources/mychat/my_adapter.py

from __future__ import annotations

import asyncio
from typing import Any

from astrbot.api.platform import (
    AstrBotMessage,
    MessageMember,
    MessageType,
    Platform,
    PlatformMetadata,
    register_platform_adapter,
)
from astrbot.core.platform.astr_message_event import MessageSesion
from astrbot.core.message.message_event import AstrMessageEvent

# ① Register the adapter – this runs at import time

@register_platform_adapter(
    "mychat",                         # unique adapter name

    "MyChat 适配器",                  # description shown in the Dashboard

    default_config_tmpl={            # optional default config template

        "type": "mychat",
        "enable": True,
        "api_key": "",
        "endpoint": "https://api.mychat.com",
    },
    adapter_display_name="MyChat",
    logo_path="logo.png",            # relative to the plugin dir, optional

)
class MyChatAdapter(Platform):
    """Concrete implementation for the MyChat messaging platform."""

    def __init__(self, platform_config: dict, platform_settings: dict, event_queue: asyncio.Queue) -> None:
        super().__init__(platform_config, event_queue)
        self.settings = platform_settings
        self.client = None                 # will hold the SDK / HTTP client

        self.shutdown_event = asyncio.Event()

    # ② Provide metadata for the Dashboard / config UI

    def meta(self) -> PlatformMetadata:
        return PlatformMetadata(
            name="mychat",
            description="MyChat 适配器",
            id=self.config.get("id", "mychat"),
            default_config_tmpl=self.config,
            support_streaming_message=False,
        )

    # ③ Core run loop – initialise client and start listening

    async def run(self) -> None:
        # Initialise the third‑party SDK / HTTP session here

        self.client = await self._init_client()
        # Register a callback for inbound messages

        self.client.on_message = self._on_message_received
        # Keep the coroutine alive until the platform is stopped

        await self.shutdown_event.wait()

    async def _init_client(self) -> Any:
        # Placeholder – replace with real SDK initialisation

        from mychat_sdk import MyChatClient
        return MyChatClient(api_key=self.config["api_key"], endpoint=self.config["endpoint"])

    async def _on_message_received(self, raw_msg: dict) -> None:
        """SDK callback – convert and forward to AstrBot."""
        abm = self.convert_message(raw_msg)
        await self.handle_msg(abm)

    # ④ Convert a raw platform payload to AstrBotMessage

    def convert_message(self, raw: dict) -> AstrBotMessage:
        msg = AstrBotMessage()
        msg.type = MessageType.GROUP_MESSAGE if raw["group"] else MessageType.FRIEND_MESSAGE
        msg.session_id = raw["session_id"]
        msg.message_str = raw["text"]
        msg.sender = MessageMember(str(raw["user_id"]), raw["nickname"])
        msg.message = []                       # Populate MessageComponent list if needed

        msg.raw_message = raw
        return msg

    # ⑤ Push the event onto the central queue

    async def handle_msg(self, message: AstrBotMessage) -> None:
        event = MyChatMessageEvent(
            message_str=message.message_str,
            message_obj=message,
            platform_meta=self.meta(),
            session_id=message.session_id,
            client=self.client,
        )
        self.commit_event(event)

    # ⑥ Optional graceful shutdown

    async def terminate(self) -> None:
        self.shutdown_event.set()
        if self.client:
            await self.client.close()

Key implementation details from the source code:

• The decorator pattern mirrors the built‑in Telegram adapter at lines 42‑44 of tg_adapter.py, where @register_platform_adapter("telegram", "telegram 适配器") is applied to the class definition.
• The meta() method structure follows the Discord adapter's implementation at lines 13‑18 of discord_platform_adapter.py, which returns a PlatformMetadata instance with fields like support_streaming_message.
• The run() method's use of await self.shutdown_event.wait() to keep the coroutine alive is consistent with both the Telegram adapter (lines 27‑31) and the Discord adapter (lines 25‑30).
• Message conversion logic in convert_message() parallels Telegram's implementation (lines 73‑130) and Discord's _convert_message_to_abm (lines 85‑127), both of which map native API payloads to the standardized AstrBotMessage dataclass.

How the Framework Loads Your Adapter

When AstrBot starts, the PlatformManager (located in astrbot/core/platform/manager.py) performs the following steps:

1. Discovery: As Python imports the adapter modules (e.g., when the sources package is loaded), the @register_platform_adapter decorator executes, populating the global platform_cls_map dictionary with adapter names as keys and class references as values.
2. Instantiation: The manager reads the platforms section from config.yaml. For each enabled platform entry, it looks up the corresponding class in platform_cls_map using the configured type name.
3. Lifecycle Management: The manager creates an instance of your adapter, passing the platform-specific configuration dictionary, settings, and the central asyncio.Queue for event distribution. It then schedules the run() coroutine as a background task.

Real-World Reference Implementations

For production-ready patterns, examine the built-in adapters:

• Telegram: astrbot/core/platform/sources/telegram/tg_adapter.py demonstrates media group handling, webhook configuration, and the send_by_session implementation for session-aware replies.
• Discord: astrbot/core/platform/sources/discord/discord_platform_adapter.py shows slash-command registration, async client wrapper patterns, and streaming message support flags.

Both files illustrate how to handle platform-specific error boundaries while maintaining the AstrBotMessage contract required by the core event bus.

Summary

Implementing a custom platform adapter for AstrBot requires understanding three core concepts:

• Inheritance: Subclass the abstract Platform base class from astrbot/core/platform/platform.py to ensure lifecycle compatibility.
• Registration: Use the @register_platform_adapter decorator from astrbot/core/platform/register.py to expose your adapter to the PlatformManager and Dashboard UI.
• Message Translation: Implement convert_message() to map native platform payloads to AstrBotMessage, and handle_msg() to push events onto the central queue via commit_event().

Once registered, your adapter becomes a first-class citizen in AstrBot's architecture, configurable via config.yaml and manageable through the web Dashboard.

Frequently Asked Questions

What is the minimum set of methods I must implement for a functional adapter?

You must implement meta(), run(), convert_message(), and handle_msg(). The meta() method provides metadata for the Dashboard, run() initializes your platform client and keeps the adapter alive, convert_message() translates native payloads into AstrBotMessage objects, and handle_msg() creates the platform-specific event and commits it to the event queue. Optional methods like terminate() and send_by_session() provide graceful shutdown and session-aware messaging capabilities.

How does AstrBot discover my custom adapter automatically?

Discovery happens through Python's import system and the registration decorator. When AstrBot starts, it imports the platform sources package, which triggers the execution of @register_platform_adapter decorators in your module. This decorator, defined in astrbot/core/platform/register.py, stores your class in the global platform_cls_map dictionary. The PlatformManager then looks up adapters by name in this registry when processing the config.yaml file.

Can I implement streaming message support in my custom adapter?

Yes. Set support_streaming_message=True in the PlatformMetadata object returned by your meta() method. This flag indicates to the Dashboard and core system that your adapter can handle chunked or streaming responses. You will then need to implement the streaming logic within your send_by_session or event handling methods, typically by yielding or sending message segments as they become available from the LLM or data source.

Where should I store platform-specific configuration defaults?

Store default configuration templates in the default_config_tmpl parameter of the @register_platform_adapter decorator. This dictionary should include all necessary fields for your platform, such as API keys, endpoints, and boolean flags. The AstrBot Dashboard uses this template to render configuration UI elements dynamically. When the PlatformManager instantiates your adapter, it passes the user-specific configuration (merged with your defaults) as the platform_config argument to your __init__ method.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}