Nekro Agent 文档站

Appearance

Agent 上下文 (AgentCtx) ​

AgentCtx 是 Nekro Agent 插件开发中最重要的概念之一。它封装了 Agent 在执行任务时所需的所有上下文信息，为插件提供了统一的接口来访问相关数据和功能。

什么是 AgentCtx？ ​

AgentCtx（Agent Context）是一个包含了当前执行环境所有关键信息的上下文对象。无论是处理来自聊天软件的消息，还是响应 Webhook 事件，AgentCtx 都提供了统一的接口让插件与 Nekro Agent 的核心功能进行交互。

在插件开发中，AgentCtx 通常作为沙盒方法的第一个参数（约定使用 _ctx 作为变量名）提供给插件使用。

核心属性 ​

基本信息属性 ​

python
# 沙盒和聊天标识
_ctx.from_chat_key: Optional[str]     # 来源聊天的唯一标识
_ctx.chat_key: Optional[str]          # 聊天频道唯一ID (同 from_chat_key)
_ctx.container_key: Optional[str]     # 沙盒容器的唯一标识

# 聊天频道信息
_ctx.channel_id: Optional[str]        # 频道的原始平台 ID（如 QQ 群号或用户 ID）
_ctx.channel_name: Optional[str]      # 频道名称（如 QQ 群名或用户名）
_ctx.channel_type: Optional[str]      # 频道类型（如 'group' 或 'private'）
_ctx.adapter_key: Optional[str]       # 适配器标识（如 'onebot_v11'）

# Webhook 相关（当由 Webhook 触发时）
_ctx.webhook_request: Optional[WebhookRequest]  # Webhook 请求数据

重要属性访问器 ​

chat_key 属性 ​

python
@property
def chat_key(self) -> str:
    """聊天频道唯一ID"""

这是当前聊天频道的唯一标识符，通常由 adapter_key 和 channel_id 组成。

使用示例：

python
@plugin.mount_sandbox_method(SandboxMethodType.TOOL, "get_chat_info", "获取当前聊天频道信息")
async def get_chat_info(_ctx: AgentCtx) -> str:
    return f"当前聊天频道: {_ctx.chat_key}"
    # 输出示例: "当前聊天频道: onebot_v11-group_12345678"

adapter 属性 ​

python
@property
def adapter(self) -> "BaseAdapter":
    """消息关联适配器实例"""

通过此适配器实例，插件可以获取适配器相关信息或调用适配器相关方法。

db_chat_channel 和 db_user 属性 ​

python
@property
def db_chat_channel(self) -> Optional["DBChatChannel"]:
    """当前聊天频道的数据库实例"""

@property
def db_user(self) -> Optional["DBUser"]:
    """触发本次 Agent 的数据库用户实例"""

这些属性提供对底层数据库模型的访问，用于高级数据操作。

核心功能模块 ​

文件系统 (fs) ​

_ctx.fs 提供了强大的文件处理能力，用于在插件和 AI 之间安全地传递文件。

python
@property
def fs(self) -> FileSystem:
    """文件系统工具"""

主要方法：

mixed_forward_file - 插件向 AI 传递文件 ​

当插件需要生成一个文件（如图片、文档）并返回给 AI 时使用：

python
@plugin.mount_sandbox_method(SandboxMethodType.TOOL, "generate_chart", "生成图表并返回给AI")
async def generate_chart(_ctx: AgentCtx, data: str) -> str:
    # 插件通过 API 获取了一张图片
    image_url = "https://example.com/chart.png"

    # 将其转换为 AI 可用的沙盒路径
    sandbox_path = await _ctx.fs.mixed_forward_file(image_url, file_name="chart.png")

    # 返回沙盒路径给 AI
    return sandbox_path  # "/app/uploads/chart.png"

get_file - AI 向插件传递文件 ​

当 AI 调用插件并传入一个沙盒文件路径作为参数时，插件使用此方法获取宿主机上的真实路径：

python
@plugin.mount_sandbox_method(SandboxMethodType.TOOL, "analyze_image", "分析图片内容")
async def analyze_image(_ctx: AgentCtx, image_path: str) -> str:
    # AI 提供的沙盒路径: "/app/shared/photo.jpg"
    # 转换为宿主机可访问的真实路径
    host_path = _ctx.fs.get_file(image_path)

    # 现在可以读取文件了
    with open(host_path, "rb") as f:
        file_content = f.read()
        # 进行图片分析...

    return "图片分析完成"

消息模块 (ms) ​

_ctx.ms 提供对底层消息模块的直接访问，主要用于需要手动指定 chat_key 的高级场景和主动触发 AI 响应。

python
@property
def ms(self):
    """消息模块"""

主要方法：

send_text - 向指定频道发送文本消息 ​

python
async def send_text(self, chat_key: str, content: str, ctx: AgentCtx) -> None:
    """向指定聊天频道发送文本消息"""

