Adapter Development Best Practices

This document provides best practice recommendations for ErisPulse adapter development.

Bot Status Management and Meta Events

Adapters should actively send meta events through adapter.emit() to allow the framework to automatically track the Bot's connection status, online/offline events, and heartbeat information.

1. When to Send Meta Events

2. Sending Meta Events

The framework provides the emit_meta() method to send meta events in a single line:

class MyAdapter(BaseAdapter):
    async def _ws_handler(self, websocket):
        bot_id = self._get_bot_id()

        # Bot online: send connect event in one line
        await self.emit_meta("connect", bot_id, user_name="MyBot", nickname="MyBot")

        try:
            while True:
                data = await websocket.receive_text()
                event = self.convert(data)
                if event:
                    await self.adapter.emit(event)
        except WebSocketDisconnect:
            pass
        finally:
            # Bot offline
            await self.emit_meta("disconnect", bot_id)

3. Heartbeat Events

Adapters should regularly send heartbeat events during the connection's active period to update the Bot's active time:

class MyAdapter(BaseAdapter):
    async def _heartbeat_loop(self, bot_id: str):
        while self._connected:
            # Send meta heartbeat to the framework (done in one line)
            await self.emit_meta("heartbeat", bot_id)
            await asyncio.sleep(30)

4. Automatic Discovery of the self Field

The framework's adapter.emit() automatically processes the self field in all events (not just meta events):

• Regular events (message/notice/request) with the self field will be automatically discovered and register the Bot.
• Extended information in the self field: Supports optional fields such as user_name, nickname, avatar, and account_id.

# Including the `self` field in the converter will automatically register the Bot
onebot_event = {
    "type": "message",
    "detail_type": "private",
    "platform": "myplatform",
    "self": {
        "platform": "myplatform",
        "user_id": "bot123",
        "user_name": "MyBot",
        "nickname": "MyBot",
    },
    # ... other fields
}
await self.adapter.emit(onebot_event)
# Bot "bot123" has been automatically registered and its active time updated

5. Bot Status Query

The framework provides the following query methods:

from ErisPulse import sdk

# Get Bot detailed information
info = sdk.adapter.get_bot_info("myplatform", "bot123")
# {"status": "online", "last_active": 1712345678.0, "info": {"nickname": "MyBot"}}

# List all Bots (grouped by platform)
all_bots = sdk.adapter.list_bots()

# List Bots for a specific platform
platform_bots = sdk.adapter.list_bots("myplatform")

# Check if a Bot is online
is_online = sdk.adapter.is_bot_online("myplatform", "bot123")

# Get a complete status summary (suitable for WebUI display)
summary = sdk.adapter.get_status_summary()
# {"adapters": {"myplatform": {"status": "started", "bots": {...}}}}

Connection Management

1. Implement Connection Retry

import asyncio

class MyAdapter(BaseAdapter):
    async def start(self):
        retry_count = 0
        max_retries = 5
        
        while retry_count < max_retries:
            try:
                await self._connect_to_platform()
                self.logger.info("Connection successful")
                break
            except Exception as e:
                retry_count += 1
                if retry_count < max_retries:
                    # Exponential backoff strategy
                    wait_time = min(60 * (2 ** retry_count), 600)
                    self.logger.warning(
                        f"Connection failed, retrying in {wait_time} seconds ({retry_count}/{max_retries}): {e}"
                    )
                    await asyncio.sleep(wait_time)
                else:
                    self.logger.error("Connection failed, maximum retry attempts reached")
                    raise

2. Connection State Management

class MyAdapter(BaseAdapter):
    async def start(self):
        self.connection = None
        self._connected = False
    
    async def _ws_handler(self, websocket: WebSocket):
        self.connection = websocket
        self._connected = True
        self.logger.info("Connection established")
        
        try:
            while True:
                data = await websocket.receive_text()
                await self._process_event(data)
        except WebSocketDisconnect:
            self.logger.info("Connection disconnected")
        finally:
            self.connection = None
            self._connected = False

3. Heartbeat Keepalive and Meta Heartbeat

The adapter's heartbeat should fulfill two tasks simultaneously: sending a keepalive heartbeat to the platform and sending a meta heartbeat event to the framework.

