Python SDK

The SDK is a thin async client. It covers what a service account needs to do: register models, send data, upload baselines. Dashboards, drift detection, and alerting happen server-side automatically.

Install

pip install yaai-monitoring          # just the client (httpx + pydantic)
pip install yaai-monitoring[gcp]     # adds google-auth for GCP service accounts

Authentication

Two options:

# Explicit API key (recommended for most setups)
client = YaaiClient("http://localhost:8000/api/v1", api_key="yaam_...")

# Google Application Default Credentials (requires yaai[gcp])
# Uses GOOGLE_APPLICATION_CREDENTIALS, workload identity, or GCP metadata server
client = YaaiClient("http://localhost:8000/api/v1")

When no api_key is passed, the client uses Google ADC and refreshes tokens automatically.

Quick start

The fastest way to get going: copy the Model ID from the UI, then use get_or_create_version. You always provide the schema info (either a sample record or an explicit field list). If the version already exists the schema params are ignored; if it doesn't exist, a new version is created. This makes the call idempotent — it always succeeds regardless of prior state.

Option 1: Auto-infer schema from sample data

import asyncio
from yaai import YaaiClient

async def main():
    async with YaaiClient("http://localhost:8000/api/v1", api_key="yaam_...") as client:

        MODEL_ID = "your-model-uuid-here"

        # Idempotent: creates version "v2" if it doesn't exist; returns it if it does.
        # Schema is inferred from the sample record.
        version = await client.get_or_create_version(
            MODEL_ID,
            "v2",
            sample_data={"inputs": {"amount": 100.0, "country": "DE"}, "outputs": {"is_fraud": "false"}},
        )

        await client.add_inference(
            model_version_id=version.id,
            inputs={"amount": 42.0, "country": "US"},
            outputs={"is_fraud": "false"},
        )

asyncio.run(main())

Option 2: Define schema explicitly

import asyncio
from yaai import YaaiClient
from yaai.schemas.model import SchemaFieldCreate

async def main():
    async with YaaiClient("http://localhost:8000/api/v1", api_key="yaam_...") as client:

        MODEL_ID = "your-model-uuid-here"

        # Same idempotent behaviour, but with an explicit schema definition.
        version = await client.get_or_create_version(
            MODEL_ID,
            "v2",
            schema_fields=[
                SchemaFieldCreate(field_name="amount", direction="input", data_type="numerical"),
                SchemaFieldCreate(field_name="country", direction="input", data_type="categorical"),
                SchemaFieldCreate(field_name="is_fraud", direction="output", data_type="categorical"),
            ],
        )

        await client.add_inference(
            model_version_id=version.id,
            inputs={"amount": 42.0, "country": "US"},
            outputs={"is_fraud": "false"},
        )

asyncio.run(main())

Note: sample_data and schema_fields are mutually exclusive — pass exactly one. If you only need to look up a version without creating it, use get_version_by_label instead.

## Usage

The client is an async context manager:

