Sandbox System

The NexAU framework provides a powerful sandbox system for secure code execution and file operations. Sandboxes isolate tool execution from the host system, improving security and enabling safe deployment in production environments.

Overview

The sandbox system consists of three main components:

1. BaseSandbox: Abstract interface defining sandbox operations
2. Sandbox Implementations: Concrete implementations (LocalSandbox, E2BSandbox)
3. SandboxManager: Manages sandbox lifecycle (start, stop, pause, restore) and state persistence

Available Sandbox Backends

• LocalSandbox: Executes commands directly on the local system (development/testing only)
• E2BSandbox: Cloud-based sandboxes using E2B (production-ready)

Bash Output Truncation

When executing bash commands, the sandbox automatically redirects stdout and stderr to temporary files on disk and applies smart truncation to prevent oversized outputs from consuming excessive LLM context tokens.

How It Works

1. Output redirection: Every execute_bash call creates a unique output directory under /tmp/nexau_bash_tool_results/{uuid}/, containing command.txt, stdout.txt, and stderr.txt.
2. Smart truncation: After command execution, smart_truncate_output() checks whether the combined stdout + stderr exceeds a character threshold. If so, each stream is independently truncated — keeping the first N and last M characters — with an omitted marker and a hint pointing to the full file.
3. Full output preserved: The untruncated content is always available in the temp files, so the agent can use read_file to access the complete output when needed.

Truncated output example:

[first 5,000 chars of stdout]

... [15,000 characters omitted] ...
(Full output: /tmp/nexau_bash_tool_results/a1b2c3d4/stdout.txt)

[last 5,000 chars of stdout]

CommandResult Fields

The CommandResult dataclass returned by execute_bash includes truncation metadata:

Field	Type	Description
stdout	str	Truncated stdout (if truncation applied)
stderr	str	Truncated stderr (if truncation applied)
truncated	bool	Whether the output was truncated
original_stdout_length	int | None	Full stdout length before truncation
original_stderr_length	int | None	Full stderr length before truncation
output_dir	str | None	Directory containing full output files
stdout_file	str | None	Full path to stdout.txt
stderr_file	str | None	Full path to stderr.txt

Configuring Truncation Parameters

Truncation thresholds are configurable via sandbox_config:

sandbox_config:
  type: local  # or e2b
  output_char_threshold: 10000   # Total char threshold for truncation (default: 10,000)
  truncate_head_chars: 5000      # Characters to keep from the start (default: 5,000)
  truncate_tail_chars: 5000      # Characters to keep from the end (default: 5,000)

Programmatic configuration:

config = AgentConfig(
    name="my_agent",
    sandbox_config={
        "type": "local",
        "output_char_threshold": 20000,
        "truncate_head_chars": 8000,
        "truncate_tail_chars": 8000,
    }
)

Implementation Differences

Aspect	LocalSandbox	E2BSandbox
Redirection	Python-level (Popen(stdout=fout))	Shell-level ({ cmd; } > stdout.txt 2> stderr.txt)
File location	Local filesystem	Remote sandbox filesystem
Reading output	Path.read_text()	sandbox._filesystem.read()
Truncation	Same smart_truncate_output() logic	Same smart_truncate_output() logic

Tip: Because execute_bash already handles its own truncation, you should add execute_bash to the bypass_tool_names list when using LongToolOutputMiddleware to avoid double truncation. See Middleware Hooks — LongToolOutputMiddleware for details.

How to Configure Sandbox

Sandboxes are configured through the agent's YAML configuration file or programmatically via AgentConfig.

YAML Configuration

Add a sandbox_config section to your agent configuration:

name: my_agent
llm_config:
  model: gpt-4

sandbox_config:
  type: local
  work_dir: /path/to/working/directory # Optional, default to current directory
  persist_sandbox: true  # Optional, default to true. If false, sandbox is destroyed after agent.run

E2B Sandbox Configuration

For production use with E2B sandboxes:

sandbox_config:
  type: e2b
  work_dir: /home/user # Optional, default to `/home/user` in E2B
  template: base  # E2B template name, default to "base"
  timeout: 300    # Timeout in seconds, default to 300
  api_key: ${env.E2B_API_KEY}  # Use environment variable
  persist_sandbox: true  # Optional, default to true. If false, sandbox is destroyed after agent.run
  metadata:
    project: my_project
    environment: production
  envs:
    CUSTOM_VAR: value

E2B Configuration Options:

• template: E2B template/image to use (default: "base")
• timeout: Sandbox timeout in seconds (default: 300)
• api_key: E2B API key (can use environment variable E2B_API_KEY)
• api_url: Custom E2B API URL (optional)
• persist_sandbox: Whether to persist sandbox state after agent.run (default: true). If false, sandbox is destroyed after each agent run
• metadata: Custom metadata dictionary for the sandbox
• envs: Environment variables to set in the sandbox

Programmatic Configuration

Same as YAML configuration, but using AgentConfig.

from nexau.archs.main_sub.config import AgentConfig

config = AgentConfig(
    name="my_agent",
    llm_config={"model": "gpt-4"},
    sandbox_config={
        "type": "local"
    }
)

Sandbox State Persistence

Sandbox state is automatically persisted to the session storage, allowing sandboxes to be restored across agent restarts.

How State Persistence Works

1. Automatic Persistence: When a sandbox is created or modified, its state is saved to the session
2. State Restoration: On agent restart, the sandbox manager attempts to restore from saved state
3. Reconnection: For E2B sandboxes, the system reconnects to existing cloud instances

Example: State Lifecycle

# Create agent with sandbox
agent = Agent.from_config("config.yaml")
# Sandbox created with ID: e2b_abc123
# State automatically saved to session

# Pause the sandbox (preserves state in session)
agent.sandbox_manager.pause()
# Sandbox paused, state persisted to session_manager

# Later, restart the sandbox
agent.sandbox_manager.start(
    session_manager=agent._session_manager,
    user_id=agent._user_id,
    session_id=agent._session_id,
    sandbox_config=agent.config.sandbox_config or {}
)
# Sandbox state loaded from session_manager
# Automatically restored and reconnected to E2B sandbox: e2b_abc123

How Pause and Start Work Internally

The BaseSandboxManager implements a sophisticated state management system:

Pause Mechanism

When pause() is called:

1. Backend-Specific Pause: Each sandbox backend implements its own pause logic:

   • E2BSandbox: Calls sandbox.beta_pause() to pause the cloud container
   • LocalSandbox: Simply returns True (no actual pause needed for local execution)

2. State Preservation: The sandbox state remains in _session_context, which stores:

   • session_manager: Reference to the session manager
   • user_id: Current user ID
   • session_id: Current session ID
   • sandbox_config: Configuration used to create the sandbox
   • upload_assets: List of uploaded assets

3. Persistent Storage: The sandbox state is already persisted to the session via persist_sandbox_state():

   sandbox_state = sandbox.dict()  # Serialize sandbox attributes
   sandbox_state["sandbox_type"] = sandbox.__class__.__name__
   await session_manager.update_session_sandbox(
       user_id=user_id,
       session_id=session_id,
       sandbox_state=sandbox_state
   )

Start Mechanism

When start() is called after a pause:

1. Load Saved State: The manager calls load_sandbox_state():

   session = await session_manager.get_session(user_id, session_id)
   if session and session.sandbox_state:
       return session.sandbox_state  # Contains sandbox_id, work_dir, etc.

2. Attempt Restoration: If state exists, try to restore the sandbox:

   • E2BSandbox: Reconnects to existing cloud instance using Sandbox.connect(sandbox_id=...)
   • LocalSandbox: Recreates sandbox instance with saved configuration

3. Fallback to New Creation: If restoration fails, creates a new sandbox instance

4. Asset Re-upload: Re-uploads any assets that were previously uploaded

Auto-Resume on Access

The instance property implements automatic resumption:

@property
def instance(self) -> TSandbox | None:
    if self.start_future is not None:
        self.start_future.result()  # Wait for async start to complete

    if not self.is_running():
        if self._session_context:
            # Auto-resume using saved context
            logger.warning("Sandbox is not running. Try resuming it...")
            self.start_no_wait(**self._session_context)
            if self.start_future is not None:
                self.start_future.result()
        else:
            raise SandboxError("Sandbox is not running. Please run `start_no_wait` first.")

    return self._instance

This means when tools try to access the sandbox after a pause, it automatically resumes without manual intervention.

Asynchronous Operations

• start_no_wait(): Starts sandbox in background thread using ThreadPoolExecutor
• pause_no_wait(): Pauses sandbox in background thread
• start_future: Tracks async start operation, allows waiting for completion

This design ensures:

• Non-blocking initialization: Agent can continue while sandbox starts
• Automatic recovery: Sandbox resumes automatically when accessed
• State continuity: All sandbox state persists across pause/resume cycles
• Session integration: State is tied to session for multi-session support

Sandbox and Session Relationship

Are Sandbox and Session Required Together?

No, they are independent but complementary:

• Sandbox without Session: You can create standalone sandboxes without sessions
• Session without Sandbox: Sessions can exist without sandboxes
• Together: When used together, sandbox state is persisted to the session for continuity

Using Sandbox with Sessions

You can manually integrate sandbox with session manager for state persistence:

from nexau.archs.sandbox import LocalSandboxManager, E2BSandboxManager
from nexau.archs.session import SessionManager
from nexau.archs.session.orm import InMemoryDatabaseEngine

# Initialize session manager
session_manager = SessionManager(
    engine=InMemoryDatabaseEngine.get_shared_instance()
)

# Create sandbox manager
sandbox_manager = E2BSandboxManager(
    template="base",
    timeout=300
)

# Start sandbox with session integration
user_id = "user_123"
session_id = "session_456"
sandbox_config = {
    "type": "e2b",
}

sandbox = sandbox_manager.start(
    session_manager=session_manager,
    user_id=user_id,
    session_id=session_id,
    sandbox_config=sandbox_config
)

# Sandbox state is automatically persisted to session
# On next start() call with same session_id, sandbox will be restored

# Use the sandbox
result = sandbox.execute_bash("echo 'Hello from sandbox'")
print(result.stdout)

# Pause to save state
sandbox_manager.pause()

# Later, restart with same session - state will be restored
sandbox = sandbox_manager.start(
    session_manager=session_manager,
    user_id=user_id,
    session_id=session_id,
    sandbox_config=sandbox_config
)
# Sandbox reconnected to existing instance

Using Sandbox without Sessions

You can use sandbox manager without session persistence by passing session_manager=None:

from nexau.archs.sandbox import LocalSandboxManager, E2BSandboxManager

# Create sandbox manager
sandbox_manager = LocalSandboxManager()

# Start sandbox without session manager (no state persistence)
sandbox = sandbox_manager.start(
    session_manager=None,  # Disables state persistence
    user_id="user_123",
    session_id="session_456",
    sandbox_config={"type": "local"}
)

# Use sandbox operations
result = sandbox.execute_bash("ls -la")
print(result.stdout)

# Pause the sandbox
sandbox_manager.pause()

# Start again - creates a NEW sandbox (state not persisted)
sandbox = sandbox_manager.start(
    session_manager=None,
    user_id="user_123",
    session_id="session_456",
    sandbox_config={"type": "local"}
)
# This is a fresh sandbox, not restored from previous state

Important: When session_manager=None:

• persist_sandbox_state() does nothing (state is not saved)
• load_sandbox_state() returns None (no state to restore)
• Each start() call creates a completely new sandbox instance
• Useful for stateless, ephemeral sandbox usage

How to Connect to an Existing E2B Sandbox

For E2B sandboxes, you can reconnect to an existing sandbox instance by configuring the sandbox_id in your configuration. The sandbox manager will automatically restore the connection.

