Nekro Agent 文档站

Appearance

文件交互 ​

插件开发中经常需要处理文件，例如读取用户上传的文件、生成图片或文档返回给 AI、处理多媒体内容等。Nekro Agent 提供了全新的文件系统 API，通过 AgentCtx 的 fs 属性，让插件和 AI 沙盒之间的文件传递变得简单高效。

文件系统概述 ​

在 Nekro Agent 中，AI 在隔离的沙盒环境中运行，而插件在主服务进程中执行。新的文件系统 API 通过 _ctx.fs 提供了两个核心方法来处理文件传递：

  • mixed_forward_file: 插件向 AI 传递文件（URL、字节数据、本地文件等 → 沙盒路径）
  • get_file: AI 向插件传递文件（沙盒路径 → 宿主机真实路径）

核心概念 ​

沙盒路径与宿主机路径 ​

  • 沙盒路径: AI 看到的文件路径，通常以 /app/uploads/ 或 /app/shared/ 开头
  • 宿主机路径: 插件实际操作的文件在宿主机上的真实路径

文件系统 API 自动处理这两种路径之间的转换，开发者无需关心底层实现细节。

插件向 AI 传递文件 ​

mixed_forward_file 方法 ​

当插件需要生成文件（图片、文档、数据等）并返回给 AI 时，使用此方法：

python
async def mixed_forward_file(
    file_source: Union[str, bytes, Path, BinaryIO],
    file_name: Optional[str] = None,
    **kwargs
) -> str

支持的文件来源类型：

  • HTTP/HTTPS URL 字符串
  • 字节数据 (bytes)
  • 本地文件路径 (Path)
  • 文件对象 (BinaryIO)

使用示例 ​

示例 1：从 URL 转发图片 ​

python
from nekro_agent.api.plugin import SandboxMethodType
from nekro_agent.api.schemas import AgentCtx

@plugin.mount_sandbox_method(SandboxMethodType.TOOL, "get_logo", "获取项目 Logo")
async def get_logo(_ctx: AgentCtx) -> str:
    """从 URL 获取 Logo 图片并返回给 AI"""

    # 从外部 URL 获取图片
    image_url = "https://nekro-agent-dev.oss-cn-beijing.aliyuncs.com/images/NA_logo.png"

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

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

示例 2：生成数据文件 ​

python
import json

@plugin.mount_sandbox_method(SandboxMethodType.TOOL, "generate_data_file", "生成数据文件")
async def generate_data_file(_ctx: AgentCtx, data: dict) -> str:
    """生成 JSON 数据文件并返回给 AI"""

    # 将数据序列化为 JSON 字节
    json_data = json.dumps(data, ensure_ascii=False, indent=2).encode('utf-8')

    # 转换为 AI 可用的沙盒路径
    sandbox_path = await _ctx.fs.mixed_forward_file(json_data, file_name="data.json")

    return sandbox_path  # "/app/uploads/data.json"

示例 3：使用插件共享目录 ​

python
@plugin.mount_sandbox_method(SandboxMethodType.TOOL, "create_temp_file", "创建临时文件")
async def create_temp_file(_ctx: AgentCtx, content: str) -> str:
    """在插件共享目录创建临时文件"""

    # 在共享目录创建文件
    temp_file = _ctx.fs.shared_path / "temp.txt"
    with open(temp_file, "w", encoding="utf-8") as f:
        f.write(content)

    # 转换为 AI 可用的沙盒路径
    sandbox_path = _ctx.fs.forward_file(temp_file)

    return sandbox_path  # "/app/shared/temp.txt"

AI 向插件传递文件 ​

get_file 方法 ​

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

python
def get_file(sandbox_path: Union[str, Path]) -> Path

使用示例 ​

示例 1：分析图片文件 ​

python
import aiofiles
from PIL import Image

@plugin.mount_sandbox_method(SandboxMethodType.TOOL, "analyze_image", "分析图片文件")
async def analyze_image(_ctx: AgentCtx, image_path: str) -> str:
    """分析 AI 提供的图片文件"""

    # AI 提供的沙盒路径: "/app/uploads/photo.jpg"
    # 转换为宿主机可访问的真实路径
    host_path = _ctx.fs.get_file(image_path)

    try:
        # 使用 PIL 打开并分析图片
        with Image.open(host_path) as img:
            width, height = img.size
            format_name = img.format
            mode = img.mode

        return f"图片分析结果：尺寸 {width}x{height}，格式 {format_name}，模式 {mode}"

    except Exception as e:
        return f"图片分析失败：{e}"

示例 2：处理文本文件 ​

python
@plugin.mount_sandbox_method(SandboxMethodType.TOOL, "process_text_file", "处理文本文件")
async def process_text_file(_ctx: AgentCtx, file_path: str) -> str:
    """处理 AI 提供的文本文件"""

    # 转换为宿主机路径
    host_path = _ctx.fs.get_file(file_path)

    try:
        # 异步读取文件内容
        async with aiofiles.open(host_path, "r", encoding="utf-8") as f:
            content = await f.read()

        # 简单的文本分析
        lines = len(content.splitlines())
        words = len(content.split())
        chars = len(content)

        return f"文件分析：{lines} 行，{words} 个词，{chars} 个字符"

    except Exception as e:
        return f"文件处理失败：{e}"

结合消息发送的文件处理 ​

使用便捷方法发送文件 ​

