PyEryx - Python Bindings for Eryx

Python bindings for the Eryx sandbox - execute untrusted Python code securely inside WebAssembly, from within a Python host.

Eryx runs full CPython 3.14 (not a limited subset), with memory and CPU limits, no filesystem or network access by default, and async host callbacks for the access you do want to allow. Used in production at Grafana.

Installation

pip install pyeryx

Note: The package is installed as pyeryx but imported as eryx.

Or build from source using maturin:

cd crates/eryx-python
maturin develop

Quick Start

import eryx

# Create a sandbox with the embedded Python runtime
sandbox = eryx.Sandbox()

# Execute Python code in complete isolation
result = sandbox.execute('''
print("Hello from the sandbox!")
x = 2 + 2
print(f"2 + 2 = {x}")
''')

print(result.stdout)
# Output:
# Hello from the sandbox!
# 2 + 2 = 4

print(f"Execution took {result.duration_ms:.2f}ms")

Features

• Complete Isolation: Sandboxed code runs in WebAssembly with no host access by default
• Controlled Network Access: Optional TCP/TLS networking with host filtering and policies
• Host Callbacks: Expose async Python functions for controlled host interaction
• Resource Limits: Configure timeouts, memory limits, and callback restrictions
• Fast Startup: Pre-initialized Python runtime embedded for ~1-5ms sandbox creation
• Pre-initialization: Custom snapshots with packages for even faster specialized sandboxes
• Package Support: Load Python packages (.whl, .tar.gz) including native extensions
• Persistent Sessions: Maintain Python state across executions with Session
• Virtual Filesystem: Sandboxed file storage with VfsStorage
• Type Safe: Full type stubs for IDE support and static analysis

Python Version

The sandbox runs CPython 3.14 compiled to WebAssembly (WASI). This is the same Python build used by componentize-py from the Bytecode Alliance.

Performance

The pyeryx package ships with a pre-initialized Python runtime embedded in the binary. This means Python's interpreter initialization (~450ms) has already been done at build time, so creating a sandbox is very fast:

import eryx
import time

# First sandbox - fast! (~1-5ms)
start = time.perf_counter()
sandbox = eryx.Sandbox()
print(f"Sandbox created in {(time.perf_counter() - start) * 1000:.1f}ms")

# Execution is also fast
start = time.perf_counter()
result = sandbox.execute('print("Hello!")')
print(f"Execution took {(time.perf_counter() - start) * 1000:.1f}ms")

For repeated sandbox creation with custom packages, see SandboxFactory below.

Each execute() on a fresh sandbox also has to instantiate the WebAssembly runtime. Eryx hides most of that by keeping a pre-instantiated instance ready in the background and handing it to the next execution; the environment variables that tune this (ERYX_WARM_INSTANCES, ERYX_ALLOCATOR, ...) are documented in the Performance Tuning guide. Set them in the environment of the Python process before importing eryx.

API Reference

Core Classes:

• Sandbox - Main class for isolated Python execution
• Session - Persistent state across executions
• ExecuteResult - Execution results with stdout and stats
• ResourceLimits - Configure execution constraints
• NetConfig - Configure network access and policies
• Callbacks - Expose host functions to sandboxed code
• SandboxFactory - Pre-initialize sandboxes with packages
• VfsStorage - Virtual filesystem for sessions

Sandbox vs Session

Feature	Sandbox	Session
State persistence	No - fresh each execute()	Yes - variables persist
Virtual filesystem	No	Yes (optional)
Use case	One-off execution	REPL, multi-step workflows
Isolation	Complete per-call	Complete from host

Use Sandbox when:

• Running untrusted code that should start fresh each time
• Each execution is independent
• You want maximum isolation between executions

Use Session when:

• Building up state across multiple executions
• You need file persistence (VFS)
• Implementing a REPL or notebook-like experience
• Performance matters (no re-initialization per call)

Sandbox

The main class for executing Python code in isolation.

