ErisPulse API 动作标准
本文档定义 ErisPulse 适配器中 OneBot12 标准 API 动作的统一接口规范,使模块开发者可以面向标准接口编程,由适配器负责映射到平台原生 API。
覆盖范围:OneBot12 标准动作中,
ApiDSL提供用户 / 群组 / 频道(Guild)/ 消息管理 / 元(Meta)常规接口的强类型方法(send_message由SendDSL.Raw_ob12承担)。文件资源动作(upload_file/get_file/ 分片)仅作 降级透传保留,见 §3.5 说明。平台扩展动作经Api.call("prefix.action", ...)逃生舱调用。动作参数与返回结构以 OneBot12 规范(仓库内onebot/specs/interface/)为准。
1. 设计背景
在 ErisPulse 中,消息段(消息收发)和事件格式已经完全遵循 OneBot12 标准,但 API 动作调用(如获取用户信息、获取群列表、撤回消息等)此前未统一——模块开发者必须为每个平台写不同的 call_api 调用。
ApiDSL 通过提供强类型的标准动作方法,解决这一问题:
模块代码(跨平台统一) 适配器实现(平台特定)
───────────────── ──────────────────
adapter.Api.get_user_info("123") → 适配器 call_api / 覆盖
adapter.Api.get_group_list() → 适配器 call_api / 覆盖
adapter.Api.delete_message("id") → 适配器 call_api / 覆盖
2. 三层 DSL 并行结构
ErisPulse 适配器有三个并行的 DSL 内部类,各司其职:
BaseAdapter
├── Send(SendDSL) ← 消息发送(Text/Image/Raw_ob12)
├── Request(RequestDSL) ← 请求操作(accept/reject)
└── Api(ApiDSL) ← 标准 API 动作(用户/群组/频道/消息管理/文件/元)★
| DSL | 职责 | 方法风格 | 返回值 |
|---|---|---|---|
Send |
发送消息 | 链式 + asyncio.Task |
标准响应 |
Request |
处理请求事件 | asyncio.Task |
标准响应 |
Api |
查询/管理操作 | async 方法 |
标准响应 |
3. 标准动作列表
3.1 用户相关
| 方法 | OB12 动作 | 参数 | data 返回 |
|---|---|---|---|
get_self_info() |
get_self_info |
无 | user_id, user_name, user_displayname |
get_user_info(user_id) |
get_user_info |
user_id: str |
user_id, user_name, user_displayname, user_remark |
get_friend_list() |
get_friend_list |
无 | list[get_user_info 响应] |
3.2 群组相关
| 方法 | OB12 动作 | 参数 | data 返回 |
|---|---|---|---|
get_group_info(group_id) |
get_group_info |
group_id: str |
group_id, group_name |
get_group_list() |
get_group_list |
无 | list[get_group_info 响应] |
get_group_member_info(group_id, user_id) |
get_group_member_info |
group_id: str, user_id: str |
user_id, user_name, user_displayname |
get_group_member_list(group_id) |
get_group_member_list |
group_id: str |
list[get_group_member_info 响应] |
set_group_name(group_id, group_name) |
set_group_name |
group_id: str, group_name: str |
无 |
leave_group(group_id) |
leave_group |
group_id: str |
无 |
3.3 消息管理
| 方法 | OB12 动作 | 参数 | 说明 |
|---|---|---|---|
delete_message(message_id) |
delete_message |
message_id: str |
撤回/删除消息 |
发送消息(
send_message)由SendDSL的Raw_ob12处理,不在ApiDSL中重复。
3.4 频道(Guild)相关
OneBot12 频道体系分两级:频道(guild) 与 子频道(channel)。
| 方法 | OB12 动作 | 参数 | data 返回 |
|---|---|---|---|
get_guild_info(guild_id) |
get_guild_info |
guild_id: str |
guild_id, guild_name |
get_guild_list() |
get_guild_list |
无 | list[get_guild_info 响应] |
set_guild_name(guild_id, guild_name) |
set_guild_name |
guild_id: str, guild_name: str |
无 |
get_guild_member_info(guild_id, user_id) |
get_guild_member_info |
guild_id: str, user_id: str |
user_id, user_name, user_displayname |
get_guild_member_list(guild_id) |
get_guild_member_list |
guild_id: str |
list[get_guild_member_info 响应] |
leave_guild(guild_id) |
leave_guild |
guild_id: str |
无 |
get_channel_info(guild_id, channel_id) |
get_channel_info |
guild_id: str, channel_id: str |
channel_id, channel_name |
get_channel_list(guild_id, *, joined_only) |
get_channel_list |
guild_id: str, joined_only: bool=false |
list[get_channel_info 响应] |
set_channel_name(guild_id, channel_id, channel_name) |
set_channel_name |
guild_id, channel_id, channel_name |
无 |
get_channel_member_info(guild_id, channel_id, user_id) |
get_channel_member_info |
guild_id, channel_id, user_id |
user_id, user_name, user_displayname |
get_channel_member_list(guild_id, channel_id) |
get_channel_member_list |
guild_id, channel_id |
list[get_channel_member_info 响应] |
leave_channel(guild_id, channel_id) |
leave_channel |
guild_id, channel_id |
无 |
频道体系与群组(group)彼此独立:Discord / QQ 频道 / Kook 等平台实现频道接口, 传统 QQ / 微信实现群组接口,两者可同时存在或仅其一。
3.5 文件资源操作
Warning
文件资源模型(file_id 两段式)在 ErisPulse 属"降级可用":
ErisPulse 的文件收发不走"先上传拿 file_id 再引用"模型——模块发文件用
SendDSL.File(file, filename)(URL / 路径 / 字节发送时直传,见
发送方法规范)。
本节 upload_file / get_file / 分片动作依赖平台特有的 file_id 文件资源
能力,通用性不足;仅当适配器后端天然具备该能力时才可透传,框架内置
适配器不实现也不建议实现,调用时通常返回 retcode=10002。
模块需要跨平台传文件时,请使用 SendDSL.File,勿依赖 file_id。
展望:file_id 资源模型标准化到框架层是未来的方向,当前版本不提供。
整包传输(小文件):
| 方法 | OB12 动作 | 参数 | data 返回 |
|---|---|---|---|
upload_file(*, type, name, ...) |
upload_file |
type, name, url/path/data, headers?, sha256? |
file_id |
get_file(file_id, type) |
get_file |
file_id: str, type: str |
name, url/path/data |
upload_file 的 type 参数:
"url":通过 URL 上传(需提供url)"path":通过本地路径上传(需提供path)"data":通过二进制数据上传(需提供data)
3.5.1 分片传输(大文件,属上述降级范围)
OneBot12 分片动作按 stage 区分阶段。ApiDSL 将同一动作的三/两阶段拆分为独立方法
(offset 为字节偏移,data 在 JSON 中为 Base64);下表仅为查阅保留,
适配器无需也不应强制实现:
分片上传三步:prepare → transfer(循环逐片)→ finish
| 方法 | 对应 stage | 参数 | data 返回 |
|---|---|---|---|
upload_file_fragmented_prepare(name, total_size) |
prepare |
name: str, total_size: int |
file_id(传输期用) |
upload_file_fragmented_transfer(file_id, offset, data) |
transfer |
file_id, offset: int, data: bytes |
无 |
upload_file_fragmented_finish(file_id, sha256) |
finish |
file_id, sha256: str(整文件校验) |
file_id |
total = os.path.getsize(path)
r = await adapter.Api.upload_file_fragmented_prepare(os.path.basename(path), total)
fid = r["data"]["file_id"]
offset = 0
with open(path, "rb") as f:
while chunk := f.read(65536):
await adapter.Api.upload_file_fragmented_transfer(fid, offset, chunk)
offset += len(chunk)
sha256 = hashlib.sha256(open(path, "rb").read()).hexdigest()
await adapter.Api.upload_file_fragmented_finish(fid, sha256)
分片下载两步:prepare → transfer(循环取片)
| 方法 | 对应 stage | 参数 | data 返回 |
|---|---|---|---|
get_file_fragmented_prepare(file_id) |
prepare |
file_id |
name, total_size, sha256 |
get_file_fragmented_transfer(file_id, offset, size) |
transfer |
file_id, offset: int, size: int |
data(本次分片字节) |
3.6 元(Meta)动作
元动作不针对具体账号,无需 Using() 指定 Bot。
| 方法 | OB12 动作 | 参数 | data 返回 |
|---|---|---|---|
get_latest_events(limit, timeout) |
get_latest_events |
limit: int=0, timeout: int=0 |
事件对象数组(不含元事件) |
get_supported_actions() |
get_supported_actions |
无 | list[str] 支持的动作名 |
get_status() |
get_status |
无 | good: bool, bots: list[{self, online, ...}] |
get_version() |
get_version |
无 | impl, version, onebot_version |
3.7 通用扩展动作
| 方法 | 说明 |
|---|---|
call(action, **params) |
平台扩展动作的逃生舱,遵循 OB12 扩展命名规则 {prefix}.{action} |
4. 使用方式
4.1 基本调用
from ErisPulse import adapter
# 获取用户信息(跨平台统一)
result = await adapter.myplatform.Api.get_user_info("123456")
if result["status"] == "ok":
user_name = result["data"]["user_name"]
print(f"用户名: {user_name}")
# 获取群列表
result = await adapter.myplatform.Api.get_group_list()
groups = result["data"]
# 撤回消息
await adapter.myplatform.Api.delete_message("msg_123456")
4.2 指定 Bot 账号(多账户模式)
# 使用指定 Bot 账号执行操作
info = await adapter.myplatform.Api.Using("bot1").get_self_info()
4.3 平台扩展动作
# 调用平台特有的扩展动作(建议使用 {prefix}.{action} 命名)
result = await adapter.telegram.Api.call(
"telegram.send_sticker",
sticker_id="CAACAgIAAxkBAA...",
)
4.4 在事件处理器中使用
from ErisPulse.Core.Event import message
@message()
async def handle(event):
# 获取发送者详细信息
user_id = event.get_user_id()
platform = event.get_platform()
result = await getattr(adapter, platform).Api.get_user_info(user_id)
if result["status"] == "ok":
user_name = result["data"]["user_name"]
await event.reply(f"你好,{user_name}!")
5. 适配器实现
5.1 默认行为(零配置)
ApiDSL 的默认实现将标准动作名作为 endpoint 直接传递给 adapter.call_api():
# ApiDSL 默认实现等价于:
async def get_user_info(self, user_id: str) -> dict:
return await self._adapter.call_api("get_user_info", user_id=user_id, account_id=self._account_id)
适用场景:当适配器的底层后端自身即遵循 OneBot12 标准动作协议时,
call_api 天然支持标准动作名(如直接对接遵循该协议的服务端)。
5.2 覆盖标准方法(映射到平台原生 API)
适配器可覆盖单个标准方法,将其映射到平台原生 API:
class MyAdapter(BaseAdapter):
class Api(BaseAdapter.Api):
"""MyPlatform 标准 API 动作实现"""
async def get_user_info(self, user_id: str) -> dict:
# 映射到平台原生 API
raw = await self._adapter._request("GET", f"/users/{user_id}")
if raw.get("code") != 0:
return self._adapter.make_error(retcode=34600, message="用户不存在")
user = raw["data"]
return self._adapter.make_response(
data={
"user_id": str(user["id"]),
"user_name": user.get("nick", ""),
"user_displayname": user.get("display_name", ""),
"user_remark": user.get("remark", ""),
},
raw=raw,
)
async def get_friend_list(self) -> dict:
raw = await self._adapter._request("GET", "/friends")
friends = [
{
"user_id": str(u["id"]),
"user_name": u.get("nick", ""),
"user_displayname": u.get("display_name", ""),
"user_remark": u.get("remark", ""),
}
for u in raw.get("data", [])
]
return self._adapter.make_response(data=friends, raw=raw)
5.3 未支持的动作
适配器未覆盖的标准方法走默认实现(委托给 call_api)。如果 call_api 也不支持该动作,应返回标准错误响应:
async def call_api(self, endpoint: str, **params):
if endpoint not in self._supported_endpoints:
return self.make_error(retcode=10002, message=f"不支持的动作: {endpoint}")
# ... 平台 API 调用
模块开发者可通过返回值的 retcode 判断是否支持:
result = await adapter.myplatform.Api.get_friend_list()
if result["retcode"] == 10002:
print("该平台不支持获取好友列表")
6. 响应格式
所有 ApiDSL 方法返回标准 API 响应格式(详见 API 响应标准):
{
"status": "ok",
"retcode": 0,
"data": { ... },
"message_id": "",
"message": "",
"myplatform_raw": { ... }
}
注意:信息查询类动作的
message_id为空字符串(仅消息发送类动作才有message_id)。
7. 与 SendDSL / RequestDSL 的关系
| 场景 | 使用 DSL | 示例 |
|---|---|---|
| 发送消息 | Send |
adapter.Send.To("group", "123").Text("hi") |
| 同意/拒绝请求 | Request |
adapter.Request("req_id").accept() |
| 获取用户/群信息 | Api |
adapter.Api.get_user_info("123") |
| 撤回消息 | Api |
adapter.Api.delete_message("msg_id") |
| 退出群 | Api |
adapter.Api.leave_group("group_id") |
8. 适配器实现检查清单
标准动作
-
call_api能处理标准动作名(或覆盖对应ApiDSL方法) - 不支持的动作返回
retcode=10002 - 返回值遵循标准 API 响应格式
-
data字段包含 OB12 标准定义的字段 - 频道平台需实现
get_guild_*/get_channel_*/leave_guild/leave_channel - 元动作(
get_status/get_version/get_supported_actions)建议实现 - 文件收发用
SendDSL.File(直传);文件资源动作(upload_file/get_file/分片)不强制实现,仅当后端具备file_id资源能力时才需透传
扩展动作
- 平台扩展动作使用
{prefix}.{action}命名 - 扩展动作的参数和响应仍遵循 OB12 动作请求/响应结构