通过E2B-SDK接入沙箱

安装 SDK

当前使用E2B Python SDK时，仅支持使用小于v2.25.0的版本：

pip install "e2b<2.25.0"   

如需使用 Code Interpreter、Desktop 等场景化能力，按需安装对应扩展包：

pip install e2b_code_interpreter   # Code Interpreter 场景
pip install e2b_desktop            # Desktop 场景

配置环境变量

import os
os.environ["E2B_DOMAIN"] = "sandbox01.cn-shanghai.pai-eas.aliyuncs.com"
os.environ["E2B_API_KEY"] = "<YOUR-API-TOKEN>"        

核心 API

示例

#!/usr/bin/env python
"""
E2B SDK Test Demo for Template
This demo demonstrates how to use the E2B SDK to test all sandbox features
(except PTY terminal).
Before running:
1. Install e2b SDK: pip install "e2b<2.25.0" 
"""
from e2b import Sandbox
import os
import time
import httpx

# Set environment variables
os.environ["E2B_DOMAIN"] = "<YOUR-DOMAIN>"        # 例如 sandbox01.cn-shanghai.pai-eas.aliyuncs.com
os.environ["E2B_API_KEY"] = "<YOUR-API-TOKEN>"

print("=" * 60)
print("E2B Sandbox Test Demo")
print("=" * 60)
# Create sandbox
print("\n[1] Creating sandbox with template: code-interpreter")
sandbox = Sandbox.create(
    template="code-interpreter",
    timeout=300,  # 5 minutes timeout
)
print(f"    ✓ Sandbox created (ID: {sandbox.sandbox_id})")
# Wait for sandbox to be ready
print("\n[2] Waiting for sandbox to be ready...")
max_attempts = 30
for attempt in range(1, max_attempts + 1):
    try:
        is_running = sandbox.is_running()
        status = "Running" if is_running else "Not Running"
        print(f"    [{attempt}/{max_attempts}] Sandbox status: {status}")
        if is_running:
            print("    ✓ Sandbox is now running!")
            break
    except Exception as e:
        print(f"    [{attempt}/{max_attempts}] Failed to check status: {e}")
    if attempt < max_attempts:
        time.sleep(2)
# Test basic command execution
print("\n[3] Testing basic command execution (SDK)...")
try:
    result = sandbox.commands.run("echo 'Hello from sandbox!'")
    if "Hello from sandbox!" in result.stdout:
        print("    ✓ Command execution works!")
        print(f"    Output: {result.stdout.strip()}")
    else:
        print(f"    ✗ Unexpected output: {result.stdout}")
except Exception as e:
    print(f"    ✗ Command execution failed: {e}")
# Test filesystem operations
print("\n[4] Testing filesystem operations (SDK)...")
try:
    test_content = "Hello from filesystem test!"
    sandbox.files.write("/tmp/test.txt", test_content)
    print("    ✓ File written successfully")
    content = sandbox.files.read("/tmp/test.txt")
    if content == test_content:
        print("    ✓ Filesystem read works!")
    files = sandbox.files.list("/tmp")
    file_names = [f.name for f in files]
    if "test.txt" in file_names:
        print("    ✓ Directory listing works!")
    sandbox.files.remove("/tmp/test.txt")
    print("    ✓ Cleanup completed")
except Exception as e:
    print(f"    ✗ Filesystem test failed: {e}")
# Test Python code execution
print("\n[5] Testing Python code execution (SDK)...")
try:
    result = sandbox.commands.run(
        "python -c \"import sys; print('Python version:', sys.version)\"",
        timeout=10
    )
    if "Python version:" in result.stdout:
        print("    ✓ Python execution works!")
        print(f"    {result.stdout.strip()[:60]}...")
    else:
        print("    ✗ Python might not be available")
except Exception as e:
    print(f"    ✗ Python test failed: {e}")
# Test envd health endpoint via direct API
print("\n[6] Testing envd health endpoint (Direct API)...")
try:
    envd_token = getattr(sandbox, '_SandboxBase__envd_access_token', None)
    if envd_token:
        envd_url = f"https://49983-{sandbox.sandbox_id}.sandbox01.cn-shanghai.pai-eas.aliyuncs.com"
        headers = {
            "Content-Type": "application/json",
            "X-Access-Token": envd_token
        }
        resp = httpx.get(f"{envd_url}/health", headers=headers, timeout=5)
        if resp.status_code in [200, 204]:
            print(f"    ✓ Health check passed! (status: {resp.status_code})")
        else:
            print(f"    ✗ Health check failed: {resp.status_code}")
    else:
        print("    ✗ envd_access_token not found in sandbox object")
except Exception as e:
    print(f"    ✗ Health check failed: {e}")
# Test data plane access
print("\n[7] Testing data plane access...")
try:
    # Install Flask
    print("    Installing Flask...")
    result = sandbox.commands.run("python -m pip install flask -q -i https://mirrors.cloud.aliyuncs.com/pypi/simple/ --trusted-host=mirrors.cloud.aliyuncs.com", timeout=60)
    if result.exit_code == 0:
        print("    ✓ Flask installed")
    # Create Flask app
    flask_code = '''
from flask import Flask
app = Flask(__name__)
@app.route('/sandbox_test')
def hello():
    return 'hello sandbox'
if __name__ == '__main__':
    app.run(host='0.0.0.0', port=50001)
'''
    sandbox.files.write("/tmp/flask_app.py", flask_code)
    print("    ✓ Flask app created")
    # Start Flask server
    print("    Starting Flask server on port 50001...")
    sandbox.commands.run("nohup python /tmp/flask_app.py > /tmp/flask.log 2>&1 &", timeout=5)
    time.sleep(2)
    # Check if server is running
    result = sandbox.commands.run("pgrep -f flask_app.py || true", timeout=5)
    if result.stdout.strip():
        print("    ✓ Flask server started")
    # Access through data plane
    envd_token = getattr(sandbox, '_SandboxBase__envd_access_token', None)
    if envd_token:
        data_plane_url = f"https://50001-{sandbox.sandbox_id}.sandbox01.cn-shanghai.pai-eas.aliyuncs.com/sandbox_test"
        print(f"    Accessing data plane URL...")
        headers = {"X-Access-Token": envd_token}
        max_retries = 5
        for attempt in range(max_retries):
            try:
                resp = httpx.get(data_plane_url, headers=headers, timeout=10)
                if resp.status_code == 200 and "hello sandbox" in resp.text:
                    print("    ✓ Data plane access successful!")
                    print(f"    Response: {resp.text.strip()}")
                    break
            except Exception as e:
                if attempt < max_retries - 1:
                    time.sleep(1)
                else:
                    print(f"    ✗ Data plane access failed: {e}")
    else:
        print("    ✗ envd_access_token not found")
except Exception as e:
    print(f"    ✗ Data plane test failed: {e}")
# Check sandbox status
print("\n[8] Testing sandbox status...")
try:
    is_running = sandbox.is_running()
    status = "Running" if is_running else "Not Running"
    print(f"    Sandbox running status: {status}")
except Exception as e:
    print(f"    Failed to check status: {e}")
# Summary
print("\n" + "=" * 60)
print("✓ All tests completed!")
print("=" * 60)
# Kill sandbox
print("\nKilling sandbox...")
sandbox.kill()
print("✓ Sandbox killed")