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

Matrix平台特性文件

MatrixAdapter 是基於 Matrix協議 建構的適配器，整合了 Matrix 協議的所有核心功能模組，提供統一的事件處理和訊息操作介面。

文件資訊

基本資訊

配置說明

MatrixAdapter 支援多帳戶配置，每個帳戶獨立配置 homeserver 和認證資訊。

# config.toml
# 帳戶1
[Matrix_Adapter.accounts.default]
homeserver = "https://matrix.org"          # Matrix 伺服器地址（必填）
access_token = "YOUR_ACCESS_TOKEN"          # 訪問權杖（與 user_id+password 二選一）
user_id = ""                                # Matrix 用戶 ID（如 @bot:matrix.org）
password = ""                               # Matrix 用戶密碼
auto_accept_invites = true                  # 是否自動接受房間邀請（可選，預設為 true）
enabled = true                              # 是否啟用（可選，預設為 true）

# 帳戶2
[Matrix_Adapter.accounts.bot2]
homeserver = "https://matrix.example.com"
access_token = "ANOTHER_TOKEN"
enabled = true

兼容舊配置：若檢測到舊的單帳戶 [Matrix_Adapter] 配置（含 access_token），會自動遷移為 accounts.default。

配置項說明（每個帳戶）：

認證方式：

v5 範式更新（4.2.0）

標準Api動作示例

from ErisPulse import sdk
matrix = sdk.adapter.get("matrix")
result = await matrix.Api.get_self_info()            # /account/whoami
result = await matrix.Api.get_group_info(room_id)    # m.room.name
result = await matrix.Api.get_group_list()           # /joined_rooms
await matrix.Api.delete_message(event_id)            # redact（登記表補全 room_id）

已對接平台能力

支援的消息傳送類型

所有傳送方法皆透過鏈式語法實現，例如：

from ErisPulse.Core import adapter
matrix = adapter.get("matrix")

await matrix.Send.To("group", room_id).Text("Hello World!")

支援的傳送類型包括：

鏈式修飾方法（可組合使用）

鏈式修飾方法返回 self，支援鏈式呼叫，必須在最終傳送方法前呼叫：

鏈式呼叫範例

# 基礎傳送
await matrix.Send.To("user", dm_room_id).Text("Hello")

# 回覆訊息
await matrix.Send.To("group", room_id).Reply("$event_id").Text("回覆訊息")

# @使用者
await matrix.Send.To("group", room_id).At("@user:matrix.org").Text("你好")

# @所有人
await matrix.Send.To("group", room_id).AtAll().Text("公告通知")

# 組合使用：回覆 + @
await matrix.Send.To("group", room_id).Reply("$event_id").At("@user:matrix.org").Text("複合訊息")

# 發送 HTML 訊息
await matrix.Send.To("group", room_id).Html("<h1>標題</h1><p>內容</p>", fallback="標題\n內容")

# 發送通知訊息
await matrix.Send.To("group", room_id).Notice("系統通知")

OneBot12 訊息支援

適配器支援傳送 OneBot12 格式的訊息，便於跨平台訊息相容：

# 發送 OneBot12 格式訊息
ob12_msg = [{"type": "text", "data": {"text": "Hello"}}]
await matrix.Send.To("user", dm_room_id).Raw_ob12(ob12_msg)

# 配合鏈式修飾
ob12_msg = [{"type": "text", "data": {"text": "回覆訊息"}}]
await matrix.Send.To("group", room_id).Reply("$event_id").Raw_ob12(ob12_msg)

# 複雜訊息
ob12_msg = [
    {"type": "text", "data": {"text": "看這張圖片："}},
    {"type": "image", "data": {"file": "https://example.com/image.png"}},
    {"type": "text", "data": {"text": "不錯吧？"}}
]
await matrix.Send.To("group", room_id).Raw_ob12(ob12_msg)

發送方法返回值

所有發送方法均返回一個 Task 對象，可以直接 await 獲取發送結果。返回結果遵循 ErisPulse 适配器标准化返回规范：

{
    "status": "ok",           // 執行狀態: "ok" 或 "failed"
    "retcode": 0,             // 返回碼
    "data": {...},            // 响应数据
    "message_id": "$event_id", // Matrix事件ID
    "message": "",            // 錯誤信息
    "matrix_raw": {...}       // 原始响应数据
}

錯誤碼說明

retcode 說明
0 成功
32000 請求超時或媒體上傳失敗
33000 API調用異常
34000 API返回了意外格式或業務錯誤

特有事件類型

需要 platform=="matrix" 檢測再使用本平台特性

核心差異點

  1. 去中心化架構：Matrix 是一個去中心化的通信協議，使用者ID格式為 @user:server.domain，房間ID格式為 !room_id:server.domain
  2. 房間概念：Matrix 不區分群聊和私聊，所有對話都是「房間」。適配器透過 DM（Direct Message）帳戶資料自動識別私聊房間
  3. Long Polling 同步：使用 /sync API 進行長輪詢以取得新事件，而非 WebSocket
  4. MXC URI：媒體檔案透過 mxc://server.domain/media_id 格式引用
  5. HTML 富文本：支援透過 formatted_body 發送 HTML 格式訊息
  6. 表情回應：支援訊息層級的表情回應（Reaction），有別於傳統的回覆訊息
  7. 訊息編輯：支援透過 m.replace 關係編輯已發送的訊息
  8. 訊息撤回：支援透過 m.room.redaction 撤回/刪除訊息

擴展欄位

特殊欄位範例

# 群組訊息
{
  "type": "message",
  "detail_type": "group",
  "user_id": "@user:matrix.org",
  "group_id": "!room_id:matrix.org",
  "matrix_room_id": "!room_id:matrix.org"
}

