Skip to content

SocktDev/sockt-python

Folders and files

NameName
Last commit message
Last commit date

Latest commit

 

History

2 Commits
 
 
 
 
 
 
 
 
 
 

Repository files navigation

Sockt Python SDK

Official Python SDK for managing Sockt sandboxes.

Package: sockt · Version: 0.2.0 · Requires: Python 3.10+

See also: TypeScript/JavaScript SDK (@sockt/client)


Table of Contents


Installation

pip install sockt

From source:

pip install -e .

Dependencies: httpx (>=0.27,<1), websockets (>=14,<16)


Quick Start

from sockt import ComputeClient

client = ComputeClient(api_key="sockt_live_...")
sandbox = client.create_sandbox(tier="nano")
sandbox.wait_until_running()

result = sandbox.exec_sync("echo hello")
print(result.stdout)  # "hello\n"

sandbox.terminate()
client.close()

Authentication

The SDK supports two authentication methods corresponding to two billing models:

API Key (Credits Billing)

Format: sockt_live_...

Used for prepaid USD credits billing. Obtain from the Sockt dashboard.

client = ComputeClient(api_key="sockt_live_abc123")

Sandbox Token (Lightning Billing)

Format: sbx_...

Per-sandbox token returned when creating a Lightning-billed sandbox. Scoped to a single sandbox.

client = ComputeClient(sandbox_token="sbx_abc123")
sandbox = client.get_sandbox("sandbox-id-here")

Environment Variables

Variable Description Used When
SOCKT_API_KEY API key for credits billing Constructor api_key not provided
SOCKT_SANDBOX_TOKEN Sandbox token for Lightning billing Constructor sandbox_token not provided
SOCKT_API_URL API base URL override Constructor base_url not provided

Priority: Constructor parameter > Environment variable > Default

export SOCKT_API_KEY="sockt_live_abc123"
export SOCKT_API_URL="https://api.sockt.dev"
client = ComputeClient()  # reads from env automatically

Core Concepts

Sandbox Lifecycle

creating → starting → running → paused → terminated
                         ↓                    ↑
                         └────────────────────┘
                              (also: failed)
State Description
starting Pod is being provisioned (30-90 seconds)
running Ready for commands, files, and shell access
paused Compute stopped (no billing), can be resumed
terminated Permanently destroyed
failed Provisioning or runtime error

Billing Methods

Credits (Prepaid USD)

  • Requires API key (sockt_live_...)
  • Automatic deduction from account balance
  • Sandbox pauses if balance depleted — top up and resume
  • Unused allocation refunded on terminate

Lightning Network (Pay-as-you-go)

  • No account required (anonymous)
  • Returns BOLT11 invoice on create — pay to start sandbox
  • Receives a sandbox_token (sbx_...) for subsequent operations
  • New invoices generated when balance runs low
  • Events fired: pending_invoice, low_balance, out_of_balance
  • Optional refund to Lightning address on terminate

Polling Model

Command execution is asynchronous:

  1. exec() submits the command and returns an Execution with an ID
  2. exec_result() polls for output chunks
  3. Repeat until status is completed, failed, or cancelled

The convenience method exec_sync() handles this polling loop automatically.


API Reference

ComputeClient

The main entry point for creating and managing sandboxes.

Constructor

ComputeClient(
    api_key: str | None = None,
    sandbox_token: str | None = None,
    base_url: str | None = None,
    timeout: float = 120.0
)
Parameter Type Default Description
api_key string SOCKT_API_KEY env API key for credits billing
sandbox_token string SOCKT_SANDBOX_TOKEN env Sandbox token for Lightning
base_url string https://api.sockt.dev API base URL
timeout float 120.0 HTTP request timeout (seconds)

list_tiers()

List available sandbox tiers with pricing.

tiers: list[Tier] = client.list_tiers()

Returns: List of Tier objects with pricing in sats and USD.


create_sandbox()

Create a new sandbox.