sandbox = eryx.Sandbox(
    resource_limits=eryx.ResourceLimits(
        execution_timeout_ms=5000,      # 5 second timeout
        max_memory_bytes=100_000_000,   # 100MB memory limit
    ),
    network=eryx.NetConfig(             # Optional: enable networking
        allowed_hosts=["api.example.com"]
    ),
    callbacks=[                          # Optional: host functions
        {"name": "get_data", "fn": get_data_fn, "description": "Fetch data"}
    ]
)

result = sandbox.execute("print('Hello!')")

Async and Cancellation

Sandbox, Session and pooled sandboxes have execute_async(), which returns an asyncio.Future so the event loop keeps running while code executes (schedule it with asyncio.ensure_future(), not create_task()). Concurrent execute_async() calls on one Session run one at a time, in call order; reset(), clear_state(), snapshot_state() and restore_state() also wait behind calls made before them.

result = await sandbox.execute_async("print('Hello!')")

Pass a CancelToken as cancel= to stop a run from any thread; the call then raises eryx.CancelledError. Cancelling the asyncio task awaiting execute_async() also stops the run, and Ctrl-C stops a blocking execute() and raises KeyboardInterrupt.

token = eryx.CancelToken()
threading.Timer(5.0, token.cancel).start()
try:
    sandbox.execute("while True: pass", cancel=token)
except eryx.CancelledError:
    print("cancelled")

A cancelled Session run loses its Python state (VFS files are kept), and the next call starts from a fresh interpreter.

Loading Packages

To use custom packages, use SandboxFactory which bundles packages into a reusable runtime snapshot:

import eryx

# Create a factory with your packages (one-time, takes 3-5 seconds)
factory = eryx.SandboxFactory(
    packages=[
        "/path/to/jinja2-3.1.2-py3-none-any.whl",
        "/path/to/markupsafe-2.1.3-wasi.tar.gz",  # WASI-compiled native extension
    ],
    imports=["jinja2"],  # Optional: pre-import for faster first execution
)

# Create sandboxes with packages already loaded (~1ms each)
sandbox = factory.create_sandbox()
result = sandbox.execute('''
from jinja2 import Template
template = Template("Hello, {{ name }}!")
print(template.render(name="World"))
''')

For packages with native extensions (like markupsafe), you need WASI-compiled versions. These are automatically late-linked into the WebAssembly component.

ExecuteResult

Returned by sandbox.execute() with execution results:

• stdout: str - Captured standard output
• stderr: str - Captured standard error
• result: Any - The script's result variable, parsed from JSON into a native Python value, or None if it was not set (see below)
• result_json: Optional[str] - The raw JSON string of the captured result
• result_error: Optional[str] - Why result capture failed (e.g. the value was not JSON-serializable), or None on success
• duration_ms: float - Execution time in milliseconds
• callback_invocations: int - Number of callback invocations
• peak_memory_bytes: Optional[int] - Peak memory usage (if available)
• fuel_consumed: Optional[int] - Fuel (WASM instructions) consumed (if available)

Returning a structured result

Assign a variable named result in your script and it is JSON-serialized and returned as ExecuteResult.result — a structured channel separate from stdout:

result = sandbox.execute('result = {"answer": 42, "items": [1, 2, 3]}')
print(result.result)  # {'answer': 42, 'items': [1, 2, 3]}

If result is not JSON-serializable, result is None and result_error explains why — execution still succeeds. To capture a different variable name, pass result_variable= to Sandbox(...) (or Session(...)):

sandbox = eryx.Sandbox(result_variable="output")

ResourceLimits

Configure execution constraints:

limits = eryx.ResourceLimits(
    execution_timeout_ms=30000,        # Max script runtime (default: 30s)
    callback_timeout_ms=10000,         # Max single callback time (default: 10s)
    max_memory_bytes=134217728,        # Max memory (default: 128MB)
    max_callback_invocations=1000,     # Max callbacks (default: 1000)
)

# Or create unlimited (use with caution!)
unlimited = eryx.ResourceLimits.unlimited()

