📋 05 - API 参考

本章把插件开发常用 API 集中放在一起，便于写插件时随手查。

🛠️ plugin_base 模块

导入方式如下。

python

from core.plugin_base import (
    text, image, image_url, record, record_url,
    segments, build_action, run_sync,
    ensure_dir, load_json, write_json, atomic_write_text,
    split_message_segments,
)

消息段构建

text(content)

创建文本消息段。

python

text("Hello World")
# 返回: {"type": "text", "data": {"text": "Hello World"}}

image(file_path)

创建本地图片消息段。

python

image("/path/to/image.png")
# 返回: {"type": "image", "data": {"file": "file:///path/to/image.png"}}

手写消息段时，推荐使用 Path(file_path).resolve().as_uri() 生成本地文件 URI。

image_url(url)

创建网络图片消息段。

python

image_url("https://example.com/pic.jpg")
# 返回: {"type": "image", "data": {"file": "https://example.com/pic.jpg"}}

record(file_path)

创建本地语音消息段。

python

record("/path/to/audio.mp3")
# 返回: {"type": "record", "data": {"file": "file:///path/to/audio.mp3"}}

同样地，手写本地语音消息段时也优先使用 Path(file_path).resolve().as_uri()。

record_url(url)

创建网络语音消息段。

python

record_url("https://example.com/audio.mp3")
# 返回: {"type": "record", "data": {"file": "https://example.com/audio.mp3"}}

segments(payload)

将任意值转换为消息段列表。

python

segments("Hello")      # -> [{"type": "text", "data": {"text": "Hello"}}]
segments(None)         # -> []
segments([text("Hi")]) # -> [{"type": "text", "data": {"text": "Hi"}}]

build_action(segs, user_id, group_id)

构建 OneBot Action。

python

segs = [text("Hello")]
build_action(segs, user_id=123, group_id=None)
# 返回: {
#   "action": "send_private_msg",
#   "params": {"user_id": 123, "message": [...]}
# }

build_action(segs, user_id=123, group_id=456)
# 返回: {
#   "action": "send_group_msg",
#   "params": {"group_id": 456, "message": [...]}
# }

异步工具

run_sync(func, *args, **kwargs)

在线程池中运行同步函数。

python

import requests

async def handle(...):
    # 避免阻塞事件循环
    response = await run_sync(requests.get, "https://api.example.com")
    return segments(response.text)

文件工具

ensure_dir(path)

确保目录存在（递归创建）。

python

ensure_dir(Path("/path/to/dir"))

load_json(path, default=None)

加载 JSON 文件。

python

data = load_json(Path("data.json"))  # 文件不存在返回 {}
data = load_json(Path("data.json"), default={"count": 0})

write_json(path, data)

写入 JSON 文件（先写临时文件再替换，防止中断损坏）。

python

write_json(Path("data.json"), {"count": 1})

atomic_write_text(path, payload)

原子写入文本文件。

python

atomic_write_text(Path("output.txt"), "内容")

消息分割

split_message_segments(segs, max_length=500)

将消息段列表按文本长度分割为多个分片，防止超长消息被 OneBot 截断。

python

long_segs = [text("很长的内容...")]
parts = split_message_segments(long_segs, max_length=500)
# parts = [[seg1, seg2], [seg3, ...], ...]

PluginContext 类

插件处理函数的 context 参数类型。

属性

属性	类型	说明
`config`	`Dict[str, Any]`	config.json 完整内容
`secrets`	`Dict[str, Any]`	secrets.json 完整内容
`plugin_name`	`str`	当前插件名
`plugin_dir`	`Path`	插件目录路径
`data_dir`	`Path`	数据目录路径
`logger`	`_RequestLogger`	日志记录器（自动附带 request_id）
`http_session`	`aiohttp.ClientSession \| None`	HTTP 客户端
`send_action`	`Callable`	发送 OneBot Action 的异步回调，可用于后台任务完成后的主动通知
`metrics`	`MetricsCollector \| None`	运行指标收集器
`current_user_id`	`int \| None`	当前消息的用户 ID
`current_group_id`	`int \| None`	当前消息的群 ID
`state`	`Dict[str, Any]`	插件私有状态（当次请求生命周期）

