Python SDK

Reference

Full Python SDK reference — error handling, async usage, SkytellsClient constructor, configuration types, enums, constants, and deprecated APIs.

Error handling

All API and network failures raise SkytellsError. Always wrap API calls in a try/except. Import ApiErrorId to match specific errors without hardcoding strings.

SkytellsError attributes

error_idstr

Machine-readable identifier (e.g. UNAUTHORIZED, MODEL_NOT_FOUND). See ApiErrorId enum.

detailsstr

Human-readable description with context.

http_statusint

HTTP status code. 0 for network/timeout errors with no HTTP response.

For the complete list of error IDs and HTTP status codes, see the Errors reference.

Error handling

Basic
from skytells import SkytellsClient, SkytellsError

client = SkytellsClient("sk-your-api-key")

try:
  prediction = client.run("flux-pro", input={"prompt": "test"})
  print(prediction.output)
except SkytellsError as e:
  print(f"Error: {e}")              # human-readable message
  print(f"Error ID: {e.error_id}")  # e.g. "UNAUTHORIZED"
  print(f"HTTP Status: {e.http_status}")  # e.g. 401
  print(f"Details: {e.details}")

Error handling — wait()

wait() timeout
from skytells import SkytellsError

bg = client.predictions.create({"model": "flux-pro", "input": {"prompt": "A cat"}})

try:
  result = client.wait(bg, options={"max_wait": 60_000})  # 1 minute max
  print(result["output"])
except SkytellsError as e:
  if e.error_id == "WAIT_TIMEOUT":
      pred_id = bg["id"]
      print(f"Timed out. Prediction ID for later: {pred_id}")

ApiErrorId

All known error_id values are in the ApiErrorId enum, which inherits str so members compare equal to their string values (e.g. ApiErrorId.UNAUTHORIZED == "UNAUTHORIZED").

API-returned errors (from the Skytells server):

error_idHTTPDescription
UNAUTHORIZED401Invalid or missing API key
INVALID_PARAMETER400Request parameter failed validation
INVALID_DATE_FORMAT400Date string is not YYYY-MM-DD
INVALID_DATE_RANGE400since is after until
MODEL_NOT_FOUND404Model slug does not exist
INTERNAL_ERROR500Unexpected server-side error
INVALID_INPUT400Model input validation failed
INSUFFICIENT_CREDITS402Account has insufficient credits
ACCOUNT_SUSPENDED403Account has been suspended
PAYMENT_REQUIRED402Payment is required to continue
SECURITY_VIOLATION403Request violated a security policy
RATE_LIMIT_EXCEEDED429Too many requests
INFRASTRUCTURE_ERROR503Platform infrastructure failure

Client-side errors (raised by the SDK):

error_idHTTPDescription
PREDICTION_FAILEDPrediction completed with status "failed"
WAIT_TIMEOUT408wait() exceeded max_wait
REQUEST_TIMEOUT408HTTP request exceeded timeout
NETWORK_ERROR0Network-level failure (no HTTP response)
SERVER_ERROR5xxNon-JSON or unexpected error response
INVALID_JSONResponse body could not be parsed as JSON

For full details and remediation, see the Errors reference.

SkytellsClient

The main synchronous client. All methods are blocking. Import it from skytells.

Constructor

SkytellsClient(
    api_key: str | None = None,
    options: ClientOptions | dict | None = None,
    *,
    base_url: str | None = None,
    timeout: int = 60_000,
    headers: dict[str, str] | None = None,
    retry: RetryOptions | dict | None = None,
)
api_keystr | None

API key starting with sk-. Required for all authenticated endpoints.

optionsClientOptions | dict | None

Client options object or dict. Alternative to keyword arguments.

base_urlstr | None

Override the API base URL.

timeoutint

Request timeout in milliseconds.

headersdict[str, str] | None

Extra headers sent with every request.

retryRetryOptions | dict | None

Retry configuration.

Sub-APIs

.predictionsPredictionsAPI
.modelsModelsAPI

SkytellsClient

Keyword args
from skytells import SkytellsClient

# Keyword arguments (recommended)
client = SkytellsClient("sk-key", timeout=30_000, retry={"retries": 3})

Async usage

Use AsyncSkytellsClient for async code. All I/O methods are coroutines. HTTP calls run in a ThreadPoolExecutor — zero external dependencies.

The constructor signature is identical to SkytellsClient. Every method mirrors the sync client but is async:

MethodReturns
await client.run(...)AsyncPrediction
await client.predict(...)dict
await client.wait(...)dict
client.queue(...)None (sync)
await client.dispatch()list[dict] — concurrent via asyncio.gather
await client.stream_prediction(id)dict
await client.cancel_prediction(id)dict
await client.delete_prediction(id)dict
await client.predictions.create(...)dict
await client.predictions.get(id)dict
await client.predictions.list(...)PaginatedResponse
await client.models.list(...)list[dict]
await client.models.get(slug, ...)dict

