HOPX.AI Python SDK

Official Python SDK for HOPX.AI Sandboxes - Execute code safely in isolated microVM environments.

Features

• 🚀 47/47 Features - 100% API coverage
• 🔒 Type Safe - Full Pydantic models with auto-complete
• ⚡ Async Support - Native async/await for all operations
• 🌊 WebSocket Streaming - Real-time code execution, terminal, file watching
• 🎨 Rich Outputs - Automatic capture of plots, images, DataFrames
• 🔐 Security First - Environment variables for secrets, execution timeouts
• 📦 Context Managers - Automatic cleanup with with statement
• 🐍 Pythonic API - Follows Python best practices

Installation

pip install hopx

For WebSocket features (optional):

pip install hopx[websockets]

Quick Start

from hopx import Sandbox

# Create sandbox
with Sandbox.create(template="code-interpreter", api_key="your-api-key") as sandbox:
    # Execute Python code
    result = sandbox.run_code("""
import pandas as pd
df = pd.DataFrame({'a': [1, 2, 3], 'b': [4, 5, 6]})
print(df.sum())
    """)
    
    print(result.stdout)
    # Output: a    6
    #         b   15

Environment Variables

API Key

Set your HOPX API key as an environment variable:

export HOPX_API_KEY="hopx_your_key_here"

Or pass it explicitly:

sandbox = Sandbox.create(
    template="code-interpreter",
    api_key="hopx_your_key_here"
)

Secrets in Code Execution

✅ ALWAYS pass secrets via environment variables:

# ✅ GOOD: Secrets via environment variables
result = sandbox.run_code(
    """
import os
api_key = os.getenv('OPENAI_API_KEY')
# Use api_key safely...
    """,
    env={"OPENAI_API_KEY": "sk-..."}
)

# ❌ BAD: Never hardcode secrets
# result = sandbox.run_code('api_key = "sk-..."')  # DON'T DO THIS!

Core Features

Code Execution

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

# With environment variables
result = sandbox.run_code(
    "import os; print(os.getenv('MY_VAR'))",
    env={"MY_VAR": "value"}
)

# Async execution (non-blocking)
execution_id = sandbox.run_code_async("import time; time.sleep(10)")

# Background execution (fire-and-forget)
execution_id = sandbox.run_code_background("# long task...")

# IPython/Jupyter-style
result = sandbox.run_ipython("df.describe()")

Real-time Streaming (WebSocket)

import asyncio

async def stream():
    async for message in sandbox.run_code_stream("""
import time
for i in range(5):
    print(f"Step {i+1}")
    time.sleep(1)
    """):
        if message['type'] == 'stdout':
            print(message['data'], end='')

asyncio.run(stream())

File Operations

# Write file
sandbox.files.write('/workspace/data.txt', 'Hello, World!')

# Read file
content = sandbox.files.read('/workspace/data.txt')

# List directory
files = sandbox.files.list('/workspace')
for file in files:
    print(f"{file.name} ({file.size_kb:.2f}KB)")

# Upload local file
sandbox.files.upload('./local.txt', '/workspace/remote.txt')

# Download remote file
sandbox.files.download('/workspace/remote.txt', './local.txt')

# Check existence
if sandbox.files.exists('/workspace/data.txt'):
    print("File exists!")

# Remove file/directory
sandbox.files.remove('/workspace/data.txt')

# Create directory
sandbox.files.mkdir('/workspace/mydir')

# Watch filesystem (WebSocket)
async for event in sandbox.files.watch('/workspace'):
    print(f"{event['event']}: {event['path']}")

Commands

# Run shell command
result = sandbox.commands.run('ls -la /workspace')
print(result.stdout)

# With environment variables
result = sandbox.commands.run(
    'echo "Key: $API_KEY"',
    env={'API_KEY': 'secret'}
)

# Custom working directory
result = sandbox.commands.run(
    'pwd',
    working_dir='/workspace/myproject'
)

Environment Variables

# Get all
env_vars = sandbox.env.get_all()

# Set all (replace)
sandbox.env.set_all({'API_KEY': 'sk-...', 'DEBUG': 'true'})

# Update (merge)
sandbox.env.update({'NEW_VAR': 'value'})

# Delete
sandbox.env.delete(['VAR_TO_DELETE'])

# Get single
value = sandbox.env.get('API_KEY')

# Set single
sandbox.env.set('API_KEY', 'sk-...')

Process Management

# List processes
processes = sandbox.list_processes()
for proc in processes:
    print(f"{proc.pid}: {proc.name} (CPU: {proc.cpu_percent}%)")

# Kill process
sandbox.kill_process(1234)

Metrics & Health

# Get metrics
metrics = sandbox.get_metrics_snapshot()
print(f"Total executions: {metrics.total_executions}")
print(f"Uptime: {metrics.uptime_seconds}s")

# Health check
health = sandbox.get_health()
print(f"Status: {health.status}")

