Skip to main content

SDK Reference

Official Python SDK (colabhive) for the ColabHive Builder APIs.

The SDK is two things at once, mirroring the platform:

  • Use inference now — call any public endpoint (LLMs, specialists, tools, generative) via client.endpoints.
  • Bring your own model — import any model from HuggingFace with client.models.hf, or train / merge / retrain your own with client.training, then serve it.

Installation​

pip install colabhive

Requires Python 3.8+. Depends on httpx and pydantic (installed automatically).

Quick Start​

from colabhive import ColabHive
import os

# Initialize client
client = ColabHive(
    api_key=os.getenv("COLABHIVE_API_KEY"),
    account_id=os.getenv("COLABHIVE_ACCOUNT_ID"),
)

# Call a public LLM (inference, no training required)
result = client.endpoints.infer(
    "mistral-7b-instruct-public",
    {"messages": [{"role": "user", "content": "Say hello in 3 languages."}]},
)
print(result["result"])

# Or train your own model
dataset = client.datasets.upload(name="my_data", file="./train.csv")
job = client.training.create(model="xgboost-regression", dataset_id=dataset.id)
job.wait()
print(job.get_metrics())

Authentication​

Get your API key and account ID from console.colabhive.com. API keys are prefixed hive_. Pass them as constructor arguments (or read them from the environment):

client = ColabHive(
    api_key="hive_...",
    account_id="0914e1c6-...",
)

The client always sends X-API-Key and sends X-Account-ID when account_id is configured. Your account is derived from the API key; if supplied, X-Account-ID must match the key's bound account or the gateway returns HTTP 403. API keys cannot switch accounts with this header. Account switching is available only to authenticated Console sessions and only after membership validation.

Scopes: an API key currently grants access to everything the account can do. Per-scope keys are on the roadmap, not enforced yet.

Pagination​

Two styles are used, depending on the resource — the SDK matches the gateway exactly:

MethodStyleParameters
datasets.listcursorlimit, cursor
training.listcursorlimit, cursor, status
models.listcursorlimit, cursor, visibility
models.versionscursorlimit, cursor
models.hf.list_registryoffsetlimit, offset

API Reference​

Client​

ColabHive(api_key, account_id, base_url=None, timeout=30.0)​

Main client for interacting with ColabHive APIs.

Parameters:

  • api_key (str): Your API key (hive_...) from console.colabhive.com
  • account_id (str): Your account ID
  • base_url (str, optional): API base URL. Defaults to $COLABHIVE_BASE_URL or https://api.colabhive.com
  • timeout (float): Request timeout in seconds (default: 30.0)

Namespaces: client.datasets, client.training, client.models (with client.models.hf), client.endpoints, client.users.

Example:

client = ColabHive(
    api_key="hive_...",
    account_id="0914e1c6-...",
    base_url="https://api.colabhive.com",
)

Import a model from HuggingFace​

ColabHive is not a fixed catalog — you can import any compatible model from HuggingFace and it becomes a live inference endpoint (cold-started on first request). Access via client.models.hf.

How HuggingFace imports are treated: models imported this way are always registered as public, candidate, base, and free (price_per_request = 0). The API accepts only visibility="public" and lifecycle_status="candidate" during import; alternatives return HTTP 422. If you need a private, account-scoped, or priced endpoint, register a trained model instead (see register_for_inference).

Search models​

search(query, task_type=None, max_size_gb=None, license=None, min_downloads=None, limit=20) → List[HFModelInfo]

results = client.models.hf.search(
    query="embedding model",
    task_type="embeddings",
    max_size_gb=2,
    min_downloads=10000,
)

# Results are dataclasses — use attribute access, not dict keys.
for m in results:
    print(m.repo_id, m.compatibility.status)

task_type uses the ColabHive vocabulary: embeddings, rerank, translation, ocr, stt, moderation, text-generation. compatibility.status is one of compatible, requires_review, incompatible.

Get model info​

info(repo_id) → HFModelDetails

details = client.models.hf.info("sentence-transformers/all-MiniLM-L6-v2")
print(details.size_mb, details.estimated_requirements)

