watasu 0.1.64

Python SDK for Watasu

Watasu Python SDK

Python SDK for Watasu.

Install

pip install watasu

Set WATASU_API_KEY before using the SDK.

Usage

from watasu import Sandbox

sbx = Sandbox()
sbx.files.write("/home/user/a.py", "print(2 + 2)")
result = sbx.commands.run("python /home/user/a.py")
print(result.stdout)
sbx.kill()

Sandbox(), Sandbox.create, and Sandbox.connect return only after the Watasu API supplies a usable data-plane session. The SDK does not poll sandbox readiness.

sbx.beta_pause()
sbx.resume(timeout=300)

Choose what happens when the sandbox timeout expires:

with Sandbox.beta_create(auto_pause=True) as sbx:
    result = sbx.commands.run("echo retained")
    print(result.stdout)

Sandbox.create(lifecycle={"on_timeout": "pause", "auto_resume": True}) keeps the retained disk after timeout and allows a later data-plane request to resume that paused sandbox automatically. The default timeout action is kill.

Mount a named persistent volume when the sandbox starts:

with Sandbox.create(
    volume_mounts={
        "/workspace/cache": "cache",
        "/data/models": {"name": "models"},
    }
) as sbx:
    sbx.commands.run("echo ready > /workspace/cache/status.txt")

Create and edit a persistent volume while it is detached:

from watasu import Volume

volume = Volume.create("cache")
volume.make_dir("/workspace")
volume.write_file("/workspace/status.txt", "ready\n", mode="0644")
print(volume.read_file("/workspace/status.txt"))
print([entry.path for entry in volume.list("/workspace")])
volume.remove("/workspace/status.txt")
volume.destroy()

from watasu import Sandbox

with Sandbox.create() as sbx:
    result = sbx.commands.run("echo hello")
    print(result.stdout)

Leaving the context manager calls kill().

Code Interpreter

pip install watasu-code-interpreter

from watasu_code_interpreter import Sandbox

with Sandbox.create() as sbx:
    context = sbx.create_code_context()
    execution = sbx.run_code(
        "print('hello')\n2 + 3",
        context=context,
        on_stdout=lambda message: print(message.line),
    )
    print(execution.text)
    sbx.remove_code_context(context)

watasu_code_interpreter.Sandbox starts the code-interpreter template by default. Code runs in persistent Python contexts and returns structured results, logs, and error fields for each execution.

MCP Gateway

import os

from watasu import Sandbox

with Sandbox.create(
    mcp={
        "github": {
            "command": "github-mcp-server",
            "args": ["stdio"],
            "env": {"GITHUB_TOKEN": os.environ["GITHUB_TOKEN"]},
        }
    }
) as sbx:
    print(sbx.get_mcp_url())
    print(sbx.get_mcp_token())

Listing Sandboxes

from watasu import Sandbox

paginator = Sandbox.list(
    query={"metadata": {"purpose": "ci"}, "state": ["running"]},
    limit=20,
)

for sandbox in paginator.list_items():
    print(sandbox.sandbox_id, sandbox.state)

Git, Watch, PTY, And Signed File URLs

from watasu import PtySize, Sandbox

with Sandbox.create() as sbx:
    sbx.git.init("/workspace/new-project", initial_branch="main")
    sbx.git.clone(
        "https://github.com/acme/project.git",
        path="/workspace/project",
        branch="main",
        depth=1,
    )
    status = sbx.git.status("/workspace/project")
    sbx.git.configure_user(
        "Watasu Bot",
        "[e-mail]",
        scope="local",
        path="/workspace/project",
    )
    sbx.git.create_branch("/workspace/project", "feature/docs")
    sbx.git.add("/workspace/project", files=["README.md"])
    sbx.git.commit(
        "/workspace/project",
        "Update docs",
        author_name="Watasu Bot",
        author_email="[e-mail]",
    )
    sbx.git.push(
        "/workspace/project",
        remote="origin",
        branch="feature/docs",
        set_upstream=True,
    )
    remote_url = sbx.git.remote_get("/workspace/project", "origin")
    sbx.git.restore("/workspace/project", paths=["README.md"])
    sbx.git.reset("/workspace/project", mode="hard", target="HEAD")

    watcher = sbx.files.watch_dir("/workspace/project", recursive=True)
    sbx.files.write_files(
        [
            {"path": "/workspace/project/a.txt", "data": "alpha"},
            {"path": "/workspace/project/b.bin", "data": b"\x00\x01\x02"},
        ]
    )

    terminal = sbx.pty.create(PtySize(rows=30, cols=100))
    terminal.send_stdin("echo hello\n")
    result = terminal.wait()

    upload_url = sbx.upload_url("/workspace/input.bin")
    download_url = sbx.download_url("/workspace/output.bin")

    events = watcher.get_new_events()
    watcher.stop()

