Python SDK - Best Practices

This page covers patterns and recommendations for building production applications with the Skytells Python SDK. Following these practices will make your integration more resilient, secure, and cost-efficient.

API Key Security

Your API key grants full access to your Skytells account, including billing. Treat it like a password.

Never:

Hard-code your key in source files
Commit it to version control
Log it or expose it in error messages
Use it in client-side code (browsers, mobile apps)

Always:

Load it from environment variables or a secrets manager
Rotate keys periodically or after any suspected exposure
Use separate keys for development and production

Loading API keys

Environment variable

import os
from skytells import SkytellsClient

# SDK auto-reads SKYTELLS_API_KEY from env if api_key is not passed
client = SkytellsClient()

# Or explicitly:
client = SkytellsClient(api_key=os.environ["SKYTELLS_API_KEY"])

Use a Singleton Client

Create the SkytellsClient once at application startup and reuse it everywhere. Avoid constructing a new client per request — each client manages connection state and retry configuration.

The SkytellsClient is thread-safe. A single instance can be shared across threads and web workers.

# services/skytells.py
from skytells import SkytellsClient, RetryOptions

_client: SkytellsClient | None = None

def get_client() -> SkytellsClient:
    global _client
    if _client is None:
        _client = SkytellsClient(
            retry=RetryOptions(retries=3, retry_delay=1000),
            timeout=120_000,
        )
    return _client

# In any module:
from services.skytells import get_client

prediction = get_client().run("truefusion-pro", input={
    "prompt": "A serene mountain lake",
})

Choose the Right Method

The SDK provides several ways to run a prediction. Choose based on your use case:

client.run()Prediction

Start + wait + return. Blocks until the prediction completes. Best for scripts, CLI tools, and simple request/response flows.

client.predict()Prediction

Start only. Returns immediately with a pending prediction. Use when you want to dispatch work in the background, use webhooks, or control polling yourself.

client.queue()Prediction

Queue and dispatch. Starts a prediction with a webhook — use for background jobs and async pipelines.

client.stream_prediction()Iterator

Streaming events. Yields SSE events as they arrive. Best for real-time progress UIs.

# Use run() for simple synchronous use cases
result = client.run("truefusion-edge", input={"prompt": "A cat"})
print(result.outputs())

# Use predict() when you want to start and check later
prediction = client.predict("mera", input={
    "prompt": "City timelapse at night",
    "seconds": "8",
    "size": "1280x720",
})
print(f"Started: {prediction.id}")
# ... do other work ...
finished = client.wait(prediction)
print(finished.outputs())

# Use queue() for webhook-driven pipelines
prediction = client.queue("beatfusion-2.0", input={
    "lyrics": "[Verse]\nBuilding the future...",
    "prompt": "Hip-hop, energetic",
}, webhook={
    "url": "https://your-app.com/webhooks/skytells",
    "events": ["completed", "failed"],
})
print(f"Queued: {prediction.id}")

Set Timeouts for Long-Running Models

Video and audio models can take minutes to complete. Set max_wait on WaitOptions and timeout on the client accordingly so your application doesn't hang indefinitely.

Model type	Typical duration	Recommended `max_wait`
Image (fast)	1–5s	30s
Image (quality)	5–30s	60s
Video (short)	30–120s	3 min
Video (long/quality)	2–8 min	10 min
Audio/music	30–90s	3 min

from skytells import SkytellsClient, WaitOptions

# Global timeout covers the HTTP request itself
client = SkytellsClient(timeout=300_000)  # 5 minutes

# max_wait covers the polling loop
prediction = client.predict("mera", input={
    "prompt": "A cinematic ocean sunset with waves",
    "seconds": "12",
    "size": "1280x720",
})

result = client.wait(prediction, WaitOptions(
    interval=10_000,    # poll every 10s for slow models
    max_wait=600_000,   # give up after 10 minutes
))
print(result.outputs())

Configure Automatic Retries