# 私聊訊息
{
  "type": "message",
  "detail_type": "private",
  "user_id": "@user:matrix.org",
  "matrix_room_id": "!dm_room_id:matrix.org"
}

# 表情回應
{
  "type": "notice",
  "detail_type": "matrix_reaction",
  "matrix_reaction_event_id": "$reacted_msg_id",
  "matrix_reaction_key": "👍"
}

# 訊息撤回
{
  "type": "notice",
  "detail_type": "matrix_redaction",
  "matrix_redacted_event_id": "$deleted_msg_id"
}

# 訊息編輯
{
  "type": "message",
  "detail_type": "group",
  "matrix_edit": True,
  "matrix_original_event_id": "$original_event_id"
}

# 線程訊息
{
  "type": "message",
  "detail_type": "group",
  "thread_id": "$thread_root_id"
}

訊息段類型

Matrix訊息根據 msgtype 自動轉換為對應的訊息段：

msgtype 轉換類型 說明
m.text text 文本訊息
m.notice text 通知訊息
m.emote text 動作訊息
m.image image 圖片訊息
m.audio voice 音訊訊息
m.video video 影片訊息
m.file file 檔案訊息
m.location location 位置訊息

訊息段結構範例：

// 文本訊息（帶HTML）
{
  "type": "text",
  "data": {
    "text": "純文本內容",
    "html": "<b>HTML內容</b>"
  }
}

// 圖片訊息
{
  "type": "image",
  "data": {
    "url": "mxc://matrix.org/abc123",
    "filename": "photo.png",
    "matrix_mxc": "mxc://matrix.org/abc123",
    "info": {
      "mimetype": "image/png",
      "w": 800,
      "h": 600,
      "size": 123456
    }
  }
}

// 位置訊息
{
  "type": "location",
  "data": {
    "latitude": 0.0,
    "longitude": 0.0,
    "matrix_geo_uri": "geo:39.9,116.4",
    "text": "北京市"
  }
}

Event Mixin 方法

MatrixAdapter 註冊了以下事件混入方法，可在事件處理中直接呼叫：

方法 回傳類型 說明
get_room_id() str 取得房間ID
get_matrix_event_type() str 取得原始Matrix事件類型
get_matrix_sender() str 取得原始發送者ID
get_reaction_key() str 取得回應表情
is_edited() bool 判斷訊息是否為編輯訊息
is_notice() bool 判斷訊息是否為 m.notice 類型 
@message.on_message()
async def handle_message(event):
    if event.get("platform") != "matrix":
        return

    room_id = event.get_room_id()
    event_type = event.get_matrix_event_type()
    sender = event.get_matrix_sender()
    is_edited = event.is_edited()
    is_notice = event.is_notice()

Sync API 連接

同步流程

  1. 使用 access_token 或 user_id + password 進行認證
  2. 調用 /_matrix/client/v3/account/whoami 以獲取 bot_user_id
  3. 發出 connect 元事件
  4. 執行初始同步（/_matrix/client/v3/sync?timeout=0）以獲取 next_batch token
  5. 發現 DM 房間（/_matrix/client/v3/user/{user_id}/account_data/m.direct）
  6. 開始 Long Polling 同步循環（/_matrix/client/v3/sync?since={next_batch}&timeout=30000）
  7. 處理每次同步返回的新事件並轉換發出

心跳機制

房間邀請

使用示例

處理群組訊息

from ErisPulse.Core.Event import message
from ErisPulse import sdk

matrix = sdk.adapter.get("matrix")

@message.on_message()
async def handle_group_msg(event):
    if event.get("platform") != "matrix":
        return
    if event.get("detail_type") != "group":
        return

    text = event.get_text()
    room_id = event.get("group_id")

    if text == "hello":
        await matrix.Send.To("group", room_id).Reply(
            event.get("message_id")
        ).Text("Hello!")

處理表情回應

from ErisPulse.Core.Event import notice

@notice.on_notice()
async def handle_reaction(event):
    if event.get("platform") != "matrix":
        return

    if event.get("detail_type") == "matrix_reaction":
        reaction_key = event.get("matrix_reaction_key")
        reacted_event_id = event.get("matrix_reaction_event_id")
        room_id = event.get_room_id()
        # 處理表情回應...

發送媒體訊息

# 發送圖片（URL）
await matrix.Send.To("group", room_id).Image("https://example.com/image.png")

# 發送圖片（MXC URI）
await matrix.Send.To("group", room_id).Image("mxc://matrix.org/abc123")

# 發送圖片（二進位數據）
with open("image.png", "rb") as f:
    image_bytes = f.read()
await matrix.Send.To("group", room_id).Image(image_bytes)

# 發送圖片（本地檔案路徑）
await matrix.Send.To("group", room_id).Image("/path/to/image.png")

# 發送檔案（帶檔案名）
await matrix.Send.To("group", room_id).File("/path/to/document.pdf", filename="文件.pdf")

處理訊息編輯

@message.on_message()
async def handle_edited_message(event):
    if event.get("platform") != "matrix":
        return

    if event.is_edited():
        original_id = event.get("matrix_original_event_id")
        # 處理編輯訊息...

監聽成員變更

@notice.on_notice()
async def handle_member_change(event):
    if event.get("platform") != "matrix":
        return

    detail_type = event.get("detail_type")

    if detail_type == "group_member_increase":
        user_id = event.get("user_id")
        nickname = event.get("user_nickname")
        print(f"用戶 {nickname} ({user_id}) 加入了房間")

    elif detail_type == "group_member_decrease":
        user_id = event.get("user_id")
        operator_id = event.get("operator_id")
        print(f"用戶 {user_id} 被移除，操作者: {operator_id}")