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

アダプター開発入門

このガイドでは、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 を実装する必要があります")

メタイベントの送信

アダプターは、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 の状態管理とメタイベントの詳細については、アダプターのベストプラクティス - Bot 状態管理とメタイベント を参照。

5. Send クラスの実装

At/AtAll/Reply 修飾子は、フレームワークの SendDSL 基底クラスに内蔵されているため、アダプターは Raw_ob12 と具体的な送信メソッドを実装するだけでよい。

フレームワークは以下の重要な補助メソッドを提供する：

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）の実装のポイント：

__getattr__ マジックメソッド：

Raw_ob12 メソッド：

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"]

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

Note

この機能は ErisPulse 2.8.0以降が必要です。

__init__ の注意点

アダプター開発において、__init__ のオーバーライドは3つのレベルで考慮する必要があります。それぞれの正しい実装方法を以下に示します。

1. BaseAdapter 層（通常はオーバーライド不要）

BaseAdapter.__init__(self, sdk=None) は Send / Request ファクトリのインスタンスを作成し、以下の自動処理を行います：

通常は __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  # プラットフォーム特有の初期化

まとめ

層 什么时候重写 必须做的事
BaseAdapter 需要自定义初始化逻辑时 super().__init__(sdk) （传入 sdk 参数）
Send 内部类 需要初始化发送相关状态时 super().__init__(adapter, target_type, target_id, account_id)
Request 内部类 需要初始化请求相关状态时 super().__init__(adapter, request_id, account_id)
三个层面 大多数情况 声明 ConfigClass 即可，不碰 __init__

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

メソッド 説明
sse.send(data, event=None, id=None, retry=None) SSE イベントを送信。str 以外の data は自動的に JSON シリアライズされる
sse.close() SSE 接続を優雅に閉じる（安全に呼び出せる、複数回呼び出せる）
sse.closed 接続が閉じられているかどうか
sse.request ベースとなるリクエストオブジェクト（クエリパラメータ、headers を読み取るのに使用可能）

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 エンドポイントを実装できます。

次のステップ