API / concurrent-imagine-video-generations

API

Run concurrent Imagine video generations

Run concurrent Imagine video generations

Fire several video.generate calls at once with xai_sdk.AsyncClient and asyncio.gather. Useful for A/B prompts or a small variation set in one wall-clock wait. Chat-completion concurrency is a different page: Send async concurrent Grok API requests. For hundreds of offline jobs, prefer the Batch API.

Async gather

import os
import asyncio
import xai_sdk

async def generate_concurrently():
    client = xai_sdk.AsyncClient(api_key=os.getenv("XAI_API_KEY"))

    prompts = [
        "A cat sitting on a sunlit windowsill, tail gently swishing.",
        "A dog sprinting through a field of tall grass at golden hour.",
        "A hummingbird hovering near a red flower in slow motion.",
    ]

    tasks = [
        client.video.generate(
            prompt=prompt,
            model="grok-imagine-video-1.5",
            duration=5,
            aspect_ratio="16:9",
            resolution="720p",
        )
        for prompt in prompts
    ]

    results = await asyncio.gather(*tasks)

    for prompt, result in zip(prompts, results):
        print(f"{prompt}: {result.url}")

asyncio.run(generate_concurrently())

Each task still polls internally until done. Raise per-call timeout / interval the same way as Customize Imagine video SDK poll timeout when you stack 1080p or 15-second clips.

Cap in-flight work

Video jobs are heavy. Wrap each generate in an asyncio.Semaphore if you need a hard ceiling (for example max two at a time) so you stay under console rate limits and avoid a burst of service_unavailable retries.

import asyncio
import os
import xai_sdk

async def capped(prompts, max_in_flight=2):
    client = xai_sdk.AsyncClient(api_key=os.getenv("XAI_API_KEY"))
    sem = asyncio.Semaphore(max_in_flight)

    async def one(prompt):
        async with sem:
            return await client.video.generate(
                prompt=prompt,
                model="grok-imagine-video-1.5",
                duration=5,
            )

    return await asyncio.gather(*(one(p) for p in prompts))

Pitfalls

  • asyncio.gather fails the whole group on the first uncaught error unless you pass return_exceptions=True and inspect each item.
  • Temporary URLs expire independently — download each result as it lands if you need archives.
  • Console API credits are a separate bill from SuperGrok's weekly pool on grok.com.