Metadata-Version: 2.4
Name: colabone-python
Version: 1.1.0
Summary: Python SDK for the ColabOne API — unified access to multiple AI models
Author-email: vishwaaareddy-commits <vishwaaareddy@gmail.com>
License: MIT
Project-URL: Homepage, https://github.com/colabintelligence-source/colabone-python
Keywords: colabone,ai,llm,sdk,openai,claude,anthropic
Classifier: Programming Language :: Python :: 3
Classifier: License :: OSI Approved :: MIT License
Classifier: Operating System :: OS Independent
Classifier: Intended Audience :: Developers
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Requires-Python: >=3.9
Description-Content-Type: text/markdown
Requires-Dist: httpx>=0.27.0

# colabone-python

A Python SDK for the ColabOne AI Gateway platform.

## Features

- 🚀 **Async First** - Built on `asyncio` and [`httpx`](https://www.python-httpx.org/) 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

```bash
pip install colabone-python
```

## Quick Start

```python
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

```python
# 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:

```python
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

```python
# 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

```python
# 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

```python
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:

```python
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.

```python
# 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

```python
# 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 `dict`s matching the
ColabOne API's JSON responses.

## License

MIT