push_system - 推送系统消息并触发 AI 响应 ​

python
async def push_system(self, chat_key: str, message: str, ctx: AgentCtx, trigger_agent: bool = False) -> None:
    """向系统推送消息，可选择是否触发 AI 响应"""

参数说明：

  • chat_key: 目标聊天频道标识
  • message: 要推送的系统消息内容
  • ctx: Agent 上下文
  • trigger_agent: 是否触发 AI 生成回复（默认 False）

使用场景：

向其他聊天频道发送消息

python
@plugin.mount_sandbox_method(SandboxMethodType.BEHAVIOR, "notify_admin", "通知管理员")
async def notify_admin(_ctx: AgentCtx, message: str) -> str:
    # 向监控频道发送状态更新
    monitor_chat_key = "onebot_v11-group_987654321"

    await _ctx.ms.send_text(monitor_chat_key, f"系统通知: {message}", _ctx)

    return "已通知管理员"

跨聊天频道触发 AI 响应

python
@plugin.mount_sandbox_method(SandboxMethodType.BEHAVIOR, "report_to_monitoring", "报告到监控群")
async def report_to_monitoring(_ctx: AgentCtx, incident: str) -> str:
    # 向监控群推送事件并触发 AI 分析
    monitoring_chat = "onebot_v11-group_monitor123"
    analysis_prompt = f"检测到系统事件：{incident}\n请分析影响并提供处理建议。"

    await _ctx.ms.push_system(
        chat_key=monitoring_chat,
        message=analysis_prompt,
        ctx=_ctx,
        trigger_agent=True  # 触发监控群中的 AI 分析
    )

    return "已报告到监控群，AI 正在分析"

便捷方法 ​

消息发送方法 ​

send_text - 发送文本消息 ​

python
async def send_text(self, content: str, *, record: bool = True):
    """发送文本消息到当前聊天频道"""

参数说明：

  • content: 要发送的文本内容
  • record: 是否记录到对话历史（默认 True）

使用示例：

python
@plugin.mount_sandbox_method(SandboxMethodType.BEHAVIOR, "send_notification", "发送通知")
async def send_notification(_ctx: AgentCtx, message: str) -> str:
    await _ctx.send_text(f"📢 通知: {message}")
    return "通知已发送"

send_image - 发送图片 ​

python
async def send_image(self, file_path: str, *, record: bool = True):
    """发送图片到当前聊天频道"""

使用示例：

python
@plugin.mount_sandbox_method(SandboxMethodType.BEHAVIOR, "send_chart", "发送图表")
async def send_chart(_ctx: AgentCtx, chart_url: str) -> str:
    # 转换为沙盒路径
    image_path = await _ctx.fs.mixed_forward_file(chart_url, file_name="chart.png")

    # 发送图片
    await _ctx.send_image(image_path)

    return "图表已发送"

send_file - 发送文件 ​

python
async def send_file(self, file_path: str, *, record: bool = True):
    """发送文件到当前聊天频道"""

使用示例：

python
@plugin.mount_sandbox_method(SandboxMethodType.BEHAVIOR, "send_report", "发送报告")
async def send_report(_ctx: AgentCtx, report_content: str) -> str:
    # 在共享目录创建文件
    file_on_host = _ctx.fs.shared_path / "report.txt"
    with open(file_on_host, "w", encoding="utf-8") as f:
        f.write(report_content)

    # 转换为沙盒路径
    file_path = _ctx.fs.forward_file(file_on_host)

    # 发送文件
    await _ctx.send_file(file_path)

    return "报告已发送"

push_system - 向当前频道推送系统消息 ​

python
async def push_system(self, message: str, *, trigger_agent: bool = False):
    """向当前聊天频道推送系统消息，可选择是否触发 AI 响应"""

参数说明：

  • message: 要推送的系统消息内容
  • trigger_agent: 是否触发 AI 生成回复（默认 False）

这是 _ctx.ms.push_system() 的便捷形式，自动使用当前的 chat_key。

使用示例：

python
@plugin.mount_sandbox_method(SandboxMethodType.BEHAVIOR, "schedule_analysis", "定时分析")
async def schedule_analysis(_ctx: AgentCtx, report_data: str) -> str:
    # 异步触发当前会话的 AI 分析（比如定时任务完成后）
    analysis_prompt = f"定时数据收集完成，以下是报告数据：\n{report_data}\n请分析并总结要点。"

    await _ctx.push_system(analysis_prompt, trigger_agent=True)

    return "数据收集完成，AI 分析已启动"

@plugin.mount_sandbox_method(SandboxMethodType.BEHAVIOR, "cross_chat_notify", "跨群通知")
async def cross_chat_notify(_ctx: AgentCtx, target_chat: str, event: str) -> str:
    # 向其他聊天群推送事件并触发 AI 处理
    notification = f"来自 {_ctx.channel_name} 的事件通知：{event}\n请评估此事件的重要性。"

    await _ctx.ms.push_system(
        chat_key=target_chat,
        message=notification,
        ctx=_ctx,
        trigger_agent=True
    )

    return f"已向 {target_chat} 发送事件通知"

