from runloop_api_client import RunloopSDK

runloop = RunloopSDK()

# Create a ready-to-use devbox
with runloop.devbox.create(name="my-devbox") as devbox:
    result = devbox.cmd.exec("echo 'Hello from Runloop!'")
    print(result.stdout())

    # Stream stdout in real time
    devbox.cmd.exec(
        "ls -la",
        stdout=lambda line: print("stdout:", line),
    )

# Blueprints
blueprint = runloop.blueprint.create(
    name="my-blueprint",
    dockerfile="FROM ubuntu:22.04\nRUN echo 'Hello' > /hello.txt\n",
)
devbox = blueprint.create_devbox(name="dev-from-blueprint")

# Storage objects
obj = runloop.storage_object.upload_from_text("Hello world!", name="greeting.txt")
print(obj.download_as_text())

import asyncio
from runloop_api_client import AsyncRunloopSDK


async def main():
    runloop = AsyncRunloopSDK()
    async with await runloop.devbox.create(name="async-devbox") as devbox:
        result = await devbox.cmd.exec("pwd")
        print(await result.stdout())

        def capture(line: str) -> None:
            print(">>", line)

        await devbox.cmd.exec("ls", stdout=capture)


asyncio.run(main())

from runloop_api_client import RunloopSDK

runloop = RunloopSDK(
    bearer_token="your-api-key",  # defaults to RUNLOOP_API_KEY env var
    # ... other options
)

# Create a new devbox
devbox = runloop.devbox.create(name="my-devbox")

# Create a devbox from a blueprint ID
devbox_from_blueprint = runloop.devbox.create_from_blueprint_id(
    blueprint_id="bpt_123",
    name="my-devbox-from-blueprint",
)

# Create a devbox from a blueprint name
devbox_from_name = runloop.devbox.create_from_blueprint_name(
    blueprint_name="my-blueprint-name",
    name="my-devbox-from-blueprint",
)

# Create a devbox from a snapshot
devbox_from_snapshot = runloop.devbox.create_from_snapshot(
    snapshot_id="snp_123",
    name="my-devbox-from-snapshot",
)

# Or get an existing one (waits for it to be running)
existing_devbox = runloop.devbox.from_id(devbox_id="dbx_123")

# List all devboxes
devboxes = runloop.devbox.list(limit=10)

# Get devbox information
info = devbox.get_info()
print(f"Devbox {info.name} is {info.status}")

# exec blocks until completion - use for commands that return immediately
result = devbox.cmd.exec("ls -la")
print("Output:", result.stdout())
print("Exit code:", result.exit_code)
print("Success:", result.success)

# exec_async returns immediately - use for long-running processes
execution = devbox.cmd.exec_async("npm run dev")

# Check execution status
state = execution.get_state()
print("Status:", state.status)

# Wait for completion and get result
result = execution.result()
print("Final output:", result.stdout())

# Kill the process
execution.kill()

# Start a long-running process
execution = devbox.cmd.exec_async("python train_model.py")

# Get the execution ID
print("Execution ID:", execution.execution_id)
print("Devbox ID:", execution.devbox_id)

# Poll for current state
state = execution.get_state()
print("Status:", state.status)  # "running", "completed", etc.
print("Exit code:", state.exit_status)  # only set when execution has completed

# Wait for completion and get results
result = execution.result()
print("Exit code:", result.exit_code)
print("Output:", result.stdout())
print("Errors:", result.stderr())

# Or kill the process early
execution.kill()

# From synchronous execution
result = devbox.cmd.exec("ls -la /tmp")

# Or from asynchronous execution
execution = devbox.cmd.exec_async("echo 'test'")
result = execution.result()

# Access execution results
print("Exit code:", result.exit_code)
print("Success:", result.success)  # True if exit code is 0
print("Failed:", result.failed)  # True if exit code is non-zero

# Get output streams
stdout = result.stdout()
stderr = result.stderr()
print("Standard output:", stdout)
print("Standard error:", stderr)

# Access raw result data
raw_result = result.raw
print("Raw result:", raw_result)

def handle_output(line: str) -> None:
    print("LOG:", line)


result = devbox.cmd.exec(
    "python train.py",
    stdout=handle_output,
    stderr=lambda line: print("ERR:", line),
    output=lambda line: print("ANY:", line),
)
print("exit code:", result.exit_code)

