Python SDK - Runcrate

The official Python SDK for Runcrate. Supports both synchronous and asynchronous usage with full type safety via Pydantic.

Installation

pip install runcrate-sdk

Requires Python 3.9+.

Quick Start

from runcrate import Runcrate

client = Runcrate(api_key="rc_live_YOUR_API_KEY")

# Chat completion
response = client.models.chat_completion(
    model="deepseek-ai/DeepSeek-V3",
    messages=[{"role": "user", "content": "Hello!"}],
)
print(response.choices[0].message.content)

# List GPU instances
instances = client.instances.list()
for inst in instances:
    print(f"{inst.name}: {inst.status}")

client.close()

Or use as a context manager:

with Runcrate(api_key="rc_live_YOUR_API_KEY") as client:
    balance = client.billing.get_balance()
    print(f"Credits: ${balance.credits_balance}")

Configuration

client = Runcrate(
    api_key="rc_live_...",                    # or RUNCRATE_API_KEY env var
    base_url="https://www.runcrate.ai",           # infrastructure API
    inference_url="https://api.runcrate.ai",  # model inference API
    timeout=30,                               # seconds
    max_retries=3,                            # retries; see Retries and launches below
    environment="production",                 # optional: target a specific environment
)

Retries and launches

Reads, updates and deletes are retried on 429, 5xx, timeouts and lost connections, honouring Retry-After up to 60 seconds. A request that creates or launches something is never sent twice by accident:

Job submits and reruns send an Idempotency-Key (a new one per call, reused on its retries), which the server honours, so they are retried like reads: a retry after a timeout answers the first job.
Instance, schedule and volume creates also send an Idempotency-Key, but those routes do not deduplicate it yet; schedules.run_now sends none. Once one of these, or a run-now, may have reached Runcrate (a timeout, a lost connection, a 500, 502 or 504), it is not sent again and the error is raised: check client.instances.list() before launching again. Only a 429 or 503, which say nothing was launched, and a connection that never opened are retried.
A launch waits at least 75 seconds for its answer (a job submit 135), whatever timeout is, so the client does not give up while the server is still launching.
A 502 with code launch_refused is never retried: the same request is refused the same way.

Environments

API keys are workspace-scoped. By default, requests target the workspace’s default environment (usually main). To target a different environment, pass environment at client construction:

# Default environment (e.g., main)
client = Runcrate(api_key="rc_live_...")

# Specific environment
staging = Runcrate(api_key="rc_live_...", environment="staging")
prod    = Runcrate(api_key="rc_live_...", environment="production")

# Each client only sees resources in its own environment
staging.instances.list()   # only staging instances
prod.instances.list()      # only production instances

What’s environment-scoped: instances, crates, storage volumes. What’s workspace-wide: SSH keys, billing, API keys, templates.

Async Support

from runcrate import AsyncRuncrate

async with AsyncRuncrate(api_key="rc_live_...") as client:
    instances = await client.instances.list()
    balance = await client.billing.get_balance()

Every method available on Runcrate has an async equivalent on AsyncRuncrate.

Model Inference

All inference methods hit api.runcrate.ai.

Chat Completions

response = client.models.chat_completion(
    model="deepseek-ai/DeepSeek-V3",
    messages=[{"role": "user", "content": "Explain quantum computing"}],
    max_tokens=500,
    temperature=0.7,
)
print(response.choices[0].message.content)

Streaming

stream = client.models.chat_completion(
    model="deepseek-ai/DeepSeek-V3",
    messages=[{"role": "user", "content": "Tell me a story"}],
    stream=True,
)
for chunk in stream:
    delta = chunk["choices"][0]["delta"]
    print(delta.get("content", ""), end="", flush=True)

Image Generation

image = client.models.generate_image(
    model="black-forest-labs/FLUX.1-schnell",
    prompt="A futuristic cityscape at sunset",
    width=1024,
    height=768,
)

# Save directly
image.data[0].save("output.png")

Model-Specific Parameters

All image methods accept extra keyword arguments that get passed through to the model. Different models support different parameters:

# Seed for reproducibility
image = client.models.generate_image(
    model="black-forest-labs/FLUX.1-schnell",
    prompt="A cat in space",
    seed=42,
    num_inference_steps=4,
    guidance=3.5,
)

# Image editing: pass a file path, URL, or base64 string
image = client.models.generate_image(
    model="black-forest-labs/FLUX.1-kontext-pro",
    prompt="Make the sky purple",
    image="./photo.png",             # file path (auto base64-encoded)
)

# Image editing with URL
image = client.models.generate_image(
    model="Wan-AI/Wan2.6-Image-Edit",
    prompt="Remove the background",
    image="https://example.com/photo.png",  # URL (passed as-is)
)