Network Policy

with Sandbox.create(
    network={
        "allow_out": lambda ctx: list(ctx.rules.keys()) + ["pypi.org:443"],
        "deny_out": ["169.254.169.254"],
        "rules": {
            "api.example.com": [
                {"transform": {"headers": {"authorization": "Bearer token"}}}
            ]
        },
        "mask_request_host": "${PORT}-sandbox.example.com",
    }
) as sbx:
    sbx.update_network(
        allow_internet_access=False,
        allow_package_registry_access=True,
        allow_out=["registry.npmjs.org:443"],
    )

Template Builds

from watasu import Template

template = (
    Template()
    .from_python_image("3.12")
    .copy("requirements.txt", "/workspace/requirements.txt")
    .apt_install(["git"])
    .pip_install(["pytest"])
    .set_envs({"PIP_DISABLE_PIP_VERSION_CHECK": "1"})
    .run_cmd("echo ready")
)

build = Template.build_in_background(
    template,
    "python-ci:stable",
    tags=["stable"],
    cpu_count=2,
    memory_mb=2048,
)
status = Template.get_build_status(build)

Template.assign_tags("python-ci:stable", ["prod"])
print(Template.exists("python-ci"))

The same builder classes are also available from the sync and async template namespaces:

from watasu.template_sync import Template
from watasu.template_async import AsyncTemplate

Template names resolve server-side. python-ci starts the latest ready build; python-ci:stable starts the tagged build.

Template(file_context_path=".").from_dockerfile("Dockerfile") parses common FROM, WORKDIR, COPY, RUN, ENV, CMD, and ENTRYPOINT instructions into Watasu's package-spec builder.

Metrics And Snapshots

import datetime

from watasu import Sandbox

with Sandbox.create() as sbx:
    metrics = sbx.get_metrics(
        start=datetime.datetime.now(datetime.timezone.utc)
        - datetime.timedelta(minutes=5),
        end=datetime.datetime.now(datetime.timezone.utc),
    )
    snapshot = sbx.create_snapshot(name="ready")
    snapshots = sbx.list_snapshots().list_items()
    all_snapshots = Sandbox.list_snapshots(limit=100).next_items()
    restored = sbx.restore(snapshot_id=snapshot.snapshot_id)
    sbx.delete_snapshot(snapshot.snapshot_id)

Watasu snapshots are backed by sandbox checkpoints. Use the returned snapshot_id when restoring from a checkpoint.

Async API

import datetime

from watasu import AsyncSandbox


async def main() -> None:
    async with await AsyncSandbox.create() as sbx:
        result = await sbx.commands.run("echo hello")
        print(result.stdout)

        metrics = await sbx.get_metrics(
            start=datetime.datetime.now(datetime.timezone.utc)
            - datetime.timedelta(minutes=5),
            end=datetime.datetime.now(datetime.timezone.utc),
        )
        snapshot = await sbx.create_snapshot(name="ready")
        snapshots = await sbx.list_snapshots().list_items()
        all_snapshots = await AsyncSandbox.list_snapshots(limit=100).list_items()
        await sbx.delete_snapshot(snapshot.snapshot_id)

Unsupported control/runtime surfaces fail with explicit errors instead of silently falling back to client-side polling.