def capture(line: str) -> None:
    # Callbacks must be synchronous
    # Use thread-safe data structures if needed
    log_queue.put_nowait(line)


await devbox.cmd.exec(
    "tail -f /var/log/app.log",
    stdout=capture,
)

# Write files
devbox.file.write(
    path="/home/user/app.js",
    contents='console.log("Hello from devbox!");',
)

# Read files
content = devbox.file.read(path="/home/user/app.js")
print(content)

# Upload files
from pathlib import Path

devbox.file.upload(
    path="/home/user/upload.txt",
    file=Path("local_file.txt"),
)

# Download files
data = devbox.file.download(path="/home/user/download.txt")
with open("local_download.txt", "wb") as f:
    f.write(data)

# Create SSH key for remote access
ssh_key = devbox.net.create_ssh_key()
print("SSH URL:", ssh_key.url)

# Create tunnel to expose port
tunnel = devbox.net.create_tunnel(port=8080)
print("Public URL:", tunnel.url)

# Remove tunnel when done
devbox.net.remove_tunnel(port=8080)

# Create a snapshot (waits for completion)
snapshot = devbox.snapshot_disk(
    name="my-snapshot",
    commit_message="Added new features",
)

# Create a snapshot asynchronously (returns immediately)
snapshot = devbox.snapshot_disk_async(
    name="my-snapshot",
    commit_message="Added new features",
)
# Wait for it to complete later
snapshot.await_completed()

# Create new devbox from snapshot
new_devbox = snapshot.create_devbox(name="devbox-from-snapshot")

# Suspend devbox (pause without losing state)
devbox.suspend()

# Resume suspended devbox
devbox.resume()

# Keep devbox alive (extend timeout)
devbox.keep_alive()

# Wait for devbox to reach running state
devbox.await_running()

# Wait for devbox to be suspended
devbox.await_suspended()

# Shutdown devbox
devbox.shutdown()

# Synchronous
with runloop.devbox.create(name="temp-devbox") as devbox:
    result = devbox.cmd.exec("echo 'Hello'")
    print(result.stdout())
# devbox is automatically shutdown when exiting the context

# Asynchronous
async with await runloop.devbox.create(name="temp-devbox") as devbox:
    result = await devbox.cmd.exec("echo 'Hello'")
    print(await result.stdout())
# devbox is automatically shutdown when exiting the context

# Get all devbox logs
logs = devbox.logs()
for log in logs.logs:
    print(f"[{log.level}] {log.message}")

# Filter logs by execution ID
result = devbox.cmd.exec('echo "hello"')
exec_logs = devbox.logs(execution_id=result.execution_id)

# Filter logs by shell name
shell_logs = devbox.logs(shell_name="my-shell")

# Create a new blueprint
blueprint = runloop.blueprint.create(
    name="my-blueprint",
    dockerfile="FROM ubuntu:22.04\nRUN apt-get update && apt-get install -y python3\n",
    system_setup_commands=["pip install numpy pandas"],
)

# Or create a blueprint with a Docker build context from a local directory
from pathlib import Path
from runloop_api_client.lib.context_loader import build_docker_context_tar

context_root = Path("./my-app")
tar_bytes = build_docker_context_tar(context_root)

build_ctx_obj = runloop.storage_object.upload_from_bytes(
    data=tar_bytes,
    name="my-app-context.tar.gz",
    content_type="tgz",
)

blueprint_with_context = runloop.blueprint.create(
    name="my-blueprint-with-context",
    dockerfile="""\
FROM node:22
WORKDIR /usr/src/app

# copy using the build context from the object
COPY package.json package.json
COPY src src

RUN npm install --only=production
CMD ["node", "src/app.js"]
""",
    # Build context
    build_context=build_ctx_obj.as_build_context(),
)

# Or get an existing one
blueprint = runloop.blueprint.from_id(blueprint_id="bpt_123")

# List all blueprints
blueprints = runloop.blueprint.list()

# Get blueprint details and build logs
info = blueprint.get_info()
logs = blueprint.logs()

# Create a devbox from this blueprint
devbox = blueprint.create_devbox(name="devbox-from-blueprint")

