python-sdk/README.md

Wrenn Python SDK

Python client for the Wrenn microVM platform. Create isolated capsules, execute commands, manage files, run interactive terminals, and execute persistent code -- all from Python.

Designed as a drop-in replacement for e2b. If you're migrating, just swap your imports.

Installation

pip install wrenn

Requires Python 3.13+.

Authentication

Set the WRENN_API_KEY environment variable:

export WRENN_API_KEY="wrn_your_api_key_here"

Optionally override the API base URL:

export WRENN_BASE_URL="https://app.wrenn.dev/api"  # default

For self-hosted deployments you can also override the capsule proxy domain (used to build {port}-{capsule_id}.<domain> URLs returned by Capsule.get_url):

export WRENN_PROXY_DOMAIN="wrenn.example.com"

Resolution order: explicit proxy_domain= kwarg → WRENN_PROXY_DOMAIN env → wrenn.dev when base_url is the default app.wrenn.dev host, else the base_url host (with port) verbatim.

You can also pass credentials directly:

from wrenn import WrennClient, Capsule

# WrennClient also accepts a timeout (httpx.Timeout or float seconds).
# Default: 30s read/write/pool, 10s connect.
client = WrennClient(
    api_key="wrn_...",
    base_url="https://...",
    proxy_domain="wrenn.example.com",  # optional override
    timeout=30.0,                       # optional override
)

capsule = Capsule(api_key="wrn_...", base_url="https://...")

---

Wrenn Capsules

Quick Start

from wrenn import Capsule

# Create a capsule (reads WRENN_API_KEY from env)
with Capsule(template="minimal") as capsule:
    result = capsule.commands.run("echo hello")
    print(result.stdout)  # "hello\n"

Creating Capsules

from wrenn import Capsule

# Direct construction (creates immediately)
capsule = Capsule()
capsule = Capsule(template="base-python", vcpus=2, memory_mb=1024, timeout=300)

# With auto-wait (blocks until capsule is running)
capsule = Capsule(template="minimal", wait=True)

# Via factory classmethod
capsule = Capsule.create(template="minimal", wait=True)

Context Manager

Use capsules as context managers for automatic cleanup (destroys capsule on exit):

with Capsule(template="minimal", wait=True) as capsule:
    capsule.commands.run("echo hello")
# capsule is automatically destroyed

Connecting to Existing Capsules

Attach to a running capsule by ID. If it's paused, it will be resumed automatically:

capsule = Capsule.connect("cl-abc123")
result = capsule.commands.run("echo still running")

For code runner capsules:

from wrenn.code_runner import Capsule as CodeCapsule

capsule = CodeCapsule.connect("cl-abc123")
result = capsule.run_code("print('reconnected')")

Lifecycle Management

# Instance methods
capsule.pause()
capsule.resume()
capsule.destroy()
capsule.ping()           # reset inactivity timer
capsule.wait_ready()     # block until running

info = capsule.get_info()
print(info.status)       # "running"
print(capsule.is_running())  # True

# Static methods (no instance needed)
Capsule.destroy("cl-abc123", api_key="wrn_...")
Capsule.pause("cl-abc123")
Capsule.resume("cl-abc123")
info = Capsule.get_info("cl-abc123")

# List all capsules
capsules = Capsule.list()

Command Execution

Commands are accessed via capsule.commands:

# Foreground (blocks until complete)
result = capsule.commands.run("python -c 'print(42)'")
print(result.stdout)       # "42\n"
print(result.stderr)       # ""
print(result.exit_code)    # 0
print(result.duration_ms)  # 35

# With options
result = capsule.commands.run(
    "python train.py",
    timeout=120,
    envs={"CUDA_VISIBLE_DEVICES": "0"},
    cwd="/app",
)

# Background process
handle = capsule.commands.run("python server.py", background=True)
print(handle.pid)  # 1234
print(handle.tag)  # "exec-abc123"

Streaming Output

import sys

