文件交互

插件经常需要与文件系统进行交互，例如读取用户上传的文件、写入处理结果、加载大型数据资源等。然而，由于 Nekro Agent 的沙盒执行机制，插件开发者在处理文件路径时必须格外小心，并理解沙盒环境与主服务执行环境之间的差异。

沙盒与主服务的路径差异

如前所述，插件的沙盒方法虽然由沙盒中的 AI 代码调用，但其实际执行发生在 Nekro Agent 的主服务进程中。这两个环境拥有不同的文件系统视图：

• 沙盒环境：通常是一个容器化的、受限的文件系统。沙盒中的代码看到的路径（例如，/app/uploads/somefile.txt 或 /app/shared/data.json）可能与主服务中的实际物理路径完全不同。
• 主服务环境：Nekro Agent 核心进程运行的环境，拥有对系统文件（如 data/ 目录）的直接访问权限。

直接在沙盒方法中返回或使用硬编码的沙盒内部路径是错误的，因为这些路径在主服务中通常无效，反之亦然。

路径转换工具

为了解决这个问题，Nekro Agent 提供了一套路径转换工具，帮助你在沙盒路径和主服务可访问路径之间进行映射。

from pathlib import Path # 建议使用 Path 对象处理路径
from nekro_agent.api.schemas import AgentCtx # 获取上下文信息，如 chat_key
from nekro_agent.tools.path_convertor import (
    convert_to_host_path,      # 将沙盒容器内路径转换为主机可访问的物理路径
    convert_to_container_path, # 将主机路径（主要指文件名）转换为沙盒容器内 /app/uploads/ 下的路径
    convert_filename_to_container_path, # 将文件名转换为沙盒容器内 /app/uploads/ 下的路径
    convert_filename_to_access_path, # 将文件名转换为用户上传目录中主机可访问的物理路径
    is_url_path,               # 检查路径是否为 URL
    PathLocation               # 枚举，定义了常见的路径位置类型 (用于概念理解)
)
# from nekro_agent.core.config import USER_UPLOAD_DIR, SANDBOX_SHARED_HOST_DIR # 底层配置

PathLocation 枚举 (概念性)

PathLocation 定义了 Nekro Agent 管理的两种主要文件存储区域。虽然在最新的路径转换函数中，这些位置通常被自动检测，但理解它们有助于开发者明白文件存储的逻辑：

• PathLocation.UPLOADS: 用户上传文件的目录。
  • 沙盒视角: 通常以 /app/uploads/ 开头，例如 /app/uploads/image.png。
  • 主机视角: 对应于 data/uploads/{chat_key}/{filename} 或类似的、按会话隔离的物理路径。
• PathLocation.SHARED: 沙盒与主机共享的目录，通常用于临时文件交换或特定沙盒实例的数据。
  • 沙盒视角: 通常以 /app/shared/ 开头，例如 /app/shared/temp_data.json。
  • 主机视角: 对应于 data/sandboxes/{container_key}/{path_after_shared_in_sandbox}，按沙盒实例隔离。例如，沙盒路径 /app/shared/output/result.txt 可能主机路径 data/sandboxes/my_container_id/output/result.txt。

路径转换函数

• convert_to_host_path(sandbox_path: Path, chat_key: str, container_key: Optional[str] = None) -> Path

  • 将沙盒容器内的路径（通常由 AI 提供，如 /app/uploads/file.txt 或 /app/shared/data.json）转换为主机操作系统可以访问的物理 Path 对象。
  • 该函数会自动检测 sandbox_path 的类型（UPLOADS 或 SHARED）基于其路径结构（例如，是否包含 /uploads/ 或 /shared/ 段）。
  • sandbox_path: 来自沙盒的 Path 对象。如果是相对路径，则会相对于沙盒的 /app 目录进行解析。
  • chat_key: 当前会话的唯一标识。对于 UPLOADS 区域的文件是必需的，以定位到特定会话的上传目录。
  • container_key: 沙盒容器的唯一标识。仅当 sandbox_path 指向 SHARED 区域时才需要，用于定位特定沙盒实例的共享目录。
  • 何时使用：当你的沙盒方法接收到一个来自 AI（在沙盒中）的文件路径参数时，你需要用此函数将其转换为主服务可以实际读写该文件的路径。

