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

Getting Started with Adapter Development

This guide helps you get started with developing ErisPulse adapters to connect new messaging platforms.

Adapter Overview

What is an Adapter

An adapter serves as the bridge between ErisPulse and various messaging platforms, responsible for:

  1. Forward Conversion: Receiving platform events and converting them into the OneBot12 standard format (Converter)
  2. Reverse Conversion: Converting OneBot12 message segments into platform API calls (Raw_ob12)
  3. Managing the connection with the platform (WebSocket/WebHook)
  4. Providing a unified SendDSL message sending interface

Adapter Architecture

flowchart LR
    subgraph receive["Forward Conversion (Receiving)"]
        direction TB
        P1["Platform Event"] --> C1["Converter.convert()"] --> O1["OneBot12 Standard Event"] --> S1["Event System"] --> M1["Module Processing"]
    end
    subgraph send["Reverse Conversion (Sending)"]
        direction TB
        M2["Module Builds Message"] --> R1["Send.Raw_ob12()"] --> N1["Platform Native API Call"] --> R2["Standard Response Format"]
    end

Directory Structure

Standard adapter package structure:

MyAdapter/
├── pyproject.toml          # Project configuration
├── README.md               # Project description
├── LICENSE                 # License
└── MyAdapter/
    ├── __init__.py          # Package entry point
    ├── Core.py               # Adapter main class
    └── Converter.py          # Event converter

Quick Start

1. Create Project

mkdir MyAdapter && cd MyAdapter

2. Create pyproject.toml

[project]
name = "ErisPulse-MyAdapter"
version = "1.0.0"
description = "MyAdapter Platform Adapter"
readme = "README.md"
requires-python = ">=3.10"
license = { file = "LICENSE" }
authors = [ { name = "yourname", email = "[email protected]" } ]

dependencies = [
    "ErisPulse>=2.4.0"  # aiohttp is built-in in ErisPulse, usually no need to depend separately
]

[project.urls]
"homepage" = "https://github.com/yourname/MyAdapter"

[project.entry-points."erispulse.adapter"]
"MyAdapter" = "MyAdapter:MyAdapter"

3. Create Adapter Main Class

The framework provides ConfigClass / AccountConfigClass for declarative configuration management. The adapter only needs to declare the configuration class to automatically load, validate, and generate the configuration template.

# MyAdapter/Core.py
from dataclasses import dataclass, field
from ErisPulse.Core import BaseAdapter
from ErisPulse.Core.Bases import BaseConfig

@dataclass
class MyAdapterConfig(BaseConfig):
    """MyAdapter Configuration"""
    api_endpoint: str = field(
        default="https://api.example.com",
        metadata={
            "description": {"i18n": "my_adapter.api_endpoint", "default": "API Address"},
            "required": False,
            "ui": {"widget": "text", "group": "connection", "order": 1},
        },
    )
    token: str = field(
        default="",
        metadata={
            "description": {"i18n": "my_adapter.token", "default": "Platform Token"},
            "required": True,
            "secret": True,
            "ui": {"widget": "password", "group": "basic", "order": 2},
        },
    )

class MyAdapter(BaseAdapter):
    ConfigClass = MyAdapterConfig  # Declare the configuration class, the framework manages it automatically
    
    # No need to override __init__! The framework handles:
    # - self.sdk / self.logger are automatically set
    # - self.cfg reads the configuration in real-time
    # - self.Send / self.Request are automatically initialized
    
    def _setup_converter(self):
        from .Converter import MyPlatformConverter
        return MyPlatformConverter()

⚠️ About __init__: In the new version, BaseAdapter.__init__(self, sdk=None) automatically handles SDK references, logging initialization, and configuration loading. Most adapters do not need to override __init__. See init Notes.

⚠️ About super().__init__(): BaseAdapter.__init__() is responsible for creating Send and Request factory instances. If you forget to call it, all message sending and request operations will raise AttributeError. See init Notes.

4. Implement Required Methods