# Stream a new command
for event in capsule.commands.stream("python", args=["-u", "train.py"]):
    match event.type:
        case "start":
            print(f"PID: {event.pid}")
        case "stdout":
            print(event.data, end="")
        case "stderr":
            print(event.data, end="", file=sys.stderr)
        case "exit":
            print(f"\nExited with code {event.exit_code}")

# Connect to a running background process
for event in capsule.commands.connect(handle.pid):
    match event.type:
        case "start":
            print(f"PID: {event.pid}")
        case "stdout":
            print(event.data, end="")

Process Management

# List running processes
for proc in capsule.commands.list():
    print(proc.pid, proc.cmd, proc.tag)

# Kill a process
capsule.commands.kill(pid=1234)

Filesystem

Files are accessed via capsule.files:

# Write and read files
capsule.files.write("/app/main.py", "print('hello')")
content = capsule.files.read("/app/main.py")        # str
raw = capsule.files.read_bytes("/app/main.py")       # bytes

# Check existence
capsule.files.exists("/app/main.py")  # True

# List directory
entries = capsule.files.list("/home/user", depth=1)
# FileEntry has: name, type (file/dir), size, modified_at
for entry in entries:
    print(entry.name, entry.type, entry.size)

# Create directory
capsule.files.make_dir("/app/data")

# Remove file or directory
capsule.files.remove("/app/old_data")

Streaming (Large Files)

# Streaming upload
def chunks():
    yield b"chunk1"
    yield b"chunk2"

capsule.files.upload_stream("/data/large.bin", chunks())

# Streaming download
for chunk in capsule.files.download_stream("/data/large.bin"):
    process(chunk)

Git

Git operations are accessed via capsule.git. All commands execute the real git binary inside the capsule:

# Initialize a repo
capsule.git.init("/app", initial_branch="main")

# Configure user
capsule.git.configure_user("Alice", "alice@example.com", cwd="/app")

# Stage and commit
capsule.git.add(all=True, cwd="/app")
capsule.git.commit("initial commit", cwd="/app")

# Check status
status = capsule.git.status(cwd="/app")
print(status.branch)    # "main"
print(status.is_clean)  # True
for f in status.files:
    print(f.path, f.index_status, f.work_tree_status)

# Branches
branches = capsule.git.branches(cwd="/app")
capsule.git.create_branch("feature", cwd="/app")
capsule.git.checkout_branch("main", cwd="/app")
capsule.git.delete_branch("feature", cwd="/app")

Clone with Authentication

# Clone a private repo (credentials are stripped from remote URL after clone)
capsule.git.clone(
    "https://github.com/org/repo.git",
    username="user",
    password="ghp_token",
    cwd="/app",
)

# Push/pull with inline credentials (temporarily embedded, then restored)
capsule.git.push("origin", "main", username="user", password="ghp_token", cwd="/app")
capsule.git.pull("origin", "main", username="user", password="ghp_token", cwd="/app")

Configuration and Remotes

capsule.git.set_config("core.autocrlf", "false", cwd="/app")
value = capsule.git.get_config("user.name", cwd="/app")  # str | None

capsule.git.remote_add("upstream", "https://github.com/org/repo.git", cwd="/app")
url = capsule.git.remote_get("origin", cwd="/app")  # str | None

# Reset and restore
capsule.git.reset(mode="hard", ref="HEAD~1", cwd="/app")
capsule.git.restore(["file.txt"], staged=True, cwd="/app")

Persistent Credential Store

For workflows that need repeated authenticated operations, you can persist credentials via the git credential store:

capsule.git.dangerously_authenticate(
    username="user",
    password="ghp_token",
    host="github.com",
    protocol="https",
)

Warning: Credentials are written in plaintext inside the capsule and are accessible to any process running there. Prefer per-operation username/password on clone, push, and pull instead.

Git errors raise GitCommandError (or GitAuthError for authentication failures), both inheriting from GitError:

from wrenn import GitCommandError, GitAuthError

try:
    capsule.git.push("origin", "main", username="user", password="bad", cwd="/app")
except GitAuthError as e:
    print(e.stderr)
    print(e.exit_code)

Interactive Terminal (PTY)

import sys

