Async Client and Performance — Generate and Edit Images with Grok

Updated November 3, 2026 based on the xAI Rate limits page and the Images API reference: Imagine model limits count as requests per second—6 at the entry tier, 100 at the highest tier—not “30 per minute”; a semaphore bounds concurrent requests, not throughput; images output as JPEG by default; and the specific timings from the old version, which came from no published measurement, give way to a calculation. Sources: Rate limits · Images API.

Optimize Performance with Async

When you need to generate or edit many images, sequential calls become a bottleneck. The xAI API async client lets you run multiple requests in parallel, significantly reducing total processing time.

The AsyncOpenAI Client

Async doesn’t make a single generation faster: a single image request takes exactly the same amount of time. What it eliminates is unnecessary waiting between requests—while the first image calculates on the server, your program can request ten others instead of staying blocked. That’s why the gain is zero for one image and dramatic for fifty.

from openai import AsyncOpenAI
import asyncio
import os

async_client = AsyncOpenAI(
    api_key=os.getenv("XAI_API_KEY"),
    base_url="https://api.x.ai/v1"
)

async def generate_image(prompt):
    """Generates an image asynchronously."""
    response = await async_client.images.generate(
        model="grok-imagine-image",
        prompt=prompt,
        extra_body={"resolution": "1k"},
    )
    return response.data[0].url

async def main():
    url = await generate_image("A lighthouse on a cliff, watercolor style")
    print(url)

asyncio.run(main())

Parallel Requests with asyncio.gather

asyncio.gather launches all requests then waits for the set to finish. The total duration becomes that of the slowest, not the sum of all—that’s the whole difference, and you see it with just three or four images.

async def generate_batch(prompts):
    """Generates images for multiple prompts in parallel."""
    tasks = [generate_image(prompt) for prompt in prompts]
    results = await asyncio.gather(*tasks)
    return results

async def main():
    prompts = [
        "Portrait of an engineer, corporate style",
        "Panoramic view of a futuristic city",
        "Isometric diagram of a data center",
        "Abstract blue and silver logo",
        "Illustration of a team in a meeting"
    ]

    urls = await generate_batch(prompts)
    for prompt, url in zip(prompts, urls):
        print(f"{prompt[:40]}... → {url}")

asyncio.run(main())

With five prompts, instead of waiting for each request to finish before launching the next, all five images calculate at the same time: total time approaches that of the slowest request.

Control Parallelism

Without a limit, gather launches everything at once—and fifty concurrent requests on a model capped at 6 requests per second at the entry tier produce an avalanche of 429 errors that costs you more time than sequential execution. The semaphore is therefore not an optional precaution: it is what makes parallelism usable. Calibrate it to the model’s rate limit, never to your machine’s capacity.

async def generate_with_limit(prompts, max_concurrent=5):
    """Generates images with a maximum number of parallel requests."""
    semaphore = asyncio.Semaphore(max_concurrent)

    async def limited_generate(prompt):
        async with semaphore:
            return await generate_image(prompt)

    tasks = [limited_generate(prompt) for prompt in prompts]
    return await asyncio.gather(*tasks)

The official sheets for grok-imagine-image and grok-imagine-image-2.0 indicate 6 requests per second at the entry tier, a limit that grows with your team’s tier up to 100. Note: a semaphore bounds concurrent requests, not throughput. With max_concurrent=5, you stay under 6 requests per second as long as each generation lasts more than one second; if your responses arrive faster, space out submissions too. Check your team’s limits in the xAI console before increasing.

Complete Async Pipeline

This pipeline chains the three operations that must go together: generate, download immediately—URLs expire—and isolate failures. On a large batch, this third point makes the difference between an incident and a total loss.

import aiohttp
import aiofiles

async def download_image(session, url, filepath):
    """Downloads an image asynchronously."""
    async with session.get(url) as resp:
        if resp.status == 200:
            async with aiofiles.open(filepath, "wb") as f:
                await f.write(await resp.read())
            return filepath
    return None

async def pipeline(prompts, output_dir="output"):
    """Complete pipeline: generation + download."""
    os.makedirs(output_dir, exist_ok=True)

    # Step 1: generate all images in parallel
    urls = await generate_with_limit(prompts, max_concurrent=5)

    # Step 2: download all images in parallel
    async with aiohttp.ClientSession() as session:
        download_tasks = []
        for i, url in enumerate(urls):
            if url:
                # JPEG by default (image/jpeg), unless quality setting produces PNG
                filepath = os.path.join(output_dir, f"image_{i:03d}.jpg")
                download_tasks.append(download_image(session, url, filepath))

        saved_files = await asyncio.gather(*download_tasks)

    count = sum(1 for f in saved_files if f)
    print(f"{count}/{len(prompts)} images generated and saved")

Synchronous vs Asynchronous Comparison

The gain calculates on paper. If one generation takes T seconds, twenty images requested one after another take about 20 × T. With five concurrent requests, they go in waves of five: about 4 × T, a gain close to five, the number of concurrent requests—provided you stay under the throughput limit. Measure T on your own account before promising a deadline: xAI’s image pages give no generation time, and it varies with model, resolution, and load.

Error Handling

gather’s default behavior is brutal: an exception in one task bubbles up the error and you lose the results of the others, including those already paid for. return_exceptions=True changes that—each task returns either its result or its exception, and you decide what to do with failures without sacrificing successes.

async def safe_generate(prompt):
    """Generates an image with error handling."""
    try:
        return await generate_image(prompt)
    except Exception as e:
        print(f"Error for '{prompt[:30]}...' : {e}")
        return None

Key Takeaways

The AsyncOpenAI client enables parallel requests via asyncio
Use asyncio.gather to launch multiple generations simultaneously
Limit concurrency with a semaphore to respect rate limits
Gain approaches the number of concurrent requests, within the authorized throughput limit (6 requests per second at the entry tier)
Handle errors individually to avoid losing the whole batch