• convert_filename_to_access_path(filename: str | Path, chat_key: str) -> Path

  • 根据给定的文件名和 chat_key，构建该文件在主机上对应会话上传区域的完整物理路径。
  • 路径格式通常为：{USER_UPLOAD_DIR}/{chat_key}/{filename}。
  • 何时使用：当你的插件在主服务中生成了一个新文件，并且希望将其保存到当前会话的用户上传目录时，使用此函数获取目标主机路径。

• convert_filename_to_container_path(filename: str | Path) -> Path

  • 将一个文件名（或包含文件名的 Path 对象）转换为沙盒内 /app/uploads/ 目录下的概念路径。
  • 例如，输入 "report.txt" 或 Path("report.txt")，将返回 Path("/app/uploads/report.txt")。
  • get_sandbox_path(filename: str | Path) -> Path 是此函数的别名。
  • 何时使用：当你的插件在主机端（通常在由 convert_filename_to_access_path 确定的路径）生成了一个文件后，如果需要将此文件的路径返回给 AI（在沙盒中），以便 AI 可以在其代码中引用此文件（假定它位于其标准上传目录），则使用此函数。

• convert_to_container_path(path: Path) -> Path

  • 这是一个更通用的函数，它接收一个主机端的 Path 对象，提取其文件名，然后将其转换为沙盒内 /app/uploads/ 目录下的概念路径。
  • 例如，输入 Path("/data/uploads/some_chat_key/report.txt")，将返回 Path("/app/uploads/report.txt")。
  • 何时使用：与 convert_filename_to_container_path 类似，但当你已经有一个主机 Path 对象而不是只有文件名时使用。主要用于将主机端已存在（或刚创建）的、逻辑上属于“上传”性质的文件，转换为 AI 可理解的沙盒内上传区路径。

• is_url_path(path: str) -> bool

  • 检查给定的字符串路径是否是一个 HTTP 或 HTTPS URL。
  • 何时使用：在处理可能是文件路径也可能是 URL 的输入时，进行判断。

重要原则：

• 输入转换（AI -> 主机）：如果沙盒方法接收来自 AI 的文件路径参数（通常是沙盒内的绝对路径，如 /app/uploads/...），立即使用 convert_to_host_path 将其转换为可操作的主机路径。
• 输出转换（主机 -> AI）：
  • 如果插件在主机端生成文件，并计划将其存放在会话的上传区（以便 AI 可以通过 /app/uploads/... 引用）：
    1. 使用 convert_filename_to_access_path 结合文件名和 chat_key 来获取文件应保存的主机物理路径。
    2. 将文件写入该主机路径。
    3. 使用 convert_filename_to_container_path (或 convert_to_container_path 如果你已有主机 Path 对象) 结合文件名来获取 AI 可在沙盒中使用的沙盒路径，然后返回此路径给 AI。
• 避免直接返回主机物理路径给 AI：AI 在沙盒中无法直接访问主机物理路径。
• 优先传递内容或引用而非路径：如果可能，优先直接处理文件内容并返回结果，或者返回一个可供 AI 理解的抽象引用（如下载链接、通过消息API发送的文件），而不是直接传递文件系统路径，以减少路径转换的复杂性和潜在错误。特别是对于不由 Agent 标准上传/共享机制管理的文件（例如插件自身数据目录中的文件）。

示例：处理 AI 提供的上传文件路径

import aiofiles
from pathlib import Path
from nekro_agent.api import core
from nekro_agent.api.schemas import AgentCtx
from nekro_agent.tools.path_convertor import convert_to_host_path

