Environments

Understanding execution environments in PandaAGI SDK

What are Environments?

Environments provide isolated execution contexts for general AI agents, allowing them to safely run code, access files, and execute shell commands. The PandaAGI SDK offers three main environment types that balance security, performance, and ease of use.

Local Environment

Direct execution on the host system with working directory isolation

Docker Environment

Containerized execution with strong isolation and port management

File Operations

Read, write, delete files with support for various formats including PDFs

Shell Execution

Execute shell commands with both blocking and non-blocking modes

E2B Environment

Cloud-based sandbox execution with secure isolation via E2B SDK

Local Environment

The LocalEnv executes operations directly on the host filesystem within a specified base directory:

from panda_agi.envs import LocalEnv

# Create local environment
local_env = LocalEnv(base_path="/path/to/workspace")

# Execute shell command
result = await local_env.exec_shell("ls -la")
print(result["stdout"])

# Write a file
await local_env.write_file("hello.txt", "Hello, World!")

# Read a file (supports PDFs too)
content = await local_env.read_file("hello.txt")
print(content["content"])

Docker Environment

The DockerEnv runs operations inside a Docker container for better isolation:

from panda_agi.envs import DockerEnv

# Create Docker environment
docker_env = DockerEnv(
    base_path="/host/workspace",
    image="python:3.9-slim",
    container_name="my-agent-env"
)

# Use as async context manager
async with docker_env as env:
    result = await env.exec_shell("python --version")
    print(result["stdout"])

Basic Setup

from panda_agi.envs import DockerEnv

env = DockerEnv(
    base_path="/workspace",
    image="python:3.9-slim",
    container_name="agent-container",
    working_dir="/workspace",  # Working directory inside container
    auto_start=True,           # Start container automatically
    auto_remove=True           # Remove container when done
)

Volume Mounting

Mount additional directories from the host:

env = DockerEnv(
    base_path="/workspace",
    image="python:3.9-slim",
    volumes={
        "/host/data": "/container/data",
        "/host/models": "/container/models"
    }
)

Environment Variables

Pass environment variables to the container:

env = DockerEnv(
    base_path="/workspace",
    image="python:3.9-slim",
    env_vars={
        "PYTHONPATH": "/workspace/lib",
        "API_KEY": "your-secret-key",
        "DEBUG": "true"
    }
)

Port Management

The Docker environment automatically exposes port 2664 (PandaAGI default) and allows custom port mappings:

# Default behavior - exposes common development ports
env = DockerEnv(
    base_path="/workspace",
    image="python:3.9-slim",
    expose_common_ports=True  # Exposes port 2664 by default
)

# Custom port mappings
env = DockerEnv(
    base_path="/workspace",
    image="python:3.9-slim",
    ports={
        8080: 8080,  # host_port: container_port
        3000: 3000,
        5432: 5432
    }
)

# Add ports dynamically (requires container restart)
env.add_port_mapping(9000, 9000)

# Check exposed ports
ports = env.get_exposed_ports()
print(f"Exposed ports: {ports}")

Network Configuration

Connect to existing Docker networks:

env = DockerEnv(
    base_path="/workspace",
    image="python:3.9-slim",
    network="my-docker-network"
)

File Operations

All environments support comprehensive file operations:

Writing Files

# Write text file
result = await env.write_file("data.txt", "Sample content")
print(f"File written: {result['path']}, Size: {result['size']} bytes")

# Write binary file
with open("image.png", "rb") as f:
    binary_data = f.read()
await env.write_file("copy.png", binary_data, mode="wb")

# Append to file
await env.write_file("log.txt", "New log entry\n", mode="a")

Reading Files

# Read text file
result = await env.read_file("data.txt")
print(result["content"])

# Read binary file
result = await env.read_file("image.png", mode="rb")
binary_content = result["content"]

# Read PDF file (Local environment only)
result = await env.read_file("document.pdf")
if result["status"] == "success":
    print(f"PDF text: {result['content']}")
    print(f"Extracted {result['extracted_text_length']} characters")

File Management

# List files in current directory
result = await env.list_files()
for file_info in result["files"]:
    print(f"{file_info['name']} ({file_info['type']}) - {file_info['size']} bytes")

# List files recursively with hidden files
result = await env.list_files(
    path="src/",
    recursive=True,
    include_hidden=True
)

# Delete file or directory
result = await env.delete_file("temp.txt")
print(result["message"])

# Change working directory
new_dir = env.change_directory("src/components")
print(f"Working directory: {new_dir}")

Shell Command Execution

Blocking Execution

