quivvo

The Python SDK for Quivvo: Linux machines for AI agents, each a Firecracker VM with root, Docker and a persistent disk at /data. Python 3.9 or later, no dependencies.

pip install quivvo            # or, from this repo: pip install ./sdk/python

from quivvo import Quivvo

q = Quivvo()                          # reads QUIVVO_API_KEY
m = q.machines.create(name="dev")
r = m.exec("uname -a")
print(r["exit_code"], r["stdout"])
m.delete()

Auth

Pass a qvk_ key as Quivvo(api_key=...) or set QUIVVO_API_KEY. Get one with quivvo login --agent or from the dashboard. It's sent as Authorization: Bearer. base_url falls back to QUIVVO_API_URL, then https://tryquivvo.com. q.whoami() shows the key's scopes and limits.

Quivvo(api_key=None, base_url=None, max_retries=4, timeout=60.0)

What comes back

The API's JSON, as plain dicts. Machines are Machine, a dict with methods (m["state"], m.exec(...), m.stop()). Calls that list return the list. quivvo.types has a TypedDict for each object, for your type checker. A machine is given by its id, its name, or the Machine itself.

Errors

from quivvo import QuivvoError, QuivvoNetworkError

try:
    q.machines.start("dev")
except QuivvoError as e:          # the API answered with an error
    print(e.status, e.code, e.message, e.request_id)
    if e.owner_action:            # only the workspace's owner can fix this
        print(e.owner_action["message"], e.owner_action["url"])
except QuivvoNetworkError as e:   # nothing answered
    print(e.message, e.maybe_acted)

QuivvoError carries status, code, message, retryable, retry_after_seconds, docs_url, request_id and owner_action. Branch on code; errors lists them. A command that exits non-zero is a result, not an error.

Retryable errors are retried up to max_retries times, after retry_after_seconds (or 0.5 s doubling to 8 s), never waiting more than 60 s. Reads, deletes, and lifecycle POSTs are retried, and lifecycle POSTs (create, start, stop, suspend, resume, restart, snapshot, fork, restore) send one Idempotency-Key for every attempt, so a retry can't act twice. Pass idempotency_key= to choose it. Exec and other POSTs are only repeated when the server says nothing ran: rate_limited, or machine_not_running while it boots.

Streaming

with q.machines.exec_stream("dev", "make test") as events:
    for ev in events:
        if ev["event"] in ("stdout", "stderr"):
            print(ev["data"], end="")
        elif ev["event"] == "exit":
            print("exit", ev["exit_code"])

There's no output cap. A chunk that isn't valid UTF-8 also has data_base64. Leaving the loop hangs up, which stops the command. A stream cut before exit raises StreamLostError.

For work longer than 280 seconds, start a job and follow it:

import sys

job = q.jobs.start("dev", "make build")["job"]
for chunk in q.jobs.follow("dev", job["id"]):
    sys.stdout.buffer.write(chunk)
print(q.jobs.wait("dev", job["id"])["exit_code"])

Webhooks

from quivvo import verify_webhook

ok = verify_webhook(secret, raw_body, headers["Quivvo-Signature"])  # rejects a t over 5 minutes old

Methods

A Machine has refresh(), start(), stop(), suspend(), resume(), restart(), delete(), exec() and exec_stream(). Start a machine from a snapshot with q.machines.create(snapshot="base"). q.snapshots.get() reads the list, as the API has no route for one snapshot.

Tests

cd sdk/python && PYTHONPATH=src python3 -m unittest discover -s tests

REST API reference · OpenAPI

License

MIT. See LICENSE.