AsyncPrediction has the same properties as Prediction. cancel() and delete() are async.

Async client

Basic
import asyncio
from skytells import AsyncSkytellsClient

async def main():
  client = AsyncSkytellsClient("sk-your-api-key")
  prediction = await client.run("flux-pro", input={"prompt": "A cat"})
  print(prediction.output)
  await prediction.delete()

asyncio.run(main())

Async client — lifecycle & errors

Async lifecycle
async def main():
  client = AsyncSkytellsClient("sk-key")
  prediction = await client.run("flux-pro", input={"prompt": "A cat"})

  # AsyncPrediction — cancel and delete are async
  await prediction.cancel()
  await prediction.delete()

  # Or by ID
  await client.cancel_prediction("pred_abc123")
  await client.delete_prediction("pred_abc123")

Type reference

Imports

from skytells import SkytellsClient          # Synchronous client (recommended)
from skytells import AsyncSkytellsClient      # Async client (asyncio)
from skytells import Prediction, AsyncPrediction
from skytells import PredictionsAPI, ModelsAPI
from skytells import SkytellsError
from skytells import API_BASE_URL

Enums and types:

from skytells import (
    ApiErrorId,
    ModelPrivacy,
    ModelType,
    PredictionSource,
    PredictionStatus,
    PredictionType,
    PricingOperator,
    PricingUnit,
)

Configuration types:

from skytells import (
    ClientOptions,
    ModelFieldsOptions,
    PaginatedResponse,
    Pagination,
    PredictionRequest,
    PredictionsListOptions,
    QueueItem,
    RetryOptions,
    RunOptions,
    WaitOptions,
    OnProgressCallback,   # Callable[[dict], None]
)

Also importable from skytells.types:

from skytells.types import (
    PredictionStatus,
    PredictionType,
    PredictionSource,
    ModelType,
    ModelPrivacy,
    PricingUnit,
    ApiErrorId,
)

Enums

All enums inherit str: PredictionStatus.SUCCEEDED == "succeeded" is True.

PredictionStatusenum

pending · starting · started · processing · succeeded · failed · cancelled

PredictionTypeenum

inference · training

PredictionSourceenum

api · cli · web

ModelTypeenum

image · video · audio · music · text · code · multimodal

ModelPrivacyenum

public · private

PricingUnitenum

image · video · second · prediction · gpu · image_megapixel · computing_second · audio_second · video_second · token · 5 seconds · minute

PricingOperatorenum

equals · ==

Constants

NameValue
API_BASE_URL"https://api.skytells.ai/v1"
__version__"1.0.0"

Key type shapes

PredictionResponse
{
  "id": str,                    # "pred_abc123"
  "status": str,                # see PredictionStatus
  "type": str,                  # "inference" | "training"
  "stream": bool,
  "input": dict,                # the input you sent
  "response": str | None,       # human-readable message (error details on failure)
  "output": str | list[str] | None,  # output URL(s), None if not complete
  "created_at": str,            # ISO 8601
  "started_at": str,
  "completed_at": str,
  "updated_at": str,
  "privacy": str,
  "source": str | None,         # "api" | "cli" | "web"
  "model": {"name": str, "type": str} | None,
  "webhook": {"url": str | None, "events": list[str]} | None,
  "metrics": {
      "image_count": int | None,
      "predict_time": float | None,   # seconds
      "total_time": float | None,     # seconds
      "asset_count": int | None,
      "progress": float | None,       # 0–100
  } | None,
  "metadata": {
      "billing": {"credits_used": int} | None,
      "storage": {
          "files": [{"name": str, "type": str, "size": int, "url": str}],
      } | None,
      "data_available": bool | None,
  } | None,
  "urls": {
      "get": str | None,
      "cancel": str | None,
      "stream": str | None,
      "delete": str | None,
  } | None,
}

Key type shapes — pagination

PaginatedResponse
result = client.predictions.list()

result.data        # list[dict] — prediction items
result.pagination  # Pagination object

result.pagination.current_page  # int
result.pagination.per_page      # int
result.pagination.total         # int
result.pagination.last_page     # int

Deprecated methods

These methods still work but emit DeprecationWarning. Migrate to their current replacements before they are removed in a future release.

DeprecatedReplacement
create_client(key)SkytellsClient(key)
client.list_models()client.models.list()
client.list_predictions()client.predictions.list()
client.get_prediction(id)client.predictions.get(id)
client.get_model(slug)client.models.get(slug)
import warnings

# Before (deprecated)
with warnings.catch_warnings():
  warnings.simplefilter("ignore", DeprecationWarning)
  models      = client.list_models()
  predictions = client.list_predictions()
  pred        = client.get_prediction("pred_abc123")
  model       = client.get_model("flux-pro")

# After (recommended)
models      = client.models.list()
predictions = client.predictions.list()
pred        = client.predictions.get("pred_abc123")
model       = client.models.get("flux-pro")

How is this guide?

On this page