適配器開發入門

本指南幫助你開始開發 ErisPulse 適配器，連接新的訊息平台。

適配器簡介

什麼是適配器

適配器是 ErisPulse 與各個訊息平台之間的橋樑，負責：

1. 正向轉換：接收平台事件並轉換為 OneBot12 標準格式（Converter）
2. 反向轉換：將 OneBot12 訊息段轉換為平台 API 調用（Raw_ob12）
3. 管理與平台的連接（WebSocket/WebHook）
4. 提供統一的 SendDSL 訊息發送介面

適配器架構

flowchart LR
    subgraph receive["正向轉換（接收）"]
        direction TB
        P1["平台事件"] --> C1["Converter.convert()"] --> O1["OneBot12 標準事件"] --> S1["事件系統"] --> M1["模組處理"]
    end
    subgraph send["反向轉換（發送）"]
        direction TB
        M2["模組建構訊息"] --> R1["Send.Raw_ob12()"] --> N1["平台原生 API 調用"] --> R2["標準回應格式"]
    end

目錄結構

標準的適配器包結構：

MyAdapter/
├── pyproject.toml          # 項目配置
├── README.md               # 項目說明
├── LICENSE                 # 授權
└── MyAdapter/
    ├── __init__.py          # 包入口
    ├── Core.py               # 適配器主類
    └── Converter.py          # 事件轉換器

快速開始

1. 建立項目

mkdir MyAdapter && cd MyAdapter

2. 建立 pyproject.toml

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

dependencies = [
    "ErisPulse>=2.4.0"  # ErisPulse 已內建 aiohttp，通常無需單獨依賴
]

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

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

3. 建立適配器主類

框架提供了 ConfigClass / AccountConfigClass 宣告式配置管理，適配器只需宣告配置類即可自動載入、驗證和產生配置範本。

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

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

class MyAdapter(BaseAdapter):
    ConfigClass = MyAdapterConfig  # 宣告配置類，框架自動管理
    
    # 不需要覆寫 __init__！框架自動處理：
    # - self.sdk / self.logger 自動設定
    # - self.cfg 實時讀取配置
    # - self.Send / self.Request 自動初始化
    
    def _setup_converter(self):
        from .Converter import MyPlatformConverter
        return MyPlatformConverter()

⚠️ **關於 __init__：新版本中 BaseAdapter.__init__(self, sdk=None) 會自動處理 SDK 引用、日誌初始化和配置載入。大多數適配器不再需要覆寫 __init__**。詳見 init 注意事項。

⚠️ **關於 super().__init__()**：BaseAdapter.__init__() 負責建立 Send 和 Request 工廠實例。如果忘記呼叫，所有訊息發送和請求操作都會報 AttributeError。詳見 init 注意事項。

4. 實現必需方法

class MyAdapter(BaseAdapter):
    # ... __init__ 代碼 ...
    
    async def start(self):
        """啟動適配器（必須實現）"""
        # 註冊 WebSocket 或 WebHook 路由
        router.register_websocket(
            module_name="myplatform",
            path="/ws",
            handler=self._ws_handler
        )
        self.logger.info("適配器已啟動")
    
    async def shutdown(self):
        """關閉適配器（必須實現）"""
        router.unregister_websocket(
            module_name="myplatform",
            path="/ws"
        )
        # 清理連接和資源
        self.logger.info("適配器已關閉")
    
    async def call_api(self, endpoint: str, **params):
        """呼叫平台 API（必須實現）"""
        raise NotImplementedError("需要實現 call_api")

主動發送 Meta 事件

適配器應主動發送 meta 事件，讓框架追蹤 Bot 的線上狀態。使用 emit_meta() 一行即可完成：

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

        # Bot 上線
        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 下線
            await self.emit_meta("disconnect", bot_id)

詳細的 Bot 狀態管理和 Meta 事件說明請參閱 適配器最佳實踐 - Bot 狀態管理與-meta-事件。

5. 實現 Send 類

At/AtAll/Reply 修飾器已由框架 SendDSL 基類內建實現，適配器只需實現 Raw_ob12 和具體的發送方法即可。

框架提供兩個關鍵輔助方法：

• self._apply_modifiers(message) — 自動合併 At/AtAll/Reply 修飾器到訊息段
• self.send_context — 取得發送上下文字典（target_type、target_id、account_id）

import asyncio

