简体中文 English 繁體中文 日本語 Русский
本文为静态镜像，内容以交互版为准 在交互式文档中心打开 →

Core Concepts of Adapters

Understanding the core concepts of ErisPulse adapters is fundamental to adapter development.

Adapter Architecture

Component Relationships

Forward Conversion (Receiving Direction)                           Reverse Conversion (Sending Direction)
─────────────────                           ─────────────────
                                             
┌──────────────────┐                        ┌──────────────────┐
│ Native Platform Event     │                        │ Module-built Message     │
└────────┬─────────┘                        └────────┬─────────┘
         │                                           │
         ↓                                           ↓
┌──────────────────┐   ┌──────────────────┐   ┌──────────────────┐
│                  │   │  Adapter (MyAdapter) │   │ Send.Raw_ob12()  │
│  Converter       │   │ ┌──────────────┐ │   │ (Reverse Conversion Entry)   │
│  (Event Converter)    │──→│ │              │ │   │                  │
│                  │   │ │              │ │   │                  │
└──────────────────┘   │ └──────────────┘ │   └────────┬─────────┘
                       └──────────────────┘            │
                                │                      ↓
                                ↓              ┌──────────────────┐
                       ┌──────────────────┐    │ Platform API Call    │
                       │ OneBot12 Standard Event │    └────────┬─────────┘
                       └────────┬─────────┘             │
                                │                      ↓
                                ↓              ┌──────────────────┐
                       ┌──────────────────┐    │ Standard Response Format     │
                       │ Event System         │    └──────────────────┘
                       └────────┬─────────┘
                                │
                                ↓
                       ┌──────────────────┐
                       │ Module (Event Handling)  │
                       └──────────────────┘

Core Symmetry:

AdapterManager Adapter Manager

AdapterManager is the core component of ErisPulse's adapter system, responsible for managing the registration, startup, shutdown, and event distribution of all platform adapters.

Core Features

Basic Usage

from ErisPulse import sdk

# Register adapters (usually done automatically by Loader)
sdk.adapter.register("myplatform", MyPlatformAdapter)

# Start all adapters
await sdk.adapter.startup()

# Start specified adapters
await sdk.adapter.startup(["myplatform"])
# Start all adapters
await sdk.adapter.startup()

# Get adapter instance
my_adapter = sdk.adapter.get("myplatform")
# Or access via attribute
my_adapter = sdk.adapter.myplatform

# Shutdown all adapters
await sdk.adapter.shutdown()

Startup and Shutdown

Start Adapters

# Start all registered adapters
await sdk.adapter.startup()

# Start specified platforms
await sdk.adapter.startup(["platform1", "platform2"])

Startup Process:

  1. Submit adapter.start lifecycle event
  2. Submit adapter.status.change event (starting)
  3. Parallel start each adapter
  4. If startup fails, automatically retry (exponential backoff strategy)
  5. After successful startup, submit adapter.status.change event (started)

Retry Mechanism:

Shutdown Adapters

# Shutdown all adapters
await sdk.adapter.shutdown()

Shutdown Process:

  1. Submit adapter.stop lifecycle event
  2. Call shutdown() method for all adapters
  3. Shutdown router server
  4. Clear event handlers
  5. Submit adapter.stopped lifecycle event

Configuration Management

Check Platform Status

# Check if platform is registered
exists = sdk.adapter.exists("myplatform")

# Check if platform is enabled
enabled = sdk.adapter.is_enabled("myplatform")

# Use in operator
if "myplatform" in sdk.adapter:
    print("Platform exists and is enabled")

List Platforms

# List all registered platforms
platforms = sdk.adapter.list_registered()

# List all platforms and their status
status_dict = sdk.adapter.list_items()
# Returns: {"platform1": true, "platform2": false, ...}

# Get list of enabled platforms
enabled_platforms = [p for p, enabled in status_dict.items() if enabled]

Event Listening

OneBot12 Standard Events

from ErisPulse import sdk

# Listen for standard message events from all platforms
@sdk.adapter.on("message")
async def handle_message(data):
    print(f"Received OneBot12 message: {data}")

# Listen for standard message events from a specific platform
@sdk.adapter.on("message", platform="myplatform")
async def handle_platform_message(data):
    print(f"Received myplatform message: {data}")

# Listen for all events
@sdk.adapter.on("*")
async def handle_any_event(data):
    print(f"Received event: {data.get('type')}")

