colabone-python 1.1.0

Python SDK for the ColabOne API — unified access to multiple AI models

colabone-python

A Python SDK for the ColabOne AI Gateway platform.

Features

• 🚀 Async First - Built on asyncio and httpx for fully async, non-blocking requests
• 🔄 Streaming Support - First-class support for streaming responses via async for
• 🏗️ Resource-Based Architecture - Clean, intuitive API design inspired by the OpenAI SDK
• 🛡️ Type Hints - Comprehensive type hints and dataclasses for editor autocomplete and static type checking (mypy/pyright)
• ⚡ Automatic Retries - Built-in retry logic with exponential backoff and jitter
• 🔐 Error Handling - Comprehensive error hierarchy with specialized error types
• 🔀 Model Fallback - Optional sequential fallback across multiple models on a single call

Installation

pip install colabone-python

Quick Start

import asyncio
import os
from colabone import ColabOne

async def main():
    client = ColabOne(api_key=os.environ["COLABONE_API_KEY"])

    # Make a chat completion request (accepts a single model string
    # or a list of model strings for parallel routing)
    response = await client.chat.completions.create(
        model="gpt-4o",  # or ["gpt-4o", "claude-3-5-sonnet"] for parallel routing
        messages=[{"role": "user", "content": "Hello!"}],
    )

    print(response["choices"][0]["message"]["content"])
    await client.close()

asyncio.run(main())

Usage Examples

Chat Completions

# Regular completion with a single model string
completion = await client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {"role": "system", "content": "You are a helpful assistant."},
        {"role": "user", "content": "What is Python?"},
    ],
    temperature=0.7,
    max_tokens=1000,
)

# Multi-model parallel routing completion (executes models in parallel)
multi_completion = await client.chat.completions.create(
    model=["gpt-4o", "claude-3-5-sonnet", "gemini-1.5-pro"],
    messages=[{"role": "user", "content": "Compare SQL and NoSQL databases."}],
)

# Streaming completion (single model string only)
stream = await client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Write a poem about Python"}],
    stream=True,
    # Optional. Defaults to ["text"] when omitted.
    modalities=["text"],
)

async for chunk in stream:
    content = chunk["choices"][0]["delta"].get("content")
    if content:
        print(content, end="", flush=True)

Model Fallback

Sequentially try a list of models, falling back to the next one if a call fails (rate limited, provider error, etc.). Works for both regular and streaming requests:

response = await client.chat.completions.create_with_fallback(
    models=["gpt-4o", "claude-3-5-sonnet"],
    messages=[{"role": "user", "content": "Hello!"}],
    stream=True,
)

async for chunk in response:
    content = chunk["choices"][0]["delta"].get("content")
    if content:
        print(content, end="", flush=True)

Models

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

# Get model details
model = await client.models.retrieve("gpt-4")

# Get pricing information
pricing = await client.models.get_pricing("gpt-4")

Usage & Credits

# Get current credit balance
balance = await client.credits.balance()
print(f"Available credits: {balance['available']}")

# Get usage statistics
usage = await client.usage.get_current()
print(f"Total tokens used: {usage['totalTokens']}")

# Get usage by model
model_usage = await client.usage.by_model("2024-01-01", "2024-01-31")

Configuration

client = ColabOne(
    api_key=os.environ["COLABONE_API_KEY"],
    base_url=None,       # Optional: override the default API base URL
    timeout=30.0,        # Optional: timeout in seconds
    max_retries=3,       # Optional: number of retries for failed requests
    user_agent=None,     # Optional: override the default User-Agent header
)

Error Handling

The SDK provides specialized error types for different scenarios:

from colabone import (
    ApiError,
    AuthenticationError,
    RateLimitError,
    InsufficientCreditsError,
    ProviderError,
    ValidationError,
)

try:
    await client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": "Hello"}],
    )
except AuthenticationError:
    print("Invalid API key")
except RateLimitError as e:
    print("Rate limited. Retry after:", e.retry_after, "seconds")
except InsufficientCreditsError as e:
    print("Insufficient credits:", {"required": e.required, "available": e.available})
except ProviderError as e:
    print("Provider error:", e.provider)
except ValidationError as e:
    print("Validation errors:", e.validation_errors)
except ApiError as e:
    print("API error:", e)

Streaming

The SDK has first-class support for streaming responses:

[!NOTE] Streaming with multiple models is not available yet. For now, streaming is supported for single-model requests only — passing a list of models with stream=True raises a ValidationError before any request is made.

# Stream chat completions (single model name string)
stream = await client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Tell me a story"}],
    stream=True,
)

# Iterate over stream chunks
async for chunk in stream:
    content = chunk["choices"][0]["delta"].get("content")
    if content:
        print(content, end="", flush=True)

# Or collect all chunks
all_chunks = await stream.collect()

Environment Support

The SDK requires Python 3.9+ and works anywhere asyncio and httpx run:

• Standard Python scripts and services
• FastAPI / Django (async views) / Starlette
• Jupyter notebooks
• AWS Lambda (async handlers)

Architecture

Dependency Flow

The SDK follows a strict dependency hierarchy to prevent circular dependencies:

Resources
    ↓
Core (Types, Errors, Config)
    ↓
HTTP (HttpClient, RequestBuilder, ResponseHandler)
    ↓
httpx

Resources

• Chat - Chat completions and related operations
• Models - Model information and availability
• Usage - Usage tracking and analytics
• Credits - Credit balance and management

Advanced Usage

Custom HTTP Configuration

# Use a custom timeout per request
completion = await client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Hello"}],
    timeout=60.0,  # 60 seconds
)

# Cancel a request with asyncio.wait_for
try:
    completion = await asyncio.wait_for(
        client.chat.completions.create(
            model="gpt-4o",
            messages=[{"role": "user", "content": "Write a long story"}],
        ),
        timeout=5.0,
    )
except asyncio.TimeoutError:
    print("Request was cancelled (timed out after 5 seconds)")

Retry Configuration

The SDK automatically retries on:

• Rate limit errors (429)
• Retryable provider errors (503)

Retries use exponential backoff with jitter to prevent thundering herd. Streaming requests are not automatically retried.

Type Hints

The SDK ships with type hints and dataclasses for the underlying API shapes (colabone.core.types, colabone.resources.*.types) to support editor autocomplete and static type checkers. Response payloads from chat.completions.create, models, usage, and credits are returned as plain dicts matching the ColabOne API's JSON responses.

License

MIT