Error Handling and Production Deployment

Updated November 2, 2026 based on xAI video documentation and xai-sdk 1.21.0 code: client.video.generate (not client.videos), result in url, failures reported by VideoGenerationError with a code (invalid_argument, service_unavailable…), rate limit received as grpc.RpcError with code RESOURCE_EXHAUSTED — the RateLimitError class and expired status do not exist. Supplement November 3, 2026: rate limit counts requests per second (10 at entry tier), not “60 RPM”, and actual cost per generation is read from usage.cost_in_usd_ticks. Sources: Video Generation · Rate limits.

Prepare Your Application for Production

A script that generates a video on your machine and a service that produces hundreds per day for real users have almost nothing in common. The latter must handle moderation refusals, quota exceedances, server incidents, and expiring URLs without ever silently losing a billed generation. This lesson covers the mechanisms that make the difference: error classification, retry strategy, queue architecture, and observability.

Two Families of Errors, Two Moments

The first family appears immediately, at the moment of the POST call. These are strict request errors, and the HTTP code is enough to decide what to do.

The second family manifests much later, during polling, after the request has been accepted and generation has started.

This distinction is not academic: a 400 must return instantly to the caller because the code is at fault, while a failed status occurs after several minutes of waiting and must be handled in the job context, not the HTTP request that triggered it.

Retry Intelligently

The common base is exponential backoff with some jitter. Doubling the delay gives a transient incident time to resolve, and jitter prevents all your processes from retrying at the same instant after a spike — a phenomenon that turns a micro-outage into a lasting failure.

import asyncio
import random

async def call_with_retry(func, max_retries=3, base_delay=2):
    """Executes a function with retry and exponential backoff."""
    for attempt in range(max_retries + 1):
        try:
            return await func()
        except Exception as e:
            if attempt == max_retries:
                raise

            # Exponential backoff with jitter
            delay = base_delay * (2 ** attempt) + random.uniform(0, 1)
            print(f"Attempt {attempt + 1} failed: {e}. Retrying in {delay:.1f}s")
            await asyncio.sleep(delay)

This generic scheme is not enough, however, because not all errors deserve the same patience. A quota exceedance resolves by waiting a full minute; a moderation refusal will never resolve, and retrying amounts to paying multiple times for the same failure. The following function differentiates the cases.

import asyncio
from datetime import timedelta

import grpc
from xai_sdk.video import VideoGenerationError

# Failure codes that a retry will not fix
REQUIRES_FIX = {"invalid_argument", "permission_denied", "failed_precondition"}


async def generate_with_retry(client, params, max_retries=3):
    """Video generation with targeted retry (client: xai_sdk.AsyncClient)."""
    for attempt in range(max_retries + 1):
        is_last = attempt == max_retries
        try:
            return await client.video.generate(**params)
        except VideoGenerationError as e:
            if e.code.lower() in REQUIRES_FIX or is_last:
                raise  # request or content refused: fix before relaunching
            await asyncio.sleep(5 * (attempt + 1))  # service overloaded, incident
        except grpc.RpcError as e:
            if e.code() != grpc.StatusCode.RESOURCE_EXHAUSTED or is_last:
                raise
            await asyncio.sleep(2 ** attempt)  # rate limit: backoff
        except TimeoutError:
            if is_last:
                raise
            # Allow more time for the next attempt
            params["timeout"] = params.get("timeout", timedelta(minutes=10)) + timedelta(minutes=2)

A Queue to Keep Pace

As soon as your service receives requests from multiple users, you must decouple request reception from execution, and above all bound concurrency to stay under your tier’s rate limit. A queue paired with a semaphore meets both needs: jobs stack freely, but only ten maximum execute at once.

import asyncio
from collections import deque

class VideoGenerationQueue:
    def __init__(self, client, max_concurrent=10):
        self.client = client
        self.queue = deque()
        self.semaphore = asyncio.Semaphore(max_concurrent)
        self.results = {}

    async def add(self, job_id, params):
        """Adds a generation to the queue."""
        self.queue.append((job_id, params))

    async def process(self):
        """Processes all pending generations."""
        tasks = []
        while self.queue:
            job_id, params = self.queue.popleft()
            task = asyncio.create_task(
                self._execute(job_id, params)
            )
            tasks.append(task)
        await asyncio.gather(*tasks, return_exceptions=True)

    async def _execute(self, job_id, params):
        """Executes a generation with concurrency control."""
        async with self.semaphore:
            try:
                result = await generate_with_retry(
                    self.client, params
                )
                self.results[job_id] = {
                    "status": "done",
                    "url": result.url
                }
            except Exception as e:
                self.results[job_id] = {
                    "status": "failed",
                    "error": str(e)
                }