Platform Native Events

# Listen for native events from a specific platform
@sdk.adapter.on("raw_event_type", raw=True, platform="myplatform")
async def handle_raw_event(data):
    print(f"Received native event: {data}")

# Listen for native events from all platforms (wildcard)
@sdk.adapter.on("*", raw=True)
async def handle_all_raw_events(data):
    print(f"Received native event: {data}")

Event Distribution Mechanism

When calling adapter.emit(event_data):

  1. Middleware Processing: Execute all OneBot12 middlewares first
  2. Standard Event Distribution: Distribute to matching OneBot12 event handlers
  3. Native Event Distribution: If raw data exists, distribute to native event handlers

Matching Rules:

Middleware

Add Middleware

@sdk.adapter.middleware
async def logging_middleware(data):
    """Logging middleware"""
    print(f"Processing event: {data.get('type')}")
    return data  # Must return data

@sdk.adapter.middleware
async def filter_middleware(data):
    """Event filtering middleware"""
    # Filter out unwanted events
    if data.get("type") == "notice":
        return None  # Returning None means the middleware chain ignores this return value, keeping original data for further processing
    return data  # Must return data to continue passing

Middleware Return Contract

Return Value Behavior
dict Rewrite event payload (subsequent handlers receive rewritten event)
None Allow passage, payload unchanged (outputs WARNING log — recommend explicit return data)
False Reject: Event is discarded, not passed to any handler, no outbound side effects

Reject is suitable for firewall, rate limiting, blacklist scenarios where events are discarded at the event level ("direct discard at event level") (previously only achievable via high-priority event handlers). When rejecting, the framework outputs TRACE log and triggers the adapter.event.blocked lifecycle hook (carrying middleware middleware name, full event, platform / event_type / detail_type), facilitating troubleshooting of "why an event received no response":

@sdk.adapter.middleware
async def rate_limit_middleware(data):
    """Rate limiting middleware"""
    if _is_rate_limited(data):
        return False  # Reject: event is discarded
    data["rate_marked"] = True
    return data

@sdk.lifecycle.on("adapter.event.blocked")
async def on_event_blocked(data):
    print(f"Event rejected by {data['middleware']}: {data['event_type']}")

Middleware Execution Order

Middlewares execute in registration order, with later-registered middlewares executing first.

Note: If a middleware returns None (e.g., forgetting to return data), the framework ignores this return value and continues passing the original data, while outputting a warning-level log. This ensures that a single middleware mistake does not interrupt the entire event chain.

# Registration order
sdk.adapter.middleware(middleware1)  # Executes last
sdk.adapter.middleware(middleware2)  # Executes in the middle
sdk.adapter.middleware(middleware3)  # Executes first

# Execution order: middleware3 -> middleware2 -> middleware1

Get Adapter Instance

get() Method

adapter = sdk.adapter.get("myplatform")
if adapter:
    await adapter.Send.To("user", "123").Text("Hello")

Attribute Access

# Access via attribute name (case-insensitive)
adapter = sdk.adapter.myplatform
await adapter.Send.To("user", "123").Text("Hello")

BaseAdapter Base Class

Basic Structure

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

@dataclass
class MyConfig(BaseConfig):
    """Adapter Configuration (declared, framework automatically manages)"""
    token: str = field(
        default="",
        metadata={
            "description": {"i18n": "my_adapter.token", "default": "Bot Token"},
            "required": True,
            "secret": True,
            "ui": {"widget": "password", "group": "basic", "order": 1},
        },
    )

class MyAdapter(BaseAdapter):
    ConfigClass = MyConfig  # Declare configuration class
    
    # No need to override __init__, framework automatically handles:
    # - self.sdk, self.logger
    # - self.cfg (type-safe configuration instance, real-time read)
    # - self.Send, self.Request
    
    async def start(self):
        """Start adapter (must implement)"""
        cfg = self.cfg  # Automatically loaded type-safe configuration
        pass
    
    async def shutdown(self):
        """Shutdown adapter (must implement)"""
        pass
    
    async def call_api(self, endpoint: str, **params):
        """Call platform API (must implement)"""
        pass

Configuration Management

The framework provides declarative configuration management, defining configuration structures via dataclass, with the framework automatically handling loading, validation, and template generation.

Single Account Configuration

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