sandbox: Sandbox = client.create_sandbox(
    tier: str,
    billing_method: str | None = None,
    prepaid_sats: int | None = None,
    initial_credits_cents: int | None = None,
    label: str | None = None
)
Parameter Type Description
tier string Tier name (e.g., "nano", "micro", "small", "gpu-small")
billing_method string "credits" or "lightning"
prepaid_sats int Prepaid satoshis (Lightning only)
initial_credits_cents int Initial credits in cents (Credits only)
label string Human-readable label

Returns: A Sandbox instance. For Lightning billing, the sandbox's pending_invoice is populated with the initial BOLT11 invoice to pay.

Example:

sandbox = client.create_sandbox(tier="small")

sandbox = client.create_sandbox(
    tier="nano",
    billing_method="lightning",
    prepaid_sats=5000
)
print(sandbox.pending_invoice.bolt11)  # Pay this invoice

get_sandbox()

Get an existing sandbox by ID.

sandbox: Sandbox = client.get_sandbox(sandbox_id: str)

Automatically calls refresh_status() to load current state.


close()

Close the HTTP client connection.

client.close()

with ComputeClient(api_key="...") as client:
    sandbox = client.create_sandbox(tier="nano")
    # client.close() called automatically

Sandbox

Represents a sandbox instance. Provides methods for command execution, file operations, and shell access.

Properties

Property Type Description
sandbox.id string Unique sandbox identifier
sandbox.status string Current lifecycle state
sandbox.pending_invoice PendingInvoice | None Active Lightning invoice

wait_until_running()

Block until the sandbox reaches "running" state.

sandbox.wait_until_running(
    timeout: float = 120.0,
    poll_interval: float = 2.0
)
Parameter Default Description
timeout 120s Maximum time to wait
poll_interval 2s Interval between status checks

Raises: SandboxNotRunningError if timeout expires or sandbox enters failed/terminated state.


refresh_status()

Fetch the latest sandbox status from the API.

status: SandboxStatus = sandbox.refresh_status()

Updates the sandbox's internal state and returns the full SandboxStatus object. Also triggers event callbacks if billing warnings are present.


pause()

Pause a running sandbox. Stops compute and billing.

sandbox.pause()

resume()

Resume a paused sandbox.

sandbox.resume()

terminate()

Permanently destroy a sandbox.

result: dict = sandbox.terminate(
    lightning_address: str | None = None
)
Parameter Description
lightning_address Lightning address for refund of unused balance

exec()

Start asynchronous command execution. Returns immediately with an execution handle.

execution: Execution = sandbox.exec(
    command: str,
    working_dir: str | None = None,
    timeout_ms: int | None = None
)
Parameter Default Description
command Shell command to execute (run via /bin/bash -lc)
working_dir "." Working directory (relative to /home/sandbox)
timeout_ms 30000 Execution timeout in milliseconds

Returns: Execution object with execution_id and status: "running".


exec_sync()

Execute a command and poll until completion. Convenience wrapper around exec() + exec_result().

result: ExecResult = sandbox.exec_sync(
    command: str,
    working_dir: str | None = None,
    timeout_ms: int | None = None,
    poll_interval: float = 0.5
)
Parameter Default Description
command Shell command to execute
working_dir "." Working directory
timeout_ms 30000 Execution timeout
poll_interval 0.5s Interval between result polls

Returns: ExecResult with full output, exit code, and status.

Example:

result = sandbox.exec_sync("pip install requests && python app.py")
print(result.stdout)
print(f"Exit code: {result.exit_code}")

exec_result()

Poll for execution output. Returns new output chunks since last poll.

result: ExecResult = sandbox.exec_result(execution_id: str)

Call repeatedly until result.status is "completed", "failed", or "cancelled".


exec_cancel()

Cancel a running execution.

sandbox.exec_cancel(execution_id: str)

write_file()

Write content to a file in the sandbox.

result: dict = sandbox.write_file(
    path: str,
    content: str,
    encoding: str = "utf8",
    create_dirs: bool = True
)
Parameter Default Description
path File path (relative to /home/sandbox or absolute)
content File content (plain text or base64-encoded)
encoding "utf8" "utf8" for text, "base64" for binary
create_dirs true Create parent directories if missing