# Delete the blueprint when done
blueprint.delete()

# Get an existing snapshot
snapshot = runloop.snapshot.from_id(snapshot_id="snp_123")

# List all snapshots
snapshots = runloop.snapshot.list()

# List snapshots for a specific devbox
devbox_snapshots = runloop.snapshot.list(devbox_id="dbx_123")

# Get snapshot details and check status
info = snapshot.get_info()
print(f"Snapshot status: {info.status}")

# Update snapshot metadata
snapshot.update(
    name="updated-snapshot-name",
    metadata={"version": "v2.0"},
)

# Wait for async snapshot to complete
snapshot.await_completed()

# Create a devbox from this snapshot
devbox = snapshot.create_devbox(name="devbox-from-snapshot")

# Delete the snapshot when done
snapshot.delete()

# Create a new storage object
storage_object = runloop.storage_object.create(
    name="my-file.txt",
    content_type="text",
    metadata={"project": "demo"},
)

# Upload content to the object
storage_object.upload_content("Hello, World!")
storage_object.complete()

# Upload from file
from pathlib import Path

uploaded = runloop.storage_object.upload_from_file(
    Path("/path/to/file.txt"),
    name="my-file.txt",
)

# Upload text content directly
uploaded = runloop.storage_object.upload_from_text(
    "Hello, World!",
    name="my-text.txt",
    metadata={"source": "text"},
)

# Upload from bytes
uploaded = runloop.storage_object.upload_from_bytes(
    b"binary content",
    name="my-file.bin",
    content_type="binary",
)

# Get object details and download
info = storage_object.refresh()
download_url = storage_object.get_download_url(duration_seconds=3600)

# Download content
text_content = storage_object.download_as_text()
binary_content = storage_object.download_as_bytes()

# List all storage objects
objects = runloop.storage_object.list()

# Delete when done
storage_object.delete()

from pathlib import Path

# Upload local file with content-type detection
obj = runloop.storage_object.upload_from_file(file_path=Path("./report.csv"))

# Manual control
obj = runloop.storage_object.create(
    name="data.bin",
    content_type="binary",
)
obj.upload_content(b"\xde\xad\xbe\xef")
obj.complete()

# Create a storage object first
storage_object = runloop.storage_object.upload_from_text(
    "Hello, World!",
    name="my-data.txt",
)

# Create a devbox and mount the storage object
devbox = runloop.devbox.create(
    name="my-devbox",
    mounts=[
        {
            "type": "object_mount",
            "object_id": storage_object.id,
            "object_path": "/home/user/data.txt",
        },
    ],
)

# The storage object is now accessible at /home/user/data.txt in the devbox
result = devbox.cmd.exec("cat /home/user/data.txt")
print(result.stdout())  # "Hello, World!"

# Mount archived objects (tar, tgz, gzip) - they get extracted to a directory
archive_object = runloop.storage_object.upload_from_file(
    Path("./project.tar.gz"),
    name="project.tar.gz",
)

devbox_with_archive = runloop.devbox.create(
    name="archive-devbox",
    mounts=[
        {
            "type": "object_mount",
            "object_id": archive_object.id,
            "object_path": "/home/user/project",  # Archive gets extracted here
        },
    ],
)

# Access extracted archive contents
result = devbox_with_archive.cmd.exec("ls -la /home/user/project/")
print(result.stdout())

# Create a new axon
axon = runloop.axon.create(name="my-axon")

# Or get an existing one
axon = runloop.axon.from_id(axon_id="axn_123")

# List all axons
axons = runloop.axon.list()

# Get axon details
info = axon.get_info()
print(f"Axon {info.name} created at {info.created_at_ms}")

result = axon.publish(
    event_type="task_completed",
    origin="AGENT_EVENT",     # EXTERNAL_EVENT | AGENT_EVENT | USER_EVENT
    payload='{"task_id": 42, "status": "done"}',
    source="my-agent",
)
print(f"Sequence: {result.sequence}, Timestamp: {result.timestamp_ms}")

with axon.subscribe_sse() as stream:
    for event in stream:
        print(f"[{event.source}] {event.event_type}: {event.payload}")
        # event.axon_id, event.origin, event.sequence, event.timestamp_ms also available