if details.compatibility.status == "compatible":
    print("Ready to register!")
else:
    for check in details.compatibility.checks:
        if not check.passed:
            print("Warning:", check.message)

Register as an endpoint​

register(repo_id, task_type, hf_revision="main", runtime_profile_id=None, resource_requirements=None, visibility="account", lifecycle_status="candidate", tags=None, display_name=None, description=None) → HFRegistrationResult

task_type is required. The call creates a model_config + inference_endpoint and returns an HFRegistrationResult with the endpoint_id you can immediately infer against.

reg = client.models.hf.register(
    repo_id="sentence-transformers/all-MiniLM-L6-v2",
    task_type="embeddings",
)

# Use it right away (cold start on the first call)
result = client.endpoints.infer(reg.endpoint_id, {"text": "hello world"})
print(result["result"])

HFRegistrationResult fields: model_config_id (a UUID), endpoint_id, repo_id, status, message, inference_url, test_request.

List your registered models​

list_registry(source=None, lifecycle_status=None, task_type=None, limit=50, offset=0) → List[RegisteredModel]

# All HuggingFace-sourced models
models = client.models.hf.list_registry(source="huggingface")
for m in models:
    print(m.repo_id, m.lifecycle_status)

# Only ready models
ready = client.models.hf.list_registry(lifecycle_status="ready")

source is hive or huggingface. This method paginates by offset.

Update lifecycle status​

update_lifecycle(model_config_id, lifecycle_status, reason=None) → dict

# Promote after testing
client.models.hf.update_lifecycle(
    model_config_id="<uuid>",
    lifecycle_status="ready",
)

# Disable a problematic model (reason is required when disabling)
client.models.hf.update_lifecycle(
    model_config_id="<uuid>",
    lifecycle_status="disabled",
    reason="High error rate in production",
)

lifecycle_status is one of candidate, ready, preferred, disabled. disabled requires a reason.

Datasets API​

client.datasets.upload(name, file, domain="text", description=None)​

Upload a dataset file.

Parameters:

  • name (str): Dataset name
  • file (str): Path to dataset file
  • domain (str): Dataset domain (text, code, vision, audio)
  • description (str, optional): Dataset description

Returns: Dataset object

dataset = client.datasets.upload(
    name="customer_data",
    file="./data.csv",
    domain="text",
    description="Customer support conversations",
)
print(dataset.id)

Accepted dataset formats: jsonl, parquet, hf_dataset, csv.

client.datasets.list(limit=20, cursor=None)​

List datasets (cursor pagination). Returns: list of Dataset.

client.datasets.get(dataset_id)​

Get a dataset. Returns: Dataset.

client.datasets.delete(dataset_id)​

Delete a dataset.

ArtifactRef​

merge and retrain-on-top reference their bases, adapters, and sources by catalog identity — never by raw storage paths. Build one with the ArtifactRef factory methods (it is exported from colabhive):

from colabhive import ArtifactRef

ArtifactRef.hf("Qwen/Qwen2.5-7B-Instruct")            # a public HuggingFace repo (revision="main")
ArtifactRef.model_version("<version_uuid>")           # any trained/merged version you own or that is public
ArtifactRef.job("<job_uuid>")                         # shortcut: uses the job's output version
ArtifactRef.storage("s3://...")                       # advanced escape-hatch

A plain gateway-shaped dict (with a type key) or a TrainingJob (coerced to a job reference) are also accepted anywhere an ArtifactRef is expected:

{"type": "hf",            "repo_id": "Qwen/Qwen2.5-7B-Instruct", "revision": "main"}
{"type": "model_version", "version_id": "<uuid>"}
{"type": "job",           "job_id": "<uuid>"}
{"type": "storage",       "storage_url": "s3://..."}

Visibility follows ColabHive's rules: public versions can be used by anyone; private versions only by their owner. See the Merge & Retrain API for the full contract.

Training API​

client.training.create(model, dataset_id, job_name=None, hyperparameters=None, target_columns=None, operating_mode="balanced", hardware_preference="auto", priority=0, base=None, parent_job=None)​

Create a training job. Pass base to retrain on top of an existing trained or merged model instead of the base defined by the model config.