方法

default_groups()

获取配置的默认群列表。

python

groups = context.default_groups()  # -> [123456, 789012]

reload_config()

重新加载配置文件。

python

context.reload_config()

reload_plugins()

重新加载所有插件。

python

context.reload_plugins()

list_commands()

获取所有已注册命令。

python

commands = context.list_commands()  # -> ["help", "echo", ...]

list_plugins()

获取所有已加载插件。

python

plugins = context.list_plugins()  # -> ["core", "echo", ...]

send_action(action)

主动发送 OneBot Action。

普通命令处理通常只需要 return segments(...)，由框架自动构建并发送响应。只有后台任务、定时任务或需要在当前 handler 返回之后再通知用户时，才直接调用 context.send_action(action)。

python

from core.plugin_base import build_action, segments

action = build_action(
    segments("[codex:main #1] 完成:\n结果内容"),
    user_id=event.get("user_id"),
    group_id=event.get("group_id"),
)
if action:
    await context.send_action(action)

长文本不需要在插件里手动截断；发送链路会通过 split_message_segments() 按文本长度分片。

会话方法

create_session(initial_data=None, timeout=300.0)

为当前用户创建会话。

python

session = await context.create_session(
    initial_data={"step": 1, "target": 50},
    timeout=180.0  # 3 分钟超时
)

参数：

initial_data - 初始会话数据
timeout - 超时时间（秒）

返回：Session 对象

get_session()

获取当前用户的会话。

python

session = await context.get_session()
if session:
    step = session.get("step")

返回：Session 或 None（无会话或已过期）

end_session()

结束当前用户的会话。

python

await context.end_session()

返回：bool - 是否成功删除

has_session()

检查当前用户是否有活跃会话。

python

if await context.has_session():
    ...

静音方法

mute_group(group_id, duration_minutes)

静音指定群。

python

context.mute_group(123456, 30)  # 静音 30 分钟

unmute_group(group_id)

解除群静音。

python

context.unmute_group(123456)

is_group_muted(group_id)

检查群是否被静音。

python

if context.is_group_muted(123456):
    ...

get_mute_remaining(group_id)

获取剩余静音时间（秒）。

python

remaining = context.get_mute_remaining(123456)  # -> 930.5 (秒)

返回值：float - 剩余静音时间（秒），0 表示未静音

Session 类

get_remaining_time() -> float

获取会话剩余时间（秒）。

python

remaining = session.get_remaining_time()
if remaining < 60:
    return segments(f"会话将在 {remaining} 秒后过期")

返回值：float - 剩余时间（秒）

is_active() -> bool

检查会话是否活跃。

python

if session.is_active():
    return segments("会话进行中")

返回值：bool - 会话是否活跃（未过期）

Dispatcher 线性消息流程

Dispatcher 的入口由 MessageParser.parse() 构建 MessageContext，随后 _process_event() 按固定 A-G 顺序处理。插件通过命令、会话、smalltalk provider 等约定函数接入。

MessageContext 关键字段

字段	类型	说明
`clean_text`	`str`	去除开头 bot_name / 命令前缀后的文本
`has_bot_name`	`bool`	原始文本任意位置包含 `bot_name`
`has_command_prefix`	`bool`	原始文本严格以命令前缀开头
`has_prefix`	`bool`	`has_command_prefix`、`has_bot_name`、`is_at_me` 的并集
`is_only_bot_name`	`bool`	只叫机器人名字或只 @ 机器人
`is_at_me`	`bool`	OneBot at 段或 raw_message 中 @ 机器人
`is_url_only`	`bool`	`clean_text.strip()` 整体匹配 `^https?://\S+$`

处理顺序