# ControlNet / Canny
image = client.models.generate_image(
    model="some-controlnet-model",
    prompt="A building",
    control_image="./edges.png",
    strength=0.8,
)

The image, start_image, mask, and control_image fields accept three formats:

File path: "./photo.png" (auto-detected, read and base64-encoded)
URL: "https://..." (passed through as-is)
Base64 string: raw base64 data (passed through as-is)

Video Generation

# Submit, poll, and save in one call
job = client.models.generate_video_and_save(
    "output.mp4",
    model="google/veo-3.0-audio",
    prompt="A drone flying over mountains",
    duration=8,
    on_status=lambda j: print(f"Status: {j.status}"),
)

Or manage the lifecycle manually:

job = client.models.generate_video(
    model="google/veo-3.0-audio",
    prompt="Ocean waves at sunset",
)

# Poll until done
import time
while job.status not in ("completed", "failed"):
    time.sleep(5)
    job = client.models.get_video_status(job.id)

# Download
video_bytes = client.models.download_video(job.id)
with open("video.mp4", "wb") as f:
    f.write(video_bytes)

Extra parameters (e.g., seed, negative_prompt, image for image-to-video models) are passed through:

job = client.models.generate_video(
    model="some-img2vid-model",
    prompt="Animate this scene",
    image="./first_frame.png",   # image-to-video
    seed=42,
    negative_prompt="blurry",
)

Text-to-Speech

audio = client.models.text_to_speech(
    model="hexgrad/Kokoro-82M",
    input="Hello from Runcrate!",
    voice="af_heart",
)
with open("speech.mp3", "wb") as f:
    f.write(audio)

Extra parameters like speed or language are passed through:

audio = client.models.text_to_speech(
    model="hexgrad/Kokoro-82M",
    input="Hello!",
    voice="af_heart",
    speed=1.5,           # model-specific
    language="en",       # model-specific
)

Transcription

with open("recording.wav", "rb") as f:
    result = client.models.transcribe(
        model="openai/whisper-large-v3",
        file=f,
        filename="recording.wav",
        language="en",              # optional hint; unrecognised values are ignored
        response_format="srt",     # json, verbose_json, text, srt, vtt
    )
print(result.text)  # for text, srt and vtt: the transcript or subtitle file itself

Without response_format you get the full object: text, duration, language and timed segments (the same as verbose_json). json gives text only.

Infrastructure Management

GPU Instances

# List instances
instances = client.instances.list()
instances = client.instances.list(search="training")

# Browse available GPU types
types = client.instances.list_types(gpu_type="A100")
for t in types:
    print(f"{t.id}: {t.gpu_type} x{t.gpu_count}, ${t.hourly_rate}/hr")

# Create an instance
instance = client.instances.create(
    name="training-run",
    ssh_key_id="your-key-id",
    gpu_type="A100",
    gpu_count=1,
    startup_commands=["pip install torch"],
    max_runtime_minutes=240,      # terminate 4 hours after it starts running (5 to 10080)
    idle_shutdown_minutes=30,     # terminate after 30 idle minutes (5 to 1440)
)
print(f"Created: {instance.id}: {instance.status}")

# Check status
status = client.instances.get_status(instance.id)
print(f"Status: {status.status}, IP: {status.ip}")

# Guardrails, once it is running
instance = client.instances.get(instance.id)
print(instance.max_runtime_minutes, instance.idle_shutdown_minutes)
print(f"Stops at: {instance.auto_stop_at}")  # datetime, or None without a max runtime

# Extend before a long job (total minutes from when it started); None turns a limit off
client.instances.update(instance.id, max_runtime_minutes=480, idle_shutdown_minutes=None)

# Terminate
client.instances.terminate(instance.id)

max_runtime_minutes and idle_shutdown_minutes are optional and off when omitted. After an instance stops, instance.stop_reason says why: user, max_runtime, idle, budget, low_balance, failed or system. See Spend guardrails.

Environment Budgets

env = client.environments.get("environment-id")
print(f"Budget: {env.monthly_budget_usd}")         # None means no cap
print(f"Spent this month: {env.month_to_date_usd}")  # UTC calendar month, instances plus inference
print(f"Remaining: {env.budget_remaining_usd}")     # None means no cap

The SDK reads budgets but cannot set them: a workspace owner or admin sets them in the dashboard. Once an environment reaches its budget, launches in it raise InsufficientCreditsError with e.code == "BUDGET_REACHED".

Schedules

A schedule starts an instance or a job later. Pass exactly one of run_at (once, a timezone-aware datetime or an ISO string with an offset) or cron (repeating, read in timezone). spec is the create body of the kind, and max_runtime_minutes is required in it.