class MyAdapter(BaseAdapter):
    # ... __init__ code ...
    
    async def start(self):
        """Start the adapter (must implement)"""
        # Register WebSocket or WebHook routes
        router.register_websocket(
            module_name="myplatform",
            path="/ws",
            handler=self._ws_handler
        )
        self.logger.info("Adapter started")
    
    async def shutdown(self):
        """Shutdown the adapter (must implement)"""
        router.unregister_websocket(
            module_name="myplatform",
            path="/ws"
        )
        # Clean up connections and resources
        self.logger.info("Adapter shutdown")
    
    async def call_api(self, endpoint: str, **params):
        """Call platform API (must implement)"""
        raise NotImplementedError("call_api needs to be implemented")

Actively Sending Meta Events

The adapter should actively send meta events to let the framework track the Bot's online status. Use emit_meta() to complete this in one line:

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

        # Bot online
        await self.emit_meta("connect", bot_id, user_name="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)

For detailed Bot status management and meta event explanations, see Adapter Best Practices - Bot Status Management and Meta Events.

5. Implement Send Class

At/AtAll/Reply decorators are already implemented by the framework's SendDSL base class. The adapter only needs to implement Raw_ob12 and specific send methods.

The framework provides two key helper methods:

import asyncio

class MyAdapter(BaseAdapter):
    # ... other code ...

    class Send(BaseAdapter.Send):

        def Raw_ob12(self, message, **kwargs):
            """
            Send OneBot12 formatted message (must implement)

            Use _apply_modifiers to automatically merge decorator states,
            Use send_context to get send context.
            """
            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 SendDSL base class,
        # Defaultly delegated to Raw_ob12, no need to reimplement.
        # If platform-specific logic is needed, override individual methods:
        # def Text(self, text: str):
        #     return self.Raw_ob12([{"type": "text", "data": {"text": text}}])

Media Send Method Implementation Points (Image/Video/File):

__getattr__ Magic Method:

Raw_ob12 Method:

6. Implement Converter

# MyAdapter/Converter.py
import time
import uuid

class MyPlatformConverter:
    def convert(self, raw_event):
        """Convert platform-native events to OneBot12 standard format"""
        if not isinstance(raw_event, dict):
            return None
        
        onebot_event = {
            "id": str(raw_event.get("event_id", uuid.uuid4())),
            "time": int(time.time()),
            "type": self._convert_event_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,
            "myplatform_raw_type": raw_event.get("type", "")
        }
        
        return onebot_event
    
    def _convert_event_type(self, event_type):
        """Convert event type"""
        type_map = {
            "message": "message",
            "notice": "notice"
        }
        return type_map.get(event_type, "unknown")
    
    def _convert_detail_type(self, raw_event):
        """Convert detail type"""
        return "private"  # Simplified example

7. Implement Request Class (Request Operations)

If your platform supports friend requests, group invitations, and other requests that require Bot decisions, you can implement the Request inner class:

from ErisPulse.Core import BaseAdapter, RequestDSL

class MyAdapter(BaseAdapter):
    # ... Send and other code ...

    class Request(RequestDSL):
        """Request operation implementation (friend requests, group invitations, etc.)"""

        def accept(self, **kwargs):
            """Accept request"""
            async def _do():
                result = await self._adapter.call_api(
                    endpoint="/set_request",
                    request_id=self._request_id,
                    approve=True,
                    **kwargs,
                )
                return {
                    "status": "ok" if result.get("code") == 0 else "failed",
                    "retcode": result.get("code", 0),
                    "data": None,
                    "message_id": "",
                    "message": result.get("message", ""),
                }
            return self._create_task(_do())

        def reject(self, **kwargs):
            """Reject request"""
            async def _do():
                result = await self._adapter.call_api(
                    endpoint="/set_request",
                    request_id=self._request_id,
                    approve=False,
                    **kwargs,
                )
                return {
                    "status": "ok" if result.get("code") == 0 else "failed",
                    "retcode": result.get("code", 0),
                    "data": None,
                    "message_id": "",
                    "message": result.get("message", ""),
                }
            return self._create_task(_do())

Module developers' usage:

from ErisPulse.Core.Event import request

@request.on_friend_request()
async def handle_friend_request(event):
    # Use Event convenience methods
    await event.approve()
    # Or operate directly through the adapter
    await adapter.myplatform.Request("req_id").accept()

If the platform does not support request operations, you can omit implementing the Request inner class. The base class defaults to returning retcode=10002 (unsupported operation). See Request Operation Specification.

8. Create Package Entry

# MyAdapter/__init__.py
from .Core import MyAdapter

Dependency Declaration (Optional, 2.8.0+)

Adapters can declare dependencies on other adapters or modules to enable adapter interconnection and optional features:

from typing import ClassVar

class MyAdapter(BaseAdapter):
    # Hard dependency: Adapter startup is skipped if dependency is missing (warning + status=skipped-dependency event)
    depends: ClassVar[dict] = {
        "adapters": ["onebot11"],   # Dependent adapters (by platform name)
        "modules": ["TranslateEngine"],  # Dependent modules (by registration name)
    }
    # Soft dependency: Missing dependency does not affect startup; callbacks are received when the module is loaded/unloaded (optional feature mode)
    optional_modules: ClassVar[list] = ["TranslateEngine"]

async def on_dependency_ready(self, module_name):
    """Soft dependency module is ready: enable corresponding optional features"""
    if module_name == "TranslateEngine":
        self._translate = self.sdk.TranslateEngine

async def on_dependency_lost(self, module_name):
    """Soft dependency module is lost: degrade features"""
    if module_name == "TranslateEngine":
        self._translate = None

Note

This feature requires ErisPulse 2.8.0+.

__init__ Notes

There are three levels in adapter development where __init__ may be overridden. Below are the correct practices for each level.

1. BaseAdapter Level (Most Cases Do Not Require Overriding)

BaseAdapter.__init__(self, sdk=None) is responsible for creating Send / Request factory instances and automatically performs the following tasks:

In most cases, there is no need to override __init__. Just declare ConfigClass:

class MyAdapter(BaseAdapter):
    ConfigClass = MyAdapterConfig  # After declaration, the framework automatically manages configurations
    
    async def start(self):
        cfg = self.cfg  # Type-safe, real-time read
        ...

If custom initialization is indeed required, call super().__init__(sdk):

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

2. Send Inner Class (Most Cases Do Not Require Overriding)

SendDSL.__init__ is responsible for state transfer in chain calls (target type, target ID, account, etc.). In most cases, you only need to override methods (Raw_ob12, Text, etc.), not __init__.

If overriding is necessary (for example, initializing platform-specific states), all parameters must be passed through:

class MyAdapter(BaseAdapter):
    class Send(BaseAdapter.Send):
        # Parameters: adapter, target_type, target_id, account_id
        def __init__(self, adapter, target_type=None, target_id=None, account_id=None):
            super().__init__(adapter, target_type, target_id, account_id)  # ← Must pass through
            self._my_state = None  # Platform-specific initialization

Why must it be passed through? Each step in the chain call creates a new instance via self.__class__(...):

adapter.Send.To("user", "123")               # → Send(adapter, "user", "123", None)
adapter.Send.To("user", "123").Using("bot1")  # → Send(adapter, "user", "123", "bot1")

If the __init__ signature does not match or super() is not called, the chain call will break.

3. Request Inner Class (Most Cases Do Not Require Overriding)

Same as Send. The parameters are adapter, request_id, account_id:

class MyAdapter(BaseAdapter):
    class Request(RequestDSL):
        # Parameters: adapter, request_id, account_id
        def __init__(self, adapter, request_id=None, account_id=None):
            super().__init__(adapter, request_id, account_id)  # ← Must pass through
            self._my_state = None  # Platform-specific initialization

Summary

Level When to Override Must Do
BaseAdapter When custom initialization logic is needed super().__init__(sdk) (pass in sdk parameter)
Send Inner Class When initializing send-related states is needed super().__init__(adapter, target_type, target_id, account_id)
Request Inner Class When initializing request-related states is needed super().__init__(adapter, request_id, account_id)
All Three Levels Most Cases Just declare ConfigClass, do not touch __init__

9. Connection Information and Route Discovery

After the adapter registers routes, the framework records all route information. Users can use the following API to view the adapter's connection address:

from ErisPulse import sdk

# Get complete connection information for the adapter
info = sdk.adapter.get_connection_info("myplatform")
# {
#   "platform": "myplatform",
#   "status": "started",
#   "connection": {
#     "base_url": "http://localhost:8080",
#     "http_routes": [
#       {"path": "/myplatform/webhook", "method": "POST",
#        "url": "http://localhost:8080/myplatform/webhook"}
#     ],
#     "websocket_routes": [
#       {"path": "/myplatform/ws",
#        "url": "ws://localhost:8080/myplatform/ws"}
#     ]
#   }
# }

# List all namespaces (adapters/modules) routes
namespaces = sdk.router.list_namespaces()
# {"myplatform": {"http": ["/myplatform/webhook"], "websocket": ["/myplatform/ws"]}}

# Get complete connection URLs for the namespace
urls = sdk.router.get_module_urls("myplatform")
# {"base_url": "http://localhost:8080", "http": [...], "websocket": [...]}

# Get detailed route information for the namespace
routes = sdk.router.get_module_routes("myplatform")
# {"http": [{"path": "/myplatform/webhook", "methods": ["POST"]}],
#  "websocket": [{"path": "/myplatform/ws", "auth": false}]}

Tip: The information returned by get_connection_info() is suitable for displaying to users (such as WebUI), helping users configure the callback address or WebSocket connection address on the platform side. The module_name registered during route registration must exactly match the platform name registered by the adapter in ErisPulse, otherwise route discovery will not be correctly associated.

10. SSE (Server-Sent Events) Support

ErisPulse has built-in, server-agnostic SSE support. Modules and adapters can register SSE endpoints using @sdk.router.sse().

Basic Usage

import asyncio
from ErisPulse import sdk

@sdk.router.sse("MyModule", "/events")
async def event_stream(sse):
    """Push SSE events"""
    count = 0
    while not sse.closed:
        await sse.send({"count": count}, event="update")
        count += 1
        await asyncio.sleep(1)

Using Request Parameters

The handler can declare a request parameter to access client request information:

@sdk.router.sse("MyModule", "/events")
async def event_stream(request, sse):
    token = request.query_params.get("token")
    if not validate_token(token):
        await sse.close()
        return

    while not sse.closed:
        data = await fetch_data(token)
        await sse.send(data)
        await asyncio.sleep(5)

SseEmitter API

Method Description
sse.send(data, event=None, id=None, retry=None) Send an SSE event. Non-string data is automatically JSON serialized
sse.close() Gracefully close the SSE connection (safe to call multiple times)
sse.closed Whether the connection is closed
sse.request The underlying request object (can be used to read query params, headers)

Using in RouteGroup

api = sdk.router.group("MyModule", "/api", version="1")

@api.sse("/events")
async def events(sse):
    await sse.send({"msg": "hello"})

Route Discovery

SSE routes will automatically appear in the route discovery API:

# list_namespaces will include the "sse" key
sdk.router.list_namespaces()
# {"MyModule": {"http": [...], "websocket": [...], "sse": ["/MyModule/events"]}}

# get_module_routes will mark streaming: true
sdk.router.get_module_routes("MyModule")
# {"http": [...], "websocket": [...], "sse": [{"path": "/MyModule/events", "streaming": true}]}

# get_module_urls will generate complete URLs
sdk.router.get_module_urls("MyModule")
# {"sse": [{"path": "/MyModule/events", "url": "http://localhost:8080/MyModule/events"}]}

Server-Agnostic Design: SseEmitter is decoupled from the underlying HTTP framework through callbacks. The framework provides register_sse() and @sse decorators as unified registration entry points, allowing adapters to implement SSE endpoints without directly depending on any underlying HTTP framework.

Next Steps