@dataclass
class TelegramConfig(BaseConfig):
    token: str = field(default="", metadata={
        "description": {"i18n": "telegram.token", "default": "Bot Token"},
        "required": True,
        "secret": True,
        "ui": {"widget": "password", "group": "basic", "order": 1},
    })
    proxy: str = field(default="", metadata={
        "description": {"i18n": "telegram.proxy", "default": "Proxy address"},
        "ui": {"widget": "text", "group": "advanced", "order": 10},
    })

class TelegramAdapter(BaseAdapter):
    ConfigClass = TelegramConfig
    
    async def start(self):
        cfg = self.cfg  # Type-safe, real-time read
        if not cfg.token:
            raise ValueError("Token not configured")
        await self._connect(cfg.token, proxy=cfg.proxy)

Multi-Account Configuration

The BotAccountConfig base class provides enabled and name fields. Most adapters can automatically obtain bot_id from the platform protocol or login response, injecting it into the account configuration during event conversion:

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

# Most adapters: bot_id is automatically obtained at runtime, no configuration needed
@dataclass
class MyBotConfig(BotAccountConfig):
    token: str = field(default="", metadata={
        "description": {"i18n": "my_adapter.bot_token", "default": "Token"},
        "required": True,
    })

# If bot_id cannot be obtained during login, allow users to fill it in configuration
@dataclass
class YunhuBotConfig(BotAccountConfig):
    bot_id: str = field(default="", metadata={
        "description": {"i18n": "yunhu.bot_id", "default": "Bot ID"},
        "required": True,
    })
    token: str = field(default="", metadata={
        "description": {"i18n": "yunhu.token", "default": "Token"},
        "required": True,
    })

class MyAdapter(BaseAdapter):
    AccountConfigClass = MyBotConfig
    
    async def start(self):
        for name, account in self.enabled_accounts.items():
            user_id = await self._login(name, account)
            await self.emit_meta("connect", user_id)

metadata Convention

Field metadata serves both TOML comment generation and WebUI form rendering:

metadata = {
    "description": str | dict,  # Field description (supports i18n)
    "required": bool,         # Whether required (validation + WebUI required marker)
    "secret": bool,           # Whether sensitive (WebUI displays as ***, logs are masked)
    "example": bool,          # Non-persistent flag: not written to config.toml (default value/template excluded),
                              # only rendered into config.full.example; schema with "example": true marked,
                              # CLI configuration wizard skips by default; user manually sets and persists normally
    "min": number, "max": number,  # Numerical range validation
    "ui": {                   # WebUI control configuration (old name "webui" still compatible)
        "widget": str,        # Control type: "text" | "switch" | "select" | "number" | "password"
        "group": str,         # Group: "basic" | "advanced" | "connection" etc.
        "order": int,         # Sorting weight (smaller is earlier)
        "options": list,      # Select control options [{label, value}], label supports i18n
        "placeholder": str | dict,  # Input box placeholder (supports i18n)
    },
    "extra": dict,            # Extra extended field (passed through to schema)
}

All user-visible text fields support i18n, uniformly using the {"i18n": "key", "default": "text"} format, pure strings are passed through as-is (backward compatibility). Supported i18n fields:

Field Location Description
description field metadata Field description
options[].label ui.options Select control option label
placeholder ui.placeholder Input box placeholder
group_labels _schema_meta Group display name (Dashboard partition title)

