Python SDK

The official Python SDK for Rexec provides an async-first, type-hinted interface for Terminal as a Service.

Installation

pip install rexec

Requirements

Python 3.9+
httpx for HTTP requests
websockets for terminal connections

Quick Start

import asyncio
from rexec import RexecClient

async def main():
    async with RexecClient("https://your-instance.com", "your-token") as client:
        # Create a container
        container = await client.containers.create(
            image="ubuntu:24.04",
            name="my-sandbox"
        )
        print(f"Created container: {container.id}")

        # Connect to terminal
        async with client.terminal.connect(container.id) as term:
            await term.write(b"echo 'Hello from Rexec!'\n")
            
            # Read output
            data = await term.read()
            print(data.decode())

        # Clean up
        await client.containers.delete(container.id)

asyncio.run(main())

Client Initialization

As Context Manager (Recommended)

from rexec import RexecClient

async with RexecClient(
    base_url="https://your-instance.com",
    token="your-api-token",
    timeout=30.0,  # optional
) as client:
    # Use client here
    ...

Manual Management

client = RexecClient("https://your-instance.com", "your-token")
try:
    # Use client
    ...
finally:
    await client.close()

Container Operations

The Containers service provides methods for managing sandboxed environments.

List Containers

containers = await client.containers.list()
for c in containers:
    print(f"{c.name}: {c.status}")

Get Container

container = await client.containers.get(container_id)
print(f"Container {container.name} is {container.status}")

Create Container

container = await client.containers.create(
    image="ubuntu:24.04",
    name="my-container",
    environment={"MY_VAR": "value", "DEBUG": "true"},
    labels={"project": "demo"}
)

print(f"Created: {container.id}")

Start Container

await client.containers.start(container_id)

Stop Container

await client.containers.stop(container_id)

Delete Container

await client.containers.delete(container_id)

File Operations

Manage files and directories within containers.

List Files

files = await client.files.list(container_id, "/home")
for f in files:
    prefix = "📁" if f.is_dir else "📄"
    print(f"{prefix} {f.name}")

Download File

content = await client.files.download(container_id, "/etc/passwd")
print(content.decode())

Upload File

await client.files.upload(
    container_id,
    "/home/script.py",
    b"print('hello world')"
)

Create Directory

await client.files.mkdir(container_id, "/home/mydir")

Delete File

await client.files.delete(container_id, "/home/script.py")

Terminal Operations

Connect to containers via WebSocket for real-time terminal access.

Connect to Terminal

# Recommended: use as context manager
async with client.terminal.connect(
    container_id,
    cols=120,
    rows=40,
    timeout=30.0
) as term:
    # Use terminal
    await term.write(b"ls -la\n")
    data = await term.read()
    print(data.decode())

Write to Terminal

await term.write(b"echo hello\n")

# Or use string
await term.write("cd /home && ls\n".encode())

Read from Terminal

# Single read
data = await term.read()
print(data.decode())

# Iterate over output
async for data in term:
    print(data.decode(), end="")

Resize Terminal

await term.resize(150, 50)

Advanced Examples

Run a Script and Capture Output

async def run_script(
    client: RexecClient,
    container_id: str,
    script: str
) -> str:
    output = []
    
    async with client.terminal.connect(container_id) as term:
        await term.write(f"{script}\nexit\n".encode())
        
        async for data in term:
            output.append(data.decode())
    
    return "".join(output)

# Usage
result = await run_script(
    client,
    container.id,
    "apt update && apt install -y curl"
)
print(result)

Batch Container Operations

import asyncio
from rexec import RexecClient, Container

async def create_batch(
    client: RexecClient,
    count: int
) -> list[Container]:
    tasks = [
        client.containers.create(
            image="ubuntu:24.04",
            name=f"worker-{i}"
        )
        for i in range(count)
    ]
    
    return await asyncio.gather(*tasks)

# Create 5 containers in parallel
containers = await create_batch(client, 5)
print(f"Created {len(containers)} containers")

File Sync

import os
from pathlib import Path

async def sync_directory(
    client: RexecClient,
    container_id: str,
    local_path: Path,
    remote_path: str
):
    """Sync a local directory to a container."""
    await client.files.mkdir(container_id, remote_path)
    
    for item in local_path.iterdir():
        remote_item = f"{remote_path}/{item.name}"
        
        if item.is_file():
            content = item.read_bytes()
            await client.files.upload(container_id, remote_item, content)
            print(f"Uploaded: {remote_item}")
        elif item.is_dir():
            await sync_directory(client, container_id, item, remote_item)

# Usage
await sync_directory(
    client,
    container.id,
    Path("./local-dir"),
    "/app"
)

Real-time Log Streaming

async def stream_logs(client: RexecClient, container_id: str, command: str):
    """Stream command output in real-time."""
    async with client.terminal.connect(container_id) as term:
        await term.write(f"{command}\n".encode())
        
        async for data in term:
            # Process each chunk as it arrives
            output = data.decode()
            print(output, end="")
            
            # You could also save to file, send to websocket, etc.
            # with open('logs.txt', 'a') as f:
            #     f.write(output)

# Usage
await stream_logs(client, container.id, "tail -f /var/log/app.log")

Error Handling

from rexec import RexecClient, RexecAPIError, RexecConnectionError

try:
    container = await client.containers.get("invalid-id")
except RexecAPIError as e:
    print(f"API Error {e.status_code}: {e.message}")
except RexecConnectionError as e:
    print(f"Connection Error: {e}")
except Exception as e:
    print(f"Unexpected error: {e}")

Type Hints

The SDK is fully typed for excellent IDE support:

from rexec import (
    RexecClient,
    Container,
    CreateContainerRequest,
    FileInfo,
    Terminal,
)

container: Container = await client.containers.create(image="ubuntu:24.04")
files: list[FileInfo] = await client.files.list(container.id, "/")

Async Context Managers

The SDK leverages Python’s async context managers for automatic cleanup:

# Client cleanup
async with RexecClient(base_url, token) as client:
    # Automatically closes connections on exit
    ...

# Terminal cleanup
async with client.terminal.connect(container_id) as term:
    # Automatically closes WebSocket on exit
    ...

Source Code

View the full source code on GitHub:

github.com/brimblehq/rexec-python

License

MIT License - see LICENSE for details.

​Python SDK

​Installation

​Requirements

​Quick Start

​Client Initialization

​As Context Manager (Recommended)

​Manual Management

​Container Operations

​List Containers

​Get Container

​Create Container

​Start Container

​Stop Container

​Delete Container

​File Operations

​List Files

​Download File

​Upload File

​Create Directory

​Delete File

​Terminal Operations

​Connect to Terminal

​Write to Terminal

​Read from Terminal

​Resize Terminal

​Advanced Examples

​Run a Script and Capture Output

​Batch Container Operations

​File Sync

​Real-time Log Streaming

​Error Handling

​Type Hints

​Async Context Managers

​Source Code

​License