Code Execution

Sandboxed code execution for agents: environments, security, patterns, and tools.

Why Code Execution?

Sandbox Options

E2B (Cloud VM)

Setup

pip install e2b
export E2B_API_KEY="your-key"

Basic Usage

from e2b import Sandbox

sandbox = Sandbox()

# Run code
result = sandbox.run_code("""
import pandas as pd
df = pd.read_csv("data.csv")
print(df.describe())
""")

print(result.stdout)
print(result.stderr)
print(result.error)

# Files
sandbox.files.write("data.csv", open("data.csv", "rb").read())
result = sandbox.run_code("print(open('data.csv').read()[:100])")

Advanced

# Persistent sandbox
sandbox = Sandbox(template="base", timeout=300)

# Install packages
sandbox.run_code("!pip install numpy scikit-learn -q")

# GPU support
sandbox = Sandbox(gpu=True)

# Internet access (default: true)
sandbox = Sandbox(internet_access=False)

# Commands
sandbox.commands.run("ls -la")

Modal (Cloud Functions)

Setup

pip install modal
modal token new

Basic Usage

import modal

app = modal.App("code-execution")

image = modal.Image.debian_slim().pip_install(
    "pandas", "numpy", "matplotlib", "scikit-learn"
)

@app.function(image=image, timeout=300, cpu=2, memory=4096)
def run_code(code: str) -> dict:
    # Execute in isolated environment
    import subprocess, sys, tempfile, os

    with tempfile.NamedTemporaryFile(mode='w', suffix='.py', delete=False) as f:
        f.write(code)
        f.flush()
        try:
            result = subprocess.run(
                [sys.executable, f.name],
                capture_output=True,
                text=True,
                timeout=60
            )
            return {
                "stdout": result.stdout,
                "stderr": result.stderr,
                "returncode": result.returncode
            }
        finally:
            os.unlink(f.name)

# Local testing
if __name__ == "__main__":
    with modal.run():
        result = run_code.remote("print('Hello from Modal!')")
        print(result)

GPU/GPU

@app.function(
    image=image,
    gpu="A10G",  # or "H100", "A100"
    timeout=600
)
def train_model(code: str):
    # GPU-accelerated execution
    ...

Daytona (Cloud Workspaces)

Setup

pip install daytona
daytona auth login

Usage

from daytona import Daytona, CreateWorkspaceParams

daytona = Daytona()

workspace = daytona.create(CreateWorkspaceParams(
    repository="github.com/user/repo",
    # or blank for empty
))

# Execute in workspace
execution = workspace.execute("python script.py")
print(execution.output)

# File operations
workspace.fs.write_file("script.py", "print('hello')")
workspace.fs.read_file("output.txt")

# Terminal access
terminal = workspace.terminal.create()
terminal.send("python -c 'print(1+1)'")
print(terminal.read())

Local Sandboxes

Docker

import docker

client = docker.from_env()

def run_in_docker(code: str, image="python:3.11-slim") -> dict:
    container = client.containers.run(
        image,
        command=["python", "-c", code],
        detach=True,
        mem_limit="512m",
        cpu_quota=50000,  # 0.5 CPU
        network_mode="none",  # No network
        remove=True
    )

    try:
        result = container.wait(timeout=60)
        logs = container.logs(stdout=True, stderr=True).decode()
        return {"exit_code": result["StatusCode"], "logs": logs}
    except Exception as e:
        container.kill()
        return {"error": str(e)}

gVisor (Stronger Isolation)

# Install gVisor
# runsc install

def run_in_gvisor(code: str) -> dict:
    # runsc provides stronger isolation than Docker
    result = subprocess.run(
        ["runsc", "run", "--", "python", "-c", code],
        capture_output=True,
        text=True,
        timeout=60
    )
    return {"stdout": result.stdout, "stderr": result.stderr}

WASM (Wasmtime)

import wasmtime

# Compile Python to WASM (using wasm32-wasi target)
# Then execute in Wasmtime

engine = wasmtime.Engine()
store = wasmtime.Store(engine)
module = wasmtime.Module.from_file(engine, "python.wasm")
instance = wasmtime.Instance(store, module, [])

# Call exported functions
result = instance.exports(store)["main"]()

Code Execution Patterns

1. REPL / Interactive

class REPLSession:
    def __init__(self, sandbox):
        self.sandbox = sandbox
        self.namespace = {}

    def execute(self, code: str) -> ExecutionResult:
        # Wrap in function to capture returns
        wrapped = f"""
import sys
import io
old_stdout = sys.stdout
sys.stdout = io.StringIO()
try:
    {code}
    result = sys.stdout.getvalue()
finally:
    sys.stdout = old_stdout
print(repr(result))
"""
        return self.sandbox.run_code(wrapped)

2. Notebook-Style (Cell-Based)

