Python SDK

The official Python SDK for the KLIQ AI API. Supports both synchronous and async usage, with full type hints and auto-pagination.

pip install kliq

KliqClient

The synchronous client for simple scripts and integrations.

Methods

from kliq import KliqClient
import os

kliq = KliqClient(api_key=os.environ["KLIQ_API_KEY"])
tenant = kliq.for_tenant("t_abc123")

AsyncKliqClient

For async applications (FastAPI, asyncio scripts, etc.):

from kliq import AsyncKliqClient
import asyncio

async def main():
    kliq = AsyncKliqClient(api_key=os.environ["KLIQ_API_KEY"])
    tenant = kliq.for_tenant("t_abc123")

    obs = await tenant.observations.create(
        location_id="loc_abc123",
        image_url="https://bucket.com/photo.jpg",
    )
    print(obs.id)

asyncio.run(main())

The AsyncKliqClient has the same interface as KliqClient, but all methods return coroutines.

TenantClient

Tenant-scoped client with access to all resource APIs.

Properties

Observations

observations.create(**kwargs)

Returns: Observation

observations.get(id: str)

Returns: Observation

observations.list(**kwargs)

Returns: PaginatedResponse[Observation]

observations.list_all(**kwargs)

Auto-paginating generator. Returns: Iterator[Observation]

for obs in tenant.observations.list_all(status="completed"):
    print(obs.id, obs.status)

observations.create_batch(items: list[dict])

Returns: list[Observation]

observations.delete(id: str)

Returns: None

observations.get_upload_url(**kwargs)

Returns: UploadUrl (with upload_url and image_url fields)

Locations

locations.create(**kwargs)

Returns: Location

locations.get(id) / locations.list(**kwargs) / locations.update(id, **kwargs) / locations.delete(id)

Standard CRUD operations with the same pagination pattern.

CvJobs

cv.start(**kwargs)

Returns: CvJob

cv.get(id: str)

Returns: CvJob

cv.wait_for_completion(id, **kwargs)

Polls until the job completes or fails.

Returns: CvJob

cv.list(**kwargs)

Returns: PaginatedResponse[CvJob]

Webhooks

webhooks.create(**kwargs)

Returns: Webhook

webhooks.list() / webhooks.delete(id)

Standard list and delete operations.

TrackedObjects (Catalog)

catalog.create(**kwargs) / catalog.import_csv(**kwargs) / catalog.list(**kwargs)

Manage the SKU/product catalog for CV model matching.

Buildings (BuildingScan)

buildings.create(**kwargs) / buildings.get(id) / buildings.list(**kwargs) / buildings.update(id, **kwargs)

Manage buildings for inspection workflows.

Assets (BuildingScan)

assets.create(**kwargs) / assets.get(id) / assets.list(**kwargs) / assets.update(id, **kwargs)

Manage individual building assets/components.

Visits

visits.create(**kwargs) / visits.get(id) / visits.list(**kwargs)

Track inspection visits and link them to observations.

Models

All response objects are dataclasses with typed fields:

Observation

@dataclass
class Observation:
    id: str
    tenant_id: str
    location_id: str
    image_url: str
    latitude: float | None
    longitude: float | None
    status: str  # "pending" | "processing" | "completed" | "failed"
    metadata: dict
    created_at: str
    updated_at: str

CvJob

@dataclass
class CvJob:
    id: str
    observation_id: str
    model: str
    status: str  # "queued" | "processing" | "completed" | "failed"
    result: CvResult | None
    error: str | None
    created_at: str
    completed_at: str | None

Detection

@dataclass
class Detection:
    object_name: str
    confidence: float
    bounding_box: BoundingBox
    attributes: dict

PaginatedResponse

@dataclass
class PaginatedResponse(Generic[T]):
    data: list[T]
    cursor: Cursor

@dataclass
class Cursor:
    next: str | None
    has_more: bool

Error Handling

The SDK raises typed exceptions:

from kliq.exceptions import (
    KliqError,           # Base exception
    AuthenticationError, # 401 responses
    ForbiddenError,      # 403 responses
    NotFoundError,       # 404 responses
    RateLimitError,      # 429 responses (auto-retried)
    ValidationError,     # 400 responses
    ServerError,         # 500+ responses
)

try:
    obs = tenant.observations.get("invalid_id")
except NotFoundError as e:
    print(f"Not found: {e.message}")
except KliqError as e:
    print(f"API error: {e.status_code} - {e.message}")