Step A: URL short-circuit（clean_text 单 URL → url_parser；mute 不影响）
Step B: 处理门控（私聊、require_bot_name_in_group=False、has_prefix、活跃 session）
Step C: is_only_bot_name → 默认回应 / call_bot_name_only
Step D: router 命中 → 执行命令
Step E: has_command_prefix 且命令未命中 → 未知命令提示
Step F: 活跃 session → 转 session 插件
Step G: 回落 smalltalk provider（mute 仅在此步阻塞）

URL 与未知命令

URL 处理改用 ctx.is_url_only，只接受完整单 URL。看看 https://example.com 不会触发 url_parser。
未知命令提示只在 has_command_prefix=True 且 router 未命中时出现。小青不存在的指令 会继续走会话或 smalltalk 回落，不会被当作 /不存在的指令。

Session 类

多轮对话的会话对象。

属性

属性	类型	说明
`user_id`	`int`	用户 ID
`group_id`	`Optional[int]`	群 ID（私聊为 None）
`plugin_name`	`str`	所属插件
`state`	`str`	会话状态
`data`	`Dict`	会话数据
`timeout`	`float`	超时时间（秒）

方法

get(key, default=None)

获取会话数据。

python

step = session.get("step", 1)

set(key, value)

设置会话数据（自动更新时间戳）。

python

session.set("step", 2)
session.set("attempts", session.get("attempts", 0) + 1)

clear()

清空会话数据。

python

session.clear()

is_expired()

检查是否过期。

python

if session.is_expired():
    ...

handle() 函数签名

插件入口函数。

python

async def handle(
    command: str,           # 命令名
    args: str,              # 参数字符串
    event: Dict[str, Any],  # 原始 OneBot 事件
    context: PluginContext  # 插件上下文
) -> List[Dict[str, Any]]:  # 返回消息段列表
    ...

event 参数常用字段

python

event = {
    "post_type": "message",
    "message_type": "group",  # 或 "private"
    "user_id": 123456,
    "group_id": 789012,       # 私聊时为 None
    "message": [              # 消息段列表
        {"type": "text", "data": {"text": "内容"}}
    ],
    "raw_message": "内容",    # 原始消息文本
    "sender": {
        "user_id": 123456,
        "nickname": "昵称",
        "card": "群名片",
        "role": "member"      # member/admin/owner
    },
    "time": 1234567890
}

handle_session() 函数签名

多轮对话处理函数（可选）。

概述

当用户处于活跃会话时，此函数会被调用处理用户的后续消息。在 dispatcher 线性流程中，Step F 负责调用此函数。

函数签名

python

async def handle_session(
    text: str,              # 用户输入的文本
    event: Dict[str, Any],  # 原始 OneBot 事件
    context: PluginContext, # 插件上下文
    session: Session        # 会话对象
) -> List[Dict[str, Any]]:  # 返回消息段列表
    ...

参数说明

参数	类型	说明
`text`	`str`	用户输入的文本（未经过任何处理）
`event`	`Dict[str, Any]`	原始 OneBot 事件
`context`	`PluginContext`	插件上下文
`session`	`Session`	当前会话对象

返回值

返回消息段列表，表示要发送的回复。

会话生命周期

用户发送命令（如 /guess）
    ↓
插件调用 context.create_session()
    ↓
会话创建，状态为 active
    ↓
用户后续消息
    ↓
Dispatcher Step F 捕获
    ↓
调用 handle_session()
    ↓
插件处理并返回回复
    ↓
会话更新（session.set()）
    ↓
┌─────────────┬─────────────┐
│ 继续对话      │ 结束对话      │
│ (返回消息段)  │ 调用 end_session())
└─────────────┴─────────────┘
    ↓              ↓
回到 handle_session()   会话被删除

使用示例

python

async def handle(command: str, args: str, event: Dict, context) -> List:
    """开始猜数字游戏"""
    target = random.randint(1, 100)
    
    # 创建会话
    await context.create_session(
        initial_data={
            "target": target,
            "attempts": 0,
            "start_time": time.time()
        },
        timeout=180  # 3 分钟超时
    )
    
    return segments(
        "🎮 猜数字游戏开始！\n"
        "我已经想好了一个 1-100 的数字\n"
        "请输入你的猜测"
    )