class MyAdapter(BaseAdapter):
    # ... 其他代碼 ...

    class Send(BaseAdapter.Send):

        def Raw_ob12(self, message, **kwargs):
            """
            發送 OneBot12 格式訊息（必須實現）

            使用 _apply_modifiers 自動合併修飾器狀態，
            使用 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 已從 SendDSL 基類繼承，
        # 預設委派給 Raw_ob12，無需重複實現。
        # 如需平台特定邏輯，可覆蓋單個方法：
        # def Text(self, text: str):
        #     return self.Raw_ob12([{"type": "text", "data": {"text": text}}])

媒體類發送方法（Image/Video/File）實現要點：

• 基類的預設實現會將 file 參數封裝為 OneBot12 訊息段傳給 Raw_ob12，適配器需在 Raw_ob12 中處理下載/上傳
• file 參數應同時支援 bytes 二進位資料和 str URL 兩種型別
• 當傳入 URL 時，需先下載檔案再上傳到平台
• 平台通常需要先呼叫上傳介面獲取檔案標識，再呼叫發送介面

__getattr__ 魔術方法：

• 實現方法名大小寫不敏感（Text、text、TEXT 都能呼叫）
• 未定義的方法應回傳提示資訊而非報錯

Raw_ob12 方法：

• 將 OneBot12 標準訊息格式轉換為平台格式發送
• 使用 self._apply_modifiers(message) 自動處理 At/AtAll/Reply 修飾器
• 使用 **self.send_context 傳遞發送目標資訊和帳號資訊

6. 實現轉換器

# MyAdapter/Converter.py
import time
import uuid

class MyPlatformConverter:
    def convert(self, raw_event):
        """將平台原生事件轉換為 OneBot12 標準格式"""
        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):
        """轉換事件類型"""
        type_map = {
            "message": "message",
            "notice": "notice"
        }
        return type_map.get(event_type, "unknown")
    
    def _convert_detail_type(self, raw_event):
        """轉換詳細類型"""
        return "private"  # 簡化範例

7. 實現 Request 類（請求操作）

如果你的平台支援好友請求、群邀請等需要 Bot 做出決策的請求，可以實現 Request 內部類：

from ErisPulse.Core import BaseAdapter, RequestDSL

class MyAdapter(BaseAdapter):
    # ... Send 和其他代碼 ...

    class Request(RequestDSL):
        """請求操作實現（好友請求、群邀請等）"""

        def accept(self, **kwargs):
            """同意請求"""
            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):
            """拒絕請求"""
            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())

模組開發者使用方式：

from ErisPulse.Core.Event import request

@request.on_friend_request()
async def handle_friend_request(event):
    # 透過 Event 便捷方法
    await event.approve()
    # 或透過適配器直接操作
    await adapter.myplatform.Request("req_id").accept()

如果平台不支援請求操作，可以不實現 Request 內部類。基類預設回傳 retcode=10002（不支援的操作）。詳見 請求操作規範。

8. 建立包入口

# MyAdapter/__init__.py
from .Core import MyAdapter

依賴宣告（可選，2.8.0+）

適配器可以宣告對其他適配器或模組的依賴，實現適配器間聯動與可選功能：

from typing import ClassVar

class MyAdapter(BaseAdapter):
    # 硬依賴：缺失時跳過啟動（警告 + status=skipped-dependency 事件）
    depends: ClassVar[dict] = {
        "adapters": ["onebot11"],   # 依賴的適配器（按平台名）
        "modules": ["TranslateEngine"],  # 依賴的模組（按註冊名）
    }
    # 軟依賴：缺失不影響啟動；模組載入/卸載時收到回調（可選功能模式）
    optional_modules: ClassVar[list] = ["TranslateEngine"]

• 啟動順序：宣告了模組硬依賴的適配器會延遲到模組初始化完成後再啟動
• 軟依賴通知：optional_modules（或模組硬依賴）中的模組被載入時呼叫 on_dependency_ready(module_name)；被卸載時呼叫 on_dependency_lost(module_name)（預設空實作，可覆寫）——覆寫晚載入與熱重載場景：

async def on_dependency_ready(self, module_name):
    """軟依賴模組就緒：啟用對應可選功能"""
    if module_name == "TranslateEngine":
        self._translate = self.sdk.TranslateEngine

async def on_dependency_lost(self, module_name):
    """軟依賴模組遺失：降級功能"""
    if module_name == "TranslateEngine":
        self._translate = None

__init__ 注意事項

適配器開發中有三個層面可能涉及 __init__ 重寫。以下是每個層面的正確做法。

1. BaseAdapter 層（大多數情況不需要重寫）

BaseAdapter.__init__(self, sdk=None) 負責建立 Send / Request 工廠實例，並自動完成以下工作：

• 接受 sdk 參數並設定 self.sdk、self.logger
• 如果宣告了 ConfigClass，可透過 self.cfg 實時讀取全域配置
• 如果宣告了 AccountConfigClass，可透過 self.accounts 實時讀取多帳號配置

**大多數情況下不需要覆寫 __init__**，只需宣告 ConfigClass 即可：

class MyAdapter(BaseAdapter):
    ConfigClass = MyAdapterConfig  # 宣告後框架自動管理配置
    
    async def start(self):
        cfg = self.cfg  # 類型安全，實時讀取
        ...

如果確實需要自訂初始化，呼叫 super().__init__(sdk) 即可：

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

2. Send 內部類（大多數情況不需要重寫）

SendDSL.__init__ 負責鏈式呼叫的狀態傳遞（目標類型、目標ID、帳號等）。大多數情況下，你只需要重寫方法（Raw_ob12、Text 等），不需要重寫 __init__。

如果確實需要（比如初始化平台特有的狀態），必須透傳所有參數：

class MyAdapter(BaseAdapter):
    class Send(BaseAdapter.Send):
        # 參數：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)  # ← 必須透傳
            self._my_state = None  # 平台特有初始化

為什麼必須透傳？ 鏈式呼叫的每一步都透過 self.__class__(...) 建立新實例：

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

如果 __init__ 簽名不匹配或沒呼叫 super()，鏈式呼叫就會中斷。

3. Request 內部類（大多數情況不需要重寫）

與 Send 同理。參數為 adapter, request_id, account_id：

class MyAdapter(BaseAdapter):
    class Request(RequestDSL):
        # 參數：adapter, request_id, account_id
        def __init__(self, adapter, request_id=None, account_id=None):
            super().__init__(adapter, request_id, account_id)  # ← 必須透傳
            self._my_state = None  # 平台特有初始化

總結

9. 連接資訊與路由發現

適配器註冊路由後，框架會記錄所有路由資訊。使用者可以透過以下 API 查看適配器的連接位址：

from ErisPulse import sdk

# 取得適配器完整連接資訊
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"}
#     ]
#   }
# }

# 列出所有命名空間（適配器/模組）的路由
namespaces = sdk.router.list_namespaces()
# {"myplatform": {"http": ["/myplatform/webhook"], "websocket": ["/myplatform/ws"]}}

# 取得命名空間的完整連接 URL
urls = sdk.router.get_module_urls("myplatform")
# {"base_url": "http://localhost:8080", "http": [...], "websocket": [...]}

# 取得命名空間的詳細路由資訊
routes = sdk.router.get_module_routes("myplatform")
# {"http": [{"path": "/myplatform/webhook", "methods": ["POST"]}],
#  "websocket": [{"path": "/myplatform/ws", "auth": false}]}

提示：get_connection_info() 回傳的資訊適合展示給使用者（如 WebUI），幫助使用者設定平台側的回呼位址或 WebSocket 連接位址。路由註冊時的 module_name 必須與適配器在 ErisPulse 中註冊的 platform 名稱完全一致，否則路由發現將無法正確關聯。

10. SSE (Server-Sent Events) 支援

ErisPulse 內建了伺服器無關的 SSE 支援，模組和適配器可以透過 @sdk.router.sse() 註冊 SSE 端點。

基本使用

import asyncio
from ErisPulse import sdk

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

使用請求參數

處理器可以宣告 request 參數來存取客戶端請求資訊：

@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

在 RouteGroup 中使用

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

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

路由發現

SSE 路由會自動出現在路由發現 API 中：

# list_namespaces 會包含 "sse" 鍵
sdk.router.list_namespaces()
# {"MyModule": {"http": [...], "websocket": [...], "sse": ["/MyModule/events"]}}

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

# get_module_urls 會產生完整 URL
sdk.router.get_module_urls("MyModule")
# {"sse": [{"path": "/MyModule/events", "url": "http://localhost:8080/MyModule/events"}]}

伺服器無關設計：SseEmitter 透過回呼與底層 HTTP 框架解耦。框架提供了 register_sse() 和 @sse 裝飾器作為統一的註冊入口，適配器無需直接依賴任何底層 HTTP 框架即可實作 SSE 端點。

下一步

• 適配器核心概念 - 了解適配器架構
• SendDSL 詳解 - 學習訊息發送
• 轉換器實現 - 了解事件轉換
• 適配器最佳實踐 - 開發高品質適配器