A Matrix protocol adapter built on the ErisPulse framework. It receives events via the Long Polling Sync API and integrates modules for multiple scenarios such as private chats and groups, providing a unified interface for event handling and message operations.
The MatrixAdapter is fully compatible with the OneBot12 standard event format, with some extension fields:
| Event Type | detail_type | Description |
|---|---|---|
| Message (Private) | private | A message sent by a user in a DM room |
| Message (Group) | group | A message sent by a user in a group room |
| Group Member Increase | group_member_increase | A user joined the room |
| Group Member Decrease | group_member_decrease | A user left or was banned |
| Member Info Update | matrix_member_update | Room member information changed |
| Matrix Reaction | matrix_reaction | A reaction to a message |
| Matrix Redaction | matrix_redaction | A message was redacted/deleted |
| Matrix Room Name Change | matrix_name | The room name changed |
| Matrix Room Topic Change | matrix_topic | The room topic changed |
| Matrix Room Avatar Change | matrix_avatar | The room avatar changed |
| Matrix Power Levels Change | matrix_power_levels | Room permissions changed |
from ErisPulse import sdk
matrix = sdk.adapter.get("matrix")
# Send a text message
await matrix.Send.To("group", room_id).Text("Hello World!")
# Send a message with @mention
await matrix.Send.To("group", room_id).At("@user:matrix.org").Text("Hello")
# Send a message mentioning everyone
await matrix.Send.To("group", room_id).AtAll().Text("Announcement")
# Send a reply message
await matrix.Send.To("group", room_id).Reply("$event_id").Text("Reply content")
# Send an image (URL)
await matrix.Send.To("group", room_id).Image("https://example.com/image.png")
# Send an image (MXC URI)
await matrix.Send.To("group", room_id).Image("mxc://matrix.org/abc123")
# Send an image (binary data)
with open("image.png", "rb") as f:
image_data = f.read()
await matrix.Send.To("group", room_id).Image(image_data)
# Send an image (local file path)
await matrix.Send.To("group", room_id).Image("/path/to/image.png")
# Send a notice message (m.notice)
await matrix.Send.To("group", room_id).Notice("System notice")
# Send an HTML-formatted message
await matrix.Send.To("group", room_id).Html("<b>bold</b> <i>italic</i>", fallback="bold italic")
# Send a file (with filename)
await matrix.Send.To("group", room_id).File("/path/to/file.pdf", filename="document.pdf")
# Combined usage: reply + mention
await matrix.Send.To("group", room_id).Reply("$event_id").At("@user:matrix.org").Text("Composite message")
# Send a OneBot12-format message using Raw_ob12
message = [
{"type": "text", "data": {"text": "First line"}},
{"type": "image", "data": {"file": "https://example.com/img.jpg"}},
{"type": "text", "data": {"text": "Second line"}}
]
await matrix.Send.To("group", room_id).Raw_ob12(message)A default configuration is generated automatically on first run. MatrixAdapter supports multi-account configuration.
# config.toml
# Account 1
[Matrix_Adapter.accounts.default]
homeserver = "https://matrix.org" # Matrix server address (required)
access_token = "YOUR_ACCESS_TOKEN" # Access token (one of access_token or user_id+password)
user_id = "" # Matrix user ID (e.g. @bot:matrix.org)
password = "" # Matrix user password
auto_accept_invites = true # Whether to auto-accept room invites (optional, default true)
enabled = true # Whether to enable (optional, default true)
# Account 2
[Matrix_Adapter.accounts.bot2]
homeserver = "https://matrix.example.com"
access_token = "ANOTHER_TOKEN"
enabled = trueBackward compatibility: If an old single-account
[Matrix_Adapter]configuration (containing access_token) is detected, it will be automatically migrated toaccounts.default.
Configuration fields (per account):
homeserver: Matrix server address (required), defaults tohttps://matrix.orgaccess_token: Access token, obtainable from a Matrix client (e.g. Element) settingsuser_id: Matrix user ID (e.g.@bot:matrix.org), used together withpasswordpassword: Matrix user password, used for auto-login to obtain an access_tokenauto_accept_invites: Whether to auto-accept room invites, defaults totrueenabled: Whether to enable this account (optional, default true)
Authentication methods:
- Method 1 (recommended): Provide
access_tokendirectly - Method 2: Provide
user_idandpassword; the adapter will call the login API to obtain a token automatically
See the Matrix platform features documentation for platform-specific capabilities, including the decentralized architecture, room concepts, Long Polling sync, MXC URIs, HTML rich text, reactions, message editing, and extension field descriptions.
For the detailed event conversion reference, see the conversion mapping documentation.
from ErisPulse.Core.Event import message, notice
@message.on_message()
async def handle_message(event):
if event["platform"] == "matrix":
detail_type = event["detail_type"]
if detail_type == "private":
# Handle private message
pass
elif detail_type == "group":
# Handle group message
pass
@notice.on_notice()
async def handle_notice(event):
if event["platform"] == "matrix":
detail_type = event["detail_type"]
if detail_type == "matrix_reaction":
# Handle reaction
reaction_key = event.get("matrix_reaction_key", "")
elif detail_type == "matrix_redaction":
# Handle message redaction
redacted_id = event.get("matrix_redacted_event_id", "")
elif detail_type == "group_member_increase":
# Handle member join
user_id = event.get("user_id", "")@sdk.adapter.on("message")
async def handle_message(event):
if event["platform"] == "matrix":
bot_id = event["self"]["user_id"]
print(f"Message from Bot: {bot_id}")
@sdk.adapter.on("notice")
async def handle_notice(event):
if event["platform"] == "matrix":
# Handle Matrix notice event
pass@message.on_message()
async def handle_message(event):
if event.get("platform") != "matrix":
return
room_id = event.get_room_id() # Get the room ID
event_type = event.get_matrix_event_type() # Get the raw Matrix event type
sender = event.get_matrix_sender() # Get the sender ID
is_edited = event.is_edited() # Whether the message is an edit
is_notice = event.is_notice() # Whether it is an m.notice type- Make sure all handlers are registered before calling
startup(). - Matrix is a decentralized protocol. User IDs use the format
@user:server.domain, and room IDs use the format!room_id:server.domain. - Matrix does not distinguish between group chats and private chats — all conversations are "rooms". The adapter identifies private chats via DM account data automatically.
- The adapter uses Long Polling (the
/syncAPI) to receive events, not WebSocket. - Media files are referenced via
mxc://URIs; the adapter supports automatic upload and download. - Call
shutdown()on program exit to ensure resources are released. - Auto-accepting room invites is supported (can be disabled via the
auto_accept_invitesconfig). - HTML-formatted messages are supported via the
.Html()method. - Message replies (
.Reply()) and user mentions (.At(),.AtAll()) are supported.
- ErisPulse Main Repository
- Matrix Protocol Specification
- Matrix Client-Server API
- Module Development Guide
MatrixAdapter 是基于 ErisPulse 架构的 Matrix 协议适配器，通过 Long Polling Sync API 接收事件，整合了私聊、群组等多种场景的功能模块，提供统一的事件处理和消息操作接口。
MatrixAdapter 适配器完全兼容 OneBot12 标准事件格式，并提供了一些扩展字段：
| 事件类型 | detail_type | 说明 |
|---|---|---|
| 消息事件（私聊） | private | DM房间中用户发送的消息 |
| 消息事件（群组） | group | 群组房间中用户发送的消息 |
| 群成员增加 | group_member_increase | 用户加入房间 |
| 群成员减少 | group_member_decrease | 用户离开/被封禁 |
| 成员信息更新 | matrix_member_update | 房间成员信息变更 |
| Matrix表情回应 | matrix_reaction | 消息表情回应 |
| Matrix消息撤回 | matrix_redaction | 消息被撤回/删除 |
| Matrix房间名称变更 | matrix_name | 房间名称变更 |
| Matrix房间话题变更 | matrix_topic | 房间话题变更 |
| Matrix房间头像变更 | matrix_avatar | 房间头像变更 |
| Matrix权限等级变更 | matrix_power_levels | 房间权限变更 |
from ErisPulse import sdk
matrix = sdk.adapter.get("matrix")
# 发送文本消息
await matrix.Send.To("group", room_id).Text("Hello World!")
# 发送带@的消息
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").Text("回复内容")
# 发送图片（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_data = f.read()
await matrix.Send.To("group", room_id).Image(image_data)
# 发送图片（本地文件路径）
await matrix.Send.To("group", room_id).Image("/path/to/image.png")
# 发送通知消息（m.notice）
await matrix.Send.To("group", room_id).Notice("系统通知")
# 发送HTML格式消息
await matrix.Send.To("group", room_id).Html("<b>加粗</b> <i>斜体</i>", fallback="加粗 斜体")
# 发送文件（带文件名）
await matrix.Send.To("group", room_id).File("/path/to/file.pdf", filename="文档.pdf")
# 组合使用：回复 + @
await matrix.Send.To("group", room_id).Reply("$event_id").At("@user:matrix.org").Text("复合消息")
# 使用 Raw_ob12 发送 OneBot12 格式消息
message = [
{"type": "text", "data": {"text": "第一行"}},
{"type": "image", "data": {"file": "https://example.com/img.jpg"}},
{"type": "text", "data": {"text": "第二行"}}
]
await matrix.Send.To("group", room_id).Raw_ob12(message)首次运行会自动生成默认配置。MatrixAdapter 支持多账户配置。
# 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。
配置项说明（每个账户）：
homeserver：Matrix服务器地址（必填），默认为https://matrix.orgaccess_token：访问令牌，可从Matrix客户端（如 Element）的设置中获取user_id：Matrix用户ID（如@bot:matrix.org），与password配合使用password：Matrix用户密码，用于自动登录获取 access_tokenauto_accept_invites：是否自动接受房间邀请，默认为trueenabled：是否启用该账户（可选，默认为true）
认证方式：
- 方式一（推荐）：直接提供
access_token - 方式二：提供
user_id和password，适配器会自动调用登录接口获取 token
请参考 Matrix平台特性文档 了解Matrix平台的特有功能，包括去中心化架构、房间概念、Long Polling同步、MXC URI、HTML富文本、表情回应、消息编辑、扩展字段说明等内容。
详细的事件转换对照请参考 转换对照文档。
from ErisPulse.Core.Event import message, notice
@message.on_message()
async def handle_message(event):
if event["platform"] == "matrix":
detail_type = event["detail_type"]
if detail_type == "private":
# 处理私聊消息
pass
elif detail_type == "group":
# 处理群组消息
pass
@notice.on_notice()
async def handle_notice(event):
if event["platform"] == "matrix":
detail_type = event["detail_type"]
if detail_type == "matrix_reaction":
# 处理表情回应
reaction_key = event.get("matrix_reaction_key", "")
elif detail_type == "matrix_redaction":
# 处理消息撤回
redacted_id = event.get("matrix_redacted_event_id", "")
elif detail_type == "group_member_increase":
# 处理成员加入
user_id = event.get("user_id", "")@sdk.adapter.on("message")
async def handle_message(event):
if event["platform"] == "matrix":
bot_id = event["self"]["user_id"]
print(f"消息来自Bot: {bot_id}")
@sdk.adapter.on("notice")
async def handle_notice(event):
if event["platform"] == "matrix":
# 处理Matrix通知事件
pass@message.on_message()
async def handle_message(event):
if event.get("platform") != "matrix":
return
room_id = event.get_room_id() # 获取房间ID
event_type = event.get_matrix_event_type() # 获取原始Matrix事件类型
sender = event.get_matrix_sender() # 获取发送者ID
is_edited = event.is_edited() # 是否为编辑消息
is_notice = event.is_notice() # 是否为 m.notice 类型- 确保在调用
startup()前完成所有处理器的注册 - Matrix 是去中心化协议，用户ID格式为
@user:server.domain，房间ID格式为!room_id:server.domain - Matrix 不区分群聊和私聊，所有会话都是"房间"。适配器通过 DM 账户数据自动识别私聊
- 适配器使用 Long Polling（
/syncAPI）获取事件，而非 WebSocket - 媒体文件通过
mxc://URI 引用，适配器支持自动上传和下载 - 程序退出时请调用
shutdown()确保资源释放 - 支持自动接受房间邀请（可通过
auto_accept_invites配置关闭） - 支持发送 HTML 格式消息，使用
.Html()方法 - 支持消息回复（
.Reply()）和用户提及（.At()、.AtAll()）