--- outline: deep --- # Developing a Platform Adapter AstrBot supports integrating platform adapters in plugin form, allowing you to connect platforms that AstrBot does not natively support — such as Lark, DingTalk, Bilibili private messages, or even Minecraft. We will use a platform called `FakePlatform` as an example. First, add `fake_platform_adapter.py` and `fake_platform_event.py` to your plugin directory. The former handles the platform adapter implementation, while the latter defines the platform event. ## Platform Adapter Assume FakePlatform's client SDK looks like this: ```py import asyncio class FakeClient(): '''Simulates a messaging platform that sends a message every 5 seconds''' def __init__(self, token: str, username: str): self.token = token self.username = username # ... async def start_polling(self): while True: await asyncio.sleep(5) await getattr(self, 'on_message_received')({ 'bot_id': '123', 'content': 'new message', 'username': 'zhangsan', 'userid': '123', 'message_id': 'asdhoashd', 'group_id': 'group123', }) async def send_text(self, to: str, message: str): print('Message sent:', to, message) async def send_image(self, to: str, image_path: str): print('Image sent:', to, image_path) ``` Now create `fake_platform_adapter.py`: ```py import asyncio from astrbot.api.platform import Platform, AstrBotMessage, MessageMember, PlatformMetadata, MessageType from astrbot.api.event import MessageChain from astrbot.api.message_components import Plain, Image, Record # Message chain components, import as needed from astrbot.core.platform.message_session import MessageSesion from astrbot.api.platform import register_platform_adapter from astrbot import logger from .client import FakeClient from .fake_platform_event import FakePlatformEvent # Register the platform adapter. First param: platform name, second: description, third: default config. @register_platform_adapter("fake", "fake adapter", default_config_tmpl={ "token": "your_token", "username": "bot_username" }) class FakePlatformAdapter(Platform): def __init__(self, platform_config: dict, platform_settings: dict, event_queue: asyncio.Queue) -> None: super().__init__(event_queue) self.config = platform_config # The default config above; filled in by the user and passed here self.settings = platform_settings # platform_settings: platform settings async def send_by_session(self, session: MessageSesion, message_chain: MessageChain): # Must be implemented await super().send_by_session(session, message_chain) def meta(self) -> PlatformMetadata: # Must be implemented. Simply return as shown below. return PlatformMetadata( "fake", "fake adapter", ) async def run(self): # Must be implemented. This is the main logic. # FakeClient is defined by us — this is just an example. This is its callback function. async def on_received(data): logger.info(data) abm = await self.convert_message(data=data) # Convert to AstrBotMessage await self.handle_msg(abm) # Initialize FakeClient self.client = FakeClient(self.config['token'], self.config['username']) self.client.on_message_received = on_received await self.client.start_polling() # Continuously listens for messages; this is a blocking call. async def convert_message(self, data: dict) -> AstrBotMessage: # Convert the platform message to AstrBotMessage. # The degree of adaptation is reflected here. Different platforms have different message # structures; convert accordingly. abm = AstrBotMessage() abm.type = MessageType.GROUP_MESSAGE # Also friend_message for private chats. Analyze per platform. Important! abm.group_id = data['group_id'] # Can be omitted for private chats abm.message_str = data['content'] # Plain text message. Important! abm.sender = MessageMember(user_id=data['userid'], nickname=data['username']) # Sender. Important! abm.message = [Plain(text=data['content'])] # Message chain. Append other message types as needed. Important! abm.raw_message = data # Raw message. abm.self_id = data['bot_id'] abm.session_id = data['userid'] # Session ID. Important! abm.message_id = data['message_id'] # Message ID. return abm async def handle_msg(self, message: AstrBotMessage): # Handle the message message_event = FakePlatformEvent( message_str=message.message_str, message_obj=message, platform_meta=self.meta(), session_id=message.session_id, client=self.client ) self.commit_event(message_event) # Submit the event to the event queue. Don't forget this! ``` `fake_platform_event.py`: ```py from astrbot.api.event import AstrMessageEvent, MessageChain from astrbot.api.platform import AstrBotMessage, PlatformMetadata from astrbot.api.message_components import Plain, Image from .client import FakeClient class FakePlatformEvent(AstrMessageEvent): def __init__(self, message_str: str, message_obj: AstrBotMessage, platform_meta: PlatformMetadata, session_id: str, client: FakeClient): super().__init__(message_str, message_obj, platform_meta, session_id) self.client = client async def send(self, message: MessageChain): for i in message.chain: # Iterate over the message chain if isinstance(i, Plain): # If it's a text message await self.client.send_text(to=self.get_sender_id(), message=i.text) elif isinstance(i, Image): # If it's an image # convert_to_file_path() resolves supported media refs through # the shared media utilities. img_path = await i.convert_to_file_path() await self.client.send_image(to=self.get_sender_id(), image_path=img_path) await super().send(message) # Must be called at the end to invoke the parent class's send method. ``` ## Media Message Handling Platform adapters do not need to reimplement media parsing for every platform. Convert the platform message into AstrBot message components, and put the media reference in the component's `file` / `url` field. The supported reference forms are: - Local path, such as `/tmp/a.jpg` - Standard `file:` URI, such as `file:///tmp/a.jpg` - HTTP(S) URL, such as `https://example.com/a.jpg` - `base64://`, such as `base64://iVBORw0KGgo...` - Data URI, such as `data:image/png;base64,iVBORw0KGgo...` - Legacy bare base64, such as `iVBORw0KGgo...`; supported for compatibility, but new code should not generate it intentionally If you already have a local file, prefer the component factory methods. They generate standard `file:` URIs: ```py from astrbot.api.message_components import Image, Record, Video abm.message.append(Image.fromFileSystem("/tmp/image.png")) abm.message.append(Record.fromFileSystem("/tmp/audio.wav")) abm.message.append(Video.fromFileSystem("/tmp/video.mp4")) ``` If the platform gives you an accessible URL, pass it directly to the component: ```py abm.message.append(Image(file=image_url, url=image_url)) abm.message.append(Record(file=audio_url, url=audio_url)) abm.message.append(Video(file=video_url, url=video_url)) ``` Before plugins and LLM providers see the event, AstrBot's preprocess stage tries to normalize media in the message chain: - `Image` is materialized through the shared media utilities and converted to JPEG when needed. - `Record` is materialized through the shared media utilities and converted to WAV when needed. - `Image` / `Record` inside `Reply` chains are normalized in the same way. - Temporary files created by core are attached to the current event and cleaned up when the event finishes. When sending messages, if the platform SDK needs a local file path, call the component's `convert_to_file_path()` instead of writing checks like `path.startswith("file://")`: ```py if isinstance(i, Image): image_path = await i.convert_to_file_path() await self.client.send_image(to=self.get_sender_id(), image_path=image_path) elif isinstance(i, Record): audio_path = await i.convert_to_file_path() await self.client.send_audio(to=self.get_sender_id(), audio_path=audio_path) elif isinstance(i, Video): video_path = await i.convert_to_file_path() await self.client.send_video(to=self.get_sender_id(), video_path=video_path) ``` If the adapter downloads platform media into AstrBot's temporary directory by itself, register the path on the event after creating it so the file does not remain after the event finishes: ```py message_event.track_temporary_local_file(temp_media_path) ``` Finally, in `main.py`, simply import the `fake_platform_adapter` module during initialization. The decorator will handle registration automatically. ```py from astrbot.api.star import Context, Star class MyPlugin(Star): def __init__(self, context: Context): from .fake_platform_adapter import FakePlatformAdapter # noqa ``` Once set up, run AstrBot: ![image](https://files.astrbot.app/docs/source/images/plugin-platform-adapter/QQ_1738155926221.png) The `fake` adapter we created now appears here. ![image](https://files.astrbot.app/docs/source/images/plugin-platform-adapter/QQ_1738155982211.png) After starting, you can see it working correctly: ![image](https://files.astrbot.app/docs/source/images/plugin-platform-adapter/QQ_1738156166893.png) If you have any questions, feel free to join the community group and ask~