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.
The client automatically retries on 429, 500, 502, 503, 504 when retries are configured. See RetryOptions.
Error handling
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()
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_id | HTTP | Description |
|---|---|---|
UNAUTHORIZED | 401 | Invalid or missing API key |
INVALID_PARAMETER | 400 | Request parameter failed validation |
INVALID_DATE_FORMAT | 400 | Date string is not YYYY-MM-DD |
INVALID_DATE_RANGE | 400 | since is after until |
MODEL_NOT_FOUND | 404 | Model slug does not exist |
INTERNAL_ERROR | 500 | Unexpected server-side error |
INVALID_INPUT | 400 | Model input validation failed |
INSUFFICIENT_CREDITS | 402 | Account has insufficient credits |
ACCOUNT_SUSPENDED | 403 | Account has been suspended |
PAYMENT_REQUIRED | 402 | Payment is required to continue |
SECURITY_VIOLATION | 403 | Request violated a security policy |
RATE_LIMIT_EXCEEDED | 429 | Too many requests |
INFRASTRUCTURE_ERROR | 503 | Platform infrastructure failure |
Client-side errors (raised by the SDK):
error_id | HTTP | Description |
|---|---|---|
PREDICTION_FAILED | — | Prediction completed with status "failed" |
WAIT_TIMEOUT | 408 | wait() exceeded max_wait |
REQUEST_TIMEOUT | 408 | HTTP request exceeded timeout |
NETWORK_ERROR | 0 | Network-level failure (no HTTP response) |
SERVER_ERROR | 5xx | Non-JSON or unexpected error response |
INVALID_JSON | — | Response 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
Access to predictions.create(), predictions.get(), and predictions.list().
.modelsModelsAPI
Access to models.list() and models.get().
SkytellsClient
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:
| Method | Returns |
|---|---|
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.
AsyncSkytellsClient.dispatch() uses asyncio.gather — all queued predictions fire concurrently. SkytellsClient.dispatch() is sequential.
Async client
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 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_URLEnums 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
| Name | Value |
|---|---|
API_BASE_URL | "https://api.skytells.ai/v1" |
__version__ | "1.0.0" |
Key type shapes
{
"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
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 # intDeprecated methods
These methods still work but emit DeprecationWarning. Migrate to their current replacements before they are removed in a future release.
| Deprecated | Replacement |
|---|---|
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")For the complete list of error IDs and prediction-level errors, see the Errors reference. For request and response shapes, see the Predictions API.
How is this guide?