Code Executor

Execute Python code snippets using either secure sandboxed execution or high-performance local execution.

Executor Types

Built-In Code Executor (Sandboxed)

Secure execution in isolated E2B containers with full feature support.

from ag_kit_py.tools.code_executers import BuiltInCodeExecutor

executor = BuiltInCodeExecutor(
    sandbox=sandbox,  # Optional: provide existing sandbox
    api_key="your_api_key",  # Optional: E2B API key
    domain="your_domain",  # Optional: E2B domain
    timeout_ms=60000
)

Unsafe Local Code Executor

Direct execution on local system with identical feature set.

from ag_kit_py.tools.code_executers import UnsafeLocalCodeExecutor

executor = UnsafeLocalCodeExecutor(
    timeout_ms=30000,
    on_stdout=lambda data: print(f"OUT: {data['line']}"),
    on_stderr=lambda data: print(f"ERR: {data['line']}")
)

Common Features

Both executors support the same feature set:

Feature	Built-In (Sandboxed)	Unsafe Local
Security	✅ Isolated	⚠️ Full access
Performance	Network overhead	✅ Maximum speed
Context Persistence	✅ Yes	✅ Yes
Real-time Output	✅ on_stdout/on_stderr	✅ on_stdout/on_stderr
Environment Variables	✅ envs	✅ envs
Timeout Control	✅ timeout_ms	✅ timeout_ms
Multi-language	✅ Python/JS/TS/Java/R/Bash	✅ Python/JS/TS/Bash

Basic Usage

from ag_kit_py.tools.code_executers import UnsafeLocalCodeExecutor

executor = UnsafeLocalCodeExecutor()

result = await executor.invoke({
    "code": 'x = "Hello World!"\nprint(x)',
    "language": "python"
})

if result.success:
    print(result.data.results[0].text)  # "Hello World!"

Advanced Features

Persistent Context

Variables persist between executions in both executors:

# First execution
await executor.invoke({
    "code": "x = 10",
    "language": "python"
})

# Second execution - x is still available
result = await executor.invoke({
    "code": "print(x)",  # Outputs: 10
    "language": "python"
})

if result.success:
    print(result.data.logs.stdout)  # ['10']

Real-time Output Streaming

executor = UnsafeLocalCodeExecutor(
    on_stdout=lambda data: print(f"OUT: {data['line']}", end=""),
    on_stderr=lambda data: print(f"ERR: {data['line']}", end="")
)

result = await executor.invoke({
    "code": """
import time
for i in range(3):
    print(f"Step {i}")
    time.sleep(1)
""",
    "language": "python"
})
# Real-time output: OUT: Step 0, OUT: Step 1, OUT: Step 2

Environment Variables

import os

# Set environment variables before execution
os.environ["API_KEY"] = "secret-key"
os.environ["DEBUG"] = "true"

# Built-In Executor
sandbox_executor = BuiltInCodeExecutor()

# Local Executor  
local_executor = UnsafeLocalCodeExecutor()

result = await executor.invoke({
    "code": "import os; print(os.environ.get('API_KEY'))",
    "language": "python"
})

if result.success:
    print(result.data.logs.stdout)  # ['secret-key']

Timeout Configuration

executor = UnsafeLocalCodeExecutor(
    timeout_ms=60000  # 60 seconds
)

# For Built-In Executor
sandbox_executor = BuiltInCodeExecutor(
    timeout_ms=60000
)

Setup Instructions

Built-In Executor Setup

export AG_KIT_SANDBOX_API_KEY=your_sandbox_api_key
export AG_KIT_SANDBOX_DOMAIN=your_sandbox_domain  # Optional

import os
from ag_kit_py.tools.code_executers import BuiltInCodeExecutor
from ag_kit_py.tools.fs.env_utils import setup_e2b_env
from e2b_code_interpreter import Sandbox

# Setup E2B environment (loads .env file automatically)
api_key = setup_e2b_env()
if not api_key:
    raise ValueError("E2B API key not found. Set E2B_API_KEY in .env file or environment")

# Custom sandbox configuration
sandbox = Sandbox.create(api_key=api_key)

executor = BuiltInCodeExecutor(sandbox=sandbox)