Parameters:

  • model (str): Model name or UUID (e.g. "xgboost-regression", "bert-classification-gpu")
  • dataset_id (str): Dataset ID to train on
  • job_name (str, optional): Job name (auto-generated if omitted)
  • hyperparameters (dict, optional): Model hyperparameters (uses defaults if omitted)
  • target_columns (list, optional): Target column names for multi-output models
  • operating_mode (str): eco, balanced, or performance (default: balanced)
  • hardware_preference (str): auto, cpu_only, gpu_only, or gpu_preferred (default: auto)
  • priority (int): Job priority 0–100 (default: 0)
  • base (ArtifactRef, optional): retrain on top of a trained/merged version, a job, or storage. Omit (or pass an hf ref) to start from the model config's base.
  • parent_job (str, optional): Explicit lineage parent job id. Derived from base if omitted.

Returns: TrainingJob

job = client.training.create(
    model="xgboost-regression",
    dataset_id="dataset-123",
    job_name="Experiment 1",
    hyperparameters={"n_estimators": 100, "max_depth": 6},
)
print(job.id, job.status)

client.training.train(model, dataset_id, base=None, parent_job=None, job_name=None, hyperparameters=None, target_columns=None, operating_mode="balanced", hardware_preference="auto", priority=0)​

First-class entry point for retrain-on-top — a thin wrapper over create with base/parent_job promoted to leading arguments. Fine-tune on top of any trained or merged model_version (or a job / storage / HF reference) instead of the model config's default base.

Returns: TrainingJob

from colabhive import ArtifactRef

job = client.training.train(
    model="qwen-2.5-7b-qlora",
    dataset_id="dataset-789",
    base=ArtifactRef.model_version("<merged_or_trained_version_uuid>"),
    job_name="domain-expert-v2",
    hyperparameters={"epochs": 3, "learning_rate": 1e-4},
)
job.wait()
# Lineage/output fields are populated by refresh()/.wait():
print(job.produces_version_id, job.lineage)

client.training.merge(name, adapters=None, sources=None, base=None, method=None, weights=None, precision=None, hardware_preference=None)​

Combine N trained models/adapters into a base to produce a new servable model_version. The merge is weight arithmetic (runs on CPU, no GPU) and the result is itself reusable as a base, with lineage recorded. Dispatches POST /training/merges; a merge is a job and counts against your training quota.

Parameters:

  • name (str): Name of the resulting model_version.
  • adapters (list, optional): 1..N ArtifactRefs — for method="adapter_merge".
  • sources (list, optional): 1..N full-model ArtifactRefs — for slerp / ties / dare.
  • base (ArtifactRef, optional): auto-detected from the first adapter/source if omitted.
  • method (str, optional): adapter_merge (default) | slerp | ties | dare. Omit for the server default.
  • weights (list, optional): per-method weights/params.
  • precision (str, optional): omit for the server default (bf16; chains are never re-quantized).
  • hardware_preference (str, optional): omit for the server default (cpu_only).

At least one of adapters or sources is required.

Returns: TrainingJob (job_type=merge, no dataset). produces_version_id and lineage are empty on the returned object and get filled in by refresh() — call .wait() to block until the merged model_version is registered, then read them.

from colabhive import ArtifactRef

# Fold an adapter into its base -> a new servable model_version
job = client.training.merge(
    name="qwen-finance-base",
    base=ArtifactRef.hf("Qwen/Qwen2.5-7B-Instruct"),
    adapters=[ArtifactRef.model_version("<adapter_version_uuid>")],
)
job.wait()
print(job.produces_version_id)   # id of the new merged model_version

# TIES merge of two full models
job = client.training.merge(
    name="ensemble-base",
    method="ties",
    sources=[
        ArtifactRef.model_version("<model_a_version_uuid>"),
        ArtifactRef.model_version("<model_b_version_uuid>"),
    ],
    weights=[0.5, 0.5],
)

client.training.list(limit=20, cursor=None, status=None)​

List training jobs (cursor pagination). Filter by status (pending, running, completed, failed, cancelled). Returns: list of TrainingJob.

client.training.get(run_id)​