class MyAdapter(BaseAdapter):
    async def start(self):
        self.connection = await self._connect_to_platform()
        self._heartbeat_task = asyncio.create_task(self._heartbeat_loop())

    async def _heartbeat_loop(self):
        while self.connection:
            try:
                # 1. Send a keepalive heartbeat to the platform
                await self.connection.send_json({"type": "ping"})

                # 2. Send a meta heartbeat event to the framework (using emit_meta in one line)
                await self.emit_meta("heartbeat", self._bot_id)

                await asyncio.sleep(30)
            except Exception as e:
                self.logger.error(f"Heartbeat failed: {e}")
                break

4. Connection Information Exposure

The routes registered by the adapter should be visible to users, facilitating the configuration of callback addresses on the platform side. It is recommended to actively output connection information within the start() method:

class MyAdapter(BaseAdapter):
    async def start(self):
        router.register_websocket(
            module_name=self.platform,
            path="/ws",
            handler=self._ws_handler
        )

        if self.sdk:
            info = self.sdk.adapter.get_connection_info(self.platform)
            if info:
                self.logger.info(f"WebSocket address: "
                    f"{info.get('connection', {}).get('base_url', '')}"
                    f"{info.get('connection', {}).get('websocket_routes', [])}")

Users can view all routes and connection addresses of the adapter through the following API:

from ErisPulse import sdk

# Adapter-level connection information (recommended)
info = sdk.adapter.get_connection_info("myplatform")

# Query at the router manager level
sdk.router.list_namespaces()              # List all namespaces
sdk.router.get_module_routes("myplatform")  # Detailed route information
sdk.router.get_module_urls("myplatform")    # Complete connection URLs

Note: The module_name used during route registration must exactly match the platform name registered by the adapter in ErisPulse; otherwise, get_connection_info() will fail to associate the route. For multi-account adapters, sub-paths (e.g., /account1/webhook, /account2/webhook) should be registered for each account instead of using different module_name values.

Event Conversion

1. Strictly Follow the OneBot12 Standard

class MyPlatformConverter:
    def convert(self, raw_event):
        """Convert event"""
        onebot_event = {
            "id": str(raw_event.get("event_id", uuid.uuid4())),
            "time": int(time.time()),
            "type": self._convert_type(raw_event.get("type")),
            "detail_type": self._convert_detail_type(raw_event),
            "platform": "myplatform",
            "self": {
                "platform": "myplatform",
                "user_id": str(raw_event.get("bot_id", ""))
            },
            "myplatform_raw": raw_event,  # Preserve raw data (required)
            "myplatform_raw_type": raw_event.get("type", "")  # Original type (required)
        }
        return onebot_event

2. Standardize Timestamps

def _convert_timestamp(self, timestamp):
    """Convert to 10-digit second-level timestamp"""
    if not timestamp:
        return int(time.time())
    
    # If it's a millisecond-level timestamp
    if timestamp > 10**12:
        return int(timestamp / 1000)
    
    # If it's a second-level timestamp
    return int(timestamp)

3. Event ID Generation

import uuid

def _generate_event_id(self, raw_event):
    """Generate event ID"""
    event_id = raw_event.get("event_id")
    if event_id:
        return str(event_id)
    # If the platform does not provide an ID, generate a UUID
    return str(uuid.uuid4())

SendDSL Implementation

The At/AtAll/Reply decorators are built into the framework's SendDSL base class. Adapters only need to implement Raw_ob12 and specific send methods. Use self._apply_modifiers(message) and self.send_context to simplify development.

1. Must Return a Task Object

class Send(BaseAdapter.Send):
    def Raw_ob12(self, message, **kwargs):
        """Recommended implementation: Use framework helper methods"""
        async def _do_send():
            segments = self._apply_modifiers(message)
            return await self._adapter.call_api(
                endpoint="/send_message",
                message=segments,
                **self.send_context,
                **kwargs
            )
        return asyncio.create_task(_do_send())

    def Text(self, text: str):
        return self.Raw_ob12([{"type": "text", "data": {"text": text}}])

2. Chainable Modifier Methods Return self

class Send(BaseAdapter.Send):

    def __init__(self, adapter, target_type=None, target_id=None, account_id=None):
        super().__init__(adapter, target_type, target_id, account_id)
        self.buttons = []

    def Button(self, content: list) -> 'Send':
        self.buttons.append(content)
        return self # Return self

3. Support Platform-Specific Methods

