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==0.11.0

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

The package also installs the colabhive command, which configures a coding agent against ColabHive: see the Coding Agents quickstart.

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 grants access to everything the account can do. Granular per-scope enforcement is not part of the public API contract.

Pagination

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

Method	Style	Parameters
datasets.list	cursor	limit, cursor
training.list	cursor	limit, cursor, status
models.list	cursor	limit, cursor, visibility
models.versions	cursor	limit, cursor
models.hf.list_registry	offset	limit, 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.accounts, client.data_protection, client.datasets, client.training, client.models (with client.models.hf), client.endpoints, client.clm, client.users.

Example:

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

---

Account preferences

Share Hive spare-capacity lending is disabled by default at the account level. Read the effective setting with client.accounts.get_preferences(). An account owner can change it explicitly:

preferences = client.accounts.get_preferences()
print(preferences["share_hive_opt_in"])

client.accounts.set_share_hive_opt_in(True)   # opt in
client.accounts.set_share_hive_opt_in(False)  # immediate rollback for new settlements

The account comes from the authenticated API key. The SDK does not accept a lender account ID for these calls, so it cannot be used to change another account's preference.

---

Memory

Memory is enabled per account by the platform, and the platform ties it to every task you submit. The caller names nothing — no namespace, no snapshot — so the only thing the SDK exposes is a read-only status:

from colabhive import ColabHive, MemoryStatusDetail

client = ColabHive(api_key=os.environ["COLABHIVE_API_KEY"])
state = client.memory.status()

if state.ready:
    ...                      # the next task you submit will carry account memory
elif state.enabled and state.detail == MemoryStatusDetail.SNAPSHOT_PENDING:
    ...                      # on, but there is no snapshot to read from yet

status() takes no arguments: the account is the one the API key belongs to, and under one namespace per account there is nothing to choose.

field	meaning
enabled	the platform enabled memory for this account
ready	a task submitted now would carry it
detail	why ready is false while enabled is true, and None otherwise