with capsule.pty(cmd="/bin/bash", cols=80, rows=24, cwd="/home/user") as term:
    term.write(b"ls -la\n")
    for event in term:
        if event.type == "output":
            sys.stdout.buffer.write(event.data)
        elif event.type == "exit":
            break

# Reconnect to an existing session
with capsule.pty_connect(term.tag) as term:
    term.write(b"echo reconnected\n")

PtySession methods:

Method	Description
write(data: bytes)	Send raw bytes to stdin
resize(cols, rows)	Resize the terminal
kill()	Send SIGKILL to the process
tag	Session tag (after started event)
pid	Process PID (after started event)

Proxy URL

Access services running inside a capsule:

url = capsule.get_url(8080)
# "wss://8080-cl-abc123.app.wrenn.dev"

Snapshots

Create reusable templates from running capsules:

template = capsule.create_snapshot(name="my-template", overwrite=True)

---

Code Runner

The wrenn.code_runner module provides a specialized capsule for stateful code execution via a persistent Jupyter kernel. Defaults to the code-runner-beta template and the wrenn Jupyter kernelspec.

The legacy module path wrenn.code_interpreter still works but emits a FutureWarning on import. Use wrenn.code_runner.

Quick Start

from wrenn.code_runner import Capsule

with Capsule(wait=True) as capsule:
    result = capsule.run_code("print('hello')")
    print("".join(result.logs.stdout))  # "hello\n"

Stateful Execution

Variables, imports, and function definitions persist across run_code calls:

from wrenn.code_runner import Capsule

with Capsule(wait=True) as capsule:
    capsule.run_code("x = 42")
    result = capsule.run_code("x * 2")
    print(result.text)  # "84"

    capsule.run_code("import math")
    result = capsule.run_code("math.pi")
    print(result.text)  # "3.141592653589793"

    capsule.run_code("def greet(name): return f'hello {name}'")
    result = capsule.run_code("greet('world')")
    print(result.text)  # "hello world"

The text property returns the text/plain value of the main execute_result (the last expression in the cell). Printed output goes to result.logs.stdout instead.

Error Handling in Code

result = capsule.run_code("1 / 0")
print(result.error.name)       # "ZeroDivisionError"
print(result.error.value)      # "division by zero"
print(result.error.traceback)  # full traceback string

Rich Output

Each call to display(), plt.show(), or similar produces a Result in execution.results. Known MIME types are unpacked into named fields:

result = capsule.run_code("""
import matplotlib.pyplot as plt
plt.plot([1, 2, 3])
plt.show()
""")
for r in result.results:
    if r.png:
        print(f"Got PNG image ({len(r.png)} bytes base64)")
    print(r.formats())  # e.g. ["text", "png"]

Streaming Callbacks

capsule.run_code(
    code,
    on_result=lambda r: print("result:", r.formats()),
    on_stdout=lambda text: print("stdout:", text),
    on_stderr=lambda text: print("stderr:", text),
    on_error=lambda err: print(f"error: {err.name}: {err.value}"),
)

Custom Templates and Kernels

By default, the code-runner-beta template and the wrenn Jupyter kernelspec are used. Override either:

capsule = Capsule(
    template="my-custom-jupyter-template",
    kernel="python3",
    wait=True,
)
result = capsule.run_code("print('running on custom template')")

Capsule reuses the first kernel matching the requested kernel name on the Jupyter server and creates one if none exists.

Execution Model

run_code() returns an Execution object:

Field	Type	Description
results	list[Result]	All rich outputs (charts, images, expression values)
logs	Logs	.stdout: list[str] and .stderr: list[str] chunks
error	ExecutionError | None	.name, .value, .traceback
execution_count	int | None	Jupyter cell execution counter
timed_out	bool	True when execution was cut short by the timeout
text	str | None	(property) text/plain of the main execute_result

Each Result has typed MIME fields: text, html, markdown, svg, png, jpeg, gif, pdf, latex, json, javascript, plotly, plus extra for unknown types. The text field is Jupyter's text/plain bundle verbatim — the Python repr() of the cell's last expression. So run_code("'hi'").text is "'hi'" (with quotes), and run_code("42").text is "42". This preserves the distinction between the string '2' and the int 2.