Get a training job. Returns: TrainingJob.

client.training.delete(run_id, force=False)​

Delete a training job. force=True deletes even if running.

client.training.model_configs(category=None, framework=None, active_only=True)​

List available model configurations for training. Returns: list of ModelConfig.

configs = client.training.model_configs(category="ml_classical")
for c in configs:
    print(c.model_name, c.display_name)

client.training.logs(run_id, follow=False)​

Get training run logs. Returns: list of log entries.

follow=True is not implemented — the backend returns HTTP 501, and the OpenAPI declares it. Poll without follow instead.

logs = client.training.logs("run-123")
for log in logs:
    print(log.get("timestamp"), log.get("message"))

client.training.events(run_id)​

Get training run events (timeline). Returns: list of event entries.

Model Artifacts (Download & Export)​

Once a run completes you can either register it for inference on ColabHive (see register_for_inference) or download the artifacts to serve them on your own infrastructure (FastAPI, TorchServe, Triton, SageMaker, …).

client.training.artifacts(run_id)​

List trained model artifacts for a completed run. Returns: list of Artifact.

for a in client.training.artifacts("run-123"):
    print(a.filename, f"{a.size_mb} MB")

client.training.download_urls(run_id, files=None, expiration=3600)​

Get presigned download URLs for artifacts (max expiration 86400 s). Returns: list of Artifact with download_url populated.

client.training.download(run_id, output_dir=".", files=None, verbose=True)​

Download artifacts to local disk. Returns: list of local file paths.

paths = client.training.download("run-123", output_dir="./my-model")
print(paths)  # ['./my-model/model.pkl', './my-model/config.json']

register_for_inference​

client.training.register_for_inference(run_id, name, description, visibility="account", display_name=None, input_schema=None, output_schema=None, example_input=None, example_output=None, tags=None, task_type=None, price_per_request=None)

Register a completed training run as an inference endpoint. This is the path for a trained model (distinct from a HuggingFace import): you control visibility (account or public) and, for public endpoints, price_per_request. Public endpoints require a complete spec and review.

Returns: Endpoint

endpoint = client.training.register_for_inference(
    run_id="run-123",
    name="my-model-endpoint",
    description="My trained XGBoost model",
)
print(endpoint.endpoint_id)

Models API​

client.models.list(limit=20, cursor=None, visibility=None)​

List your trained/registered models (cursor pagination). Returns: list of Model.

client.models.get(model_id)​

Get a model. Returns: Model.

client.models.download(model_id, output_path)​

Download a trained model file. Returns: path to the downloaded file.

client.models.versions(model_id, limit=20, cursor=None)​

List model versions (cursor pagination). Each version exposes framework, precision, base_model_version_id, and training_job_id for lineage. Returns: list of version dicts.

client.models.lineage(model_id)​

Get the base→derivatives lineage graph for a model. Provenance holds even after models are renamed, because identity is the version_id.

Returns: a dict with model_id, model_name, root_version_ids, nodes (each with version_id and base_model_version_id; richer metadata is present only for versions you may view), and edges ({"from": <base>, "to": <derived>}).

graph = client.models.lineage("model-123")
for edge in graph["edges"]:
    print(edge["from"], "->", edge["to"])

client.models.rename(model_id, version_id, name)​

Rename a model version. The name is a mutable label — renaming never changes the version's identity (version_id) or its lineage. Only the owner can rename. Returns: the updated version dict.

client.models.rename(
    model_id="model-123",
    version_id="<version_uuid>",
    name="finance-expert-v3",
)

client.models.delete(model_id)​

Delete a model.

Endpoints API​

client.endpoints.list(visibility=None, task_type=None, limit=100)​

List available endpoints (public LLMs, tools, specialists, generative — plus your private endpoints).

Parameters:

  • visibility (str, optional): public, private, or account
  • task_type (str, optional): e.g. chat, code, tool, specialist
  • limit (int): maximum results

Returns: list of Endpoint

llms = client.endpoints.list(visibility="public", task_type="chat")
for llm in llms:
    print(llm.name, llm.display_name)

client.endpoints.get(endpoint_id)​

Get endpoint details. Returns: Endpoint.

