ColabHive Python SDK

Official Python client for ColabHive Builder APIs.

Train machine learning models on distributed GPUs without managing infrastructure.

Installation

pip install colabhive

Quick Start

from colabhive import ColabHive

# Initialize client
client = ColabHive(
    api_key="your_api_key_here",
    account_id="your_account_id_here"
)

# Upload dataset
dataset = client.datasets.upload(
    name="my_training_data",
    file="./data.csv"
)
print(f"Dataset uploaded: {dataset.id}")

# Train model
job = client.training.create(
    model="xgboost-regression",
    dataset_id=dataset.id,
    job_name="My First Model"
)
print(f"Training started: {job.id}")

# Wait for completion
job.wait()

if job.status == "completed":
    print("Training complete!")
    print(f"Metrics: {job.metrics}")
else:
    print(f"Training failed: {job.error_message}")

Authentication

Get your API key and account ID from console.colabhive.com.

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

Features

Datasets

# Upload
dataset = client.datasets.upload(name="data", file="./train.csv")

# List
datasets = client.datasets.list(limit=10)

# Get
dataset = client.datasets.get("dataset-id")

# Delete
client.datasets.delete("dataset-id")

Training

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

# List jobs
jobs = client.training.list(limit=10, status="running")

# Get job
job = client.training.get("run-id")

# Wait for completion
job.wait(poll_interval=5, timeout=3600, verbose=True)

# Get metrics
metrics = job.metrics
print(metrics)

# Delete job
client.training.delete("run-id")

Models

# List models
models = client.models.list()

# Get model
model = client.models.get("model-id")

# Download model
path = client.models.download("model-id", "./my_model.pkl")

# Delete model
client.models.delete("model-id")

Model Configurations

# List available model configs
configs = client.training.model_configs(category="ml_classical")

for config in configs:
    print(config.model_name, config.display_name)
    print(config.default_hyperparameters)

Cohort Latent Fabric (0.7.0 production artifact)

Cohort execution is additive, production-deployed and hidden unless the authenticated account and causal/autoregressive LLM endpoint both pass capability checks. Classic inference is unchanged when execution is omitted. Always preflight the specific endpoint before opting in:

from threading import Event

endpoint_id = "00000000-0000-4000-8000-000000000001"
client.cohorts.require_available(endpoint_id)

accepted = client.cohorts.run(
    endpoint_id=endpoint_id,
    input_data={"text": "bounded input"},
    execution={
        "mode": "cohort",
        "fallback": "single",
        "latency_budget_ms": 30_000,
        "cost_budget_credits": 1.0,
    },
    idempotency_key="unique-request-key",
)

stop = Event()
for event in client.cohorts.iter_events(
    accepted.task_id,
    follow=True,
    timeout=300,
    cancel_event=stop,
):
    print(event.type)

summary = client.cohorts.wait(accepted.task_id, timeout=300)

The follow iterator resumes by cursor, deduplicates reconnect overlap, retries transient 429/5xx responses and remains bounded by timeout or cancellation. Public models expose only redacted lifecycle, aggregate latency/cost and the terminal answer; intermediate tensors are never SDK responses.

Advanced Usage

Context Manager

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

Custom Base URL

# For production
client = ColabHive(
    api_key="...",
    account_id="...",
    base_url="https://api.colabhive.com"
)

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

Error Handling

from colabhive import (
    APIError,
    ColabHive,
    ConflictError,
    NotFoundError,
    RateLimitError,
    ValidationError,
)

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

try:
    dataset = client.datasets.upload("data", "./nonexistent.csv")
except ValidationError as e:
    print(f"Invalid request: {e.message}")
except NotFoundError as e:
    print(f"Not found: {e.message}")
except APIError as e:
    print(f"API error: {e.message} (status: {e.status_code})")

ConflictError, ValidationError, RateLimitError and APIError represent 409, 400/422, 429 and 5xx responses respectively.

Requirements

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

Documentation

• Full Documentation
• API Reference
• Examples

Support

• Discord: discord.gg/colabhive
• Email: support@colabhive.com
• Issues: GitHub Issues

License

MIT License - see LICENSE file for details.