Configure sandbox_id in YAML

sandbox_config:
  type: e2b
  sandbox_id: e2b_abc123xyz  # ID of existing E2B sandbox
  template: base
  timeout: 300

Programmatic Configuration

from nexau.archs.sandbox import E2BSandboxManager
from nexau.archs.session import SessionManager
from nexau.archs.session.orm import InMemoryDatabaseEngine

# Initialize session manager
session_manager = SessionManager(
    engine=InMemoryDatabaseEngine.get_shared_instance()
)

# Create sandbox manager
sandbox_manager = E2BSandboxManager(
    template="base",
    timeout=300
)

# Start with existing sandbox_id - will reconnect instead of creating new
sandbox_config = {
    "type": "e2b",
    "sandbox_id": "e2b_abc123xyz"  # Existing E2B sandbox ID
}

sandbox = sandbox_manager.start(
    session_manager=session_manager,
    user_id="user_123",
    session_id="session_456",
    sandbox_config=sandbox_config
)
# Sandbox manager detects sandbox_id and reconnects to existing instance

How It Works

When sandbox_id is provided in the configuration:

1. E2B Reconnection: The E2BSandboxManager calls Sandbox.connect(sandbox_id=...) to reconnect to the existing cloud instance
2. State Preservation: All files, environment variables, and running processes in the sandbox are preserved
3. No New Creation: A new sandbox is not created; the existing one is reused
4. Automatic Fallback: If the sandbox_id is invalid or the sandbox no longer exists, an error is raised

This is useful for:

• Long-running tasks: Reconnect to sandboxes with ongoing processes
• Shared sandboxes: Multiple agents/sessions can connect to the same sandbox
• Cost optimization: Reuse existing sandboxes instead of creating new ones

Configuring Sandbox Templates (E2B)

E2B sandboxes use templates (Docker images) to define the execution environment.

Using Built-in Templates

E2B provides several pre-built templates:

sandbox_config:
  type: e2b
  template: base  # Basic Ubuntu environment

Using Custom Templates

Build custom E2B templates from Docker images using the E2B Python SDK:

import asyncio
from e2b import AsyncTemplate
from dotenv import load_dotenv

load_dotenv()

# Create template from Docker image
template = AsyncTemplate().from_image(
    image="mp-bp-cn-shanghai.cr.volces.com/north-prod-images/sandbox_fusion/bp_sandbox:2.0",
    username="xxxxx", # harbor username
    password="xxxxx" # harbor password
)

async def main():
    # Build the template with custom configuration
    await AsyncTemplate.build(
        template,
        alias="north_test",  # Template name/alias
        memory_mb=1024,      # Memory allocation
        skip_cache=True,     # Skip Docker cache
        on_build_logs=lambda log: print(str(log)),  # Build progress logs
    )

if __name__ == "__main__":
    try:
        asyncio.run(main())
    except Exception as e:
        print(f"An error occurred: {e}")

After building, reference the template in your configuration:

sandbox_config:
  type: e2b
  template: north_test  # Use the alias from build
  api_key: ${env.E2B_API_KEY}

Using Sandbox in Custom Tools

Tools can access the sandbox instance to perform secure file operations and code execution.

Basic Tool with Sandbox

from nexau.archs.sandbox import BaseSandbox
from nexau.archs.main_sub.agent_state import AgentState

def file_analyzer(file_path: str, agent_state: AgentState) -> dict:
    """Analyze a file in the sandbox."""

    # Get sandbox instance from agent_state
    sandbox: BaseSandbox | None = agent_state.get_sandbox()

    # Check if file exists
    if not sandbox.file_exists(file_path):
        return {"error": f"File not found: {file_path}"}

    # Read file content
    result = sandbox.read_file(file_path)
    if result.status != SandboxStatus.SUCCESS:
        return {"error": result.error}

    # Get file info
    info = sandbox.get_file_info(file_path)

    return {
        "path": file_path,
        "size": info.size,
        "encoding": info.encoding,
        "lines": len(result.content.split("\n")) if result.content else 0
    }