NetConfig

Configure network access for sandboxed code. By default, all network access is disabled. Enable networking by creating a NetConfig and passing it to the sandbox. A NetConfig allows only the hosts in allowed_hosts; pass allow_all_hosts=True to allow any host that isn't blocked.

import eryx

# Allowlist specific hosts; localhost/private networks stay blocked
config = eryx.NetConfig(
    max_connections=10,                    # Max concurrent connections
    connect_timeout_ms=30000,              # Connection timeout (30s)
    io_timeout_ms=60000,                   # I/O timeout (60s)
    allowed_hosts=["api.example.com"],     # Whitelist specific hosts
    blocked_hosts=[]                       # Override default blocks
)

sandbox = eryx.Sandbox(network=config)
result = sandbox.execute("""
import urllib.request
response = urllib.request.urlopen("https://api.example.com/data")
print(response.read().decode())
""")

Security Defaults

By default, NetConfig blocks localhost and private networks to prevent SSRF attacks:

• localhost, 127.*, [::1]
• Private networks: 10.*, 172.16.*-172.31.*, 192.168.*, 169.254.*

Those patterns match the hostname. Every connection is also checked by the address it resolves to against blocked_networks (CIDRs; by default every non-public range, including cloud metadata), so a public name pointing at an internal address is refused too. To reach an internal service on purpose, use allow_network("10.96.0.0/12") or allow_internal_host("svc.internal.example.com").

Permissive Configuration

For testing or development, use .permissive() to allow all hosts including localhost:

# WARNING: Allows sandbox to access local services
config = eryx.NetConfig.permissive()
sandbox = eryx.Sandbox(network=config)

Host Filtering

Control which hosts sandboxed code can connect to using patterns with wildcards:

config = eryx.NetConfig(
    allowed_hosts=[
        "api.example.com",           # Exact host
        "*.googleapis.com",          # Wildcard subdomain
        "api.*.com",                 # Wildcard in middle
    ]
)

Only matching hosts are allowed (allow_host() and assigning allowed_hosts both turn allow_all_hosts off). Blocked hosts are checked first. To allow any public host:

config = eryx.NetConfig(allow_all_hosts=True)

Builder Methods

Chain methods for convenient configuration:

config = (eryx.NetConfig()
    .allow_host("api.example.com")
    .allow_host("*.openai.com")
    .allow_localhost()               # Remove localhost from blocked list
    .allow_network("10.96.0.0/12")   # Exempt a CIDR from blocked_networks
    .with_root_cert(cert_der_bytes)) # Add custom CA cert for self-signed certs

Custom Certificates

Add custom root certificates for testing with self-signed certificates:

# Load certificate in DER format
with open("ca-cert.der", "rb") as f:
    cert_der = f.read()

config = eryx.NetConfig(allowed_hosts=["internal.example.com"]).with_root_cert(cert_der)
sandbox = eryx.Sandbox(network=config)

Supported Protocols

Network-enabled sandboxes support:

• HTTP/HTTPS via urllib.request, http.client
• Raw TCP/TLS via socket module
• Async networking via asyncio streams
• Third-party libraries like requests, httpx (when loaded via SandboxFactory)

Callbacks

Expose host functions to sandboxed code as async Python functions. Callbacks enable controlled interaction between sandboxed code and the host environment.

Dict-Based API

Simple and explicit, good for dynamic callback registration:

import eryx

def get_time():
    import time
    return {"timestamp": time.time()}

def fetch_user(user_id: int):
    # Call database, API, etc. from host
    return {"id": user_id, "name": "Alice", "email": "alice@example.com"}

sandbox = eryx.Sandbox(
    callbacks=[
        {
            "name": "get_time",
            "fn": get_time,
            "description": "Returns current Unix timestamp"
        },
        {
            "name": "fetch_user",
            "fn": fetch_user,
            "description": "Fetches user data from database"
        }
    ]
)

