sockt 0.2.0

Sockt Agent SDK — manage sandboxes via REST

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
• Quick Start
• Authentication
• Core Concepts
• API Reference
  • ComputeClient
  • Sandbox
  • ShellSession
  • Models
  • Errors
• Configuration
• Examples
• REST API Endpoints
• Troubleshooting

---

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="[e-mail]")
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