class Send(BaseAdapter.Send):
    def Sticker(self, sticker_id: str):
        """Send a sticker pack"""
        return asyncio.create_task(
            self._adapter.call_api(
                endpoint="/send_sticker",
                message=[{"type": "sticker", "data": {"id": sticker_id}}],
                **self.send_context
            )
        )
    
    def Card(self, card_data: dict):
        """Send a card message"""
        return asyncio.create_task(
            self._adapter.call_api(
                endpoint="/send_card",
                message=[{"type": "card", "data": card_data}],
                **self.send_context
            )
        )

API Response

1. Standardized Response Format

The framework provides make_response() and make_error() methods to construct standardized responses:

async def call_api(self, endpoint: str, **params):
    try:
        raw_response = await self._platform_api_call(endpoint, **params)
        
        if raw_response.get("success"):
            return self.make_response(
                data=raw_response.get("data"),
                message_id=raw_response.get("data", {}).get("message_id", ""),
                raw=raw_response,
            )
        else:
            return self.make_error(
                retcode=raw_response.get("code", 10001),
                message=raw_response.get("message", ""),
                raw=raw_response,
            )
    except Exception as e:
        return self.make_error(message=str(e))

make_response() automatically generates a response dictionary containing the {platform}_raw key. make_error() defaults to retcode=34000 (Platform Error).

2. Error Code Specification

Follow the OneBot12 standard error codes:

# 1xxxx - Action Request Errors
10001: Bad Request
10002: Unsupported Action
10003: Bad Param

# 2xxxx - Action Handler Errors
20001: Bad Handler
20002: Internal Handler Error

# 3xxxx - Action Execution Errors
31000: Database Error
32000: Filesystem Error
33000: Network Error
34000: Platform Error
35000: Logic Error

Multi-Account Support

1. Declarative Configuration (Recommended)

After declaring the AccountConfigClass, the framework automatically manages multi-account loading, validation, and template generation. The BotAccountConfig base class provides the enabled and name fields, which do not need to be declared by the adapter:

from dataclasses import dataclass, field
from ErisPulse.Core.Bases import BotAccountConfig

@dataclass
class MyBotConfig(BotAccountConfig):
    token: str = field(default="", metadata={
        "description": {"i18n": "my_adapter.bot_token", "default": "Bot Token"},
        "required": True,
        "secret": True,
    })

class MyAdapter(BaseAdapter):
    AccountConfigClass = MyBotConfig
    
    async def start(self):
        for name, account in self.enabled_accounts.items():
            self.logger.info(f"Starting account {name}")
            await self._connect(name, account.token)
            # bot_id is automatically retrieved from the platform protocol/login response and filled back
    
    async def call_api(self, endpoint: str, **params):
        account_id = params.pop("account_id", None)
        name, account = self._resolve_account(account_id)
        # name: account name, account: MyBotConfig instance

The configuration file is automatically generated as:

[MyAdapter.accounts.default]
token = ""
enabled = true
name = ""

2. Account Selection Mechanism

The framework includes the _resolve_account() method, with matching priority as follows:

1. Account Name — exact match with the configuration key name
2. bot_id Field — automatically retrieved bot_id (i.e., event["self"]["user_id"])
3. Any str Field — other string fields in the configuration
4. Fallback — the first enabled account

# Match by account name
name, account = self._resolve_account("account1")

# Match by bot_id (most commonly used, from event)
name, account = self._resolve_account("bot_123")

# Get the first enabled account (pass None)
name, account = self._resolve_account(None)

Error Handling

1. Categorized Exception Handling

Use make_error() to construct standardized error responses. When making requests via sdk.client, catch ErisPulse exceptions:

from ErisPulse.Core.Bases.errors import ClientError, ClientTimeoutError

async def call_api(self, endpoint: str, **params):
    try:
        from ErisPulse.Core import client
        resp = await client.post(
            f"https://api.platform.com/{endpoint}",
            json=params,
            max_retries=2,
        )
        response = await resp.json()
        return self.make_response(data=response, raw=response)
    except ClientTimeoutError:
        self.logger.error(f"Request timeout: {endpoint}")
        return self.make_error(retcode=32000, message="Request timeout")
    except ClientError as e:
        self.logger.error(f"Network error: {e}")
        return self.make_error(retcode=33000, message="Network request failed")
    except json.JSONDecodeError:
        self.logger.error("JSON parsing failed")
        return self.make_error(retcode=10006, message="Response format error")
    except Exception as e:
        self.logger.error(f"Unknown error: {e}", exc_info=True)
        return self.make_error(message=str(e))

Backward Compatibility: Old adapter code that directly uses aiohttp is unaffected and can still catch aiohttp.ClientError. Exception translation only takes effect when requests are made through sdk.client.