Enable automatic retries on the client to handle transient failures (rate limits, service hiccups) without writing retry logic yourself.

The SDK uses linear backoff: attempt n waits retry_delay × n ms. Attempt 1 → 1s, attempt 2 → 2s, attempt 3 → 3s.

Retries apply to HTTP-level failures (429, 500, 502, 503, 504). PREDICTION_FAILED is not retried automatically — handle it in your own code.

from skytells import SkytellsClient, RetryOptions

client = SkytellsClient(
    retry=RetryOptions(
        retries=4,
        retry_delay=2000,    # 2s, 4s, 6s, 8s
        retry_on=[429, 500, 502, 503, 504],
    )
)

Always Use `outputs()` Over `.output`

Different models return different output shapes — some return a string URL, others return a list. The outputs() method normalises this for you:

Single-element lists are unwrapped to a string
Multi-element lists are returned as-is
None is returned when there's no output

This means your code works correctly across all model types without special-casing.

prediction = client.run("truefusion-pro", input={
    "prompt": "A coastal town at sunset",
    "num_outputs": 1,
})

result = prediction.outputs()

if result is None:
    print("No output")
elif isinstance(result, str):
    print("Single image:", result)
elif isinstance(result, list):
    for i, url in enumerate(result):
        print(f"Image {i + 1}:", url)

Use Async for Concurrent Predictions

When you need to run multiple predictions at the same time, use AsyncSkytellsClient with asyncio.gather(). This is significantly faster than running them sequentially because predictions run in parallel.

AsyncSkytellsClient uses a ThreadPoolExecutor under the hood — no extra dependencies needed.

import asyncio
from skytells import AsyncSkytellsClient

client = AsyncSkytellsClient()

async def main():
    prompts = [
        "A sunset over mountains",
        "A futuristic cityscape at night",
        "An ancient forest with mist",
        "A beach at golden hour",
    ]

    # Run all predictions in parallel
    tasks = [
        client.run("truefusion-pro", input={
            "prompt": prompt,
            "aspect_ratio": "16:9",
        })
        for prompt in prompts
    ]

    results = await asyncio.gather(*tasks, return_exceptions=True)

    for prompt, result in zip(prompts, results):
        if isinstance(result, Exception):
            print(f"Failed '{prompt}': {result}")
        else:
            print(f"'{prompt}': {result.outputs()}")

asyncio.run(main())

Clean Up Predictions

Predictions and their output assets are stored on the Skytells platform. If you generate outputs that should not be retained (e.g. user-submitted content, test runs), delete them after you've processed the output.

prediction.delete() removes the prediction record and all associated output assets
prediction.cancel() cancels a prediction that hasn't completed yet

Deleting a prediction is irreversible. Download or copy output URLs before calling delete().

from skytells import SkytellsClient
import urllib.request

client = SkytellsClient()

prediction = client.run("truefusion-pro", input={
    "prompt": "A photorealistic landscape",
})

output_url = prediction.outputs()

# Download first
urllib.request.urlretrieve(output_url, "output.webp")
print("Saved to output.webp")

# Then delete
prediction.delete()
print("Prediction and assets deleted")

Webhook-Driven Pipelines

For long-running predictions in web applications, avoid holding an open HTTP connection while you wait. Use webhooks instead:

Start the prediction with client.queue() and a webhook URL
Return a response to your user immediately
The Skytells platform POSTs the result to your webhook when complete

This is the recommended pattern for video and audio generation in web apps.

See the Predictions guide for full webhook details.

from skytells import SkytellsClient
from fastapi import FastAPI

app = FastAPI()
client = SkytellsClient()

@app.post("/generate")
async def generate(prompt: str, user_id: str):
    prediction = client.queue("truefusion-video-pro", input={
        "prompt": prompt,
        "aspect_ratio": "16:9",
        "duration": "5",
    }, webhook={
        "url": "https://your-app.com/webhooks/skytells",
        "events": ["completed", "failed"],
    })

    # Return immediately — don't wait
    return {"prediction_id": prediction.id, "status": "queued"}