from datetime import datetime
from zoneinfo import ZoneInfo

# A job every night at 02:00 London time
schedule = client.schedules.create(
    kind="job",
    name="nightly-train",
    cron="0 2 * * *",
    timezone="Europe/London",
    spec={"command": "python train.py", "gpu_type": "H100", "max_runtime_minutes": 120},
)
print(f"Next run: {schedule.next_run_at}")

# An instance once, at a set time
client.schedules.create(
    kind="instance",
    run_at=datetime(2026, 10, 21, 9, 0, tzinfo=ZoneInfo("America/New_York")),
    timezone="America/New_York",
    spec={"name": "demo-box", "gpu_type": "A100", "max_runtime_minutes": 480},
)

# List, inspect, pause, resume
schedules = client.schedules.list()
s = client.schedules.get(schedule.id)
print(s.last_status, s.last_error, s.run_count)
client.schedules.update(schedule.id, enabled=False)
client.schedules.update(schedule.id, enabled=True)

# Launch once now, then delete
run = client.schedules.run_now(schedule.id)
client.schedules.delete(schedule.id)

Jobs

A job runs one command on its own fresh GPU machine, which stops (and stops billing) when the command ends. Files the command writes to $RUNCRATE_OUTPUT_DIR are kept 30 days.

import os

job = client.jobs.run(
    command="python train.py --epochs 3 && cp model.pt $RUNCRATE_OUTPUT_DIR/",
    gpu_type=["A4000", "L40S"],     # tried in order; an entry may carry its count, "H100:2"
    max_runtime_minutes=60,         # required, 1 to 10080; each attempt stops after this long
    template="ubuntu-train",
    env={"HF_TOKEN": os.environ["HF_TOKEN"]},
    tags={"project": "vision"},
    max_retries=2,                  # retries a machine failure; add retry_on=["nonzero_exit"] for the command's own
    code={"git": {"url": "https://github.com/acme/train", "ref": "main"}},
)

# The output as it is written, until the job ends
for chunk in client.jobs.follow_logs(job.id):
    print(chunk.text, end="")

# Polls with a doubling gap (2 s to 30 s); raises JobWaitTimeoutError past the timeout, and the job keeps running
done = client.jobs.wait(job.id, timeout=3600)
print(done.status, done.exit_code, done.cost_usd)

for artifact in client.jobs.artifacts.list(job.id):
    client.jobs.artifacts.download(job.id, artifact.path, to=os.path.basename(artifact.path))

Send local code with upload = client.jobs.uploads.upload("./my-project"): it packs the directory as a tar.gz (without .git, __pycache__ and node_modules; a .tar.gz file or bytes work too, at most 1 GB) and sends it. Run with code={"upload_id": upload.upload_id} within 24 hours; the command runs in /workspace/code.

Method	What it does
`jobs.run(**spec, idempotency_key=None)`	Runs one job.
`jobs.run_group(**spec, count=, max_parallel=)` or `params=[...]`	Runs `count` copies, or one job per `params` entry (each merged over `env`), at most `max_parallel` at once. Answers the group with its jobs. See Arrays.
`jobs.get(id)`	One job: status, attempts, exit code, cost, tags, code, `artifacts_count`, GPU use.
`jobs.list(status=, tags=, group_id=, q=, gpu_type=, created_after=, limit=, cursor=)`	One page of jobs, newest first (`has_more`, `cursor`). `jobs.list_all(...)` pages through.
`jobs.cancel(id)`	Stops the job’s machine; the job becomes `cancelled`.
`jobs.delete(id)`	Deletes a job that ended, with its logs and outputs (a live one raises `ConflictError`).
`jobs.rerun(id, **overrides)`	Runs the spec again as a new job with its env values; `env` is merged over them.
`jobs.logs(id, after=0, limit=None, follow=False)`	One page of output; pass `cursor` back as `after`.
`jobs.follow_logs(id, after=0)`	Every chunk as it is written, until the job ends.
`jobs.events(id)`	The timeline: queued, launch attempts, machine up, started, exited, retrying, stopped.
`jobs.metrics(id, step=None)`	GPU use while the command ran, one point per `step` seconds.
`jobs.wait(id, timeout=None, poll_interval=2, max_poll_interval=30)`	Polls until the job ends and answers it.
`jobs.artifacts.list(id)`, `.url(id, path)`, `.download(id, path, to=None)`	The outputs (bytes, or streamed to the file `to`); `url` is a signed link valid an hour. The API key is never sent to it.
`jobs.uploads.upload(source)`	Sends your code as a tar.gz.
`job_groups.get(id)`, `job_groups.cancel(id)`	A group’s counts per status and its cost; cancels its live jobs.