async with await axon.subscribe_sse() as stream:
    async for event in stream:
        print(f"[{event.source}] {event.event_type}: {event.payload}")

# Create a table
axon.sql.query(sql="CREATE TABLE tasks (id INTEGER PRIMARY KEY, name TEXT, done INTEGER)")

# Insert with parameterized queries (? placeholders)
axon.sql.query(
    sql="INSERT INTO tasks (id, name, done) VALUES (?, ?, ?)",
    params=[1, "Write docs", 0],
)

# Query
result = axon.sql.query(sql="SELECT * FROM tasks WHERE done = ?", params=[0])
for row in result.rows:
    print(row)           # e.g. [1, "Write docs", 0]
print(result.columns)    # [SqlColumnMetaView(name="id", type="INTEGER"), ...]
print(result.meta)       # SqlResultMetaView(changes=0, duration_ms=0.5, rows_read_limit_reached=False)

result = axon.sql.batch(
    statements=[
        {"sql": "CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT)"},
        {"sql": "INSERT INTO users (id, name) VALUES (?, ?)", "params": [1, "Alice"]},
        {"sql": "INSERT INTO users (id, name) VALUES (?, ?)", "params": [2, "Bob"]},
        {"sql": "SELECT * FROM users ORDER BY id"},
    ],
)

# Each statement gets a result
for step in result.results:
    if step.success:
        print(f"Rows: {step.success.rows}, Changes: {step.success.meta.changes}")
    elif step.error:
        print(f"Error: {step.error}")

runloop = RunloopSDK()
raw_devbox = runloop.api.devboxes.create()

from runloop_api_client import RunloopSDK
import runloop_api_client

runloop = RunloopSDK()

try:
    devbox = runloop.devbox.create(name="example-devbox")
    result = devbox.cmd.exec("invalid-command")
except runloop_api_client.APIConnectionError as e:
    print("The server could not be reached")
    print(e.__cause__)  # an underlying Exception, likely raised within httpx.
except runloop_api_client.RateLimitError as e:
    print("A 429 status code was received; we should back off a bit.")
except runloop_api_client.APIStatusError as e:
    print("Another non-200-range status code was received")
    print(e.status_code)
    print(e.response)

import httpx
from runloop_api_client import RunloopSDK, DefaultHttpxClient

runloop = RunloopSDK(
    bearer_token="your-api-key",  # defaults to RUNLOOP_API_KEY env var
    base_url="https://api.runloop.ai",  # or use RUNLOOP_BASE_URL env var
    timeout=60.0,  # 60 second timeout (default is 30)
    max_retries=3,  # Retry failed requests (default is 5)
    default_headers={
        "X-Custom-Header": "value",
    },
    # Custom HTTP client with proxy
    http_client=DefaultHttpxClient(
        proxy="http://my.test.proxy.example.com",
        transport=httpx.HTTPTransport(local_address="0.0.0.0"),
    ),
)

import asyncio
from runloop_api_client import AsyncRunloopSDK


async def main():
    runloop = AsyncRunloopSDK()

    # All the same operations, but with await
    async with await runloop.devbox.create(name="async-devbox") as devbox:
        result = await devbox.cmd.exec("pwd")
        print(await result.stdout())

        # Streaming (note: callbacks must be synchronous)
        def capture(line: str) -> None:
            print(">>", line)

        await devbox.cmd.exec("ls", stdout=capture)

        # Async file operations
        await devbox.file.write(path="/tmp/test.txt", contents="Hello")
        content = await devbox.file.read(path="/tmp/test.txt")

        # Async network operations
        tunnel = await devbox.net.create_tunnel(port=8080)
        print("Tunnel URL:", tunnel.url)


asyncio.run(main())

from runloop_api_client.lib.polling import PollingConfig

# Create devbox with custom polling
devbox = runloop.devbox.create(
    name="my-devbox",
    polling_config=PollingConfig(
        timeout_seconds=300.0,  # Wait up to 5 minutes
        interval_seconds=2.0,  # Poll every 2 seconds
    ),
)

# Wait for snapshot completion with custom polling
snapshot.await_completed(
    polling_config=PollingConfig(
        timeout_seconds=600.0,  # Wait up to 10 minutes
        interval_seconds=5.0,  # Poll every 5 seconds
    ),
)