@app.post("/webhooks/skytells")
async def handle_webhook(payload: dict):
    pred_id = payload["id"]
    status = payload["status"]

    if status == "succeeded":
        output_url = payload["output"]
        # Save output, notify user, etc.
        print(f"Prediction {pred_id} succeeded: {output_url}")
    elif status == "failed":
        print(f"Prediction {pred_id} failed: {payload.get('error')}")

    return {"ok": True}

Validate Inputs Before Sending

Fetch a model's input_schema before running in production to validate your inputs locally. This avoids paying for a prediction that will fail validation on the server.

Use the schema's required list to ensure all required fields are present, and check properties for type and enum constraints.

from skytells import SkytellsClient

client = SkytellsClient()

# Fetch schema once at startup (cache it!)
model = client.models.get("truefusion-pro", fields=["input_schema"])
schema = model["input_schema"]

required_fields = schema.get("required", [])
properties = schema.get("properties", {})

def validate_input(inp: dict) -> list[str]:
    errors = []
    for field in required_fields:
        if field not in inp:
            errors.append(f"Missing required field: {field}")
    for key, value in inp.items():
        prop = properties.get(key)
        if prop and prop.get("type") == "integer" and not isinstance(value, int):
            errors.append(f"'{key}' must be an integer")
    return errors

user_input = {"prompt": "A mountain range at dawn", "aspect_ratio": "16:9"}
errs = validate_input(user_input)
if errs:
    print("Input errors:", errs)
else:
    prediction = client.run("truefusion-pro", input=user_input)
    print(prediction.outputs())

Use Enums for Status Checks

Always use PredictionStatus enum values when checking prediction statuses instead of string literals. This prevents bugs from typos and makes your code resilient to future API changes.

All SDK enums inherit from str, so PredictionStatus.SUCCEEDED == "succeeded" is True.

from skytells import SkytellsClient, PredictionStatus

client = SkytellsClient()

prediction = client.predict("truefusion-pro", input={
    "prompt": "A cinematic landscape",
})

# Poll manually
import time
while True:
    current = client.predictions.get(prediction.id)
    status = current["status"]

    if status == PredictionStatus.SUCCEEDED:
        print("Done:", current["output"])
        break
    elif status == PredictionStatus.FAILED:
        print("Failed:", current.get("error"))
        break
    elif status in (
        PredictionStatus.PENDING,
        PredictionStatus.STARTING,
        PredictionStatus.PROCESSING,
    ):
        print(f"Status: {status} — waiting...")
        time.sleep(5)
    else:
        print(f"Unexpected status: {status}")
        break

Logging and Observability

Add structured logging around predictions to track latency, cost, and failures in production:

import logging
import time
from skytells import SkytellsClient, SkytellsError

logger = logging.getLogger(__name__)
client = SkytellsClient(retry={"retries": 3})

def run_tracked(model: str, input: dict) -> str | None:
    start = time.monotonic()
    try:
        prediction = client.run(model, input=input)
        elapsed = time.monotonic() - start

        raw = prediction.raw()
        credits = raw.get("metadata", {}).get("billing", {}).get("credits_used")
        predict_time = raw.get("metrics", {}).get("predict_time")

        logger.info(
            "prediction.succeeded",
            extra={
                "prediction_id": prediction.id,
                "model": model,
                "elapsed_s": round(elapsed, 2),
                "predict_time_s": predict_time,
                "credits_used": credits,
            },
        )
        return prediction.outputs()

    except SkytellsError as e:
        elapsed = time.monotonic() - start
        logger.error(
            "prediction.failed",
            extra={
                "model": model,
                "error_id": e.error_id,
                "status_code": e.status_code,
                "elapsed_s": round(elapsed, 2),
            },
        )
        return None

How is this guide?

Python SDK - Best Practices

Loading API keys

On this page