AsyncRuncrate has the same methods; follow_logs and list_all are async iterators. Pass your own idempotency_key to make a submit safe to repeat across processes: the same key with the same request answers the first job for 24 hours (see Idempotency). A low balance raises InsufficientCreditsError, a reached environment budget the same with code == "BUDGET_REACHED", and a bad field UnprocessableEntityError.

Secrets

The names of the secrets every machine in the client’s environment gets. Values are set in the dashboard and never returned. A value passed for one launch (env_vars) wins over a secret of the same name.

for secret in client.secrets.list():
    print(secret.key, secret.sensitive, secret.updated_at)

SSH Keys

# List keys
keys = client.ssh_keys.list()

# Add a key
key = client.ssh_keys.create(
    name="my-laptop",
    public_key="ssh-ed25519 AAAA...",
)

# Delete a key
client.ssh_keys.delete(key.id)

Storage

Storage volumes are environment-scoped. Your workspace’s storage must be set up in the dashboard first; the SDK picks it up automatically.

# List available regions (with friendly names)
regions = client.storage.list_regions()
for r in regions:
    print(f"{r.name} ({r.provider})")

# List volumes in the current environment
volumes = client.storage.list()

# Get a specific volume
volume = client.storage.get("volume-id")

# Create a 100GB volume
volume = client.storage.create(
    name="datasets",
    size_gb=100,
    region="us-east-1",
)

# Launch an H100 with the volume attached (files in /workspace)
instance = client.instances.create(
    name="training-run",
    ssh_key_id="your-key-id",
    gpu_type="H100",
    storage_id=volume.id,
)

# Resize (increase capacity only)
client.storage.resize(volume.id, size_gb=200)

# Delete (refunds unused prepaid days pro-rata)
result = client.storage.delete(volume.id)
print(f"Refunded ${result.refund_amount}")

Billing: $0.03/GB/month, charged weekly in advance. Deletion refunds the unused portion of the current billing week.

Billing

# Check balance
balance = client.billing.get_balance()
print(f"Credits: ${balance.credits_balance}")

# List transactions
txns = client.billing.list_transactions(limit=20)
for t in txns:
    print(f"{t.type}: ${t.amount}")
print(f"Has more: {txns.has_more}")

# Usage summary
usage = client.billing.usage(from_date="2025-01-01", to_date="2025-01-31")
print(f"Total cost: ${usage.total_cost}")

Templates

# List templates with search
templates = client.templates.list(search="finetune", category="ml")
for t in templates:
    print(f"{t.name}: {t.category}")
print(f"Total: {templates.total}")

Error Handling

All API errors include the actual error message from the server, never a generic fallback.

from runcrate import (
    NotFoundError,
    AuthenticationError,
    RateLimitError,
    InsufficientCreditsError,
    BadRequestError,
    UnprocessableEntityError,
)

try:
    client.instances.get("nonexistent")
except NotFoundError as e:
    print(f"Not found: {e.message}")
except AuthenticationError:
    print("Invalid API key")
except RateLimitError:
    print("Rate limited, retry later")
except InsufficientCreditsError as e:
    if e.code == "BUDGET_REACHED":
        print(f"Environment budget reached: {e.message}")
    else:
        print(f"Not enough credits: {e.message}")
except UnprocessableEntityError as e:
    print(f"Validation error: {e.message}")  # e.g. invalid model params
except BadRequestError as e:
    print(f"Bad request: {e.message}")

Every error exposes:

e.message: human-readable error description from the API
e.status_code: HTTP status code
e.code: machine-readable error code (e.g. not_found, rate_limited)
e.details: additional details (when available)

Error Hierarchy

Exception	Status Code	Description
`BadRequestError`	400	Invalid parameters
`AuthenticationError`	401	Invalid or missing API key
`InsufficientCreditsError`	402	Not enough credits, or the environment’s monthly budget is reached (`code` is `BUDGET_REACHED`)
`PermissionDeniedError`	403	Insufficient permissions
`NotFoundError`	404	Resource not found
`ConflictError`	409	Resource conflict
`UnprocessableEntityError`	422	Validation error (e.g. invalid model params)
`RateLimitError`	429	Rate limit exceeded
`InternalServerError`	500	Server error
`ConnectionError`	None	Network failure
`TimeoutError`	None	Request timed out

​Installation

​Quick Start

​Configuration

​Retries and launches

​Environments

​Async Support

​Model Inference

​Chat Completions

​Streaming

​Image Generation

​Model-Specific Parameters

​Video Generation

​Text-to-Speech

​Transcription

​Infrastructure Management

​GPU Instances

​Environment Budgets

​Schedules

​Jobs

​Secrets

​SSH Keys

​Storage

​Billing

​Templates

​Error Handling

​Error Hierarchy