File Operations in Tools

from nexau.archs.sandbox import BaseSandbox, SandboxStatus
from nexau.archs.main_sub.agent_state import AgentState

def process_data(input_file: str, output_file: str, agent_state: AgentState) -> str:
    """Process data file in sandbox."""

    # Get sandbox instance from agent_state
    sandbox: BaseSandbox | None = agent_state.get_sandbox()

    # Read input file
    read_result = sandbox.read_file(input_file)
    if read_result.status != SandboxStatus.SUCCESS:
        return f"Error reading file: {read_result.error}"

    # Process content
    content = read_result.content
    processed = content.upper()  # Example processing

    # Write output file
    write_result = sandbox.write_file(
        output_file,
        processed,
        create_directories=True
    )

    if write_result.status != SandboxStatus.SUCCESS:
        return f"Error writing file: {write_result.error}"

    return f"Processed {input_file} -> {output_file}"

Code Execution in Tools

from nexau.archs.sandbox import BaseSandbox, CodeLanguage
from nexau.archs.main_sub.agent_state import AgentState

def run_analysis(data: dict, agent_state: AgentState) -> dict:
    """Run Python analysis in sandbox."""

    # Get sandbox instance from agent_state
    sandbox: BaseSandbox | None = agent_state.get_sandbox()

    # Generate Python code
    code = f"""
import json
data = {data}
result = {{
    'count': len(data),
    'keys': list(data.keys())
}}
print(json.dumps(result))
"""

    # Execute in sandbox
    result = sandbox.execute_code(
        code=code,
        language=CodeLanguage.PYTHON,
        timeout=30000  # 30 seconds
    )

    if result.status == SandboxStatus.SUCCESS:
        # Parse output
        import json
        output = result.outputs[0]['text'] if result.outputs else "{}"
        return json.loads(output)
    else:
        return {"error": result.error_value}

Advanced Tool Example

from nexau.archs.sandbox import BaseSandbox, SandboxStatus
from pathlib import Path
from nexau.archs.main_sub.agent_state import AgentState

def project_builder(
    project_name: str,
    files: dict[str, str],
    agent_state: AgentState
) -> dict:
    """Build a project structure in sandbox."""

    # Get sandbox instance from agent_state
    sandbox: BaseSandbox | None = agent_state.get_sandbox()

    project_dir = f"/home/user/projects/{project_name}"

    # Create project directory
    try:
        sandbox.create_directory(project_dir, parents=True)
    except Exception as e:
        return {"error": f"Failed to create directory: {e}"}

    # Create all files
    created_files = []
    for file_path, content in files.items():
        full_path = f"{project_dir}/{file_path}"

        result = sandbox.write_file(
            full_path,
            content,
            create_directories=True
        )

        if result.status == SandboxStatus.SUCCESS:
            created_files.append(file_path)
        else:
            return {
                "error": f"Failed to create {file_path}: {result.error}",
                "created": created_files
            }

    # List project files
    files_list = sandbox.list_files(project_dir, recursive=True)

    return {
        "project": project_name,
        "location": project_dir,
        "files_created": len(created_files),
        "total_files": len(files_list),
        "structure": [f.path for f in files_list if f.is_file]
    }

Integrating New Sandbox Backends

You can create custom sandbox implementations by extending the base classes.

Step 1: Implement BaseSandbox

Create a new sandbox class that implements all abstract methods:

from dataclasses import dataclass
from nexau.archs.sandbox import (
    BaseSandbox,
    CommandResult,
    CodeExecutionResult,
    FileOperationResult,
    FileInfo,
    SandboxStatus,
    CodeLanguage
)

