Python SDK
Build Python applications with the Archipelag.io SDK
Python SDK
The official Python SDK for Archipelag.io. Supports both synchronous and asynchronous usage.
Installation
pip install archipelag
Quick Start
from archipelag import Client
client = Client(api_key="ak_xxx")
# Simple chat
result = client.chat("What is the capital of France?")
print(result.content)
# Streaming chat
for event in client.chat_stream("Tell me a story"):
if event.type == "token":
print(event.content, end="", flush=True)
# Image generation
image = client.generate_image("a sunset over mountains")
Configuration
client = Client(
api_key="ak_xxx", # Required
base_url="https://api.archipelag.io", # Optional
timeout=60.0, # Optional, seconds
)
Chat API
Simple Chat
result = client.chat(
prompt="Hello!",
system_prompt="You are a helpful assistant",
max_tokens=500,
temperature=0.7,
)
print(result.content) # Response text
print(result.usage.total_tokens) # Token count
Streaming Chat
for event in client.chat_stream(
prompt="Explain quantum physics",
system_prompt="You are a physics teacher",
max_tokens=500,
):
if event.type == "token":
print(event.content, end="", flush=True)
elif event.type == "done":
print(f"\nTotal tokens: {event.usage.total_tokens}")
Image Generation
image = client.generate_image(
prompt="a cat wearing sunglasses",
negative_prompt="blurry, low quality",
width=1024,
height=1024,
steps=30,
guidance_scale=7.5,
seed=42, # For reproducibility
)
# Save the image
import base64
with open("output.png", "wb") as f:
f.write(base64.b64decode(image.image_data))
print(f"Generated {image.width}x{image.height} image")
Job Management
For more control, use the low-level job API:
# Create a job
job = client.create_job(
workload="llm-chat",
input={"prompt": "Hello", "max_tokens": 100}
)
# Check status
job = client.get_job(job.id)
print(job.status) # JobStatus enum
# Wait for completion
completed = client.wait_for_job(
job.id,
poll_interval=1.0, # Check every second
timeout=120.0, # Give up after 2 minutes
)
# Stream job output
for event in client.stream_job(job.id):
if event.type == "token":
print(event.content, end="")
# Cancel a job
client.cancel_job(job.id)
Batch Operations
Process multiple jobs efficiently:
jobs = client.batch([
{"workload": "sdxl", "input": {"prompt": "a cat"}},
{"workload": "sdxl", "input": {"prompt": "a dog"}},
{"workload": "sdxl", "input": {"prompt": "a bird"}},
])
results = client.wait_all(jobs, timeout=300)
for job in results:
print(f"{job.id}: {job.status}")
Async Usage
All methods have async equivalents:
import asyncio
from archipelag import AsyncClient
async def main():
async with AsyncClient(api_key="ak_xxx") as client:
# Async chat
result = await client.chat("Hello!")
print(result.content)
# Async streaming
async for event in client.chat_stream("Tell a story"):
if event.type == "token":
print(event.content, end="", flush=True)
# Async image generation
image = await client.generate_image("a sunset")
asyncio.run(main())
Context Managers
Both clients support context managers:
# Synchronous
with Client(api_key="ak_xxx") as client:
result = client.chat("Hello!")
# client.close() called automatically
# Asynchronous
async with AsyncClient(api_key="ak_xxx") as client:
result = await client.chat("Hello!")
# client.aclose() called automatically
Error Handling
from archipelag import (
ArchipelagError,
AuthenticationError,
RateLimitError,
InsufficientCreditsError,
JobFailedError,
)
try:
result = client.chat("Hello!")
except AuthenticationError:
print("Invalid API key")
except InsufficientCreditsError as e:
print(f"Need {e.required} credits, have {e.available}")
except RateLimitError as e:
print(f"Rate limited. Retry after {e.retry_after}s")
import time
time.sleep(e.retry_after)
except JobFailedError as e:
print(f"Job {e.job_id} failed: {e.message}")
except ArchipelagError as e:
print(f"API error: {e.message}")
Account Management
# Get account info
account = client.get_account()
print(f"Email: {account.email}")
print(f"Credits: {account.credits}")
# List API keys
keys = client.list_api_keys()
for key in keys:
print(f"{key.name}: {key.prefix}...")
# Create new API key
key_info, full_key = client.create_api_key(name="dev-key")
print(f"New key: {full_key}") # Only shown once!
# Delete API key
client.delete_api_key(key_info.id)
Cargo Information
# List available Cargos
workloads = client.list_workloads()
for w in workloads:
print(f"{w.slug}: {w.name}")
print(f" Price: {w.price_per_job} credits/job")
# Get Cargo details
llm = client.get_workload("llm-chat")
print(llm.description)
Type Hints
The SDK is fully typed with Pydantic models:
from archipelag.models import Job, JobStatus, ChatResult, ImageResult, Usage
# All models support:
job.model_dump() # Convert to dict
Job.model_validate(d) # Create from dict