result = sandbox.execute("""
# Callbacks are available as async functions
t = await get_time()
print(f"Time: {t['timestamp']}")

user = await fetch_user(user_id=42)
print(f"User: {user['name']} ({user['email']})")
""")

Decorator-Based API

More Pythonic, uses CallbackRegistry:

import eryx

registry = eryx.CallbackRegistry()

@registry.callback(description="Greets a person by name")
def greet(name: str, formal: bool = False):
    if formal:
        return {"greeting": f"Good day, {name}"}
    return {"greeting": f"Hey {name}!"}

@registry.callback(name="calc", description="Performs calculation")
def calculate(op: str, a: float, b: float):
    ops = {"add": a + b, "sub": a - b, "mul": a * b, "div": a / b}
    return {"result": ops[op]}

sandbox = eryx.Sandbox(callbacks=registry)

result = sandbox.execute("""
greeting = await greet(name="Alice", formal=True)
print(greeting['greeting'])

result = await calc(op="add", a=10, b=32)
print(f"10 + 32 = {result['result']}")
""")

Callback Requirements

• Must accept JSON-serializable arguments
• Must return JSON-serializable values (typically a dict)
• Can be sync or async (both work the same from Python's perspective)
• Are called as await callback_name(...) from sandboxed code

Discovering Callbacks

Sandboxed code can introspect available callbacks:

result = sandbox.execute("""
import _callbacks
callbacks = _callbacks.list()
for cb in callbacks:
    print(f"{cb['name']}: {cb['description']}")
""")

Error Handling

Callbacks can raise exceptions that propagate to sandboxed code:

def may_fail(should_fail: bool):
    if should_fail:
        raise ValueError("Operation failed!")
    return {"status": "ok"}

sandbox = eryx.Sandbox(
    callbacks=[{"name": "may_fail", "fn": may_fail, "description": "May fail"}]
)

result = sandbox.execute("""
try:
    await may_fail(should_fail=True)
except Exception as e:
    print(f"Caught: {e}")  # Caught: execution failed: ValueError: Operation failed!
""")

Sandboxed code sees only the exception type and message. The host-side traceback is logged on the eryx logger at INFO level.

SandboxFactory

For use cases with custom packages, SandboxFactory lets you create a reusable factory with your packages pre-loaded and pre-imported.

Note: For basic usage without packages, eryx.Sandbox() is already fast (~1-5ms) because the base runtime ships pre-initialized. Use SandboxFactory only when you need to bundle custom packages.

Use cases for SandboxFactory:

• Load packages (jinja2, numpy, etc.) once and create many sandboxes from the factory
• Pre-import modules to eliminate import overhead on first execution
• Pre-execute setup code (environment creation, filter registration) baked into the snapshot
• Save/load factory state to disk for persistence across process restarts

import eryx

# One-time factory creation with packages (takes 3-5 seconds)
factory = eryx.SandboxFactory(
    packages=[
        "/path/to/jinja2-3.1.2-py3-none-any.whl",
        "/path/to/markupsafe-2.1.3-wasi.tar.gz",
    ],
    imports=["jinja2"],  # Pre-import modules
)

# Create sandboxes with packages already loaded (~1ms each)
sandbox = factory.create_sandbox()
result = sandbox.execute('''
from jinja2 import Template
print(Template("Hello {{ name }}").render(name="World"))
''')

# Persistent state with the same preinitialized packages
session = factory.create_session()
session.execute("counter = 1")
session.execute("print(counter)")

# Create many sandboxes from the same factory
for i in range(100):
    sandbox = factory.create_sandbox()
    sandbox.execute(f"print('Sandbox {i}')")

Setup Code

Beyond pre-importing modules, you can run arbitrary Python code during factory construction. The resulting state is captured in the memory snapshot — each sandbox starts with it in copy-on-write memory, preserving full isolation.

factory = eryx.SandboxFactory(
    packages=["jinja2.whl", "markupsafe.whl"],
    imports=["jinja2", "jinja2.sandbox"],
    setup_code="""
from jinja2.sandbox import SandboxedEnvironment

def my_filter(value):
    return value.upper()

env = SandboxedEnvironment()
env.filters['shout'] = my_filter
""",
)

# Every sandbox starts with `env` and `my_filter` already defined
sandbox = factory.create_sandbox()
result = sandbox.execute('''
template = env.from_string("{{ name | shout }}")
print(template.render(name="world"))
''')
print(result.stdout)  # "WORLD"

This reduces per-sandbox execution time by avoiding repeated setup work. The factory build cost (one-time) is negligibly higher since the setup code runs during pre-initialization.

Saving and Loading

Save factories to disk for instant startup across process restarts:

# Save the factory (includes pre-compiled WASM state + package state)
factory.save("/path/to/jinja2-factory.bin")

# Later, in another process - loads in ~10ms (vs 3-5s to recreate)
factory = eryx.SandboxFactory.load("/path/to/jinja2-factory.bin")
sandbox = factory.create_sandbox()

Properties and Methods

• factory.size_bytes - Size of the pre-compiled factory in bytes
• factory.create_sandbox(resource_limits=...) - Create a new sandbox
• factory.create_session(...) - Create a persistent session from the factory
• factory.save(path) - Save factory to a file
• factory.to_bytes() - Get factory as bytes
• SandboxFactory.load(path) - Load factory from a file

Factory sessions default to ResourceLimits() (30-second execution timeout, 128 MB memory, 10-second callback timeout, 1000 callback invocations). A caller-provided VfsStorage retains its own quota (VfsStorage(max_bytes=...), 64 MiB by default); max_vfs_bytes applies to storage created internally for the session. Extracted package storage is kept alive by each session, so imports continue to work after dropping the factory.

Session

Unlike Sandbox which runs each execution in isolation, Session maintains persistent Python state across multiple execute() calls. This is useful for:

• Interactive REPL-style execution
• Building up state incrementally
• Faster subsequent executions (no Python initialization overhead per call)

Like Sandbox, a Session takes resource_limits= and defaults to ResourceLimits() (30-second execution timeout, 128 MB memory).

import eryx

session = eryx.Session()

# State persists across executions
session.execute("x = 42")
session.execute("y = x * 2")
result = session.execute("print(f'{x} * 2 = {y}')")
print(result.stdout)  # "42 * 2 = 84"

# Functions and classes persist too
session.execute("""
def greet(name):
    return f"Hello, {name}!"
""")
result = session.execute("print(greet('World'))")
print(result.stdout)  # "Hello, World!"

Session with Virtual Filesystem

Sessions can optionally use a virtual filesystem (VFS) for persistent file storage that survives across executions and even session resets:

import eryx

# Create shared storage
storage = eryx.VfsStorage()

# Create session with VFS enabled
session = eryx.Session(vfs=storage)

# Write files to the virtual filesystem
session.execute("""
with open('/data/config.json', 'w') as f:
    f.write('{"setting": "value"}')
""")

# Files persist across executions
result = session.execute("""
import json
with open('/data/config.json') as f:
    config = json.load(f)
print(config['setting'])
""")
print(result.stdout)  # "value"

# Files even persist across session.reset()
session.reset()
result = session.execute("print(open('/data/config.json').read())")
# File still exists!

Sharing Storage Between Sessions

Multiple sessions can share the same VfsStorage for inter-session communication:

import eryx

# Shared storage instance
storage = eryx.VfsStorage()

# Session 1 writes data
session1 = eryx.Session(vfs=storage)
session1.execute("open('/data/shared.txt', 'w').write('from session 1')")

# Session 2 reads it
session2 = eryx.Session(vfs=storage)
result = session2.execute("print(open('/data/shared.txt').read())")
print(result.stdout)  # "from session 1"

Custom Mount Path

By default, VFS files are accessible under /data. You can customize this:

session = eryx.Session(vfs=storage, vfs_mount_path="/workspace")
session.execute("open('/workspace/file.txt', 'w').write('custom path')")

State Snapshots

Capture and restore Python state for checkpointing:

session = eryx.Session()
session.execute("x = 42")
session.execute("data = [1, 2, 3]")

# Capture state as bytes (uses pickle internally)
snapshot = session.snapshot_state()

# Clear state
session.clear_state()

# Restore from snapshot
session.restore_state(snapshot)
result = session.execute("print(x, data)")
print(result.stdout)  # "42 [1, 2, 3]"

# Snapshots can be saved to disk and restored in new sessions
with open("state.bin", "wb") as f:
    f.write(snapshot)

Session Properties and Methods

• session.execute(code, cancel=None) - Execute code, returns ExecuteResult
• session.execute_async(code, cancel=None) - Awaitable execute() for asyncio
• session.reset() - Reset Python state (VFS persists)
• session.clear_state() - Clear variables without full reset
• session.snapshot_state() - Capture state as bytes
• session.restore_state(snapshot) - Restore from snapshot
• session.execution_count - Number of executions performed
• session.execution_timeout_ms - Get/set timeout in milliseconds
• session.vfs - Get the VfsStorage (if enabled)
• session.vfs_mount_path - Get the VFS mount path (if enabled)

VfsStorage

In-memory virtual filesystem storage. Files written to the VFS are completely isolated from the host filesystem - sandboxed code cannot access real files.

import eryx

# Create storage (can be shared across sessions)
storage = eryx.VfsStorage()

# Use with Session
session = eryx.Session(vfs=storage)

# Cap the total size of all files (default 64 MiB); writes past it raise
# OSError inside the sandbox
small = eryx.VfsStorage(max_bytes=1024 * 1024)

The VFS supports standard Python file operations:

• open(), read(), write() - File I/O
• os.makedirs(), os.listdir(), os.remove() - Directory operations
• os.path.exists(), os.path.isfile() - Path checks
• pathlib.Path - Full pathlib support

Exceptions

• eryx.EryxError - Base exception for all Eryx errors
• eryx.ExecutionError - Execution failed, either because the code raised or the sandbox failed
  • eryx.PythonExceptionError - The sandboxed Python code raised an uncaught exception
• eryx.InitializationError - Sandbox failed to initialize
• eryx.ResourceLimitError - Resource limit exceeded
• eryx.TimeoutError - Execution timed out (also a built-in TimeoutError)
• eryx.CancelledError - Execution was cancelled

Package Loading

Supported Formats

• .whl - Standard Python wheels (zip archives)
• .tar.gz / .tgz - Tarballs (used by wasi-wheels project)
• Directories - Pre-extracted package directories

Native Extensions

Packages containing native Python extensions (.so files compiled for WASI) are automatically detected and late-linked into the WebAssembly component. This allows packages like numpy, markupsafe, and others to work in the sandbox.

Note: You need WASI-compiled versions of native extensions, not regular Linux/macOS/Windows binaries.

Error Handling

import eryx

sandbox = eryx.Sandbox()

try:
    result = sandbox.execute("raise ValueError('oops')")
except eryx.ExecutionError as e:
    print(f"Code failed: {e}")

try:
    sandbox = eryx.Sandbox(
        resource_limits=eryx.ResourceLimits(execution_timeout_ms=100)
    )
    result = sandbox.execute("while True: pass")
except eryx.TimeoutError as e:
    print(f"Timed out: {e}")

Using eryx with os.fork()

Sandboxes, sessions, factories, pools and MCP managers run on a process-wide async runtime whose worker threads don't survive fork(). A forked child gets a fresh runtime the first time it creates one of these objects, so create new ones in the child; objects inherited from the parent must not be used there. The same applies to multiprocessing's fork start method; spawn and forkserver children start clean.

Development

Building

# Install maturin
pip install maturin

# Build and install in development mode
maturin develop

# Build release wheel
maturin build --release

Testing

pip install pytest
pytest

License

MIT OR Apache-2.0