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 (in the "default" project; pass project_id=... to choose another one)
        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, description=None, project_id=None)	Register a new model in a project (the default project if omitted)
get_model(model_id)	Fetch model details
list_models(project_id=None)	List the models you can access, optionally of one project
list_projects()	List the projects you can access, including your role (my_role) and model count
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

Server version: project_id and list_projects() require a server version with projects (introduced together with this SDK release). The SDK does not filter client-side: an older server ignores the project_id query parameter of list_models() and returns the models of all projects, and list_projects() fails with a 404. Upgrade the server before using these features. Model names are unique per project, so look models up by ID (or by name within one project) when the same name can exist in several projects.

Full API reference

Auto-generated from source:

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

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. 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 user ADC (ID token)
async with YaaiClient("http://localhost:8000/api/v1") as client:
    model = await client.create_model("my-model")

# With a longer timeout for slow server responses
async with YaaiClient("http://localhost:8000/api/v1", api_key="yaam_...", timeout=60.0) 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,
    timeout: float = 30.0,
) -> 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, timeout=timeout)

create_model(name, description=None, project_id=None) async

Create a model, optionally inside a project (the default project if omitted).

Source code in yaai/client.py

async def create_model(
    self,
    name: str,
    description: str | None = None,
    project_id: uuid.UUID | None = None,
) -> ModelRead:
    """Create a model, optionally inside a project (the default project if omitted)."""
    payload = ModelCreate(name=name, description=description, project_id=project_id)
    resp = await self._request("POST", "/models", json=payload.model_dump(mode="json"))
    return ModelRead.model_validate(resp.json()["data"])

list_models(project_id=None) async

List the models you can access, optionally restricted to one project.

Source code in yaai/client.py

async def list_models(self, project_id: uuid.UUID | None = None) -> list[ModelRead]:
    """List the models you can access, optionally restricted to one project."""
    params = {"project_id": str(project_id)} if project_id is not None else None
    return [ModelRead.model_validate(m) for m in await self._get_all_pages("/models", params)]

list_projects() async

List the projects you can access, including your role (my_role) and model counts.

Source code in yaai/client.py

async def list_projects(self) -> list[ProjectRead]:
    """List the projects you can access, including your role (``my_role``) and model counts."""
    resp = await self._request("GET", "/projects")
    return [ProjectRead.model_validate(p) for p in resp.json()["data"]]

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