Example:

sandbox.write_file("app.py", "print('hello')")

import base64
with open("image.png", "rb") as f:
    b64 = base64.b64encode(f.read()).decode()
sandbox.write_file("image.png", b64, encoding="base64")

read_file()

Read a file from the sandbox.

content: str = sandbox.read_file(
    path: str,
    encoding: str = "utf8",
    max_bytes: int | None = None
)
Parameter Default Description
path File path to read
encoding "utf8" "utf8" returns text, "base64" returns base64-encoded content
max_bytes Maximum bytes to read (truncates if exceeded)

Returns: File content as a string. For base64 encoding, returns base64-encoded string.


list_files()

List files in a sandbox directory.

entries: list[FileEntry] = sandbox.list_files(
    path: str = ".",
    recursive: bool = False,
    max_depth: int | None = None
)
Parameter Default Description
path "." Directory path to list
recursive false Include subdirectories
max_depth Maximum depth for recursive listing

Returns: List of FileEntry objects.


shell()

Create an interactive WebSocket shell session.

session: ShellSession = sandbox.shell()

Returns: A ShellSession object. Call connect() or use as a context manager.


on() / off()

Register/unregister event listeners for billing events.

sandbox.on(event: str, callback: Callable)
sandbox.off(event: str, callback: Callable)

Events:

Event Fired When Callback Data
"pending_invoice" New invoice needs payment PendingInvoice object
"low_balance" Balance running low Warning data
"out_of_balance" Balance depleted, sandbox will pause Warning data

Example:

def on_invoice(invoice):
    print(f"Pay this invoice: {invoice.bolt11}")
    print(f"Amount: {invoice.amount_msats} msats")
    print(f"Expires: {invoice.expires_at}")

sandbox.on("pending_invoice", on_invoice)
sandbox.refresh_status()

ShellSession

WebSocket-based interactive terminal session.

connect()

Establish the WebSocket connection.

session.connect()

close()

Close the WebSocket connection.

session.close()

send()

Send raw text to the shell.

session.send(text: str)

send_line()

Send text followed by a newline character.

session.send_line(text: str)

resize()

Resize the remote terminal.

session.resize(cols: int, rows: int)

recv()

Receive the next frame from the shell. Blocks until a frame is available or timeout.

frame: dict = session.recv(timeout: float = 30.0)

Returns: A frame object with type and data fields.

Frame types:

Type Description
"stdout" Standard output data
"stderr" Standard error data
"billing" Billing status update
"low_balance" Low balance warning
"deposit_required" Payment needed to continue
"out_of_balance" Session will terminate
"error" Error message

on() / off()

Register/unregister listeners for specific frame types.

session.on(event: str, callback: Callable)
session.off(event: str, callback: Callable)

Context Manager

with sandbox.shell() as sh:
    sh.send_line("ls -la")
    frame = sh.recv()
    print(frame["data"])

Models

Tier

Field Type Description
name string Tier identifier (e.g., "nano", "micro")
msats_per_second int Rate in millisatoshis per second
sats_per_second float Rate in satoshis per second
usd_cents_per_second float Rate in USD cents per second
cost_per_minute_sats float Cost per minute in sats
cost_per_hour_sats float Cost per hour in sats
cost_per_day_sats float Cost per day in sats
cost_per_hour_usd float Cost per hour in USD

SandboxStatus

Field Type Description
sandbox_id string Unique identifier
status string Lifecycle state
tier string Compute tier
runtime string Runtime environment
billing_method string "credits" or "lightning"
template string Base template
consumed_msats int Total consumed millisatoshis
msats_per_second int | None Current billing rate
sats_per_second float | None Current rate in sats
prepaid_balance_msats int | None Remaining prepaid (Lightning)
seconds_remaining int | None Estimated seconds of compute left
warning_level string | None "low_balance" or "out_of_balance"
next_poll_secs int | None Recommended poll interval
pending_invoice PendingInvoice | None Invoice awaiting payment
sandbox_token string | None Sandbox auth token (Lightning)
raw dict Full raw API response