async def handle_session(text: str, event: Dict, context, session) -> List:
    """处理游戏中的消息"""
    
    # 退出命令
    if text.lower() in ["退出", "quit", "q"]:
        target = session.get("target")
        await context.end_session()
        return segments(f"游戏结束，答案是 {target}")
    
    # 解析猜测
    try:
        guess = int(text.strip())
    except ValueError:
        return segments("请输入有效的数字")
    
    # 更新尝试次数
    attempts = session.get("attempts", 0) + 1
    session.set("attempts", attempts)
    
    # 获取目标数字
    target = session.get("target")
    
    # 判断结果
    if guess < target:
        return segments(f"太小了！（{attempts} 次尝试）")
    elif guess > target:
        return segments(f"太大了！（{attempts} 次尝试）")
    else:
        # 猜对了，结束会话
        elapsed = int(time.time() - session.get("start_time"))
        await context.end_session()
        return segments(
            f"🎉 恭喜你猜对了！\n"
            f"答案：{target}\n"
            f"尝试次数：{attempts}\n"
            f"用时：{elapsed} 秒"
        )

架构特性

在 dispatcher 线性流程中：

优先级明确：会话处理在命令匹配之后、闲聊之前
绕过普通触发条件：群聊普通文本没有 has_prefix 时，只要活跃会话存在仍会处理
独立处理：会话处理不依赖 bot_name；只有 is_only_bot_name 会先走只叫名字回应，避免打断“叫机器人”语义

注意事项

会话超时
- 超过 timeout 时间后，会话自动过期
- 用户下次发送消息时会创建新会话
每个用户独立
- 每个 (user_id, group_id) 组合有独立的会话
- 私聊和群聊的会话互不影响
手动结束
- 游戏或对话结束时，必须调用 context.end_session()
- 否则用户需要等待超时才能开始新会话
数据更新
- 使用 session.set() 更新数据会自动刷新 updated_at 时间戳
- 这会延长会话的有效期

handle_url() 函数签名

URL 自动解析函数（可选）。

Dispatcher 只在 Step A 调用此函数：ctx.clean_text.strip() 必须整体匹配 ^https?://\S+$。含附加文字或多个 URL 的消息不会进入 handle_url()。

python

async def handle_url(
    url: str,               # clean_text 中的完整单 URL
    event: Dict[str, Any],  # 原始 OneBot 事件
    context: PluginContext  # 插件上下文
) -> List[Dict[str, Any]]:  # 返回消息段列表
    ...

handle_smalltalk() 函数签名

闲聊处理函数（可选）。

概述

当插件被配置为 smalltalk_provider 时，此函数会被调用处理闲聊消息。

函数签名

python

async def handle_smalltalk(
    text: str,              # 用户消息文本（已去除前缀）
    event: Dict[str, Any],  # 原始 OneBot 事件
    context: PluginContext  # 插件上下文
) -> List[Dict[str, Any]]:  # 返回消息段列表，或 None
    ...

返回值

返回值	说明
`List[Dict]`	返回消息段列表，表示需要回复
`None` 或 `[]`	不回复，传递给后续处理或直接返回空

使用示例

python

async def handle_smalltalk(text: str, event: Dict, context) -> List:
    """简单规则闲聊"""
    
    # 1. 检查是否应该回复
    if not should_reply(text, event):
        return None  # 不回复
    
    # 2. 生成回复
    if "你好" in text or "hello" in text.lower():
        return segments("你好！有什么我可以帮助你的吗？")
    
    if "名字" in text or "你是谁" in text:
        bot_name = context.config.get("bot_name", "小青")
        return segments(f"我叫 {bot_name}~")
    
    # 3. 不回复其他消息
    return None

xiaoqing_chat 特殊处理

当 smalltalk_provider 配置为 xiaoqing_chat 时：

