Skip to content
CAVS Node

Python SDK

Terminal window
pip install cavs

Python 3.9+. Built on httpx, streams everything, verifies every download against its content hash.

Configure

Section titled “Configure”
Terminal window
export CAVS_TOKEN=cavs_pat_...
export CAVS_API=https://cavsnode.com
export CAVS_ORG=acme-ai            # optional default organization
from cavs import CAVS


client = CAVS.from_env()
# or explicitly
client = CAVS(
    token="cavs_ci_...",
    api="https://cavsnode.com",
    org="acme-ai",
    timeout=30.0,
)

Token resolution: explicit argument → $CAVS_TOKEN. The token is redacted from repr(), exception messages and logs.

The client is a context manager, and it pools connections — build one and reuse it:

with CAVS.from_env() as client:
    ...

Upload

Section titled “Upload”
result = client.artifacts.upload(
    path="./checkpoints/epoch-42",     # a file or a whole directory
    project="vision-models",           # repository slug, name or UUID
    name="resnet50",
    kind="model",
    version="1.4.0",
    metadata={"framework": "pytorch", "epochs": 42, "val_acc": 0.942},
    labels=["production", "eu"],
)


print(result.reference)   # cavs://acme-ai/vision-models/model/resnet50:1.4.0
print(result.version)
print(result.raw)         # the raw finalize payload

upload runs the full upload-session choreography: hash every file (streamed, never buffered whole), open a session with an idempotency key, authorize, PUT to presigned URLs, complete, finalize.

Progress, cancellation, idempotency

Section titled “Progress, cancellation, idempotency”
def on_progress(done: int, total: int) -> None:
    print(f"{done}/{total} bytes")


client.artifacts.upload(
    path="./big-dataset",
    project="datasets", name="train", kind="dataset", version="3.0.0",
    on_progress=on_progress,
    idempotency_key="nightly-2026-07-28",   # a retry resumes instead of duplicating
)

Re-running an interrupted upload is safe: the authorize step reports which oids are already stored, so it skips them.

Download

Section titled “Download”
# By bare oid — project is required
client.artifacts.download(
    "9f2c8b0e0a1d4f6b7c3e5a9d8f1b2c4e6a8d0f2b4c6e8a0d2f4b6c8e0a2d4f6b",
    dest="./restored/model.safetensors",
    project="vision-models",
)


# By immutable reference
client.artifacts.download(
    "cavs://acme-ai/vision-models/model/resnet50@sha256:9f2c8b0e…",
    dest="./restored",
)

Downloads stream to a temp file while hashing, verify SHA-256 == oid, then rename atomically. A mismatch raises ChecksumError and the temp file is deleted — a bad download never reaches its destination path.

What works against the deployed API

Section titled “What works against the deployed API”
Call Status
client.me() ✓
client.resolve_repo(project) ✓ — resolves a slug to a UUID
client.artifacts.upload(...) ✓ (dedup stats absent)
client.artifacts.download(oid | @sha256 ref) ✓
client.artifacts.download(logical ref) ✗ needs /resolve
client.artifacts.list(...) ✓ — the organization registry
client.artifacts.get(id) / .versions(id) ✗ needs /artifacts/{id}
client.usage.get() ✓
client.snapshots.list() / .create() ✓
client.releases.list() / .get(tag) ✓
client.lineage.graph() / .add_edge() ✓
client.runs.* ✗ not deployed
client.service_accounts.* ✗ not deployed

The ✗ rows raise NotFoundError. See Choosing an integration.

The rest of the surface

Section titled “The rest of the surface”
# Registry
artifacts = client.artifacts.list(type="model", q="resnet", limit=20)
for a in artifacts:
    print(a.name, a.size, a.oid)


# Usage
usage = client.usage.get()
print(usage.logical_storage_bytes, usage.physical_storage_bytes)


# Releases
for rel in client.releases.list("vision-models"):
    print(rel.tag, rel.lfs_file_count)
release = client.releases.get("vision-models", "v1.4.0")


# Snapshots
client.snapshots.create("vision-models", idempotency_key="monthly-2026-07")
snapshots = client.snapshots.list("vision-models")