# Cache management
cache_stats = sandbox.cache.stats()
print(f"Cache size: {cache_stats.cache.size}")

sandbox.cache.clear()

Desktop Automation

# Note: Requires 'desktop' template
sandbox = Sandbox.create(template='desktop', api_key='...')

# Mouse operations
sandbox.desktop.mouse_click(x=500, y=300, button='left')
sandbox.desktop.mouse_move(x=600, y=400)
sandbox.desktop.drag_drop(from_x=100, from_y=100, to_x=300, to_y=300)

# Keyboard
sandbox.desktop.keyboard_type('Hello, World!')
sandbox.desktop.keyboard_press('Return')
sandbox.desktop.keyboard_press('Control_L+c')  # Ctrl+C

# Clipboard
sandbox.desktop.clipboard_set('text to copy')
content = sandbox.desktop.clipboard_get()

# Screenshots
screenshot = sandbox.desktop.screenshot()
with open('screenshot.png', 'wb') as f:
    f.write(screenshot)

# OCR (text recognition)
text = sandbox.desktop.ocr()
print(f"Found text: {text}")

# Find elements
element = sandbox.desktop.find_element(text='Button')
if element:
    sandbox.desktop.mouse_click(x=element['x'], y=element['y'])

# VNC connection
vnc_info = sandbox.desktop.get_vnc_url()
print(f"VNC URL: {vnc_info['url']}")

Interactive Terminal (WebSocket)

import asyncio

async def terminal_session():
    async with sandbox.terminal.connect() as term:
        # Send commands
        await sandbox.terminal.send_input(term, 'ls -la\n')
        
        # Receive output
        async for message in sandbox.terminal.iter_output(term):
            if message['type'] == 'output':
                print(message['data'], end='')

asyncio.run(terminal_session())

Advanced Configuration

sandbox = Sandbox.create(
    template='code-interpreter',
    api_key='your-api-key',
    vcpu=4,              # 4 vCPUs
    memory_mb=4096,      # 4GB RAM
    disk_gb=20,          # 20GB disk
    region='us-west-2',  # Specific region
    timeout=600,         # 10 minute timeout
    env_vars={           # Pre-set environment variables
        'DATABASE_URL': 'postgres://...',
        'API_KEY': 'sk-...'
    }
)

Error Handling

from hopx.errors import (
    HopxError,
    AuthenticationError,
    NotFoundError,
    FileNotFoundError,
    CodeExecutionError,
    RateLimitError,
    ServerError
)

try:
    with Sandbox.create(template='code-interpreter', api_key='...') as sandbox:
        result = sandbox.run_code("print('Hello')")
        
except AuthenticationError as e:
    print(f"Auth failed: {e.message}")
    print(f"Request ID: {e.request_id}")
    
except FileNotFoundError as e:
    print(f"File not found: {e.path}")
    
except CodeExecutionError as e:
    print(f"Code execution failed: {e.message}")
    print(f"Exit code: {e.exit_code}")
    
except RateLimitError as e:
    print(f"Rate limited: {e.message}")
    
except HopxError as e:
    print(f"API error: {e.message}")
    print(f"Status code: {e.status_code}")

Best Practices

1. Always Use Context Managers

# ✅ GOOD: Automatic cleanup
with Sandbox.create(...) as sandbox:
    # Work here
    pass
# Sandbox automatically deleted

# ❌ BAD: Manual cleanup (easy to forget)
sandbox = Sandbox.create(...)
# ... work ...
sandbox.kill()  # Easy to forget if error occurs!

2. Set Timeouts

# Prevent infinite execution
result = sandbox.run_code(code, timeout=30)

3. Optimize Performance

# Set environment variables once
sandbox.env.set_all({'API_KEY': 'sk-...', 'DB_URL': '...'})

# Now available in all executions
for i in range(100):
    sandbox.run_code('...')  # Env vars already set

Use Cases

OpenAI: ChatGPT Code Interpreter

def execute_user_code(user_code: str):
    with Sandbox.create(template="code-interpreter") as sandbox:
        result = sandbox.run_code(user_code, timeout=30)
        return {
            "output": result.stdout,
            "error": result.stderr,
            "rich_outputs": result.rich_outputs  # Images, plots
        }

Stripe: Payment Reports

with Sandbox.create(template="code-interpreter") as sandbox:
    sandbox.env.set('STRIPE_SECRET_KEY', os.getenv('STRIPE_SECRET_KEY'))
    
    result = sandbox.run_code("""
import stripe
stripe.api_key = os.environ['STRIPE_SECRET_KEY']
charges = stripe.Charge.list(limit=100)
# Generate report...
    """)

Documentation

• Cookbook: See cookbook/python for 10 complete examples
• API Reference: docs.hopx.ai
• GitHub: github.com/hopx-ai/hopx

License

MIT License - See LICENSE file for details.

Support

• Issues: github.com/hopx-ai/hopx/issues
• Email: support@hopx.ai
• Documentation: docs.hopx.ai