@dataclass(kw_only=True)
class CustomSandbox(BaseSandbox):
    """Custom sandbox implementation."""

    # Add custom configuration fields
    custom_param: str = "default"

    def execute_bash(
        self,
        command: str,
        timeout: int | None = None,
    ) -> CommandResult:
        """Execute bash command in custom environment."""
        # Implement command execution
        # Return CommandResult with status, stdout, stderr, etc.
        pass

    def execute_code(
        self,
        code: str,
        language: CodeLanguage | str,
        timeout: int | None = None,
    ) -> CodeExecutionResult:
        """Execute code in custom environment."""
        # Implement code execution
        pass

    def read_file(
        self,
        file_path: str,
        encoding: str = "utf-8",
        binary: bool = False,
    ) -> FileOperationResult:
        """Read file from custom storage."""
        # Implement file reading
        pass

    def write_file(
        self,
        file_path: str,
        content: str | bytes,
        encoding: str = "utf-8",
        binary: bool = False,
        create_directories: bool = True,
    ) -> FileOperationResult:
        """Write file to custom storage."""
        # Implement file writing
        pass

    # Implement all other abstract methods:
    # - delete_file
    # - list_files
    # - file_exists
    # - get_file_info
    # - create_directory
    # - edit_file
    # - glob
    # - upload_file
    # - download_file
    # - upload_directory
    # - download_directory

Step 2: Implement BaseSandboxManager

Create a manager to handle sandbox lifecycle:

from dataclasses import dataclass, field
from typing import Any
from nexau.archs.sandbox import BaseSandboxManager
from nexau.archs.session import SessionManager

@dataclass(kw_only=True)
class CustomSandboxManager(BaseSandboxManager[CustomSandbox]):
    """Manager for custom sandbox lifecycle."""

    # Configuration fields
    custom_param: str = "default"

    def start(
        self,
        session_manager: SessionManager,
        user_id: str,
        session_id: str,
        sandbox_config: dict[str, Any]
    ) -> CustomSandbox:
        """Start a custom sandbox."""

        # Try to restore from saved state
        sandbox_state = self.load_sandbox_state(
            session_manager, user_id, session_id
        )

        if sandbox_state and sandbox_state.get("sandbox_id"):
            # Restore existing sandbox
            sandbox = CustomSandbox(**sandbox_state)
            # Reconnect to existing instance if needed
            return sandbox

        # Create new sandbox
        sandbox = CustomSandbox(
            sandbox_id=f"custom_{session_id}",
            custom_param=self.custom_param,
            **sandbox_config
        )

        # Initialize sandbox resources
        # ...

        # Persist state
        self.persist_sandbox_state(
            session_manager, user_id, session_id, sandbox
        )

        return sandbox

    def stop(self) -> bool:
        """Stop the sandbox."""
        if self._instance:
            # Cleanup resources
            return True
        return False

    def pause(self) -> bool:
        """Pause the sandbox."""
        # Implement pause logic
        return True

    def is_running(self) -> bool:
        """Check if sandbox is running."""
        return self._instance is not None

Step 3: Register Sandbox Type

Add your sandbox to the sandbox alias mapping:

# In your initialization code
from nexau.archs.sandbox.base_sandbox import SandboxAlias

SandboxAlias["custom"] = "CustomSandbox"

Step 4: Use Custom Sandbox

Configure your agent to use the custom sandbox:

sandbox_config:
  type: custom
  custom_param: my_value
  work_dir: /path/to/workdir

Or programmatically:

from nexau import Agent

config = AgentConfig.from_yaml("config.yaml")
config.sandbox_config = {
    "type": "custom",
    "custom_param": "my_value"
}
agent = Agent(config=config)

Complete Example: Docker Sandbox

Here's a simplified example of a Docker-based sandbox:

import docker
from dataclasses import dataclass, field