# Lineage
client.lineage.add_edge(
    "vision-models",
    from_type="artifact", from_key=model_oid,
    to_type="artifact",   to_key=dataset_oid,
    relation="trained-with",
)
graph = client.lineage.graph("vision-models")

Hashing helpers, for when you want the oid without uploading:

from cavs import sha256_file, sha256_reader


oid = sha256_file("./model.safetensors")

URI parsing:

from cavs import parse, format_uri


p = parse("cavs://acme-ai/vision-models/model/resnet50:1.4.0")
print(p.org, p.project, p.kind, p.name, p.version, p.immutable)

Errors

Section titled “Errors”

Every exception derives from CAVSError.

Exception Cause
AuthenticationError 401 — missing or invalid token
AuthorizationError 403 — role or scope insufficient
NotFoundError 404
ConflictError 409
QuotaExceededError 402 / 413 — a plan limit
RateLimitError 429 — carries .retry_after
ChecksumError downloaded bytes didn’t match the oid
UploadError / DownloadError transport failure during transfer
TemporaryServiceError 5xx or network error after retries 
from cavs import CAVS, QuotaExceededError, RateLimitError, ChecksumError


try:
    client.artifacts.upload(path="./model", project="p", name="n", kind="model", version="1")
except QuotaExceededError as e:
    print("plan limit hit:", e)
except RateLimitError as e:
    print("throttled, retry after", e.retry_after)
except ChecksumError:
    print("corrupt transfer — the file was not written")

Retries

Section titled “Retries”

Up to 4 attempts on 429, 5xx and transport errors, with exponential backoff plus jitter, honouring Retry-After (capped around 30 seconds).

from cavs import CAVS
from cavs._http import RetryConfig


client = CAVS(retry=RetryConfig(max_attempts=6))

Testing

Section titled “Testing”

Inject an httpx transport — no network, no mocking library:

import httpx
from cavs import CAVS
from cavs._http import Transport


def handler(request: httpx.Request) -> httpx.Response:
    if request.url.path == "/api/v1/users/me":
        return httpx.Response(200, json={"user": {"id": "u1", "email": "[email protected]"}})
    return httpx.Response(404, json={"error": {"code": "not_found", "message": "nope"}})


client = CAVS(
    token="cavs_pat_test",
    transport=Transport(api="https://test", token="cavs_pat_test",
                        client=httpx.Client(transport=httpx.MockTransport(handler))),
)
assert client.me().email == "[email protected]"

In a training job

Section titled “In a training job”
import os
from cavs import CAVS, sha256_file


client = CAVS.from_env()
PROJECT = "vision-models"


# 1. Pull the dataset by its pinned oid — reproducible by construction.
client.artifacts.download(os.environ["DATASET_OID"], dest="./data", project=PROJECT)


# 2. Train.
train("./data", out="./out/model.safetensors")


# 3. Publish, keyed so a retried job doesn't duplicate work.
run_id = os.environ["CI_RUN_ID"]
result = client.artifacts.upload(
    path="./out/model.safetensors",
    project=PROJECT, name="resnet50", kind="model",
    version=os.environ["GIT_SHA"],
    metadata={"run": run_id, "dataset_oid": os.environ["DATASET_OID"]},
    idempotency_key=f"train:{run_id}",
)


# 4. Record the oid downstream consumers should pin.
model_oid = sha256_file("./out/model.safetensors")
print(f"::set-output name=model_oid::{model_oid}")


# 5. Declare lineage against oids, not names.
client.lineage.add_edge(
    PROJECT,
    from_type="artifact", from_key=model_oid,
    to_type="artifact",   to_key=os.environ["DATASET_OID"],
    relation="trained-with",
)

Security notes

Section titled “Security notes”
  • Never hard-code the token; read it from $CAVS_TOKEN fed by a secret store.
  • Custom CA bundle: CAVS(ca_bundle="/path/to/ca.pem") or $CAVS_CA_BUNDLE.
  • The SDK never logs presigned URLs or full request headers.
  • Directory uploads guard against path traversal when expanding entries.

Next

Section titled “Next”