Environment Setup: Create a .env file in your project root:
E2B_API_KEY=your_e2b_api_key_here
Get your API key from E2B Dashboard.

Local Executor Setup

Ensure Python is installed:

python3 --version  # Python 3.8+
node --version     # Node.js
bash --version     # Bash

API Reference

Common Interface

from typing import Optional, Callable, Dict, Literal, Any
from pydantic import BaseModel

class OutputMessage(BaseModel):
    """Output message from code execution"""
    line: str
    timestamp: Optional[float] = None

class CodeExecutorInput(BaseModel):
    """Input for code execution"""
    code: str
    language: Optional[Literal["python", "js", "ts", "java", "r", "bash"]] = "python"

class CodeExecutorOptions(BaseModel):
    """Configuration options for code executor"""
    timeout_ms: Optional[int] = 30000
    envs: Optional[Dict[str, str]] = None
    on_stdout: Optional[Callable[[OutputMessage], None]] = None
    on_stderr: Optional[Callable[[OutputMessage], None]] = None
    on_result: Optional[Callable[[Any], None]] = None
    on_error: Optional[Callable[[Exception], None]] = None

Constructor Signatures

# Built-In Executor
BuiltInCodeExecutor(
    sandbox: Optional[Sandbox] = None,
    api_key: Optional[str] = None,
    domain: Optional[str] = None,
    timeout_ms: int = 3600 * 1000
)

# Local Executor
UnsafeLocalCodeExecutor(
    timeout_ms: int = 30000,
    on_stdout: Optional[Callable[[Dict], None]] = None,
    on_stderr: Optional[Callable[[Dict], None]] = None
)

Execution Result

from typing import Optional
from e2b_code_interpreter import Execution

class ToolResult:
    """Wrapper for execution result"""
    success: bool  # Whether execution succeeded
    data: Execution  # E2B Execution object
    error: Optional[str] = None  # Error message if failed

# E2B Execution object structure:
class Execution:
    results: List[Result]  # Execution outputs
    logs: Logs  # Captured logs
    error: Optional[ExecutionError] = None  # Error if execution failed
    execution_count: int  # Execution counter

class Result:
    text: str  # Text representation
    json_data: Optional[Any]  # Parsed JSON if applicable
    # ... other fields

class Logs:
    stdout: List[str]  # Standard output lines
    stderr: List[str]  # Standard error lines

class ExecutionError:
    name: str  # Error type
    value: str  # Error message
    traceback: str  # Full traceback

Language Support

Language	Built-In	Local	Notes
Python	✅	✅	Full support
JavaScript	✅	✅	Full support
TypeScript	✅	✅	Full support
Bash	✅	✅	Full support
Java	✅	❌	Sandbox only
R	✅	❌	Sandbox only

Examples

Basic Computation

from ag_kit_py.tools.code_executers import UnsafeLocalCodeExecutor

executor = UnsafeLocalCodeExecutor()

# Simple calculation
result = await executor.invoke({
    "code": "print(2 + 2)",
    "language": "python"
})

if result.success:
    print(result.data.logs.stdout[0])  # "4"

Data Processing

# Process data
result = await executor.invoke({
    "code": """
import json

data = [1, 2, 3, 4, 5]
result = {
    'sum': sum(data),
    'avg': sum(data) / len(data),
    'max': max(data)
}
print(json.dumps(result, indent=2))
""",
    "language": "python"
})

if result.success:
    print(result.data.logs.stdout)  # JSON output as list of strings

Function Definition and Usage

# Define and use functions
result = await executor.invoke({
    "code": """
def process_data(items):
    return [item.upper() for item in items if len(item) > 3]

data = ['hello', 'hi', 'world', 'foo']
result = process_data(data)
print(result)
""",
    "language": "python"
})

if result.success:
    print(result.data.logs.stdout)  # ["['HELLO', 'WORLD']"]

Environment-Based Selection

import os
from ag_kit_py.tools.code_executers import BuiltInCodeExecutor, UnsafeLocalCodeExecutor

def create_executor():
    use_local = os.environ.get('USE_LOCAL_EXECUTOR') == 'true'
    
    if use_local:
        return UnsafeLocalCodeExecutor(
            timeout_ms=60000,
            on_stdout=lambda data: print(data['line'], end="")
        )
    else:
        return BuiltInCodeExecutor(
            timeout_ms=60000,
            api_key=os.environ.get('E2B_API_KEY')
        )