进入 smalltalk 回落时会调用此函数
- random_reply_rate 不参与 dispatcher 分发
- 由插件内部控制 attention、回复频率、普通插话概率、PFC planner 和 reply checker
插件可以自主决定是否回复
- 返回 None 或 [] 表示不回复
- 返回消息段列表表示需要回复
xiaoqing_chat 的 directed attention 会跳过普通概率门
- /xc、私聊、@、直接叫名字、只喊名字后的追问、reply 引用小青、以及有近期上下文锚点的“她/ta”共指召唤会走 forced 路径
- 普通群聊消息才走 reply_probability_base、heartflow 和硬频控

可以实现更复杂的逻辑

python

async def handle_smalltalk(text: str, event: Dict, context) -> List:
    # 1. 获取用户历史
    user_id = event.get("user_id")
    history = await get_user_history(user_id, context)
    
    # 2. 情绪分析
    sentiment = analyze_sentiment(text)
    
    # 3. 根据情绪和历史决定是否回复
    if sentiment < 0 and history["negative_count"] > 3:
        return None  # 用户情绪不好，暂不回复
    
    # 4. 生成回复
    response = await generate_llm_response(text, history, context)
    
    # 5. 保存到历史
    await save_to_history(user_id, text, response)
    
    return segments(response)

配置为 smalltalk_provider

在 config.json 中配置：

json

{
  "plugins": {
    "smalltalk_provider": "your_plugin_name"
  }
}

在 secrets.json 中配置插件私有配置：

json

{
  "plugins": {
    "your_plugin_name": {
      "api_key": "your-api-key",
      "api_base": "https://api.example.com"
    }
  }
}

init() / shutdown() 钩子

python

async def init(context: PluginContext) -> None:
    """插件加载时调用"""
    ...

async def shutdown(context: PluginContext) -> None:
    """插件卸载时调用"""
    ...

定时任务处理函数

python

async def handler_name(context: PluginContext) -> List[Dict[str, Any]]:
    """定时任务处理函数"""
    return segments("定时消息")

Inbound Server API

POST /event

接收 OneBot 事件推送。

请求头：

Authorization: Bearer <inbound_token>
Content-Type: application/json

请求体（OneBot 事件）：

json

{
  "post_type": "message",
  "message_type": "group",
  "group_id": 123456,
  "user_id": 789,
  "message": [{"type": "text", "data": {"text": "/help"}}]
}

响应体：

json

{
  "actions": [
    {
      "action": "send_group_msg",
      "params": {
        "group_id": 123456,
        "message": [{"type": "text", "data": {"text": "帮助信息"}}]
      }
    }
  ]
}

WebSocket /ws

WebSocket 端点，用于持久连接。

连接：

ws://127.0.0.1:12000/ws
Header: Authorization: Bearer <token>

消息格式：同 POST /event

GET /health

健康检查。

响应：

json

{"status": "ok"}

OneBot Action 格式

XiaoQing 返回的 Action 遵循 OneBot 协议。

send_group_msg

发送群消息。

json

{
  "action": "send_group_msg",
  "params": {
    "group_id": 123456,
    "message": [{"type": "text", "data": {"text": "内容"}}]
  }
}

send_private_msg

发送私聊消息。

json

{
  "action": "send_private_msg",
  "params": {
    "user_id": 789,
    "message": [{"type": "text", "data": {"text": "内容"}}]
  }
}

core.args 模块

python

from core.args import parse, ParsedArgs

parse(raw) -> ParsedArgs

解析命令参数字符串，返回 ParsedArgs 对象。

python

parsed = parse("add 完成报告 --cat=工作 -p 2")

ParsedArgs

属性/方法	说明
`parsed.first`	第一个位置参数（property）
`parsed.second`	第二个位置参数（property）
`parsed.get(i, default="")`	获取第 i 个位置参数
`parsed.rest(start=0)`	从第 start 个参数开始拼接剩余参数
`parsed.opt(key, default="")`	获取选项值（`--key=val` 或 `-k val`）
`parsed.has(key)`	检查选项/标志是否存在
`len(parsed)`	位置参数数量
`bool(parsed)`	参数字符串是否非空
`parsed.raw`	原始参数字符串
`parsed.tokens`	位置参数列表
`parsed.options`	选项字典