@dataclass(kw_only=True)
class DockerSandbox(BaseSandbox):
    """Docker container sandbox."""

    image: str = "python:3.11"
    container_id: str | None = None
    _client: Any = field(default=None, init=False, repr=False)

    def __post_init__(self):
        self._client = docker.from_env()
        if not self.container_id:
            # Create new container
            container = self._client.containers.run(
                self.image,
                detach=True,
                tty=True,
                working_dir=str(self.work_dir)
            )
            self.container_id = container.id

    def execute_bash(self, command: str, timeout: int | None = None) -> CommandResult:
        container = self._client.containers.get(self.container_id)
        exit_code, output = container.exec_run(
            f"bash -c '{command}'",
            workdir=str(self.work_dir)
        )

        return CommandResult(
            status=SandboxStatus.SUCCESS if exit_code == 0 else SandboxStatus.ERROR,
            stdout=output.decode('utf-8'),
            exit_code=exit_code
        )

    # Implement other methods...

API Reference

BaseSandbox Interface

The BaseSandbox abstract class defines the core interface that all sandbox implementations must follow.

Attributes

• sandbox_id (str | None): Unique identifier for the sandbox instance
• work_dir (str): Working directory path in the sandbox (default: current directory)
• work_dir (Path): Property that returns work_dir as a Path object

Command Execution Methods

execute_bash(command: str, timeout: int | None = None) -> CommandResult

Execute a bash command in the sandbox.

Parameters:

• command (str): The bash command to execute
• timeout (int | None): Optional timeout in milliseconds (overrides default)

Returns: CommandResult containing execution results

Raises: SandboxError if command execution fails

execute_code(code: str, language: CodeLanguage | str, timeout: int | None = None) -> CodeExecutionResult

Execute code in the specified programming language.

Parameters:

• code (str): The code to execute
• language (CodeLanguage | str): Programming language (CodeLanguage enum or string)
• timeout (int | None): Optional timeout in milliseconds (overrides default)

Returns: CodeExecutionResult containing execution results and outputs

Raises:

• SandboxError if code execution fails
• ValueError if language is not supported

File Operation Methods

read_file(file_path: str, encoding: str = "utf-8", binary: bool = False) -> FileOperationResult

Read a file from the sandbox.

Parameters:

• file_path (str): Path to the file in the sandbox
• encoding (str): File encoding (default: "utf-8")
• binary (bool): Whether to read file in binary mode

Returns: FileOperationResult containing file content

Raises: SandboxError if file read fails

write_file(file_path: str, content: str | bytes, encoding: str = "utf-8", binary: bool = False, create_directories: bool = True) -> FileOperationResult

Write content to a file in the sandbox.

Parameters:

• file_path (str): Path to the file in the sandbox
• content (str | bytes): Content to write (string or bytes)
• encoding (str): File encoding (default: "utf-8")
• binary (bool): Whether to write file in binary mode
• create_directories (bool): Whether to create parent directories if they don't exist

Returns: FileOperationResult containing operation status

Raises: SandboxError if file write fails

delete_file(file_path: str) -> FileOperationResult

Delete a file from the sandbox.

Parameters:

• file_path (str): Path to the file in the sandbox

Returns: FileOperationResult containing operation status

Raises: SandboxError if file deletion fails

list_files(directory_path: str, recursive: bool = False, pattern: str | None = None) -> list[FileInfo]

List files in a directory in the sandbox.

Parameters:

• directory_path (str): Path to the directory in the sandbox
• recursive (bool): Whether to list files recursively
• pattern (str | None): Optional glob pattern to filter files

Returns: List of FileInfo objects for matching files

Raises: SandboxError if directory listing fails

file_exists(file_path: str) -> bool

Check if a file exists in the sandbox.

Parameters:

• file_path (str): Path to the file in the sandbox

Returns: True if file exists, False otherwise

get_file_info(file_path: str) -> FileInfo

Get information about a file in the sandbox.

Parameters:

• file_path (str): Path to the file in the sandbox

Returns: FileInfo object containing file metadata

Raises: SandboxError if file info retrieval fails

create_directory(directory_path: str, parents: bool = True) -> bool

Create a directory in the sandbox.

Parameters:

• directory_path (str): Path to the directory to create
• parents (bool): Whether to create parent directories if they don't exist

Returns: True if directory created successfully, False otherwise

Raises: SandboxFileError if directory creation fails

edit_file(file_path: str, old_string: str, new_string: str) -> FileOperationResult

