File Interaction

File handling is often required in plugin development, such as reading user-uploaded files, generating images or documents to return to AI, processing multimedia content, etc. Nekro Agent provides a brand new file system API through the fs attribute of AgentCtx, making file transfer between plugins and AI sandboxes simple and efficient.

File System Overview

In Nekro Agent, AI runs in an isolated sandbox environment, while plugins execute in the main service process. The new file system API provides two core methods through _ctx.fs to handle file transfer:

• mixed_forward_file: Plugin transfers files to AI (URL, byte data, local files, etc. → sandbox path)
• get_file: AI transfers files to plugin (sandbox path → actual path on host machine)

Core Concepts

Sandbox Path vs. Host Path

• Sandbox Path: File path seen by AI, usually starting with /app/uploads/ or /app/shared/
• Host Path: Actual path on the host machine where the plugin actually operates on files

The file system API automatically handles conversion between these two path types, so developers don't need to worry about underlying implementation details.

Plugin Transferring Files to AI

mixed_forward_file Method

When a plugin needs to generate files (images, documents, data, etc.) and return them to AI, use this method:

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

Supported File Source Types:

• HTTP/HTTPS URL strings
• Byte data (bytes)
• Local file paths (Path)
• File objects (BinaryIO)

Usage Examples

Example 1: Forwarding Images from URLs

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

@plugin.mount_sandbox_method(SandboxMethodType.TOOL, "get_logo", "Get project logo")
async def get_logo(_ctx: AgentCtx) -> str:
    """Get logo image from URL and return to AI"""

    # Get image from external URL
    image_url = "https://nekro-agent-dev.oss-cn-beijing.aliyuncs.com/images/NA_logo.png"

    # Convert to sandbox path usable by AI
    sandbox_path = await _ctx.fs.mixed_forward_file(image_url, file_name="logo.png")

    # Return sandbox path to AI
    return sandbox_path  # "/app/uploads/logo.png"

Example 2: Generating Data Files

import json

@plugin.mount_sandbox_method(SandboxMethodType.TOOL, "generate_data_file", "Generate data file")
async def generate_data_file(_ctx: AgentCtx, data: dict) -> str:
    """Generate JSON data file and return to AI"""

    # Serialize data to JSON bytes
    json_data = json.dumps(data, ensure_ascii=False, indent=2).encode('utf-8')

    # Convert to sandbox path usable by AI
    sandbox_path = await _ctx.fs.mixed_forward_file(json_data, file_name="data.json")

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

Example 3: Using Plugin Shared Directory

@plugin.mount_sandbox_method(SandboxMethodType.TOOL, "create_temp_file", "Create temporary file")
async def create_temp_file(_ctx: AgentCtx, content: str) -> str:
    """Create temporary file in plugin shared directory"""

    # Create file in shared directory
    temp_file = _ctx.fs.shared_path / "temp.txt"
    with open(temp_file, "w", encoding="utf-8") as f:
        f.write(content)

    # Convert to sandbox path usable by AI
    sandbox_path = _ctx.fs.forward_file(temp_file)

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

AI Transferring Files to Plugin

get_file Method

When AI calls a plugin and passes a file path parameter, the plugin uses this method to get the actual path on the host machine:

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

Usage Examples

Example 1: Analyzing Image Files

import aiofiles
from PIL import Image

@plugin.mount_sandbox_method(SandboxMethodType.TOOL, "analyze_image", "Analyze image file")
async def analyze_image(_ctx: AgentCtx, image_path: str) -> str:
    """Analyze image file provided by AI"""

    # Sandbox path provided by AI: "/app/uploads/photo.jpg"
    # Convert to actual path accessible by host machine
    host_path = _ctx.fs.get_file(image_path)

    try:
        # Use PIL to open and analyze image
        with Image.open(host_path) as img:
            width, height = img.size
            format_name = img.format
            mode = img.mode

        return f"Image analysis result: size {width}x{height}, format {format_name}, mode {mode}"

    except Exception as e:
        return f"Image analysis failed: {e}"

Example 2: Processing Text Files

@plugin.mount_sandbox_method(SandboxMethodType.TOOL, "process_text_file", "Process text file")
async def process_text_file(_ctx: AgentCtx, file_path: str) -> str:
    """Process text file provided by AI"""

    # Convert to host path
    host_path = _ctx.fs.get_file(file_path)

    try:
        # Asynchronously read file content
        async with aiofiles.open(host_path, "r", encoding="utf-8") as f:
            content = await f.read()

        # Simple text analysis
        lines = len(content.splitlines())
        words = len(content.split())
        chars = len(content)

        return f"File analysis: {lines} lines, {words} words, {chars} characters"

    except Exception as e:
        return f"File processing failed: {e}"