```python
import asyncio
from yaai import YaaiClient
from yaai.schemas.model import SchemaFieldCreate

async def main():
    async with YaaiClient("http://localhost:8000/api/v1", api_key="yaam_...") as client:

        # Register a model
        model = await client.create_model("fraud-detector")

        # Create a version with schema
        version = await client.create_model_version(
            model_id=model.id,
            version="v1.0",
            schema_fields=[
                SchemaFieldCreate(field_name="amount", direction="input", data_type="numerical"),
                SchemaFieldCreate(field_name="country", direction="input", data_type="categorical"),
                SchemaFieldCreate(field_name="is_fraud", direction="output", data_type="categorical"),
            ],
        )

        # Log a single inference
        await client.add_inference(
            model_version_id=version.id,
            inputs={"amount": 150.0, "country": "DE"},
            outputs={"is_fraud": "false"},
        )

        # Log a batch (up to 10,000 records)
        await client.add_inferences(
            model_version_id=version.id,
            records=[
                {"inputs": {"amount": 42.0, "country": "US"}, "outputs": {"is_fraud": "false"}},
                {"inputs": {"amount": 9001.0, "country": "NG"}, "outputs": {"is_fraud": "true"}},
            ],
        )

        # Upload reference data (training distribution)
        await client.add_reference_data(
            model_id=model.id,
            model_version_id=version.id,
            records=[
                {"inputs": {"amount": 85.0, "country": "DE"}, "outputs": {"is_fraud": "false"}},
                # ... typically hundreds or thousands of records
            ],
        )

asyncio.run(main())

Methods at a glance

Method	What it does
create_model(name)	Register a new model
get_model(model_id)	Fetch model details
list_models()	List all models
delete_model(model_id)	Delete a model and all its data
create_model_version(model_id, version, schema_fields)	Create a versioned schema
get_version(model_id, version_id)	Fetch full version details
get_version_by_label(model_id, version)	Look up a version by label (returns None if not found)
get_or_create_version(model_id, version, *, sample_data, schema_fields)	Idempotent upsert — always requires sample_data or schema_fields; returns existing or creates new
add_inference(model_version_id, inputs, outputs)	Log one inference
add_inferences(model_version_id, records)	Log a batch of inferences
add_reference_data(model_id, model_version_id, records)	Upload baseline data
add_ground_truth(inference_id, label)	Attach ground truth to an inference
get_version_job(model_id, model_version_id)	Get the drift job for a version
get_job(job_id)	Fetch job details
update_job(job_id, **fields)	Update job configuration
trigger_job(job_id)	Trigger a drift detection run
backfill_job(job_id)	Trigger historical drift backfill
infer_schema(sample)	Infer schema from a single sample
infer_schema_batch(samples)	Infer schema from multiple samples
validate_schema(schema_fields, inputs, outputs)	Validate a record against an inline schema
validate_schema_batch(schema_fields, records)	Validate a batch against an inline schema
validate_model_version_schema(model_id, version_id, inputs, outputs)	Validate a record against a version's schema
validate_model_version_schema_batch(model_id, version_id, records)	Validate a batch against a version's schema

Full API reference

Auto-generated from source:

yaai.client.YaaiClient(base_url, *, api_key=None, target_audience=None)

Async client for the yaai monitoring API.

Authenticates with either an explicit API key or Google Application Default Credentials (ADC). When no api_key is provided the client tries to obtain credentials via google.auth.default() — install yaai[gcp] to enable this.

When target_audience is provided (recommended for Google SA auth), the client requests an ID token scoped to that audience instead of a generic access token. The audience must match the server's configured GOOGLE_SA_AUDIENCE.

Usage::

# With API key
async with YaaiClient("http://localhost:8000/api/v1", api_key="yaam_...") as client:
    model = await client.create_model("my-model")

# With Google ADC + ID token (recommended for service accounts)
async with YaaiClient("http://localhost:8000/api/v1", target_audience="https://yaai.example.com") as client:
    model = await client.create_model("my-model")

# With Google ADC (access token fallback)
async with YaaiClient("http://localhost:8000/api/v1") as client:
    model = await client.create_model("my-model")

Source code in yaai/client.py

def __init__(self, base_url: str, *, api_key: str | None = None, target_audience: str | None = None) -> None:
    self._base_url = base_url.rstrip("/")
    self._credentials = None
    self._google_request = None

    if api_key is not None:
        headers: dict[str, str] = {"X-API-Key": api_key}
    else:
        headers = self._init_google_credentials(target_audience)

    self._client = httpx.AsyncClient(
        base_url=self._base_url,
        headers=headers,
    )

get_version(model_id, version_id) async

Fetch full details for a specific model version.

Source code in yaai/client.py

async def get_version(
    self,
    model_id: uuid.UUID,
    version_id: uuid.UUID,
) -> ModelVersionRead:
    """Fetch full details for a specific model version."""
    resp = await self._request("GET", f"/models/{model_id}/versions/{version_id}")
    return ModelVersionRead.model_validate(resp.json()["data"])

get_version_by_label(model_id, version) async

Look up a model version by its label.

Returns the :class:ModelVersionSummary if a version with the given label exists, or None otherwise.

Source code in yaai/client.py

async def get_version_by_label(
    self,
    model_id: uuid.UUID,
    version: str,
) -> ModelVersionSummary | None:
    """Look up a model version by its label.

    Returns the :class:`ModelVersionSummary` if a version with the given
    label exists, or ``None`` otherwise.
    """
    model = await self.get_model(model_id)
    return next((v for v in model.versions if v.version == version), None)

get_or_create_version(model_id, version, *, sample_data=None, schema_fields=None, description=None, keep_previous_active=False) async

Idempotent version upsert: return existing or create new.

Always requires exactly one of sample_data or schema_fields so the call is deterministic regardless of server state. If the version already exists the creation parameters are ignored and the existing :class:ModelVersionSummary is returned. If it doesn't exist, a new version is created from the provided schema info.

For a pure lookup without creation parameters, use :meth:get_version_by_label instead.

Parameters:

Name	Type	Description	Default
model_id	UUID	The UUID of the parent model.	required
version	str	Human-readable version label (e.g. "v2.0").	required
sample_data	dict[str, dict] | None	A dict with "inputs" and "outputs" keys used to infer the schema. Mutually exclusive with schema_fields.	None
schema_fields	list[SchemaFieldCreate] | None	An explicit list of :class:SchemaFieldCreate objects defining the version's schema. Mutually exclusive with sample_data.	None
description	str | None	Optional description for the new version.	None
keep_previous_active	bool	If True, existing active versions stay active when creating a new version.	False

Returns:

Type	Description
ModelVersionRead | ModelVersionSummary	The matched :class:ModelVersionSummary if the version already
ModelVersionRead | ModelVersionSummary	exists, or a freshly created :class:ModelVersionRead.

Raises:

Type	Description
ValueError	If both sample_data and schema_fields are provided, or if neither is provided.

Source code in yaai/client.py

async def get_or_create_version(
    self,
    model_id: uuid.UUID,
    version: str,
    *,
    sample_data: dict[str, dict] | None = None,
    schema_fields: list[SchemaFieldCreate] | None = None,
    description: str | None = None,
    keep_previous_active: bool = False,
) -> ModelVersionRead | ModelVersionSummary:
    """Idempotent version upsert: return existing or create new.

    Always requires exactly one of ``sample_data`` or ``schema_fields``
    so the call is deterministic regardless of server state.  If the
    version already exists the creation parameters are ignored and the
    existing :class:`ModelVersionSummary` is returned.  If it doesn't
    exist, a new version is created from the provided schema info.

    For a pure lookup without creation parameters, use
    :meth:`get_version_by_label` instead.

    Args:
        model_id: The UUID of the parent model.
        version: Human-readable version label (e.g. ``"v2.0"``).
        sample_data: A dict with ``"inputs"`` and ``"outputs"`` keys used
            to infer the schema.  Mutually exclusive with
            ``schema_fields``.
        schema_fields: An explicit list of :class:`SchemaFieldCreate`
            objects defining the version's schema.  Mutually exclusive
            with ``sample_data``.
        description: Optional description for the new version.
        keep_previous_active: If *True*, existing active versions stay
            active when creating a new version.

    Returns:
        The matched :class:`ModelVersionSummary` if the version already
        exists, or a freshly created :class:`ModelVersionRead`.

    Raises:
        ValueError: If both ``sample_data`` and ``schema_fields`` are
            provided, or if neither is provided.
    """
    if sample_data is not None and schema_fields is not None:
        msg = "sample_data and schema_fields are mutually exclusive. Provide one or the other, not both."
        raise ValueError(msg)

    if sample_data is None and schema_fields is None:
        msg = (
            "Either sample_data or schema_fields must be provided. "
            "Pass sample_data={'inputs': {...}, 'outputs': {...}} to "
            "auto-infer the schema, or pass schema_fields=[...] to "
            "define it explicitly. For a pure lookup use "
            "get_version_by_label() instead."
        )
        raise ValueError(msg)

    model = await self.get_model(model_id)
    for v in model.versions:
        if v.version == version:
            return v

    # Version doesn't exist — create it
    if schema_fields is not None:
        return await self.create_model_version(
            model_id,
            version,
            schema_fields,
            description=description,
            keep_previous_active=keep_previous_active,
        )

    inferred = await self.infer_schema(sample_data)
    return await self.create_model_version(
        model_id,
        version,
        inferred.schema_fields,
        description=description,
        keep_previous_active=keep_previous_active,
    )

get_version_job(model_id, model_version_id) async

Get the single job for a model version, or None if none exists.

Source code in yaai/client.py

async def get_version_job(
    self,
    model_id: uuid.UUID,
    model_version_id: uuid.UUID,
) -> dict | None:
    """Get the single job for a model version, or None if none exists."""
    resp = await self._request("GET", f"/models/{model_id}/versions/{model_version_id}/jobs")
    jobs = resp.json()["data"]
    return jobs[0] if jobs else None