CALIBER Python SDK

caliber-sdk is a typed Python client for the CALIBER management API. Installing it does not install the CALIBER server — no mlflow, no sqlalchemy, no starlette. That isolation is the reason it is a separate distribution, and CI fails the build if a server dependency ever appears in the wheel metadata.

Every code block on this page is extracted from sdk/caliber-sdk/examples/ at build time. The SDK test suite executes those functions, so a snippet here cannot drift from working code: if a signature changes, either the docs change with it or the build fails.

This page explains the Python abstraction. If you need the raw wire contract, use the REST API overview, the authentication and conventions guide, and the HTTP reference.

At a glance#

Install#

pip install caliber-sdk

Quickstart#

Construct a client, confirm who you are, and read what the deployment supports:

def quickstart(caliber: CaliberClient) -> dict[str, Any]:
    """Report who you are and which API surfaces are GA on this deployment."""
    identity = caliber.me.get()
    if identity.is_anonymous:
        # /me answers "who am I" rather than requiring a credential, so an
        # invalid token shows up here as anonymous instead of an exception.
        raise SystemExit("no usable credential — check CALIBER_TOKEN")

    capabilities = caliber.capabilities_api.get()
    return {
        "user_id": identity.user_id,
        "scopes": identity.scopes,
        "ga_surfaces": sorted(capabilities.sdk_stability.get("ga", [])),
        "queue_enabled": capabilities.workflow_runs.queue_enabled,
    }

Two things worth noticing. me.get() reports identity rather than requiring it — an invalid or revoked credential comes back as an anonymous identity, not an exception — so the check is on the value, not a try. And capabilities answers "may I call this?" without downloading the full OpenAPI document.

Configuration#

Everything can come from the environment, so a CI job needs no argument plumbing of its own:

An explicit argument always wins over the environment, and a token always wins over a trusted header — the token is a real credential and the header is only an assertion, so silently preferring the weaker one would be the wrong surprise.

If you are wiring the SDK into CI/CD, the configuration usually lands in a YAML workflow rather than an interactive shell:

name: release-readiness
on:
  workflow_dispatch:

jobs:
  score-release:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.12"
      - run: pip install caliber-sdk
      - env:
          CALIBER_BASE_URL: https://caliber.example.com
          CALIBER_TOKEN: ${{ secrets.CALIBER_TOKEN }}
          CALIBER_PROJECT: release-governance
        run: python scripts/score_release.py

That example stays deliberately small: the point is that the SDK can be dropped into a normal build job without a server-side Python environment.

Project access#

Projects now report the caller's effective role and permissions, and the sync client includes typed membership management. Use the project response as the authorization-aware feature check before showing a write or publish action:

project = client.projects.get("PRJ-1")
if "resource.write" not in project.permissions:
    raise RuntimeError(f"project role {project.access_role!r} cannot write")

members = client.projects.list_members(project.project_id)
client.projects.update_member(project.project_id, "@bob", role="reviewer")

The four project roles are owner, editor, reviewer, and viewer. Only an owner can add, update, or remove members. remove_member() deactivates the membership and returns True when the server confirms the removal. The async client intentionally keeps a narrower typed surface; use its raw transport for these membership routes until an async project resource is added.

Authentication#

Personal access tokens are the supported credential for automation. Sessions are the wrong tool for a script: they come from password login, expire on a human timescale, and carry the user's full authority.

def issue_scoped_token(caliber: CaliberClient, *, name: str = "ci") -> dict[str, Any]:
    """Create a token limited to operator scope.

    The scope list is a *ceiling*: the effective authority is the intersection
    with what the owner holds when the request is made. Requesting more than
    you hold is refused outright rather than silently narrowed, so a token
    never claims authority it cannot exercise.
    """
    issued = caliber.auth.tokens.create(name, scopes=["caliber.operator"])
    # The plaintext exists exactly once. There is no endpoint that returns it
    # again — store it now or rotate to get a new one.
    secret = issued.token

    live = [token.token_id for token in caliber.auth.tokens.list() if token.active]
    caliber.auth.tokens.revoke(issued.token_id)
    return {"token_id": issued.token_id, "secret_len": len(secret), "live_before_revoke": live}

Scopes are a ceiling, not a grant. The effective authority is what the token requested intersected with what its owner holds at request time. So a token cannot exceed its owner, and demoting the owner narrows every token they hold — immediately, without revoking anything. Omit scopes to inherit the owner's.

Requesting a scope you do not hold is refused at issue time rather than quietly narrowed, because a token that claims authority it can never exercise fails far from its cause.

CSRF#