2. Logging

The framework automatically creates a sub-logger for adapters (sdk.logger.get_child("MyAdapter")), so manual initialization is not required:

class MyAdapter(BaseAdapter):
    # ConfigClass = ...  # After declaring the configuration class, self.logger is automatically available
    
    async def start(self):
        self.logger.info("Adapter starting...")
        # ...
        self.logger.info("Adapter started")
    
    async def shutdown(self):
        self.logger.info("Adapter shutting down...")
        # ...
        self.logger.info("Adapter shutdown complete")

Testing

1. Unit Tests

import pytest
from ErisPulse.Core.Bases import BaseAdapter

class TestMyAdapter:
    def test_converter(self):
        """Test the converter"""
        converter = MyPlatformConverter()
        raw_event = {"type": "message", "content": "Hello"}
        result = converter.convert(raw_event)
        assert result is not None
        assert result["platform"] == "myplatform"
        assert "myplatform_raw" in result
    
    def test_api_response(self):
        """Test API response format"""
        adapter = MyAdapter()
        response = adapter.call_api("/test", param="value")
        assert "status" in response
        assert "retcode" in response

2. Integration Tests

@pytest.mark.asyncio
async def test_adapter_start():
    """Test adapter startup"""
    adapter = MyAdapter()
    await adapter.start()
    assert adapter._connected is True

@pytest.mark.asyncio
async def test_send_message():
    """Test sending messages"""
    adapter = MyAdapter()
    await adapter.start()
    
    result = await adapter.Send.To("user", "123").Text("Hello")
    assert result is not None

Reverse Conversion and Message Building

Raw_ob12 is a method that adapters must implement, serving as the unified entry point for reverse conversion (OneBot12 → Platform). Standard methods (e.g., Text, Image, etc.) should delegate to Raw_ob12, and modifier states (e.g., At/Reply/AtAll) must be merged into message segments within Raw_ob12.

MessageBuilder is a message segment builder tool designed to be used with Raw_ob12, supporting fluent chaining and rapid construction.

For the complete implementation specification, code examples, and usage methods, please refer to:

• Send Method Specification §6 Reverse Conversion Specification
• Send Method Specification §11 MessageBuilder

Platform Event Method Extensions

Adapters can register platform-specific methods for Event wrapper classes, allowing module developers to more easily access platform-specific data.

1. Using Mixin Classes for Batch Registration (Recommended)

When a platform has multiple specific methods, it is recommended to use a Mixin class:

# Register at the adapter's start() or module level
from ErisPulse.Core.Event import register_event_mixin

class MyPlatformEventMixin:
    def get_chat_name(self):
        """Get the chat name"""
        return self.get("myplatform_raw", {}).get("chat", {}).get("name", "")

    def is_official_message(self):
        """Determine if the message is an official message"""
        raw = self.get("myplatform_raw", {})
        return raw.get("sender", {}).get("is_official", False)

    def get_message_type(self):
        """Get the platform message type"""
        return self.get("myplatform_raw", {}).get("msg_type", "text")

# Batch registration
register_event_mixin("myplatform", MyPlatformEventMixin)

2. Registering Individual Methods Using a Decorator

from ErisPulse.Core.Event import register_event_method

@register_event_method("myplatform")
def get_chat_name(self):
    return self.get("myplatform_raw", {}).get("chat", {}).get("name", "")

3. Cleanup on Adapter Shutdown

from ErisPulse.Core.Event import unregister_platform_event_methods

class MyAdapter(BaseAdapter):
    async def shutdown(self):
        # Clean up platform event method registrations
        unregister_platform_event_methods("myplatform")
        # ... other cleanup

For more detailed information about registration and unregistration, please refer to Event System API - Registering Platform Extension Methods.

Documentation Maintenance

1. Maintaining Platform-Specific Documentation

In docs/en/platform-guide/, create a {platform}.md document (other language versions will be automatically generated):

# Platform Name Adapter Documentation

## Basic Information
- Corresponding Module Version: 1.0.0
- Maintainer: Your Name

## Supported Message Sending Types
...

## Unique Event Types
...

## Configuration Options
...

2. Updating Version Information

When releasing a new version, update the version information in the documentation:

[project]
version = "2.0.0"  # Update the version number

Related Documentation

• Getting Started with Adapter Development - Create your first adapter
• Core Concepts of Adapters - Understand the adapter architecture
• SendDSL Explained - Learn how to send messages