Cluster Protocol Python SDK

Python SDK for the Cluster Protocol Decentralized AI Inference Gateway. OpenAI-compatible API with typed responses, automatic retries, and async support.

Installation

pip install cluster-sdk

Quick Start

from cluster_sdk import ClusterClient

client = ClusterClient(api_key="cp_your_key_here")

response = client.chat.completions.create(
    model="deepseek-ai/DeepSeek-V3",
    messages=[{"role": "user", "content": "Hello!"}],
)
print(response.choices[0].message.content)

Streaming

stream = client.chat.completions.create(
    model="deepseek-ai/DeepSeek-V3",
    messages=[{"role": "user", "content": "Write a poem"}],
    stream=True,
)
for chunk in stream:
    print(chunk.choices[0].delta.content or "", end="")

Embeddings

result = client.embeddings.create(
    input="The quick brown fox jumps over the lazy dog",
    model="BAAI/bge-large-en-v1.5",
)
print(f"Dimensions: {len(result.data[0].embedding)}")

# Batch embedding
result = client.embeddings.create(
    input=["First sentence", "Second sentence"],
    model="BAAI/bge-large-en-v1.5",
)
for emb in result.data:
    print(f"[{emb.index}] {len(emb.embedding)}d vector")

Dataset Marketplace

# Browse datasets
listing = client.datasets.list(search="sentiment", category="nlp")
for ds in listing.datasets:
    print(f"{ds.name} — {ds.pricingType} — {ds.downloadCount} downloads")

# Get dataset details
dataset = client.datasets.get("imdb-reviews-v2")
print(f"{dataset.name}: {dataset.description}")

# Purchase a dataset
receipt = client.datasets.purchase(dataset.id)
print(f"Purchase status: {receipt.status}")

# Check access
access = client.datasets.check_access(dataset.id)
print(f"Has access: {access.hasAccess}")

Async Client

import asyncio
from cluster_sdk import AsyncClusterClient

async def main():
    async with AsyncClusterClient(api_key="cp_your_key_here") as client:
        # Chat
        response = await client.chat.completions.create(
            model="deepseek-ai/DeepSeek-V3",
            messages=[{"role": "user", "content": "Hello!"}],
        )
        print(response.choices[0].message.content)

        # Embeddings
        emb = await client.embeddings.create(
            input="Hello world",
            model="BAAI/bge-large-en-v1.5",
        )
        print(f"Embedding dim: {len(emb.data[0].embedding)}")

        # Datasets
        listing = await client.datasets.list(search="finance")
        print(f"Found {listing.pagination.total} datasets")

asyncio.run(main())

Models & Balance

# List models
for model in client.models.list():
    print(f"{model.id}: {model.owned_by}")

# Check balance
balance = client.balance.get()
print(f"${balance.usd:.2f}")

# Usage history
usage = client.usage.get()
print(f"Total cost: ${usage.total_cost:.4f}")

Error Handling

from cluster_sdk import ClusterClient
from cluster_sdk.exceptions import (
    AuthenticationError,
    InsufficientBalanceError,
    RateLimitError,
    ModelNotFoundError,
)

try:
    response = client.chat.completions.create(
        model="deepseek-ai/DeepSeek-V3",
        messages=[{"role": "user", "content": "Hello"}],
    )
except AuthenticationError:
    print("Invalid API key — check your cp_ key")
except InsufficientBalanceError:
    print("Top up your balance at clusterprotocol.ai")
except RateLimitError as e:
    print(f"Rate limited: {e}")  # Auto-retried with Retry-After header
except ModelNotFoundError:
    print("Model not found — check client.models.list()")

Rate-limited (429) and server error (5xx) requests are automatically retried with exponential backoff. The SDK respects the Retry-After header when present.

Configuration

License

Apache-2.0