class NotebookSession:
    def __init__(self, sandbox):
        self.sandbox = sandbox
        self.cells = []
        self.outputs = []

    def add_cell(self, code: str, cell_type="code"):
        self.cells.append({"code": code, "type": cell_type})

    def execute_cell(self, index: int) -> CellOutput:
        cell = self.cells[index]
        # Build context from previous cells
        context = "\n".join(c["code"] for c in self.cells[:index])
        full_code = context + "\n" + cell["code"]
        result = self.sandbox.run_code(full_code)
        self.outputs.append(result)
        return result

3. Code Generation + Verification

async def generate_and_verify(generator, verifier, max_attempts=3):
    for attempt in range(max_attempts):
        code = await generator.generate()

        # Syntax check
        try:
            compile(code, "<generated>", "exec")
        except SyntaxError as e:
            continue

        # Run tests
        test_result = await run_tests(code)
        if test_result.passed:
            return code

        # Feedback to generator
        await generator.feedback(test_result.failures)

    raise GenerationFailed("Max attempts exceeded")

Security

Threat Model

Hardening Checklist

SANDBOX_CONFIG = {
    # Resource limits
    "cpu_quota": 50000,        # 0.5 CPU
    "memory_limit": "512m",
    "pids_limit": 100,
    "timeout": 60,

    # Filesystem
    "read_only_root": True,
    "tmpfs": ["/tmp", "/home/user"],
    "no_new_privs": True,

    # Network
    "network": "none",  # or "bridge" with egress filter

    # Capabilities
    "drop_all_caps": True,
    "add_caps": [],  # None

    # User
    "user": "1000:1000",  # Non-root
    "working_dir": "/home/user",

    # Output
    "stdout_limit": "10MB",
    "stderr_limit": "10MB"
}

Language Support

Python

# Standard
result = sandbox.run_code("print('Hello')")

# With data science stack
image = modal.Image.debian_slim().pip_install(
    "pandas", "numpy", "matplotlib", "seaborn",
    "scikit-learn", "torch", "transformers"
)

JavaScript/TypeScript

// Node.js sandbox
const result = await sandbox.runCode(`
const fetch = require('node-fetch');
const response = await fetch('https://api.example.com/data');
console.log(await response.json());
`);

SQL

# DuckDB for analytical SQL
result = sandbox.run_code("""
import duckdb
conn = duckdb.connect()
result = conn.execute("SELECT * FROM 'data.parquet' LIMIT 10").fetchdf()
print(result.to_markdown())
""")

R, Julia, Go, Rust

# Any language with container/WASM support
# Just need appropriate base image or WASM target

Output Handling

Capture & Format

class CodeExecutor:
    def __init__(self, sandbox):
        self.sandbox = sandbox

    def execute(self, code: str, capture="all") -> ExecutionResult:
        # Wrap to capture stdout/stderr/return value
        wrapped = self._wrap_for_capture(code)
        result = self.sandbox.run_code(wrapped)

        return ExecutionResult(
            stdout=result.stdout,
            stderr=result.stderr,
            return_value=self._parse_return(result.stdout),
            error=result.error,
            execution_time=result.execution_time
        )

    def _wrap_for_capture(self, code: str) -> str:
        return f"""
import sys, io, json, traceback
old_stdout, old_stderr = sys.stdout, sys.stderr
sys.stdout, sys.stderr = io.StringIO(), io.StringIO()
try:
    result = {code}
    output = {{"stdout": sys.stdout.getvalue(), "stderr": sys.stderr.getvalue(), "return": result}}
except Exception as e:
    output = {{"stdout": sys.stdout.getvalue(), "stderr": traceback.format_exc(), "return": None}}
finally:
    sys.stdout, sys.stderr = old_stdout, old_stderr
print(json.dumps(output))
"""

Streaming Output

async def execute_streaming(code: str) -> AsyncIterator[OutputChunk]:
    # For long-running code with progress
    process = await asyncio.create_subprocess_exec(
        "python", "-c", code,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE
    )

    async def read_stream(stream, type_):
        async for line in stream:
            yield OutputChunk(type=type_, content=line.decode())

    async with asyncio.TaskGroup() as tg:
        tg.create_task(read_stream(process.stdout, "stdout"))
        tg.create_task(read_stream(process.stderr, "stderr"))

    await process.wait()

Related Notes

• Function Calling — Code execution as tool
• MCP Protocol — Code execution via MCP
• Browser Automation — Browser automation
• API Integration — API calls from code
• Custom Tools — Building execution tools
• OpenAI Agents SDK — Code interpreter tool
• LangGraph — Code execution in graphs
• AgentKit — Code execution in AgentKit
• PydanticAI — Code execution tools
• Coding Agents — Code agents
• Data Analysis Agents — Data analysis agents
• Secure Coding — Sandbox security
• Security — Sandbox security
• Observability — Execution observability

Sources: E2B docs, Modal docs, Daytona docs, Docker security, gVisor, Firecracker, Wasmtime, Python sandboxing, modal.com, e2b.dev, daytona.io, NIST sandboxing guidelines.