@plugin.mount_sandbox_method(SandboxMethodType.TOOL, "process_uploaded_image", "处理用户上传的图片文件。")
async def process_image(_ctx: AgentCtx, image_sandbox_path_str: str) -> str:
    """接收沙盒中的图片路径字符串，在主服务中处理它。

    Args:
        image_sandbox_path_str (str): AI 在沙盒中看到的图片路径 (例如 "/app/uploads/my_image.jpg")。

    Returns:
        str: 处理结果的描述。
    """
    try:
        image_sandbox_path = Path(image_sandbox_path_str)
        # 1. 将沙盒路径转换为主机可访问路径
        # chat_key 通常从 _ctx.from_chat_key 获取
        # 如果是 /app/shared/ 路径，且需要 container_key, _ctx 通常也会提供
        # container_key = _ctx.container_key
        host_image_path = convert_to_host_path(
            sandbox_path=image_sandbox_path,
            chat_key=_ctx.from_chat_key
            # container_key=container_key # 如果是 shared 路径则需要
        )
        core.logger.info(f"接收到沙盒路径 '{image_sandbox_path_str}'，转换为主机路径 '{host_image_path}'")

        # 2. 在主服务中读取和处理文件
        if not host_image_path.exists():
            return f"错误：文件在主机路径 '{host_image_path}' 未找到。可能是路径转换错误或文件确实不存在。"

        async with aiofiles.open(host_image_path, "rb") as f:
            image_data = await f.read()
        
        # ... 在这里进行图片处理 (例如：分析、缩放、添加滤镜等) ...
        # processed_result = await image_processing_library(image_data)
        
        return f"图片 '{host_image_path.name}' 处理成功。大小: {len(image_data)} 字节。"
    
    except ValueError as ve: # 特别捕捉路径转换相关的 ValueError
        core.logger.error(f"路径转换失败 '{image_sandbox_path_str}': {ve}")
        return f"处理图片失败，路径无效: {ve}"
    except Exception as e:
        core.logger.error(f"处理图片 '{image_sandbox_path_str}' 失败: {e}")
        return f"处理图片失败: {e}"

示例：插件生成文件并返回沙盒路径给 AI

import aiofiles
import time
from pathlib import Path
from nekro_agent.api import core
from nekro_agent.api.schemas import AgentCtx
from nekro_agent.tools.path_convertor import (
    convert_filename_to_access_path,
    convert_filename_to_container_path
)
# from nekro_agent.services.message import message # 如果要通过消息发送

@plugin.mount_sandbox_method(SandboxMethodType.TOOL, "generate_report_file", "生成一份报告并返回其沙盒路径。")
async def generate_report(_ctx: AgentCtx, report_data: str) -> str:
    """生成报告文件，将其保存在会话的上传区，并返回 AI 可在沙盒中使用的路径。

    Args:
        report_data (str): 报告的内容。

    Returns:
        str: 生成的报告文件在沙盒中的可访问路径 (例如 "/app/uploads/report_1678886400.txt")。
    """
    try:
        report_filename = f"report_{int(time.time())}.txt"
        
        # 1. 使用 convert_filename_to_access_path 确定文件在主机上的保存路径
        # 这会将其放入 data/uploads/{chat_key}/report_filename.txt
        host_report_path = convert_filename_to_access_path(
            filename=report_filename,
            chat_key=_ctx.from_chat_key
        )
        
        # 确保目录存在
        host_report_path.parent.mkdir(parents=True, exist_ok=True)

        # 2. 在主服务中生成并写入文件
        async with aiofiles.open(host_report_path, "w", encoding='utf-8') as f:
            await f.write(report_data)
        core.logger.info(f"报告已生成在主机路径: {host_report_path}")

        # 3. 使用 convert_filename_to_container_path 将主机文件名转换为沙盒可访问的 /app/uploads/ 路径
        container_report_path = convert_filename_to_container_path(report_filename)
        
        core.logger.info(f"返回给AI的沙盒路径: {container_report_path}")
        return str(container_report_path)

    except Exception as e:
        core.logger.error(f"生成报告 '{report_data[:50]}...' 失败: {e}")
        return f"生成报告失败: {e}"

    # 或者，考虑更稳健的方式，特别是如果文件不适合放在通用的“上传”区域，
    # 或者你想避免让AI直接操作文件路径：
    # 
    # 1. 将文件保存到插件专属目录:
    #    plugin_data_dir = plugin.get_plugin_path() 
    #    my_custom_report_path = plugin_data_dir / "reports" / report_filename
    #    my_custom_report_path.parent.mkdir(parents=True, exist_ok=True)
    #    async with aiofiles.open(my_custom_report_path, "w", encoding='utf-8') as f:
    #        await f.write(report_data)
    #
    # 2. 通过消息服务发送文件给用户:
    #    await message.send_file(_ctx.from_chat_key, str(my_custom_report_path), _ctx, filename=report_filename)
    #    return f"报告 '{report_filename}' 已生成并通过消息发送给您。"
    #
    # 3. 或者返回一个抽象的引用/描述给 AI:
    #    return f"报告 '{report_filename}' 已在服务器端生成（位于插件专属目录）。"

异步文件操作

由于 Nekro Agent 是基于异步 (asyncio) 构建的，所有耗时的 I/O 操作（包括文件读写）都应该使用异步方式进行，以避免阻塞主事件循环。推荐使用 aiofiles 库进行异步文件操作。

