Metadata-Version: 2.5
Name: quivvo
Version: 0.1.0
Summary: Python SDK for Quivvo: Linux machines for AI agents.
Project-URL: Homepage, https://tryquivvo.com
Project-URL: Documentation, https://tryquivvo.com/docs/sdks
License-Expression: MIT
License-File: LICENSE
Keywords: ai-agents,cloud-machines,firecracker,sandbox,sdk
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3 :: Only
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Typing :: Typed
Requires-Python: >=3.9
Description-Content-Type: text/markdown

# quivvo

The Python SDK for [Quivvo](https://tryquivvo.com): 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.

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

```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.

```python
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

```python
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](https://tryquivvo.com/docs/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

```python
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:

```python
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

```python
from quivvo import verify_webhook

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

## Methods

| Method | API |
| --- | --- |
| `q.whoami()` | `GET /whoami` |
| `q.catalog()` | `GET /catalog` |
| `q.usage()` | `GET /usage` |
| `q.queue()` | `GET /queue` |
| `q.feedback(message, machine=, context=)` | `POST /feedback` |
| `q.machines.list(state=, all=)` | `GET /machines` |
| `q.machines.create(name, template=, size=, vcpus=, memory_mib=, disk_gib=, region=, snapshot=, setup=, env=, auto_suspend_minutes=, delete_after_minutes=, wait=)` | `POST /machines` |
| `q.machines.get(m)` | `GET /machines/:id` |
| `q.machines.update(m, name=, auto_suspend_minutes=, vcpus=, memory_mib=, disk_gib=, delete_after_minutes=)` | `PATCH /machines/:id` |
| `q.machines.delete(m, confirm=)` | `DELETE /machines/:id` |
| `q.machines.start/resume/restart(m, wait=True)`, `stop(m)`, `suspend(m)` | `POST /machines/:id/<action>` |
| `q.machines.exec(m, cmd, argv=, cwd=, env=, timeout_ms=, stdin=, tail_bytes=)` | `POST /machines/:id/exec` |
| `q.machines.exec_stream(m, cmd, ...)` | `POST /machines/:id/exec`, `Accept: text/event-stream` |
| `q.machines.status(m)`, `ps(m)` | `GET /machines/:id/status`, `/ps` |
| `q.machines.logs(m, lines=, format=)` | `GET /machines/:id/logs` |
| `q.machines.history(m, limit=)`, `exec_output(m, exec_id)` | `GET /machines/:id/history`, `/history/:exec/output` |
| `q.machines.connect(m)` | `POST /machines/:id/connect` |
| `q.machines.fork(m, count=, name=)` | `POST /machines/:id/fork` |
| `q.machines.restore(m, snapshot)` | `POST /machines/:id/restore` |
| `q.jobs.start(m, cmd, argv=, cwd=, env=, timeout_ms=)` | `POST /machines/:id/exec` with `background` |
| `q.jobs.list(m)`, `get(m, job, wait=)`, `wait(m, job, timeout=)` | `GET /machines/:id/jobs`, `/jobs/:job` |
| `q.jobs.logs(m, job, offset=, tail_bytes=)`, `follow(m, job, offset=)` | `GET /machines/:id/jobs/:job/logs` |
| `q.jobs.kill(m, job, signal=)` | `POST /machines/:id/jobs/:job/kill` |
| `q.files.read(m, path)`, `list(m, path)` | `GET /machines/:id/files` |
| `q.files.write(m, path, data, mode=, append=)` | `PUT /machines/:id/files` |
| `q.files.delete(m, path, recursive=)` | `DELETE /machines/:id/files` |
| `q.ports.list(m)`, `publish(m, port, public=)`, `unpublish(m, port)` | `/machines/:id/ports` |
| `q.snapshots.create(m, name)` | `POST /machines/:id/snapshots` |
| `q.snapshots.list()`, `get(snapshot)`, `delete(snapshot, confirm=)` | `GET /snapshots`, `DELETE /snapshots/:id` |
| `q.keys.list()`, `create(name, scopes=)`, `revoke(key)` | `/keys` |
| `q.webhooks.list()`, `create(url, events=)`, `delete(webhook)` | `/webhooks` |

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

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

[REST API reference](https://tryquivvo.com/docs/api) · [OpenAPI](https://tryquivvo.com/api/v1/openapi.json)

## License

MIT. See [LICENSE](LICENSE).