配置访问 ​

get_core_config - 获取核心配置 ​

python
async def get_core_config(self) -> CoreConfig:
    """获取当前生效的核心配置实例"""

核心配置由三层配置混合生成：系统基本设定 → 适配器设定 → 聊天频道设定，优先级依次递增。

使用示例：

python
@plugin.mount_sandbox_method(SandboxMethodType.TOOL, "check_cloud_status", "检查云服务状态")
async def check_cloud_status(_ctx: AgentCtx) -> str:
    config = await _ctx.get_core_config()

    if config.ENABLE_NEKRO_CLOUD:
        return "云服务已启用"
    else:
        return "云服务已禁用"

特定适配器方法 ​

get_onebot_v11_bot - 获取 OneBot V11 Bot ​

python
async def get_onebot_v11_bot(self) -> OneBotV11Bot:
    """获取 OneBot V11 Bot 实例（仅适用于 OneBot V11 适配器）"""

使用示例：

python
@plugin.mount_sandbox_method(SandboxMethodType.BEHAVIOR, "send_private_msg", "发送私聊消息")
async def send_private_message(_ctx: AgentCtx, user_id: int, message: str) -> str:
    if _ctx.adapter_key == "onebot_v11":
        bot = await _ctx.get_onebot_v11_bot()
        await bot.send_private_msg(user_id=user_id, message=message)
        return f"已向用户 {user_id} 发送私聊消息"
    else:
        return "当前适配器不支持此功能"

最佳实践 ​

1. 错误处理 ​

始终检查关键属性是否存在：

python
@plugin.mount_sandbox_method(SandboxMethodType.TOOL, "get_channel_info", "获取频道信息")
async def get_channel_info(_ctx: AgentCtx) -> str:
    if not _ctx.channel_id:
        return "错误：无法获取频道信息"

    return f"频道ID: {_ctx.channel_id}, 频道名: {_ctx.channel_name}"

2. 文件路径处理 ​

使用 fs 工具进行所有文件操作：

python
# 正确的做法
sandbox_path = await _ctx.fs.mixed_forward_file(url, file_name="image.png")

# 错误的做法 - 不要直接返回 URL 或宿主机路径给 AI
# return url  # AI 可能会在访问外部 URL 时出现困难
# return "/host/path/file.png"  # AI 无法访问宿主机路径

3. 消息记录控制 ​

对于仅面向用户提示性消息，可以设置 record=False：

python
# 进度提示消息不记录到 AI 对话上下文
await _ctx.send_text("正在处理，请稍候...", record=False)

# 需要 AI 参与或知晓结果的提示性消息记录到 AI 对话上下文
await _ctx.send_text("处理完成！结果如下：...", record=True)

4. 适配器兼容性 ​

在使用特定适配器功能前检查适配器类型：

python
if _ctx.adapter_key == "onebot_v11":
    # 使用 OneBot V11 特定功能
    bot = await _ctx.get_onebot_v11_bot()
    # ...
else:
    # 使用通用功能
    await _ctx.send_text("此功能需要 OneBot V11 适配器")

5. 主动触发 AI 响应的使用场景 ​

合理使用 push_system 和 trigger_agent=True 参数：

python
# ✅ 适合触发 AI 的场景：

# 1. 异步数据处理完成，需要 AI 分析结果
await _ctx.push_system("数据采集完成，请分析趋势...", trigger_agent=True)

# 2. 跨聊天频道事件通知
await _ctx.ms.push_system(other_chat, "收到紧急事件，需要处理建议", _ctx, trigger_agent=True)

# 3. 定时任务触发 AI 汇总
await _ctx.push_system("日报数据已准备就绪，请生成总结", trigger_agent=True)

# 4. Webhook 事件需要 AI 响应
await _ctx.push_system("外部系统状态变更，请评估影响", trigger_agent=True)

# ❌ 不适合触发 AI 的场景：
# - 纯状态通知或日志记录
await _ctx.push_system("插件状态更新完成", trigger_agent=False)

# - 简单的用户操作确认
await _ctx.push_system("操作已完成", trigger_agent=False)

注意事项：

  • trigger_agent=True 会消耗 AI 模型调用配额，请合理使用
  • 主要用于异步触发（定时任务、Webhook 等）和跨聊天应用场景
  • 避免在同一个执行流程中自触发，防止循环调用
  • 确保推送的消息内容能够为 AI 提供有价值的上下文信息

通过合理使用 AgentCtx，插件可以与 Nekro Agent 的各个组件无缝集成，提供强大而稳定的功能扩展。

发现文档问题？报告问题