detail is a closed vocabulary, never free text — snapshot_pending (on, but nothing to read yet), provisioning (the namespace keys are still being set up) or unavailable (the account's memory cannot be resolved and support has to look at it). A value a newer gateway introduces and this wheel does not know yet stays the raw string instead of raising, so an older client keeps working; MemoryStatusDetail is a str enum, so both compare equal to the wire value.

An account without memory is an answer, not an error: it returns enabled=False. The one error to handle separately is FeatureUnavailable, which means this deployment does not serve the route at all — a self-hosted install with memory off, not an account without it.

from colabhive import FeatureUnavailable

try:
    state = client.memory.status()
except FeatureUnavailable:
    state = None             # this deployment has no memory surface

There is no SDK call to turn memory on, to name a namespace, or to read what memory contains. The platform owns the binding. See Memory for the model and MCP get_memory_status for the same answer from an MCP client.

---

Reasoning traces

A thinking model returns its reasoning in its own field, separate from the answer. Read reasoning, and fall back to the legacy alias:

result = client.endpoints.infer(
    "hf-Qwen-Qwen3-8B",
    {
        "messages": [{"role": "user", "content": "Cuánto es 7+5?"}],
        # The trace spends the same budget as the answer: too small a limit and
        # the model can spend it all reasoning and return an empty `content`.
        "max_tokens": 2048,
        "include_reasoning": True,
    },
)

message   = result["result"]["choices"][0]["message"]
answer    = message["content"]
reasoning = message.get("reasoning") or message.get("reasoning_content") or ""

reasoning is canonical. reasoning_content is the older name and is kept populated, but a client that reads only the old name can get an empty string in silence from engines that have renamed it — which is why the fallback goes in that order and not the other way.

In a stream both arrive as their own delta keys and must go to separate buffers:

answer, reasoning = [], []
for chunk in stream:                      # any SSE source for /v1/chat/completions
    delta = chunk["choices"][0].get("delta") or {}
    if delta.get("content"):
        answer.append(delta["content"])
    trace = delta.get("reasoning") or delta.get("reasoning_content")
    if trace:
        reasoning.append(trace)

Streaming is not exposed by client.endpoints.infer, which returns the complete result. For token-by-token output use the OpenAI-compatible endpoint with stream: true and any OpenAI client; the delta keys are the ones above.

If you asked for tools and tool_calls comes back empty while content reads like a train of thought, that is the reasoning not being separated, not the model refusing to use its tools. See Reasoning.

---

Data protection and placement

client.data_protection calls the control plane for every effective-policy and preflight decision; it does not reproduce the policy hierarchy locally.

from colabhive import NodeEligibility, WorkloadType

client.data_protection.set_account_policy(NodeEligibility.OWN_ACCOUNT_ONLY)
client.data_protection.set_endpoint_policy(
    endpoint_id,
    NodeEligibility.OWN_HARDWARE_ONLY,
)

effective = client.data_protection.effective(
    workload_type=WorkloadType.INFERENCE_ENDPOINT,
    workload_id=endpoint_id,
)
print(effective["binding_level"], effective["can_relax_to"])

preflight = client.data_protection.placement_preflight(
    workload_type=WorkloadType.INFERENCE_ENDPOINT,
    workload_id=endpoint_id,
    node_eligibility=NodeEligibility.OWN_HARDWARE_ONLY,
    explain=True,
)

The three policy writers are set_account_policy, set_endpoint_policy, and set_training_policy. Passing None to a workload writer restores inheritance. Use placement_attestation(window_from=..., window_to=...) to download past placement evidence.

For one inference task, pass node_eligibility=NodeEligibility.OWN_HARDWARE_ONLY to client.endpoints.infer. For a training task, pass it to client.training.create or client.training.train. The typed DataProtectionErrorCode enum contains the stable error and exclusion vocabulary. The SDK never retries a policy rejection with a more permissive value.

See Data Protection & Placement for semantics and boundaries.

---

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 direct-upload formats: jsonl, parquet, csv.

client.datasets.list(limit=20, offset=0)

List datasets (offset 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("state") == "succeeded":
        print("Result:", task.get("result"))
        break
    if task.get("state") == "failed":
        # failure_origin: "platform" (the model never ran) | "model" | "unattributed"
        print("Error:", task.get("failure_origin"), task.get("reason_code"), task.get("error"))
        break
    time.sleep(1)

state is the stable lifecycle (queued, running, succeeded, failed). While queued, task["reason_code"] says why the task is waiting. All fields: Get Task Result.

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.

---

CLM Scorer API

client.clm calls the CLM-8B scorer, which answers typed questions about a state and ranks candidates; it generates no text. The question types keep the names and wire format of the upstream CLM client, and plain dicts in that format are accepted too.

client.clm.system_one(model, state, questions, temperature=None, timeout=None)

• model (str): the CLM endpoint — its id or its label (endpoint_name or display_name), e.g. "clm-v0.1-8b" or the endpoint of a head you trained. Never a Hugging Face repo id.
• state: a string, an object or an array.
• questions: {id: Noul | Choice | Score | dict}.
• temperature (float, optional): above 1 flattens, below 1 sharpens; the server default is 1.
• timeout (float, optional): seconds to wait; default 300, the API edge limit (not the client's 30 s).

Returns: SystemOneResponse with answers ({id: NoulAnswer | ChoiceAnswer | ScoreAnswer}), usage, latency_ms (from X-CLM-Latency-Ms) and colabhive (endpoint_id, task_id, head_sha256, encoder_identity, runtime_vendor).

from colabhive.clm import Choice, Noul, Score

r = client.clm.system_one(
    model="clm-v0.1-8b",
    state="Customer: my invoice was charged twice and nobody answers the phone!",
    questions={
        "urgency": Noul(instructions="Is this urgent?"),
        "department": Choice(instructions="Which team should handle this?",
                             criteria={"billing": "Charges, invoices, refunds",
                                       "technical": "Bugs and outages"}),
        "frustration": Score(instructions="How frustrated is the customer?",
                             criteria=["Calm", "Frustrated", "Very angry"]),
    },
)
r.answers["urgency"].noul            # P(true), 0.0-1.0
r.answers["department"].choice       # the most probable key
r.answers["frustration"].score       # expected level, 0.0-2.0 here

client.clm.rank(model, context, question, answers, temperature=None, timeout=None)

Ranks answers (a sequence of non-empty strings; a single string is refused) for context and question. Returns: RankResponse with ranked ([RankedCandidate(rank, candidate, prob)], best first), latency_ms and colabhive.

ranked = client.clm.rank(model="clm-v0.1-8b", context="Tides are caused mainly by",
                         question="Which answer completes the sentence?",
                         answers=["the gravity of the Moon", "the wind over the ocean"])
print(ranked.ranked[0].candidate, ranked.ranked[0].prob)

Errors raise the usual exceptions: a :trained: runtime key or a Hugging Face repo id as model is a 400, too many candidates, questions or tokens a 413 (limits on the model page). Train a head of your own with client.training.create(model="clm-head-infonce", ...) — see Fine-tune a CLM head.

---

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

• PyPI Package: https://pypi.org/project/colabhive/
• Documentation: https://docs.colabhive.com
• Support: support@colabhive.com