Error Handling

from ag_kit_py.tools.code_executers import UnsafeLocalCodeExecutor

executor = UnsafeLocalCodeExecutor()

try:
    result = await executor.invoke({
        "code": user_code,
        "language": "python"
    })

    if result.success:
        if result.data.error:
            print(f"Execution failed: {result.data.error.value}")
            print(f"Traceback: {result.data.error.traceback}")
        else:
            print("Execution succeeded!")
            if result.data.results:
                print(f"Result: {result.data.results[0].text}")
    else:
        print(f"System error: {result.error}")
except Exception as e:
    print(f"Unexpected error: {e}")

Security Considerations

Built-In Executor

✅ Safe for untrusted code
✅ Network isolation
✅ Filesystem restrictions
✅ Resource limits

Local Executor

⚠️ Full system access
⚠️ Can modify files
⚠️ Can make network requests
⚠️ Can execute system commands

Security Guidelines:

Use Built-In for untrusted code
Validate input before local execution
Run in restricted environments when possible
Monitor resource usage

Performance Tips

Built-In Executor

Reuse sandbox instances
Batch multiple executions
Use connection pooling

Local Executor

Leverage persistent context
Use real-time callbacks for long operations
Optimize interpreter startup time

Cleanup

# Clean up Built-In Executor (closes sandbox)
if isinstance(executor, BuiltInCodeExecutor):
    await executor.close()

# Local executor doesn't require explicit cleanup

Troubleshooting

Built-In Executor

Sandbox creation failed: Check AG_KIT_SANDBOX_API_KEY
Network timeout: Increase timeout or check connectivity

Local Executor

Python not found: Install Python 3.8+
Permission denied: Check file/directory permissions
Context lost: Session may have terminated, restart executor

Best Practices

Set reasonable timeouts: Prevent infinite loops
Validate input: Sanitize code before execution
Use sandboxed execution: For untrusted code
Monitor resources: Track memory and CPU usage
Handle errors gracefully: Catch and log execution errors
Clean up resources: Call cleanup() when done

Examples Repository

Find complete working examples at:

code_executor.py - Comprehensive code executor examples
langchain_adapter.py - LangChain integration
llamaindex_adapter.py - LlamaIndex integration

Bash Tools

Execute shell commands

Filesystem Tools

File system operations

Framework Adapters

LangChain & LlamaIndex integration

MCP Integration

Model Context Protocol

TypeScript SDK

Python SDK

Headless UI Components

CLI Tools

​Code Executor

​Executor Types

​Built-In Code Executor (Sandboxed)

​Unsafe Local Code Executor

​Common Features

​Basic Usage

​Advanced Features

​Persistent Context

​Real-time Output Streaming

​Environment Variables

​Timeout Configuration

​Setup Instructions

​Built-In Executor Setup

​Local Executor Setup

​API Reference

​Common Interface

​Constructor Signatures

​Execution Result

​Language Support

​Examples

​Basic Computation

​Data Processing

​Function Definition and Usage

​Environment-Based Selection

​Error Handling

​Security Considerations

​Built-In Executor

​Local Executor

​Performance Tips

​Built-In Executor

​Local Executor

​Cleanup

​Troubleshooting

​Built-In Executor

​Local Executor

​Best Practices

​Examples Repository

​Related Resources

Bash Tools

Filesystem Tools

Framework Adapters

MCP Integration

Code Executor

Executor Types

Built-In Code Executor (Sandboxed)

Unsafe Local Code Executor

Common Features

Basic Usage

Advanced Features

Persistent Context

Real-time Output Streaming

Environment Variables

Timeout Configuration

Setup Instructions

Built-In Executor Setup

Local Executor Setup

API Reference

Common Interface

Constructor Signatures

Execution Result

Language Support

Examples

Basic Computation

Data Processing

Function Definition and Usage

Environment-Based Selection

Error Handling

Security Considerations

Built-In Executor

Local Executor

Performance Tips

Built-In Executor

Local Executor

Cleanup

Troubleshooting

Built-In Executor

Local Executor

Best Practices

Examples Repository

Related Resources