# Simple command execution
result = await env.exec_shell("ls -la")
print(f"Exit code: {result['return_code']}")
print(f"Output: {result['stdout']}")
print(f"Errors: {result['stderr']}")
print(f"Execution time: {result['execution_time']} seconds")

# Command with timeout
result = await env.exec_shell(
    "sleep 10",
    timeout=5.0
)
if result["status"] == "timeout":
    print("Command timed out")

Non-blocking Execution

Perfect for long-running processes or interactive commands:

# Start non-blocking process
result = await env.exec_shell(
    "python server.py",
    blocking=False
)

if result["status"] == "success":
    session_id = result["session_id"]
    print(f"Process started with session ID: {session_id}")

    # Check process status
    status = env.get_process_status(session_id)
    print(f"Process running: {status['running']}")

    # Get process output
    output = env.get_process_output(session_id)
    print(f"Current output: {output['stdout']}")

    # Write to process (Local environment only)
    if hasattr(env, 'write_to_process'):
        env.write_to_process(session_id, "help", press_enter=True)

    # Terminate process
    env.terminate_process(session_id)

E2B Environment

The E2BEnv executes operations in a secure cloud-based sandbox using the E2B Code Interpreter SDK, providing strong isolation without requiring Docker:

from panda_agi.envs import E2BEnv

# Create E2B sandbox environment
e2b_env = E2BEnv(base_path="/workspace", timeout=3600)

# Execute shell command in the sandbox
result = await e2b_env.exec_shell("python --version")
print(result["stdout"])

# Write a file to the sandbox
await e2b_env.write_file("hello.txt", "Hello from E2B sandbox!")

# Read a file from the sandbox
content = await e2b_env.read_file("hello.txt")
print(content["content"])

# Create directories with parent support
await e2b_env.mkdir("path/to/new/dir", parents=True, exist_ok=True)

Key Features

• Cloud-based Execution: All operations run in a secure, isolated cloud environment
• No Local Installation: No need to install Docker or other dependencies locally
• Automatic Cleanup: Sandbox resources are automatically cleaned up after use
• Consistent Interface: Uses the same API as other environment types for seamless switching

Container Lifecycle Management

For Docker environments, you can control the container lifecycle:

# Manual container management
env = DockerEnv(
    base_path="/workspace",
    image="python:3.9-slim",
    auto_start=False
)

# Start container
result = await env.start_container()
print(f"Container ID: {result['container_id']}")
print(f"Exposed ports: {result['exposed_ports']}")

# Check if container is running
if env.is_running:
    print("Container is active")

# Stop container (with optional removal)
await env.stop_container(remove=True)

Best Practices

Environment Selection

Choose the right environment for your use case:

# Development and testing - Local environment
dev_env = LocalEnv(base_path="./workspace")

# Production or isolation needs - Docker environment  
prod_env = DockerEnv(
    base_path="./workspace",
    image="python:3.9-slim",
    auto_remove=True
)

# Cloud-based sandbox isolation - E2B environment
cloud_env = E2BEnv(
    base_path="/workspace",
    timeout=3600
)

Error Handling

Always check operation results: python result = await env.exec_shell("command") if result["status"] != "success": logger.error(f"Command failed: {result.get("message", result.get("stderr"))}") return # Process successful result print(result["stdout"])

Path Management

Use relative paths within the environment:

# Good - relative to working directory
await env.write_file("data/output.txt", content)

# Good - change working directory first
env.change_directory("data")
await env.write_file("output.txt", content)

# Avoid - absolute paths outside base_path
# await env.write_file("/tmp/output.txt", content)

Troubleshooting

Docker Container Issues

Common Docker-related problems:

# Check if Docker daemon is running
try:
    env = DockerEnv(base_path="./workspace")
    result = await env.start_container()
    if result["status"] == "error":
        print(f"Docker error: {result['message']}")
except RuntimeError as e:
    print(f"Docker daemon issue: {e}")

Port Conflicts

Handle port binding conflicts:

try:
    env = DockerEnv(
        base_path="./workspace",
        ports={8080: 8080}
    )
except RuntimeError as e:
    if "already in use" in str(e):
        print("Port 8080 is busy, trying alternative...")
        env = DockerEnv(
            base_path="./workspace", 
            ports={8081: 8080}
        )

Process Management

Handle process lifecycle issues:

# Always check if processes are still running
session_id = "your-session-id"
status = env.get_process_status(session_id)

if status.get("status") == "error":
    print(f"Process session lost: {status['message']}")
elif not status.get("running", False):
    print("Process has terminated")
    # Get final output
    output = env.get_process_output(session_id)
    print(f"Final output: {output.get('stdout', '')}")

Next Steps

Quickstart

Get started quickly with PandaAGI

API Key

Learn how to obtain and use your API key