➡️ 下一步

配置详解 → 06-configuration.md
高级主题 → 07-advanced.md

📋 05 - API 参考 ​

🛠️ plugin_base 模块 ​

消息段构建 ​

text(content) ​

image(file_path) ​

image_url(url) ​

record(file_path) ​

record_url(url) ​

segments(payload) ​

build_action(segs, user_id, group_id) ​

异步工具 ​

run_sync(func, *args, **kwargs) ​

文件工具 ​

ensure_dir(path) ​

load_json(path, default=None) ​

write_json(path, data) ​

atomic_write_text(path, payload) ​

消息分割 ​

split_message_segments(segs, max_length=500) ​

PluginContext 类 ​

属性 ​

方法 ​

default_groups() ​

reload_config() ​

reload_plugins() ​

list_commands() ​

list_plugins() ​

send_action(action) ​

会话方法 ​

create_session(initial_data=None, timeout=300.0) ​

get_session() ​

end_session() ​

has_session() ​

静音方法 ​

mute_group(group_id, duration_minutes) ​

unmute_group(group_id) ​

is_group_muted(group_id) ​

get_mute_remaining(group_id) ​

Session 类 ​

get_remaining_time() -> float ​

is_active() -> bool ​

Dispatcher 线性消息流程 ​

MessageContext 关键字段 ​

处理顺序 ​

URL 与未知命令 ​

Session 类 ​

属性 ​

方法 ​

get(key, default=None) ​

set(key, value) ​

clear() ​

is_expired() ​

handle() 函数签名 ​

event 参数常用字段 ​

handle_session() 函数签名 ​

概述 ​

函数签名 ​

参数说明 ​

返回值 ​

会话生命周期 ​

使用示例 ​

架构特性 ​

注意事项 ​

handle_url() 函数签名 ​

handle_smalltalk() 函数签名 ​

概述 ​

函数签名 ​

返回值 ​

使用示例 ​

xiaoqing_chat 特殊处理 ​

配置为 smalltalk_provider ​

init() / shutdown() 钩子 ​

定时任务处理函数 ​

Inbound Server API ​

POST /event ​

WebSocket /ws ​

GET /health ​

OneBot Action 格式 ​

send_group_msg ​

send_private_msg ​

📋 05 - API 参考

🛠️ plugin_base 模块

消息段构建

text(content)

image(file_path)

image_url(url)

record(file_path)

record_url(url)

segments(payload)

build_action(segs, user_id, group_id)

异步工具

run_sync(func, *args, **kwargs)

文件工具

ensure_dir(path)

load_json(path, default=None)

write_json(path, data)

atomic_write_text(path, payload)

消息分割

split_message_segments(segs, max_length=500)

PluginContext 类

属性

方法

default_groups()

reload_config()

reload_plugins()

list_commands()

list_plugins()

send_action(action)

会话方法

create_session(initial_data=None, timeout=300.0)

get_session()

end_session()

has_session()

静音方法

mute_group(group_id, duration_minutes)

unmute_group(group_id)

is_group_muted(group_id)

get_mute_remaining(group_id)

Session 类

get_remaining_time() -> float

is_active() -> bool

Dispatcher 线性消息流程

MessageContext 关键字段

处理顺序

URL 与未知命令

Session 类

属性

方法

get(key, default=None)

set(key, value)

clear()

is_expired()

handle() 函数签名

event 参数常用字段

handle_session() 函数签名

概述

函数签名

参数说明

返回值

会话生命周期

使用示例

架构特性

注意事项

handle_url() 函数签名

handle_smalltalk() 函数签名

概述

函数签名

返回值

使用示例

xiaoqing_chat 特殊处理

配置为 smalltalk_provider

init() / shutdown() 钩子

定时任务处理函数

Inbound Server API

POST /event

WebSocket /ws

GET /health

OneBot Action 格式

send_group_msg

send_private_msg