Handled for you. If a write is refused for want of a token, the client fetches one and replays the request exactly once. A 403 that is not about CSRF is not mistaken for one, so a genuine permission failure surfaces immediately instead of looping.

Authoring versus deploying#

The refinement loop depends on being able to register a candidate without it going live. The SDK keeps that separation explicit: registering a version never moves an alias, and promotion is its own call and its own audit event.

def prompt_lifecycle(
    caliber: CaliberClient, *, agent_id: str = "intake-classifier"

Evidence and scoring#

def build_and_score(caliber: CaliberClient, *, owner: str = "@you") -> dict[str, Any]:
    """Create a dataset, add a row, define a judge, and run an evaluation."""
    dataset = caliber.datasets.create("intake-golden", owner=owner)
    caliber.datasets.add_example(
        dataset.dataset_id,
        inputs={"ticket": "I was charged twice"},
        expected={"intent": "billing"},
    )

    # Instructions must reference an evaluation variable. A judge with no
    # variable grades nothing — it returns the same verdict every time — so
    # the server rejects it rather than letting you collect meaningless scores.
    judge = caliber.judges.create(
        "valid-intent",
        instructions="Given {{ inputs }} and {{ outputs }}, return true if intent is allowed.",
        feedback_value_type="bool",
    )

    evaluation = caliber.evaluations.create(dataset.dataset_id, judge_id=judge.judge_id)
    return {
        "dataset_id": dataset.dataset_id,
        "judge_id": judge.judge_id,
        "evaluation_id": evaluation.evaluation_id,
    }

Judge instructions must reference at least one evaluation variable — {{ inputs }}, {{ outputs }}, {{ expectations }}, {{ conversation }}, or {{ trace }}. The server enforces it because a judge with no variable grades nothing: it returns the same verdict for every example, and the resulting scorecard would look like evidence while measuring nothing.

Running workflows#

def run_and_wait(
    caliber: CaliberClient, *, workflow_id: str, alias: str = "prod"

Targeting an alias rather than a version id is the point of deploying one: the caller does not change when a new version is promoted.

Submission is the single mutating call the SDK will not retry for you. It cannot know whether a failure happened before or after the run was created, and duplicating a run is worse than surfacing an error — pass an idempotency_key to make your own retry safe.

Waiting on long-running work#

Runs, calibration jobs, and evaluations are asynchronous. The waiters poll with capped exponential backoff and never sleep past the deadline you gave them.

They differ on failure, deliberately:

A script whose run failed almost always wants to stop. A calibration score or an evaluation metric, by contrast, is the measurement — a low one is the result, not an error in the call.

Errors#

CaliberError
├── CaliberConfigError        the client was built with unusable configuration
├── CaliberTransportError     no HTTP response at all (DNS, refused, timeout)
└── CaliberAPIError           the server answered with a non-2xx
    ├── CaliberValidationError    400 with a structured field list
    ├── CaliberAuthenticationError 401
    ├── CaliberPermissionError     403
    ├── CaliberNotFoundError       404
    ├── CaliberConflictError       409
    ├── CaliberRateLimitError      429
    └── CaliberServerError         5xx

CaliberTransportError is separate from CaliberAPIError because there is no server verdict to inspect — and because retrying is often right in one case and wrong in the other.

A validation failure prints the field and the server's reason:

[400] request body validation failed (POST .../judges) request_id=… —
instructions: instructions must reference at least one evaluation variable

The structured body behind that exception is JSON, and the SDK preserves it on the typed error object:

{
  "detail": "request body validation failed",
  "status_code": 400,
  "errors": [
    {
      "loc": ["instructions"],
      "msg": "instructions must reference at least one evaluation variable",
      "type": "value_error"
    }
  ]
}

loc is a path, not a field name — it is a list because the failure may sit inside a nested object, as in ["manifest", "nodes", 0, "tool_ref"]. That is why CaliberValidationError.errors hands you the list unchanged, and why its __str__ joins the path with dots rather than printing only the last segment.

Anything not yet modelled#

client.raw reaches any endpoint under /ajax-api/2.0/mlflow/caliber:

caliber.raw.get("/observability/traces")
caliber.raw.post("/cookbooks/01/install", json={"name": "My install"})
for item in caliber.raw.paginate("/workflows", limit=50):
    ...

This is permanent, not scaffolding. A typed façade that lags the server would otherwise make new endpoints unreachable until the SDK caught up.

Forward compatibility#

Unknown response fields are kept in extra rather than dropped, and missing ones fall back to defaults rather than raising. A newer server does not break an older client, and an older server does not break a newer one.