Code Runner + Commands/Files

The code runner capsule inherits all standard capsule features:

from wrenn.code_runner import Capsule

with Capsule(wait=True) as capsule:
    # Use run_code for Jupyter execution
    capsule.run_code("import pandas as pd; df = pd.DataFrame({'a': [1,2,3]})")
    capsule.run_code("df.to_csv('/tmp/data.csv', index=False)")

    # Use standard file operations
    content = capsule.files.read("/tmp/data.csv")
    print(content)

    # Use standard command execution
    result = capsule.commands.run("wc -l /tmp/data.csv")
    print(result.stdout)

---

Async Support

All operations have async variants via AsyncCapsule:

Async Capsule

from wrenn import AsyncCapsule

async with await AsyncCapsule.create(template="minimal", wait=True) as capsule:
    result = await capsule.commands.run("echo hello")
    print(result.stdout)

    await capsule.files.write("/app/file.txt", "data")
    entries = await capsule.files.list("/app")

    await capsule.pause()
    await capsule.resume()

Async Code Runner

from wrenn.code_runner import AsyncCapsule

async with await AsyncCapsule.create(wait=True) as capsule:
    result = await capsule.run_code("2 + 2")
    print(result.text)  # "4"

Async PTY

async with capsule.pty(cmd="/bin/bash") as term:
    await term.write(b"ls -la\n")
    async for event in term:
        if event.type == "output":
            sys.stdout.buffer.write(event.data)

---

Error Handling

The SDK maps server error codes to typed exceptions:

from wrenn import (
    WrennError,
    WrennValidationError,       # 400
    WrennAuthenticationError,   # 401
    WrennForbiddenError,        # 403
    WrennNotFoundError,         # 404
    WrennConflictError,         # 409
    WrennHostHasCapsulesError,  # 409 (host has running capsules)
    WrennInternalError,         # 500
    WrennAgentError,            # 502
    WrennHostUnavailableError,  # 503
)

try:
    Capsule.get_info("nonexistent")
except WrennNotFoundError as e:
    print(e.code)         # "not_found"
    print(e.message)      # "capsule not found"
    print(e.status_code)  # 404

All exceptions inherit from WrennError and expose .code, .message, and .status_code.

---

Migrating from e2b

Replace your imports:

# Before
from e2b import Sandbox
sandbox = Sandbox()

# After
from wrenn import Capsule
capsule = Capsule()

For code interpreter:

# Before
from e2b_code_interpreter import Sandbox
sandbox = Sandbox()
result = sandbox.run_code("print('hello')")

# After
from wrenn.code_interpreter import Capsule
capsule = Capsule()
result = capsule.run_code("print('hello')")

The Sandbox name is available as a deprecated alias in both modules:

from wrenn import Sandbox                    # works, emits FutureWarning
from wrenn.code_interpreter import Sandbox   # works, emits FutureWarning

---

Low-Level Client

For direct API access, use WrennClient / AsyncWrennClient:

from wrenn import WrennClient

with WrennClient(api_key="wrn_...") as client:
    capsule = client.capsules.create(template="minimal")
    client.capsules.pause(capsule.id)
    client.capsules.resume(capsule.id)
    client.capsules.ping(capsule.id)
    client.capsules.destroy(capsule.id)

    # Snapshots
    template = client.snapshots.create(capsule_id="cl-abc", name="my-snap")
    templates = client.snapshots.list(type="custom")  # optional type filter
    client.snapshots.delete("my-snap")

---

Development

This project uses uv for dependency management.

# Install dependencies
uv sync

# Run linting
make lint

# Run unit tests
make test

# Run all tests (including integration)
make test-integration

Running Integration Tests

Integration tests require a live Wrenn server. Set credentials via environment or a .env file at the project root:

# Option 1: environment variable
export WRENN_API_KEY="wrn_..."

# Option 2: .env file
echo 'WRENN_API_KEY=wrn_...' > .env

Then run:

make test-integration

Tests are automatically skipped when WRENN_API_KEY is not available.

License

MIT