Execution

Field Type Description
execution_id string Unique execution identifier
status string "running", "completed", "failed", "cancelled"
sandbox_id string Parent sandbox ID
command string The submitted command

ExecResult

Field Type Description
execution_id string Execution identifier
status string Final status
output list Output chunks: `[{stream: "stdout"
exit_code int | None Process exit code (null if still running)
error string | None Error message (if failed)
pending_invoice PendingInvoice | None Invoice if payment needed

Convenience properties:

result.stdout  # All stdout chunks joined as a string
result.stderr  # All stderr chunks joined as a string

FileEntry

Field Type Description
name string File or directory name
size int Size in bytes
mod_ts int Modification timestamp (Unix seconds)
is_dir bool Whether entry is a directory

PendingInvoice

Field Type Description
bolt11 string BOLT11 payment request string
payment_hash string Payment hash for verification
amount_msats int Invoice amount in millisatoshis
expires_at string | None Expiration time (ISO 8601)

Errors

All SDK errors inherit from SocktError.

Error Hierarchy

SocktError (base)
├── AuthenticationError      (HTTP 401/403)
├── NoCapacityError          (HTTP 503 with "no_capacity")
├── SandboxNotRunningError   (sandbox not in running state)
├── TopUpFailedError         (HTTP 402)
├── HostUnreachableError     (HTTP 502/503)
├── RateLimitedError         (HTTP 429)
└── APIError                 (other HTTP 4xx/5xx)

Error Properties

Property Type Description
message string Human-readable error description
slug string Machine-readable error code
status_code int HTTP status code

Error Mapping

HTTP Status Slug Error Class When
401, 403 AuthenticationError Invalid or missing credentials
402 TopUpFailedError Insufficient credits/balance
429 RateLimitedError Too many requests (300 req/min default)
502 HostUnreachableError Runtime pod connection error
503 no_capacity NoCapacityError No pods available for requested tier
503 other HostUnreachableError Pod temporarily unreachable
Other 4xx/5xx APIError Generic API error

Error Handling Example

from sockt import (
    ComputeClient,
    AuthenticationError,
    NoCapacityError,
    RateLimitedError,
    SocktError
)

try:
    client = ComputeClient(api_key="sockt_live_...")
    sandbox = client.create_sandbox(tier="gpu-small")
except AuthenticationError:
    print("Invalid API key")
except NoCapacityError:
    print("No GPU pods available, try again later")
except RateLimitedError:
    print("Too many requests, slow down")
except SocktError as e:
    print(f"API error ({e.status_code}): {e.message}")

Configuration

Default Constants

Constant Value Description
Base URL https://api.sockt.dev API endpoint
Request timeout 120s HTTP timeout
Poll interval 0.5s exec_sync poll interval
Wait poll interval 2.0s wait_until_running poll interval
Wait timeout 120s wait_until_running timeout
Pod starting backoff 2.0s Retry delay on 503 pod_starting
Pod starting max retries 3 Max retries on 503 pod_starting

Auto-Retry Behavior

The HTTP client automatically retries when receiving HTTP 503 with a "pod_starting" indication:

  • Up to 3 retry attempts
  • 2-second backoff between retries
  • Applies to all API calls transparently

Examples

Full Credits Workflow

from sockt import ComputeClient

with ComputeClient(api_key="sockt_live_...") as client:
    sandbox = client.create_sandbox(tier="small")
    sandbox.wait_until_running()

    result = sandbox.exec_sync("python --version")
    print(result.stdout)

    sandbox.write_file("hello.py", 'print("Hello from Sockt!")')

    result = sandbox.exec_sync("python hello.py")
    print(result.stdout)

    entries = sandbox.list_files()
    for entry in entries:
        print(f"{'d' if entry.is_dir else '-'} {entry.size:>8} {entry.name}")

    sandbox.terminate()

Full Lightning Workflow

from sockt import ComputeClient

client = ComputeClient()

sandbox = client.create_sandbox(
    tier="nano",
    billing_method="lightning",
    prepaid_sats=3000
)

print(f"Pay this invoice to start: {sandbox.pending_invoice.bolt11}")
print(f"Amount: {sandbox.pending_invoice.amount_msats} msats")

def on_invoice(invoice):
    print(f"\nNew invoice! Pay to continue: {invoice.bolt11}")

sandbox.on("pending_invoice", on_invoice)
sandbox.wait_until_running(timeout=300)

result = sandbox.exec_sync("echo 'Paid with Lightning!'")
print(result.stdout)

sandbox.terminate(lightning_address="user@getalby.com")
client.close()

Running Multiple Commands

commands = [
    "apt-get update -y",
    "apt-get install -y curl jq",
    "curl -s https://api.github.com/repos/python/cpython | jq .stargazers_count",
]

for cmd in commands:
    result = sandbox.exec_sync(cmd, timeout_ms=60000)
    if result.exit_code != 0:
        print(f"Command failed: {result.stderr}")
        break
    print(result.stdout)

File Upload and Download

files = {
    "main.py": open("main.py").read(),
    "requirements.txt": open("requirements.txt").read(),
    "config.json": '{"debug": true, "port": 8080}',
}

for path, content in files.items():
    sandbox.write_file(path, content)

output = sandbox.read_file("results/output.json")
print(output)

Binary File Operations (Base64)

import base64

with open("model.bin", "rb") as f:
    encoded = base64.b64encode(f.read()).decode()
sandbox.write_file("model.bin", encoded, encoding="base64")

b64_content = sandbox.read_file("output.png", encoding="base64")
with open("output.png", "wb") as f:
    f.write(base64.b64decode(b64_content))

Interactive Shell Session

with sandbox.shell() as sh:
    sh.send_line("cd /tmp && mkdir myproject")
    frame = sh.recv(timeout=5.0)
    print(frame.get("data", ""))

    sh.send_line("echo $PWD")
    frame = sh.recv(timeout=5.0)
    print(frame.get("data", ""))

    sh.resize(cols=120, rows=40)

    sh.send_line("for i in 1 2 3; do echo $i; sleep 1; done")
    for _ in range(3):
        frame = sh.recv(timeout=5.0)
        print(frame.get("data", ""), end="")

Handling Lightning Payment Events

from sockt import ComputeClient

client = ComputeClient()
sandbox = client.create_sandbox(
    tier="nano",
    billing_method="lightning",
    prepaid_sats=1000
)

def on_invoice(invoice):
    print(f"Invoice: {invoice.bolt11}")
    print(f"   Amount: {invoice.amount_msats} msats")
    print(f"   Expires: {invoice.expires_at}")

def on_low_balance(data):
    print("Balance running low!")

def on_out_of_balance(data):
    print("Out of balance — sandbox will pause")

sandbox.on("pending_invoice", on_invoice)
sandbox.on("low_balance", on_low_balance)
sandbox.on("out_of_balance", on_out_of_balance)

sandbox.wait_until_running()
result = sandbox.exec_sync("long-running-job")

Error Handling Patterns

from sockt import (
    ComputeClient,
    SocktError,
    AuthenticationError,
    NoCapacityError,
    SandboxNotRunningError,
)
import time

client = ComputeClient(api_key="sockt_live_...")

for attempt in range(3):
    try:
        sandbox = client.create_sandbox(tier="gpu-small")
        break
    except NoCapacityError:
        if attempt == 2:
            raise
        time.sleep(5 * (attempt + 1))

try:
    sandbox.wait_until_running(timeout=180)
except SandboxNotRunningError as e:
    print(f"Sandbox failed to start: {e.message}")
    sandbox.terminate()
    raise

result = sandbox.exec_sync("might-fail-command")
if result.exit_code != 0:
    print(f"Command failed with exit code {result.exit_code}")
    print(f"stderr: {result.stderr}")

Context Manager / Cleanup Pattern

from sockt import ComputeClient

def run_in_sandbox(api_key: str, command: str, tier: str = "nano") -> str:
    with ComputeClient(api_key=api_key) as client:
        sandbox = client.create_sandbox(tier=tier)
        try:
            sandbox.wait_until_running()
            result = sandbox.exec_sync(command)
            return result.stdout
        finally:
            sandbox.terminate()

REST API Endpoints

The SDK wraps these control-plane REST endpoints at https://api.sockt.dev:

Method Path SDK Method Description
GET /v1/sandboxes/tiers list_tiers() List pricing tiers
POST /v1/sandboxes create_sandbox() Create a sandbox
GET /v1/sandboxes/{id} refresh_status() Get sandbox status
POST /v1/sandboxes/{id}/exec exec() Start command execution
GET /v1/executions/{id} exec_result() Poll execution output
DELETE /v1/executions/{id} exec_cancel() Cancel execution
POST /v1/sandboxes/{id}/files/write write_file() Write file
POST /v1/sandboxes/{id}/files/read read_file() Read file
GET /v1/sandboxes/{id}/files list_files() List directory
POST /v1/sandboxes/{id}/pause pause() Pause sandbox
POST /v1/sandboxes/{id}/resume resume() Resume sandbox
DELETE /v1/sandboxes/{id} terminate() Terminate sandbox
GET /v1/sandboxes/{id}/shell-url shell() (internal) Get WebSocket URL

Authentication header: Authorization: Bearer <token>

For sandbox token auth, the token is also sent as a query parameter: ?sandbox_token=sbx_...


Troubleshooting

Pod Starting (503 Retries)

Symptom: Requests fail with HTTP 503 and a "pod_starting" error.

Cause: The sandbox's compute pod is still booting (typically 30-90 seconds after creation).

Solution: The SDK automatically retries up to 3 times with 2-second backoff. If you still get this error, call wait_until_running() before making requests, or increase your timeout.

Sandbox Terminated Unexpectedly

Symptom: Sandbox status shows "terminated" or "paused" unexpectedly.

Cause: Balance depleted (credits exhausted or Lightning prepaid ran out).

Solution:

  • Credits: Top up your account balance and create a new sandbox
  • Lightning: Pay the pending_invoice from the status response, then call resume()
  • Register event handlers (on("low_balance", ...)) for proactive notifications

Timeout Waiting for Running

Symptom: wait_until_running() raises SandboxNotRunningError with timeout.

Cause: Pod provisioning took longer than the timeout period.

Solution:

  • Increase the timeout: wait_until_running(timeout=300) (5 minutes)
  • For Lightning: ensure the invoice has been paid before waiting
  • Check if the tier has available capacity (list_tiers())

WebSocket Connection Issues

Symptom: Shell session fails to connect or disconnects immediately.

Cause: Sandbox not in "running" state, invalid token, or network issues.

Solution:

  • Verify sandbox status is "running" with refresh_status()
  • Ensure your token (API key or sandbox token) is valid
  • Check that WebSocket connections are not blocked by your network/firewall
  • For Lightning: ensure balance is not depleted

Authentication Failures (401/403)

Symptom: AuthenticationError on API calls.

Cause: Invalid, expired, or revoked token; or token doesn't own the sandbox.

Solution:

  • Verify the token format: API keys start with sockt_live_, sandbox tokens with sbx_
  • API keys can access all sandboxes owned by the account
  • Sandbox tokens can only access their specific sandbox
  • Check constructor priority: parameter > env var

Rate Limiting (429)

Symptom: RateLimitedError thrown.

Cause: Exceeded 300 requests per minute.

Solution:

  • Reduce polling frequency (increase poll_interval in exec_sync)
  • Use next_poll_secs from SandboxStatus to pace status checks
  • For Lightning: the API suggests optimal polling intervals

Execution Concurrency Limit

Symptom: HTTP 429 with "too_many_executions" when calling exec().

Cause: More than 10 concurrent executions per credential.

Solution:

  • Wait for existing executions to complete before starting new ones
  • Use exec_cancel() to clean up abandoned executions
  • Design workflows to run commands sequentially rather than in parallel

About

No description, website, or topics provided.

Resources

License

Stars

Watchers

Forks

Releases

No releases published

Packages

 
 
 

Contributors

Languages