File Handling Combined with Message Sending

Using Convenience Methods to Send Files

AgentCtx provides convenient message sending methods that can directly send files to users:

@plugin.mount_sandbox_method(SandboxMethodType.BEHAVIOR, "send_report", "Generate and send report")
async def send_report(_ctx: AgentCtx, title: str, content: str) -> str:
    """Generate report file and send to user"""

    # Create report content
    report_content = f"# {title}\n\n{content}\n\nGenerated time: {datetime.now()}"
    report_bytes = report_content.encode('utf-8')

    # Convert to sandbox path
    sandbox_path = await _ctx.fs.mixed_forward_file(report_bytes, file_name=f"{title}.md")

    # Send file to user
    await _ctx.send_file(sandbox_path)

    return f"Report '{title}' has been generated and sent"

Sending Processed Images

from PIL import Image, ImageFilter

@plugin.mount_sandbox_method(SandboxMethodType.BEHAVIOR, "apply_blur_filter", "Apply blur filter")
async def apply_blur_filter(_ctx: AgentCtx, image_path: str, radius: float = 2.0) -> str:
    """Apply blur filter to image and send result"""

    # Get host path of original image
    host_path = _ctx.fs.get_file(image_path)

    try:
        # Open and process image
        with Image.open(host_path) as img:
            # Apply blur filter
            blurred = img.filter(ImageFilter.GaussianBlur(radius=radius))

            # Save to temporary location
            output_path = _ctx.fs.shared_path / "blurred_image.jpg"
            blurred.save(output_path, "JPEG")

        # Convert to sandbox path and send
        sandbox_path = _ctx.fs.forward_file(output_path)
        await _ctx.send_image(sandbox_path)

        return f"Applied blur filter (radius {radius}) and sent result image"

    except Exception as e:
        return f"Image processing failed: {e}"

File System Properties

Common Path Properties

_ctx.fs provides some useful path properties:

# Plugin shared directory (for temporary files)
shared_path = _ctx.fs.shared_path

# User upload directory
uploads_path = _ctx.fs.uploads_path

# Chat channel identifier
chat_key = _ctx.fs.chat_key

# Container identifier
container_key = _ctx.fs.container_key

Utility Methods

# Convert host file to sandbox path
sandbox_path = _ctx.fs.forward_file(host_file_path)

# Check if file exists
if _ctx.fs.get_file(sandbox_path).exists():
    # File exists
    pass

Best Practices

1. Error Handling

Always perform appropriate error handling for file operations:

@plugin.mount_sandbox_method(SandboxMethodType.TOOL, "safe_file_process", "Safe file processing")
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 "Error: Specified file does not exist"

        # File processing logic...
        return "Processing successful"

    except Exception as e:
        return f"File processing failed: {e}"

2. File Type Validation

For specific types of files, it is recommended to perform type validation:

import mimetypes

@plugin.mount_sandbox_method(SandboxMethodType.TOOL, "process_image_only", "Process image files only")
async def process_image_only(_ctx: AgentCtx, file_path: str) -> str:
    host_path = _ctx.fs.get_file(file_path)

    # Check file type
    mime_type, _ = mimetypes.guess_type(str(host_path))
    if not mime_type or not mime_type.startswith('image/'):
        return "Error: Please provide an image file"

    # Image processing logic...
    return "Image processing completed"

3. Large File Handling

For large files, use streaming processing:

@plugin.mount_sandbox_method(SandboxMethodType.TOOL, "hash_large_file", "Calculate large file hash")
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()

    # Stream read large file
    async with aiofiles.open(host_path, "rb") as f:
        while chunk := await f.read(8192):  # 8KB chunks
            sha256_hash.update(chunk)

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

4. Temporary File Cleanup

Use context managers or clean up temporary files at appropriate times:

import tempfile
import os

@plugin.mount_sandbox_method(SandboxMethodType.TOOL, "create_temp_processed_file", "Create temporary processed file")
async def create_temp_processed_file(_ctx: AgentCtx, input_data: str) -> str:
    try:
        # Create temporary file
        with tempfile.NamedTemporaryFile(mode='w', suffix='.txt', delete=False) as temp_file:
            temp_file.write(f"Processing result: {input_data}")
            temp_path = temp_file.name

        # Convert to sandbox path
        sandbox_path = await _ctx.fs.mixed_forward_file(temp_path, file_name="processed.txt")

        # Clean up temporary file
        os.unlink(temp_path)

        return sandbox_path

    except Exception as e:
        return f"Temporary file creation failed: {e}"

By using the new file system API, plugin developers can easily implement complex file processing functions without worrying about underlying path conversion and environment differences.