AgentCtx 提供了便捷的消息发送方法，可以直接发送文件给用户：

python
@plugin.mount_sandbox_method(SandboxMethodType.BEHAVIOR, "send_report", "生成并发送报告")
async def send_report(_ctx: AgentCtx, title: str, content: str) -> str:
    """生成报告文件并发送给用户"""

    # 创建报告内容
    report_content = f"# {title}\n\n{content}\n\n生成时间：{datetime.now()}"
    report_bytes = report_content.encode('utf-8')

    # 转换为沙盒路径
    sandbox_path = await _ctx.fs.mixed_forward_file(report_bytes, file_name=f"{title}.md")

    # 发送文件给用户
    await _ctx.send_file(sandbox_path)

    return f"报告《{title}》已生成并发送"

发送处理后的图片 ​

python
from PIL import Image, ImageFilter

@plugin.mount_sandbox_method(SandboxMethodType.BEHAVIOR, "apply_blur_filter", "应用模糊滤镜")
async def apply_blur_filter(_ctx: AgentCtx, image_path: str, radius: float = 2.0) -> str:
    """对图片应用模糊滤镜并发送结果"""

    # 获取原始图片的宿主机路径
    host_path = _ctx.fs.get_file(image_path)

    try:
        # 打开并处理图片
        with Image.open(host_path) as img:
            # 应用模糊滤镜
            blurred = img.filter(ImageFilter.GaussianBlur(radius=radius))

            # 保存到临时位置
            output_path = _ctx.fs.shared_path / "blurred_image.jpg"
            blurred.save(output_path, "JPEG")

        # 转换为沙盒路径并发送
        sandbox_path = _ctx.fs.forward_file(output_path)
        await _ctx.send_image(sandbox_path)

        return f"已应用模糊滤镜（半径 {radius}）并发送结果图片"

    except Exception as e:
        return f"图片处理失败：{e}"

文件系统属性 ​

常用路径属性 ​

_ctx.fs 提供了一些有用的路径属性：

python
# 插件共享目录（用于临时文件）
shared_path = _ctx.fs.shared_path

# 用户上传目录
uploads_path = _ctx.fs.uploads_path

# 聊天频道标识
chat_key = _ctx.fs.chat_key

# 容器标识
container_key = _ctx.fs.container_key

实用方法 ​

python
# 将宿主机文件转换为沙盒路径
sandbox_path = _ctx.fs.forward_file(host_file_path)

# 检查文件是否存在
if _ctx.fs.get_file(sandbox_path).exists():
    # 文件存在
    pass

最佳实践 ​

1. 错误处理 ​

始终对文件操作进行适当的错误处理：

python
@plugin.mount_sandbox_method(SandboxMethodType.TOOL, "safe_file_process", "安全的文件处理")
async def safe_file_process(_ctx: AgentCtx, file_path: str) -> str:
    try:
        host_path = _ctx.fs.get_file(file_path)

        if not host_path.exists():
            return "错误：指定的文件不存在"

        # 文件处理逻辑...
        return "处理成功"

    except Exception as e:
        return f"文件处理失败：{e}"

2. 文件类型验证 ​

对于特定类型的文件，建议进行类型验证：

python
import mimetypes

@plugin.mount_sandbox_method(SandboxMethodType.TOOL, "process_image_only", "仅处理图片文件")
async def process_image_only(_ctx: AgentCtx, file_path: str) -> str:
    host_path = _ctx.fs.get_file(file_path)

    # 检查文件类型
    mime_type, _ = mimetypes.guess_type(str(host_path))
    if not mime_type or not mime_type.startswith('image/'):
        return "错误：请提供图片文件"

    # 图片处理逻辑...
    return "图片处理完成"

3. 大文件处理 ​

对于大文件，使用流式处理：

python
@plugin.mount_sandbox_method(SandboxMethodType.TOOL, "hash_large_file", "计算大文件哈希")
async def hash_large_file(_ctx: AgentCtx, file_path: str) -> str:
    import hashlib

    host_path = _ctx.fs.get_file(file_path)

    sha256_hash = hashlib.sha256()

    # 流式读取大文件
    async with aiofiles.open(host_path, "rb") as f:
        while chunk := await f.read(8192):  # 8KB 块
            sha256_hash.update(chunk)

    return f"文件 SHA256: {sha256_hash.hexdigest()}"

4. 临时文件清理 ​

使用上下文管理器或在适当时机清理临时文件：

python
import tempfile
import os

@plugin.mount_sandbox_method(SandboxMethodType.TOOL, "create_temp_processed_file", "创建临时处理文件")
async def create_temp_processed_file(_ctx: AgentCtx, input_data: str) -> str:
    try:
        # 创建临时文件
        with tempfile.NamedTemporaryFile(mode='w', suffix='.txt', delete=False) as temp_file:
            temp_file.write(f"处理结果：{input_data}")
            temp_path = temp_file.name

        # 转换为沙盒路径
        sandbox_path = await _ctx.fs.mixed_forward_file(temp_path, file_name="processed.txt")

        # 清理临时文件
        os.unlink(temp_path)

        return sandbox_path

    except Exception as e:
        return f"临时文件创建失败：{e}"

通过使用新的文件系统 API，插件开发者可以轻松实现复杂的文件处理功能，而无需关心底层的路径转换和环境差异问题。

发现文档问题？报告问题