When using i18n, translation keys must be pre-registered in the i18n system (see [i18n documentation](../../advanced/i18n.md#Configuration Field Multilingual)).

description / placeholder / options label example:

token: str = field(
    default="",
    metadata={
        "description": {"i18n": "my_adapter.token", "default": "Bot Token"},
        "ui": {
            "widget": "text",
            "placeholder": {"i18n": "my_adapter.token.ph", "default": "Enter Token"},
        },
    },
)
mode: str = field(
    default="a",
    metadata={
        "description": {"i18n": "my_adapter.mode", "default": "Mode"},
        "ui": {
            "widget": "select",
            "options": [
                {"label": {"i18n": "my_adapter.mode.a", "default": "Option A"}, "value": "a"},
                {"label": "Pure string label", "value": "b"},  # Pure string passed through as-is
            ],
        },
    },
)

group_labels example (declare after configuration class definition):

MyConfig._schema_meta = {
    "group_labels": {
        "basic": {"i18n": "my_adapter.group.basic", "default": "Basic Settings"},
        "advanced": {"i18n": "my_adapter.group.advanced", "default": "Advanced Settings"},
    }
}

The framework's resolve_config_schema() automatically resolves all i18n keys in these fields according to the current language; get_config_schema() passes through the i18n dictionary as-is, to be parsed by the frontend.

docstring Automatically Generates Field Descriptions (v2.8.0+)

Fields without description declared in metadata, the framework automatically extracts field descriptions from the configuration class docstring as a fallback, supporting two common styles (can be mixed):

@dataclass
class MyConfig(BaseConfig):
    """
    MyAdapter Configuration

    :ivar endpoint: Platform API address        # reST style
    :ivar timeout: Request timeout in seconds
    """

    endpoint: str = "https://api.example.com"   # No metadata description → comment/description from docstring
    timeout: int = 30

    # Google style is also supported (Attributes: section):
    # Attributes:
    #     endpoint: Platform API address

Priority: metadata description > docstring field description > empty. i18n dictionary form of description is unaffected (always prioritized).

Nested Configuration (v2.8.0+)

When the field type is a nested dataclass, the framework recursively processes: schema uses "type": "table" + "fields" subtree (WebUI renders as a collapsible nested group), TOML template renders as [sub-table] section, default values / filling / validation / i18n resolution all recursively apply.

@dataclass
class RetryConfig(BaseConfig):
    """Retry Strategy

    :ivar max_retries: Maximum retry count
    """
    max_retries: int = 3
    backoff: float = 0.5

@dataclass
class MyConfig(BaseConfig):
    """MyAdapter Configuration"""
    endpoint: str = "https://api.example.com"
    retry: RetryConfig = field(default_factory=RetryConfig)   # Nested configuration section

Generated TOML template:

endpoint = "https://api.example.com"

[retry]
# Maximum retry count
max_retries = 3
backoff = 0.5

For nested types, it is recommended to use direct type annotations; string annotations (e.g., for deferred evaluation scenarios) must ensure the type can be globally resolved from the configuration class's module, __qualname__ outer class namespace, or class attribute by name.

Non-persistent example Fields (v2.8.0+)

gc_interval: int = field(default=300, metadata={"example": True})

Fields with example: True:

Suitable for "complex and rarely touched" advanced configuration items, keeping the user's config.toml minimal.

⚠️ _schema_meta is class-level metadata (not a configuration field). If declared inside the dataclass body, it must be annotated with ClassVar (_schema_meta: ClassVar[dict] = {...}), otherwise it will be treated as a regular field by dataclass. The framework defensively excludes fields with leading underscores from any schema/template/default value/validation output, but it is still recommended to declare them properly.

Declarative Translation Keys (v2.7.0+)

Adapters can declare translation keys centrally by nesting the I18nClass class, similar to declaring ConfigClass. The framework automatically registers all declared translation keys during __init__ phase (before configuration template generation), ensuring that i18n keys referenced in configuration descriptions are available when generating templates.

from ErisPulse.Core.Bases import BaseAdapter, BaseI18n, I18nKey

class MyAdapter(BaseAdapter):
    class I18nClass(BaseI18n):
        endpoint: I18nKey = I18nKey(
            default="API Endpoint",
            zh_CN="API 地址",
            zh_TW="API 位址",
            en="API Endpoint",
            ja="APIアドレス",
            ru="API адрес",
        )
        token: I18nKey = I18nKey(
            default="Platform Token",
            zh_CN="平台 Token",
            zh_TW="平台權杖",
            en="Platform Token",
            ja="プラットフォームトークン",
            ru="Токен платформы",
        )

I18nKey.default is a language-agnostic fallback text and is not registered in any language. To make translations effective, at least one language parameter must be explicitly provided.

For detailed usage (key path rules, explicit key parameters, etc.), see [i18n documentation](../../advanced/i18n.md#Recommended Writing Method Through I18nClass Declaration of Translation Keys v270).

Declarative Event Extension Methods (v2.7.0+)

Adapters can declare platform-specific event extension methods in EventMixin, which the framework automatically registers to the current platform.

from ErisPulse.Core import BaseAdapter

class MyAdapter(BaseAdapter):
    class EventMixin:
        def get_chat_name(self):
            """Get chat name"""
            return self.get("myplatform_raw", {}).get("chat", {}).get("name", "")

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

After registration, event objects can directly call these methods:

@message.on_group_message()
async def handler(event):
    if event.is_official_message():
        chat_name = event.get_chat_name()
        await event.reply(f"[{chat_name}] Official message received")

Adapter's event extension methods are registered to its own platform (self._platform). Modules needing cross-platform event extensions should use the original register_event_mixin() API.

Account Resolution

Multi-account adapters can use _resolve_account() to automatically resolve target accounts:

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: configuration instance

Resolution strategy: account name match → bot_id field match → other str field match → first enabled account.

Configuration Hot Update

Subclasses can override on_config_update() to respond to configuration changes:

class MyAdapter(BaseAdapter):
    ConfigClass = MyConfig
    
    def on_config_update(self, old_config, new_config):
        if old_config.token != new_config.token:
            self.logger.info("Token has been updated, reconnecting")

Initialization Process

The framework automatically performs the following tasks in BaseAdapter.__init__(self, sdk=None):

  1. SDK Reference: Set self.sdk, self.logger
  2. Send/Request Factory: Create self.Send and self.Request
  3. Configuration Template: If ConfigClass is declared, automatically generate default configuration template (first time)
  4. Account Template: If AccountConfigClass is declared, automatically generate default account template (first time)
  5. EventMixin Registration: If EventMixin is declared, automatically register in AdapterManager after platform name injection

Configuration is read in real-time via self.cfg / self.accounts (each access reads the latest value from configuration storage). self.config as a compatibility alias for self.cfg is still usable.

Most adapters do not need to override __init__. If custom initialization is required:

class MyAdapter(BaseAdapter):
    ConfigClass = MyConfig
    
    def __init__(self, sdk=None):
        super().__init__(sdk)  # Pass sdk
        self.converter = self._setup_converter()
        self.convert = self.converter.convert

Send Message Sending DSL

Inheritance Structure

class MyAdapter(BaseAdapter):
    class Send(BaseAdapter.Send):
        """Send nested class, inherits from BaseAdapter.Send"""
        pass

Available Properties

The Send class automatically sets the following properties when called:

Property Description Setting Method
_target_id Target ID To(id) or To(type, id)
_target_type Target Type To(type, id)
_target_to Simplified Target ID To(id)
_account_id Sending Account ID Using(account_id)
_adapter Adapter Instance Automatically set
_at_user_ids @ User List At(user_id)
_reply_message_id ID of replied message Reply(message_id)
_at_all Whether to @ all AtAll()

Recommendation: Use self.send_context property to get target_type, target_id, account_id at once, which is clearer than directly accessing instance variables.

Framework Helper Methods

Method/Property Description
self._apply_modifiers(message) Merge At/AtAll/Reply modifier states into message segment list
self.send_context Return {target_type, target_id, account_id} dictionary

Basic Methods

Adapters only need to implement Raw_ob12, standard methods (Text/Image/Voice/Video/File) are inherited from the SendDSL base class and default to delegating to it:

class Send(BaseAdapter.Send):
    def Raw_ob12(self, message, **kwargs):
        """Must implement: OneBot12 message segment → platform API"""
        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())

    # Text/Image/Voice/Video/File are inherited from base class, automatically delegate Raw_ob12, no need to implement repeatedly
    # If platform-specific logic is needed, override individual methods:
    # def Text(self, text: str):
    #     return self.Raw_ob12([{"type": "text", "data": {"text": text}}])

Chaining Modifier Methods

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

Event Converter

Conversion Flow

Platform Raw Event
    ↓
Converter.convert()
    ↓
OneBot12 Standard Event

Required Fields

All converted events must include:

{
    "id": "Event Unique Identifier",
    "time": 1234567890,           # 10-digit Unix timestamp
    "type": "message/notice/request/meta",
    "detail_type": "Event Detailed Type",
    "platform": "Platform Name",
    "self": {
        "platform": "Platform Name",
        "user_id": "Bot ID"     # Must match bot_id
    },
    "{platform}_raw": {...},       # Raw data (must)
    "{platform}_raw_type": "..."    # Raw type (must)
}

Converter Example

class MyPlatformConverter:
    def convert(self, raw_event):
        """Convert platform raw event to OneBot12 standard format"""
        if not isinstance(raw_event, dict):
            return None
        
        # Generate event ID
        event_id = raw_event.get("event_id") or str(uuid.uuid4())
        
        # Convert timestamp
        timestamp = raw_event.get("timestamp")
        if timestamp and timestamp > 10**12:
            timestamp = int(timestamp / 1000)
        else:
            timestamp = int(timestamp) if timestamp else int(time.time())
        
        # Convert event type
        event_type = self._convert_type(raw_event.get("type"))
        detail_type = self._convert_detail_type(raw_event)
        
        # Build standard event
        onebot_event = {
            "id": str(event_id),
            "time": timestamp,
            "type": event_type,
            "detail_type": detail_type,
            "platform": "myplatform",
            "self": {
                "platform": "myplatform",
                "user_id": str(raw_event.get("bot_id", ""))
            },
            "myplatform_raw": raw_event,
            "myplatform_raw_type": raw_event.get("type", "")
        }
        
        return onebot_event

Connection Management

WebSocket Connection

class MyAdapter(BaseAdapter):
    async def start(self):
        """Register WebSocket route"""
        router.register_websocket(
            module_name="myplatform",
            path="/ws",
            handler=self._ws_handler,
            auth_handler=self._auth_handler
        )
    
    async def _ws_handler(self, websocket):
        """WebSocket connection handler"""
        self.connection = websocket
        
        try:
            while True:
                data = await websocket.receive_text()
                onebot_event = self.convert(data)
                if onebot_event:
                    await self.adapter.emit(onebot_event)
        except WebSocketDisconnect:
            self.logger.info("Connection disconnected")
        finally:
            self.connection = None
    
    async def _auth_handler(self, websocket) -> bool:
        """WebSocket authentication"""
        token = websocket.query_params.get("token")
        return token == "valid_token"

WebHook Connection

class MyAdapter(BaseAdapter):
    async def start(self):
        """Register WebHook route"""
        router.register_http_route(
            module_name="myplatform",
            path="/webhook",
            handler=self._webhook_handler,
            methods=["POST"]
        )
    
    async def _webhook_handler(self, request):
        """WebHook request handler"""
        data = await request.json()
        onebot_event = self.convert(data)
        if onebot_event:
            await self.adapter.emit(onebot_event)
        return {"status": "ok"}

Route Information Query: Adapter-registered routes (HTTP, WebSocket, SSE) can be queried for complete connection addresses (including base_url + path) via sdk.adapter.get_connection_info(platform) and sdk.router.get_module_urls(module_name). See [Getting Started - Adapter Development - Connection Information and Route Discovery](getting-started.md#9-Connection Information and Route Discovery) and SSE Support.

API Response Standard

The framework provides make_response() and make_error() methods to construct standardized responses, eliminating the need to manually build response dictionaries.

Success Response

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

Manual Response Construction (Legacy Method Still Compatible)

async def call_api(self, endpoint: str, **params):
    return {
        "status": "ok",
        "retcode": 0,
        "data": {...},
        "message_id": "msg_id",
        "message": "",
        "myplatform_raw": raw_response
    }

Multi-Account Support

After declaring AccountConfigClass, the framework automatically manages multi-account loading, validation, and template generation:

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

@dataclass
class MyBotConfig(BotAccountConfig):
    bot_id: str = field(default="", metadata={"description": "Bot ID", "required": True})
    token: str = field(default="", metadata={"description": "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}: {account.bot_id}")
            await self._connect(name, account)
    
    async def call_api(self, endpoint: str, **params):
        account_id = params.pop("account_id", None)
        name, account = self._resolve_account(account_id)
        # Use account.token, account.bot_id, etc.

Account Configuration File

[MyAdapter.accounts.account1]
bot_id = "bot_001"
token = "token1"
enabled = true

[MyAdapter.accounts.account2]
bot_id = "bot_002"
token = "token2"
enabled = true

Specifying Account for Sending

# Use Using method to specify account
my_adapter = adapter.get("myplatform")

# Through event's self.user_id (recommended, most general)
await my_adapter.Send.Using(event["self"]["user_id"]).To("user", "123").Text("Hello")

# Through account name
await my_adapter.Send.Using("account1").To("user", "123").Text("Hello")

Relationship between self.user_id and Using

The framework's event reply mechanism automatically extracts account_id (if present) or user_id from the event's self field, as the Using parameter. Adapter developers need to ensure the Converter correctly sets self.user_id so that _resolve_account() can match the correct account.

Framework Internal Behavior:

# Framework extraction logic for bot_id
bot_id = self.get("self", {}).get("account_id", "") or self.get("self", {}).get("user_id", "")

# Only call Using if bot_id is non-empty
if bot_id:
    send_chain = send_chain.Using(bot_id)

Key Point: Even if the adapter uses a single Bot configuration, as long as the Converter correctly sets self.user_id, the framework will use it as the Using parameter. The adapter must ensure self.user_id matches the identifier field (e.g., bot_id) in AccountConfigClass so that _resolve_account() can match the correct account. If self.user_id is empty, the framework will not call Using, in which case call_api receives account_id as None, and _resolve_account(None) returns the first enabled account.

Error Handling

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()
                break
            except Exception as e:
                retry_count += 1
                if retry_count < max_retries:
                    wait_time = min(60 * (2 ** retry_count), 600)
                    self.logger.warning(f"Connection failed, retrying in {wait_time} seconds")
                    await asyncio.sleep(wait_time)
                else:
                    raise

API Error Handling

async def call_api(self, endpoint: str, **params):
    try:
        # Recommended to use SDK built-in client
        from ErisPulse.Core import client
        from ErisPulse.Core.Bases.errors import ClientError, ClientTimeoutError
        resp = await client.post(
            f"https://api.platform.com/{endpoint}",
            json=params,
            max_retries=2,
        )
        response = await resp.json()
        return self._standardize_response(response)
    except ClientTimeoutError:
        self.logger.error(f"Request timeout: {endpoint}")
        return self._error_response("Request timeout", 32000)
    except ClientError as e:
        self.logger.error(f"Network error: {e}")
        return self._error_response("Network request failed", 33000)
    except Exception as e:
        self.logger.error(f"Unknown error: {e}")
        return self._error_response(str(e), 34000)

Backward Compatibility: Adapters using aiohttp.ClientSession directly are unaffected and can still catch aiohttp.ClientError. Both methods can coexist. New code is recommended to use sdk.client with ErisPulse's exception system.

Bot Status Management

AdapterManager includes a built-in bot status tracking system, automatically maintaining the online status, active time, and metadata of all registered bots.

Automatic Discovery Mechanism

When the adapter emits an event via adapter.emit(), the framework automatically checks the event's self field:

# All events containing self field trigger automatic discovery
await self.adapter.emit({
    "type": "message",
    "platform": "myplatform",
    "self": {"platform": "myplatform", "user_id": "bot123"},
    # ...
})
# Bot "bot123" is automatically registered (if first appearance) and active time is updated

Meta Event Types

detail_type Description Framework Behavior
connect Bot connects Register bot and trigger adapter.bot.online lifecycle event
disconnect Bot disconnects Mark bot as offline and trigger adapter.bot.offline lifecycle event
heartbeat Bot heartbeat Update bot active time and metadata

Adapter Sending Meta Events

Use emit_meta() to send meta events in one line:

class MyAdapter(BaseAdapter):
    async def _on_bot_connect(self, bot_id: str):
        # Send connect event in one line
        await self.emit_meta("connect", bot_id, user_name="MyBot", nickname="My Bot")

    async def _on_bot_disconnect(self, bot_id: str):
        await self.emit_meta("disconnect", bot_id)

Also supports manual construction (legacy method still compatible):

await self.adapter.emit({
    "type": "meta",
    "detail_type": "connect",
    "platform": "myplatform",
    "self": {"platform": "myplatform", "user_id": bot_id}
})

Extended Information in self Field

Besides the required platform and user_id, the self field supports the following optional fields:

Field Description
user_name Bot username
nickname Bot nickname
avatar Bot avatar URL
account_id Multi-account identifier

Bot Status Query

from ErisPulse import sdk

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

# List all bots
all_bots = sdk.adapter.list_bots()

# List bots of specified platform
platform_bots = sdk.adapter.list_bots("myplatform")

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

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

Listening to Bot Lifecycle

from ErisPulse import sdk

@sdk.lifecycle.on("adapter.bot.online")
async def on_bot_online(data):
    platform = data.get("platform")
    bot_id = data.get("bot_id")
    sdk.logger.info(f"Bot online: {platform}/{bot_id}")

@sdk.lifecycle.on("adapter.bot.offline")
async def on_bot_offline(data):
    platform = data.get("platform")
    bot_id = data.get("bot_id")
    sdk.logger.info(f"Bot offline: {platform}/{bot_id}")