Persistent volumes on BoxLite Cloud

# cloud_volume.py — create a managed volume, mount it, write and read through it
# Run: python cloud_volume.py
import asyncio
import os
import time

from boxlite import (
    ApiKeyCredential,
    Boxlite,
    BoxOptions,
    BoxliteRestOptions,
)

IMAGE = "ghcr.io/boxlite-ai/boxlite-agent-base:v0.1.0"

api_key = os.environ.get("BOXLITE_API_KEY")
if not api_key:
    raise SystemExit("Set BOXLITE_API_KEY to your blk_live_... key before running this.")


async def main() -> None:
    rt = Boxlite.rest(
        BoxliteRestOptions(
            url=os.environ.get("BOXLITE_REST_URL", "https://app.boxlite.ai/api"),
            credential=ApiKeyCredential(api_key),
        )
    )

    volume = None
    box = None
    try:
        # rt.volumes is a property, and create() takes no arguments.
        # The id you mount by comes back on the returned VolumeInfo.
        volume = await rt.volumes.create()
        print(f"Created volume {volume.id} at {volume.created_at}")

        box = await rt.create(
            BoxOptions(
                image=IMAGE,
                # (managed volume id, mount path inside the box)
                volumes=[(volume.id, "/data")],
            ),
            name=f"volume-demo-{int(time.time())}",
        )
        await box.start()

        # Write through the mount, not to the box's own disk.
        write = await box.exec(
            "sh",
            args=["-c", "echo 'subtitle model v3' > /data/notes.txt"],
        )
        write_result = await write.wait()
        if write_result.exit_code != 0:
            print(f"write failed with exit code {write_result.exit_code}")
            return

        read = await box.exec("cat", args=["/data/notes.txt"])
        content = ""
        async for line in read.stdout():
            content += line
        read_result = await read.wait()

        print(f"Exit code: {read_result.exit_code}")
        print(content)
    except Exception as exc:
        # Auth failures, creation failures, and backends without named volume
        # support all surface here.
        print(f"volume run failed: {exc!r}")
    finally:
        # Teardown in finally, so a failure above cannot leave a box billing.
        if box is not None:
            await rt.remove(box.id, force=True)
        if volume is not None:
            await rt.volumes.remove(volume.id)


if __name__ == "__main__":
    asyncio.run(main())

# Requires BOXLITE_API_KEY. Create a key in the console: /cloud/api-keys
BOXLITE_REST_URL="${BOXLITE_REST_URL:-https://app.boxlite.ai/api}"

curl -fsS "${BOXLITE_REST_URL}/v1/volumes" \
  -H "Authorization: Bearer ${BOXLITE_API_KEY}" \
  | jq .

import asyncio
import os
import time

from boxlite import (
    ApiKeyCredential,
    Boxlite,
    BoxliteRestOptions,
    BoxOptions,
)

VOLUME_ID = os.environ.get("BOXLITE_VOLUME_ID", "<YOUR_VOLUME_ID>")
IMAGE = "ghcr.io/boxlite-ai/boxlite-agent-base:v0.1.0"


async def run_in_fresh_box(rt, name, script):
    """Create a box with the volume mounted, run one shell script, then remove the box."""
    box = await rt.create(
        BoxOptions(image=IMAGE, volumes=[(VOLUME_ID, "/data")]),
        name=name,
    )
    try:
        await box.start()
        execution = await box.exec("sh", args=["-c", script])
        output = ""
        async for line in execution.stdout():
            output += line
        result = await execution.wait()
        return result.exit_code, output
    finally:
        # The box is gone after this line; the volume is not.
        await rt.remove(box.id, force=True)


async def main():
    rt = Boxlite.rest(BoxliteRestOptions(
        url=os.environ.get("BOXLITE_REST_URL", "https://app.boxlite.ai/api"),
        credential=ApiKeyCredential(os.environ["BOXLITE_API_KEY"]),
    ))
    stamp = int(time.time())

    try:
        # Box A writes, then is destroyed.
        code, _ = await run_in_fresh_box(
            rt,
            f"volume-writer-{stamp}",
            "echo 'produced by box A' > /data/handoff.txt",
        )
        if code != 0:
            print(f"box A write failed with exit code {code}")
            return

        # Box B is a different box on the same volume.
        code, output = await run_in_fresh_box(
            rt,
            f"volume-reader-{stamp}",
            "cat /data/handoff.txt",
        )
        print(f"box B exit code: {code}")
        print(f"box B read back: {output}")
    except Exception as exc:
        print(f"handoff failed: {exc}")


asyncio.run(main())

# cloud_volume_delete.py — remove a volume, then wait for it to leave the listing
# Run: python cloud_volume_delete.py
import asyncio
import os

from boxlite import ApiKeyCredential, Boxlite, BoxliteRestOptions

VOLUME_ID = os.environ.get("BOXLITE_VOLUME_ID", "<YOUR_VOLUME_ID>")

api_key = os.environ.get("BOXLITE_API_KEY")
if not api_key:
    raise SystemExit("Set BOXLITE_API_KEY to your blk_live_... key before running this.")


async def main() -> None:
    rt = Boxlite.rest(
        BoxliteRestOptions(
            url=os.environ.get("BOXLITE_REST_URL", "https://app.boxlite.ai/api"),
            credential=ApiKeyCredential(api_key),
        )
    )

    # This script creates no box and no volume, so removal is the only teardown
    # it performs — and removal is what it is here to demonstrate.
    try:
        await rt.volumes.remove(VOLUME_ID)
        print(f"Deletion accepted for {VOLUME_ID}")

        # Bounded poll: up to 60s for the volume to leave the listing.
        for _ in range(12):
            volumes = await rt.volumes.list()
            if all(volume.id != VOLUME_ID for volume in volumes):
                print("volume reclaimed")
                return
            await asyncio.sleep(5)

        print("volume still listed after 60s — reclamation runs on the platform's cycle")
    except Exception as exc:
        print(f"volume deletion failed: {exc!r}")


if __name__ == "__main__":
    asyncio.run(main())