client.endpoints.find_by_name(name)​

Find an endpoint by name (e.g. "mistral-7b-instruct-public"). Returns: Endpoint or None.

client.endpoints.infer(endpoint_id, input_data, timeout=None, sync=True, sync_timeout=30.0)​

Run inference on an endpoint. endpoint_id may be a UUID or a name (auto-resolved).

Parameters:

  • endpoint_id (str): Endpoint UUID or name
  • input_data (dict): Input for the model
  • timeout (float, optional): HTTP request timeout override
  • sync (bool): If True (default), the server waits and returns the result directly. If False, it returns a task_id immediately for polling.
  • sync_timeout (float): Max seconds the server waits in sync mode (default: 30)

Returns: a result dict. In sync mode it includes result (the model output) plus status, inference_time_ms, and metrics. If the model can't be produced within sync_timeout (e.g. a cold model), the response comes back with status: "queued" and a task_id to poll with get_task.

Example — sync (default):

# Embeddings — result returned directly
result = client.endpoints.infer("embeddings-public", {"text": "Hello world"})
print(result["result"])

# Chat with an LLM
result = client.endpoints.infer(
    "mistral-7b-instruct-public",
    {"messages": [{"role": "user", "content": "Hello!"}], "max_tokens": 100},
)
print(result["result"])

Example — async (sync=False) for long-running or cold models:

result = client.endpoints.infer(
    "deepseek-distilled-7b-public",
    {"messages": [{"role": "user", "content": "Write a poem"}]},
    sync=False,
)
task_id = result["task_id"]
# Poll with get_task()...

client.endpoints.create(...)​

Not available yet. The Builder Gateway does not expose POST /endpoints; calling this raises NotImplementedError. To create an endpoint from a trained model use register_for_inference, or provision endpoints in the Console.

client.endpoints.get_task(task_id)​

Get an inference task's status and result (for polling async inference).

result = client.endpoints.infer(endpoint_id, input_data, sync=False)
task_id = result["task_id"]

import time
while True:
    task = client.endpoints.get_task(task_id)
    if task.get("status") == "completed":
        print("Result:", task.get("result"))
        break
    if task.get("status") in ("failed", "cancelled"):
        print("Error:", task.get("error"))
        break
    time.sleep(1)

client.endpoints.upload_input(endpoint_id, file_path)​

Upload a binary input file (image / audio / video) and get a presigned URL to pass as image_url / audio_url in the infer input_data. Max upload size is 50 MB. Returns: the URL string.

image_url = client.endpoints.upload_input(endpoint_id, "./chair.png")
result = client.endpoints.infer(endpoint_id, {"image_url": image_url, "format": "glb"})

client.endpoints.download_artifact(url, output_path)​