Note that each job keeps its outcome in results, success or failure: a generation that disappears without a trace is an incident no one can diagnose.

The second architecture reflex concerns storage. Since video URLs are temporary, downloading must be automatic and immediate, with timestamped naming that guarantees no collision between two simultaneous generations.

import httpx
from pathlib import Path
from datetime import datetime

async def save_video(video_url, output_dir="./videos"):
    """Downloads and saves a generated video."""
    Path(output_dir).mkdir(parents=True, exist_ok=True)
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    file_path = f"{output_dir}/video_{timestamp}.mp4"

    async with httpx.AsyncClient() as client:
        resp = await client.get(video_url)
        resp.raise_for_status()
        Path(file_path).write_bytes(resp.content)

    return file_path

Verify Your Key Before Depending on It

The GET /v1/api-key endpoint informs you about your key’s state. Calling it at service startup avoids discovering a blocked key after twenty failed requests, while your users wait.

resp = httpx.get(
    "https://api.x.ai/v1/api-key",
    headers={"Authorization": f"Bearer {api_key}"}
).json()

if resp.get("api_key_blocked") or resp.get("api_key_disabled") or resp.get("team_blocked"):
    raise RuntimeError("API key or team blocked, or key disabled")

See What Happens in Production

Five indicators suffice to pilot a video generation service. The success rate, meaning the share of generations ending in done, measures overall health. The average generation time, measured between submission and done status, serves to calibrate your timeouts and inform your users. The error rate broken down by type — failed by error code, timeout, rate limit — tells you where to focus effort: a rise in timeouts without a rise in failed signals generations slower than your wait deadline, for example, not failures. The cumulative cost, obtained by summing the usage.cost_in_usd_ticks field from each result (one dollar equals 10 billion ticks), or alternatively by multiplying generated durations by the model’s price per second, must be monitored daily. Finally, quota usage, in requests per second relative to your tier limit (10 at entry tier for video models), anticipates saturation before it translates into 429s.

These metrics build naturally from a structured log, where each generation leaves a line usable by your analysis tool.

import logging
import json

logger = logging.getLogger("video-generation")

def log_generation(request_id, status, duration_s, cost, time_ms):
    logger.info(json.dumps({
        "event": "video_generation",
        "request_id": request_id,
        "status": status,
        "duration_seconds": duration_s,
        "cost_usd": cost,
        "processing_time_ms": time_ms
    }))

Before opening your service to the public, review the whole thing from the start. Retry with exponential backoff and jitter must be in place, and concurrency bounded by a semaphore to stay under the rate limit. Video downloads must trigger automatically as soon as they are ready, without depending on human action. Each generation must leave a log line carrying its request_id, status, and cost, otherwise the error rate and average generation time will remain unverifiable. Check the API key state at startup rather than at the first failure. And set a budget with overrun alerts: at $0.05 per second, an unbounded retry loop consumes faster than you think.

Pricing verified November 3, 2026 — prices evolve regularly: before any budget calculation, verify the current grid on the official xAI models and pricing page.

Key Takeaways

• Errors split into synchronous (request) and asynchronous (generation)
• Not all errors merit a retry: moderation is definitive
• Use a semaphore to limit concurrent generations
• Temporary URLs require immediate and systematic downloading
• Track cost, success, and generation time metrics to pilot your service

Test Your Knowledge

Generative video in production: validate the full workflow.

1. What are the particularities of a video prompt?

Answer: Describe the action and movement, not just the scene: subject, displacement, ambiance, framing — the temporal dimension is what distinguishes the video prompt from the image prompt.

2. What does image-to-video allow?

Answer: Animate an existing image, or bring given people and products into a new scene using reference images — and, on grok-imagine-video-1.5, pin the first, last, or key images: your visual identity becomes a starting point, not a chance occurrence.

3. Editing and extension: what uses?

Answer: /videos/edits modifies a generated video, /videos/extensions prolongs it — and both combine to build long sequences by iterations.

4. Why is the workflow asynchronous?

Answer: Video generation takes time: you submit a job, poll its status, then retrieve the result — the xAI SDK automates this polling.

5. What to plan for production?

Answer: Error handling and failed jobs, immediate download of outputs, and cost tracking (resolution, duration, iterations) — video is the most expensive modality: it must be budgeted.

Temporal prompt, visual reference, mastered async: generative video becomes a pipeline — not a lottery.