import aiofiles
from nekro_agent.api import core # 假设 core.logger 可用

async def read_file_content(file_path: str) -> str: # 或 Path
    try:
        async with aiofiles.open(file_path, mode="r", encoding="utf-8") as f:
            content = await f.read()
        return content
    except FileNotFoundError:
        core.logger.error(f"文件未找到: {file_path}")
        return ""
    except Exception as e:
        core.logger.error(f"读取文件 {file_path} 失败: {e}")
        return ""

数据序列化与文件交互

当沙盒方法通过 RPC 与主服务通信时，其参数和返回值会通过 pickle 进行序列化和反序列化。这带来一些限制：

• 可序列化类型：只支持 Python 内置的基本类型（str, int, list, dict 等）以及正确实现了 __getstate__ 和 __setstate__ 的自定义类的实例。
• 不可直接传递对象：文件句柄、网络连接、数据库连接等包含运行时状态的复杂对象不能直接在沙盒方法参数或返回值中传递。
• 数据大小限制：避免通过 RPC 传递非常大的数据块（如整个大文件的内容），这可能导致性能问题或超出序列化限制。对于大型数据，应考虑：
  1. 将数据保存到文件系统（使用上述路径转换和异步操作）。
  2. 在沙盒方法的参数或返回值中传递文件的引用（如转换后的沙盒路径或下载 URL）。
  3. AI 或插件的另一部分再根据此引用去读取文件内容。

错误示例（传递文件句柄）：

# 错误！文件句柄不可序列化
@plugin.mount_sandbox_method(SandboxMethodType.TOOL, "get_file_handle_WRONG", "")
async def get_file_handle(_ctx: AgentCtx, host_path_str: str): # host_path_str 应为转换后的路径
    f = open(host_path_str, "r") # 同步打开也是问题
    return f # 这是错误的！

正确做法（传递内容或路径）：

import aiofiles
from pathlib import Path
from nekro_agent.api.schemas import AgentCtx
from nekro_agent.tools.path_convertor import convert_to_host_path

@plugin.mount_sandbox_method(SandboxMethodType.TOOL, "get_file_content_CORRECT", "")
async def get_file_content(_ctx: AgentCtx, sandbox_path_str: str) -> str:
    # sandbox_path_str 示例: "/app/uploads/document.txt"
    sandbox_path = Path(sandbox_path_str)
    host_path = convert_to_host_path(sandbox_path, _ctx.from_chat_key) # 假设是 UPLOADS
    
    async with aiofiles.open(host_path, "r", encoding="utf-8") as f: # 使用 aiofiles
        content = await f.read()
    return content # 返回可序列化的字符串内容

最佳实践总结

1. 始终转换路径：
   • 从 AI 获取沙盒路径（如 /app/uploads/... 或 /app/shared/...）后，使用 convert_to_host_path 转换为可操作的主机路径。
   • 当插件在主机端生成文件并希望 AI 能通过沙盒的 /app/uploads/ 路径访问时，先用 convert_filename_to_access_path 获取主机保存路径，写入文件后，再用 convert_filename_to_container_path 生成沙盒路径返回给 AI。
2. 理解文件区域：理解 UPLOADS (会话相关) 和 SHARED (沙盒实例相关) 区域的概念，以及它们在沙盒和主机上的大致路径结构，有助于正确使用转换工具和定位文件。convert_to_host_path 会自动检测这些。
3. 使用插件专属目录：通过 plugin.get_plugin_path() 获取插件的数据目录，用于存储插件自身的持久化文件或不适合放入公共 UPLOADS/SHARED 区的文件。若要让 AI 访问这些文件，通常应通过消息API发送或提供下载链接，而非直接转换其路径。
4. 异步操作优先：对所有文件I/O使用 aiofiles 等异步库，避免阻塞。
5. 谨慎传递路径给 AI：优先传递文件内容、处理结果或可通过其他方式（如下载链接、消息发送）访问的引用。如果确实需要返回路径，确保是 AI 可理解的沙盒路径（通常是 /app/uploads/...）。
6. 注意序列化限制：不要在 RPC 调用中传递大型数据或不可序列化的对象（如文件句柄）；应通过文件系统作为中介，传递路径或引用。
7. 权限和安全：确保插件只在授权的目录中读写文件，并对用户输入的文件名/路径进行适当的清理和验证，防止路径遍历等安全风险。