Edit a file by replacing old_string with new_string.

Supports three operations:

1. CREATE: Set old_string to empty string to create a new file
2. UPDATE: Provide both old_string and new_string to update existing content
3. REMOVE_CONTENT: Set new_string to empty string to remove the old_string content

Parameters:

• file_path (str): Path to the file to edit
• old_string (str): String to replace (empty for file creation)
• new_string (str): Replacement string (empty for content removal)

Returns: FileOperationResult containing operation status and details

Raises: SandboxFileError if edit operation fails

glob(pattern: str, recursive: bool = True) -> list[str]

Find files matching a glob pattern.

Parameters:

• pattern (str): Glob pattern (e.g., '.py', '**/.txt')
• recursive (bool): Whether to search recursively (default: True)

Returns: List of file paths matching the pattern

Raises: SandboxFileError if glob operation fails

File Upload/Download Methods

upload_file(local_path: str, sandbox_path: str, create_directories: bool = True) -> FileOperationResult

Upload a file from the local filesystem to the sandbox.

Parameters:

• local_path (str): Path to the file on the local filesystem
• sandbox_path (str): Destination path in the sandbox
• create_directories (bool): Whether to create parent directories if they don't exist

Returns: FileOperationResult containing operation status

Raises: SandboxError if file upload fails

download_file(sandbox_path: str, local_path: str, create_directories: bool = True) -> FileOperationResult

Download a file from the sandbox to the local filesystem.

Parameters:

• sandbox_path (str): Path to the file in the sandbox
• local_path (str): Destination path on the local filesystem
• create_directories (bool): Whether to create parent directories if they don't exist

Returns: FileOperationResult containing operation status

Raises: SandboxError if file download fails

upload_directory(local_path: str, sandbox_path: str) -> bool

Upload a directory from the local filesystem to the sandbox.

Parameters:

• local_path (str): Path to the directory on the local filesystem
• sandbox_path (str): Destination path in the sandbox

Returns: True if directory uploaded successfully, False otherwise

Raises: SandboxError if directory upload fails

download_directory(sandbox_path: str, local_path: str) -> bool

Download a directory from the sandbox to the local filesystem.

Parameters:

• sandbox_path (str): Path to the directory in the sandbox
• local_path (str): Destination path on the local filesystem

Returns: True if directory downloaded successfully, False otherwise

Raises: SandboxError if directory download fails

---

BaseSandboxManager Interface

The BaseSandboxManager abstract class manages sandbox lifecycle and state persistence.

Abstract Methods

start(session_manager: SessionManager | None, user_id: str, session_id: str, sandbox_config: dict[str, Any]) -> TSandbox

Start a sandbox for a session.

Parameters:

• session_manager (SessionManager | None): Session manager for state persistence (None disables persistence)
• user_id (str): User identifier
• session_id (str): Session identifier
• sandbox_config (dict[str, Any]): Configuration dictionary for sandbox initialization

Returns: Started sandbox instance

Behavior:

• Attempts to load and restore sandbox state from session if session_manager is not None
• Creates new sandbox if no saved state exists
• Persists sandbox state to session after creation

stop() -> bool

Stop the sandbox.

Returns: True if sandbox stopped successfully, False otherwise

pause() -> bool

Pause the sandbox for a session.

Returns: True if sandbox paused successfully, False otherwise

Behavior:

• Backend-specific pause implementation (E2B pauses cloud instance, Local does nothing)
• Sandbox state remains in _session_context for auto-resume

is_running() -> bool

Check if the sandbox is running.

Returns: True if sandbox is running, False otherwise

Summary

The NexAU sandbox system provides:

• Flexible configuration through YAML or programmatic API
• Automatic state persistence when used with sessions
• Multiple backends (Local for development, E2B for production)
• Easy tool integration with automatic sandbox injection
• Extensible architecture for custom implementations

For production deployments, use E2B sandboxes with proper templates and resource limits. For development and testing, LocalSandbox provides a simple, fast alternative.