Download a binary artifact from a presigned URL (from an inference result's output_artifacts). Returns: the output_path.

client.endpoints.delete(endpoint_id)​

Delete an endpoint.

Users API​

Access your profile, account info, and manage API keys programmatically.

client.users.me()​

Get the current user profile. Returns: dict with user_id, email, name, account_id, account_name, account_role.

client.users.credentials()​

Get credentials info for SDK setup. Returns: dict with account_id and setup instructions.

client.users.list_api_keys()​

List API keys for the account (without the key values). Returns: list of dicts.

client.users.create_api_key(name, scopes=None, environment="live", expires_in_days=None)​

Create a new API key. The key value is returned only once — save it immediately.

Returns: dict including api_key (the value), id, name, scopes, expires_at.

key = client.users.create_api_key(name="Production Server", expires_in_days=365)
print("Save this key:", key["api_key"])

scopes are stored but not yet enforced granularly (a key currently grants full account access).

client.users.revoke_api_key(key_id)​

Revoke an API key. Immediate and irreversible.

Data Types​

Dataset​

class Dataset:
    dataset_id: str
    dataset_name: str
    domain: Optional[str]
    num_samples: Optional[int]
    total_size_bytes: Optional[int]
    total_size_gb: Optional[float]
    storage_format: Optional[str]       # jsonl | parquet | hf_dataset | csv
    checksum_sha256: Optional[str]
    visibility: Optional[str]
    status: Optional[str]
    created_at: Optional[str]

    @property
    def id(self) -> str: ...             # alias for dataset_id
    @property
    def name(self) -> str: ...           # alias for dataset_name

TrainingJob​

class TrainingJob:
    job_id: Optional[str]                # SDK also accepts legacy run_id
    job_name: Optional[str]
    status: Optional[str]
    progress_percent: Optional[float]
    model_config_id: Optional[str]
    dataset_id: Optional[str]
    hyperparameters: Optional[dict]
    metrics_summary: Optional[dict]
    error_message: Optional[str]
    job_type: Optional[str]              # "training" | "merge"
    # Lineage / output (populated by refresh()/.wait()):
    base: Optional[Any]
    parent_job_id: Optional[str]
    produces_version_id: Optional[str]   # the resulting model_version, once the job completes
    lineage: Optional[dict]

    @property
    def id(self) -> str: ...             # job_id or run_id
    def get_metrics(self) -> dict: ...   # fetch metrics from the server
    def refresh(self): ...               # refresh status + lineage/output fields
    def wait(self, poll_interval=5, timeout=3600, verbose=True): ...
job = client.training.create(...)
job.wait(verbose=True)          # blocks, prints progress
if job.status == "completed":
    print(job.get_metrics())    # or job.metrics_summary

Artifact​

class Artifact:
    filename: str
    size_bytes: int
    download_url: Optional[str]          # populated by download_urls()
    last_modified: Optional[str]
    expires_in_seconds: Optional[int]

    @property
    def size_mb(self) -> float: ...

Model​

class Model:
    model_id: str
    model_name: str
    model_type: Optional[str]
    model_category: Optional[str]
    capabilities: Optional[list | dict]
    framework: Optional[str]
    visibility: Optional[str]
    storage_url: Optional[str]
    created_at: Optional[str]

    @property
    def id(self) -> str: ...             # alias for model_id

Endpoint​

class Endpoint:
    endpoint_id: str
    name: str
    display_name: Optional[str]
    visibility: Optional[str]
    status: Optional[str]
    task_type: Optional[str]
    is_base_model: Optional[bool]
    model_config_id: Optional[str]
    price_per_request: Optional[float]
    price_currency: Optional[str]        # "USD"

    @property
    def id(self) -> str: ...             # alias for endpoint_id

ArtifactRef​

from colabhive import ArtifactRef

ArtifactRef.hf(repo_id, revision="main")     # type="hf"
ArtifactRef.model_version(version_id)        # type="model_version"
ArtifactRef.job(job_id)                      # type="job"
ArtifactRef.storage(storage_url)             # type="storage"

See ArtifactRef above for usage in merge / train.

Error Handling​

from colabhive import (
    ColabHive,
    ColabHiveError,
    AuthenticationError,
    NotFoundError,
    ValidationError,
    RateLimitError,
    APIError,
)

client = ColabHive(api_key="hive_...", account_id="...")

try:
    dataset = client.datasets.upload("data", "./file.csv")
except AuthenticationError as e:
    print("Auth failed:", e)
except NotFoundError as e:
    print("Not found:", e)
except ValidationError as e:
    print("Invalid request:", e)
except RateLimitError as e:
    print("Rate limit exceeded:", e)
except APIError as e:
    print("API error:", e)
except ColabHiveError as e:
    print("Error:", e)

Advanced Usage​

Context Manager​

with ColabHive(api_key="hive_...", account_id="...") as client:
    dataset = client.datasets.upload("data", "./train.csv")
    job = client.training.create("xgboost-regression", dataset.id)
    job.wait()
# Client automatically closed

Custom Base URL​

# Production (default)
client = ColabHive(api_key="hive_...", account_id="...",
                   base_url="https://api.colabhive.com")

# Local development
client = ColabHive(api_key="hive_...", account_id="...",
                   base_url="http://localhost:8014")

Polling Configuration​

job = client.training.create(...)
job.wait(poll_interval=10, timeout=7200, verbose=True)

Requirements​

  • Python 3.8+
  • httpx >= 0.24.0
  • pydantic >= 2.0.0

Resources​