Concurrent Google AI API Calls with Python asyncio in Antigravity — A Practical Guide

When I was building a content quality-checking tool in Antigravity — sending 100 product descriptions to the Gemini API one by one — I watched the progress bar crawl for nearly 15 minutes. Switching to async processing with asyncio cut that down to under three minutes. Same workload, same API, just a different execution model.

This guide walks through using Python's asyncio with the google-genai SDK inside Antigravity IDE, covering everything from the basic concurrent pattern to Semaphore-based rate limiting and a real batch analysis pipeline you can drop into your own projects.

Why Synchronous Processing Hits a Wall

Every call to the Gemini API takes a few seconds to return a response. In synchronous code, each request blocks execution until it completes, so 100 requests means roughly 100× the single-request latency.

# ❌ Synchronous — each request waits for the previous one
import google.generativeai as genai
 
def analyze_reviews_sync(reviews: list[str]) -> list[str]:
    model = genai.GenerativeModel("gemini-2.0-flash")
    results = []
    for review in reviews:
        response = model.generate_content(
            f"Analyze the sentiment of this review: {review}"
        )
        results.append(response.text)
    return results

100 items × ~2 seconds each = 200 seconds. The async version sends all requests in parallel, so you're waiting for the slowest response rather than the sum of all responses.

Setting Up in Antigravity

Create a Python project in Antigravity, then install the required packages:

python -m venv .venv
source .venv/bin/activate  # Windows: .venv\Scripts\activate
 
pip install google-genai aiohttp python-dotenv

Set your API key in a .env file:

GEMINI_API_KEY=YOUR_GEMINI_API_KEY

One Antigravity-specific tip: add a .antigravity/rules.md file to help the AI assistant understand your project's patterns.

# .antigravity/rules.md
This project uses Python asyncio for concurrent API processing.
- Prefer google-genai's AsyncClient (client.aio.models)
- All async functions must include type hints
- Handle google.api_core.exceptions for error recovery

With this in place, Inline Chat (⌘I) suggestions will align with your async coding style rather than defaulting to synchronous patterns.

The Core Pattern: asyncio.gather()

The google-genai SDK v0.8+ includes an async client under client.aio. Here's the fundamental concurrent processing pattern:

import asyncio
import os
from google import genai
from google.genai import types
from dotenv import load_dotenv
 
load_dotenv()
 
client = genai.Client(api_key=os.environ["GEMINI_API_KEY"])
 
async def analyze_single(review: str, index: int) -> tuple[int, str]:
    """Analyze one review asynchronously."""
    try:
        response = await client.aio.models.generate_content(
            model="gemini-2.0-flash",
            contents=f"Classify this review as positive, negative, or neutral: {review}",
            config=types.GenerateContentConfig(
                max_output_tokens=50,
                temperature=0.1,
            ),
        )
        return index, response.text.strip()
    except Exception as e:
        # Return the error without stopping other tasks
        return index, f"ERROR: {str(e)}"
 
async def analyze_reviews_async(reviews: list[str]) -> list[str]:
    """Process all reviews concurrently."""
    tasks = [analyze_single(review, i) for i, review in enumerate(reviews)]
    results = await asyncio.gather(*tasks)
    # Restore original order
    return [text for _, text in sorted(results, key=lambda x: x[0])]
 
async def main():
    reviews = [
        "Absolutely love this product — exceeded my expectations\!",
        "The packaging was damaged when it arrived.",
        "It's fine. Does what it says.",
    ]
    results = await analyze_reviews_async(reviews)
    for review, sentiment in zip(reviews, results):
        print(f"{review[:40]}... → {sentiment}")
 
if __name__ == "__main__":
    asyncio.run(main())

The key difference from the sync version: asyncio.gather() launches all tasks simultaneously and collects results as they complete. We sort by index at the end to preserve the original order regardless of which response arrived first.

Rate Limiting with Semaphore

Firing hundreds of requests at once will hit Google AI's rate limits (measured in requests per minute). A Semaphore lets you cap concurrency while keeping requests parallel:

import asyncio
from google import genai
from google.genai import types
import os
from dotenv import load_dotenv
 
load_dotenv()
 
client = genai.Client(api_key=os.environ["GEMINI_API_KEY"])
 
# Free tier: ~15 RPM. Paid tier: up to 2000 RPM.
# Setting MAX_CONCURRENT to 10 gives a comfortable buffer on paid plans.
MAX_CONCURRENT = 10
 
async def analyze_with_semaphore(
    semaphore: asyncio.Semaphore,
    review: str,
    index: int,
) -> tuple[int, str]:
    """Rate-limited async analysis."""
    async with semaphore:
        try:
            response = await client.aio.models.generate_content(
                model="gemini-2.0-flash",
                contents=f"Sentiment (positive/negative/neutral): {review}",
                config=types.GenerateContentConfig(max_output_tokens=50),
            )
            return index, response.text.strip()
        except Exception as e:
            # Back off and retry on rate limit errors
            if "RESOURCE_EXHAUSTED" in str(e):
                await asyncio.sleep(5)
                return await analyze_with_semaphore(semaphore, review, index)
            return index, f"ERROR: {str(e)}"
 
async def batch_analyze(reviews: list[str]) -> list[str]:
    """Batch process with controlled concurrency."""
    semaphore = asyncio.Semaphore(MAX_CONCURRENT)
    tasks = [
        analyze_with_semaphore(semaphore, review, i)
        for i, review in enumerate(reviews)
    ]
    results = await asyncio.gather(*tasks)
    return [text for _, text in sorted(results)]

In practice, I've found MAX_CONCURRENT = 10–15 to be stable across most paid-tier Gemini API setups. If you're seeing intermittent RESOURCE_EXHAUSTED errors, drop it to 5 and see if that resolves them.

A Production-Ready Batch Analysis Pipeline

Here's a complete pipeline that reads from CSV, runs concurrent analysis, and saves structured JSON output:

import asyncio
import csv
import json
from pathlib import Path
from google import genai
from google.genai import types
import os
from dotenv import load_dotenv
 
load_dotenv()
 
client = genai.Client(api_key=os.environ["GEMINI_API_KEY"])
MAX_CONCURRENT = 10
 
PROMPT = """
Analyze the following product review and respond in JSON format:
- sentiment: "positive" / "negative" / "neutral"
- score: integer 1–5 (intensity of the sentiment)
- key_points: list of up to 2 main points
 
Review: {review}
 
Output JSON only (no code fences).
"""
 
async def analyze_structured(
    semaphore: asyncio.Semaphore,
    review: str,
    index: int,
) -> tuple[int, dict]:
    """Return structured analysis for one review."""
    async with semaphore:
        try:
            response = await client.aio.models.generate_content(
                model="gemini-2.0-flash",
                contents=PROMPT.format(review=review),
                config=types.GenerateContentConfig(
                    max_output_tokens=200,
                    temperature=0.1,
                    response_mime_type="application/json",  # Forces valid JSON output
                ),
            )
            data = json.loads(response.text)
            return index, {"review": review, **data, "status": "ok"}
        except json.JSONDecodeError:
            return index, {"review": review, "status": "parse_error"}
        except Exception as e:
            if "RESOURCE_EXHAUSTED" in str(e):
                await asyncio.sleep(5)
                return await analyze_structured(semaphore, review, index)
            return index, {"review": review, "status": "error", "error": str(e)}
 
async def run_pipeline(input_csv: str, output_json: str):
    reviews = []
    with open(input_csv, encoding="utf-8") as f:
        for row in csv.DictReader(f):
            reviews.append(row["review_text"])
 
    print(f"Processing {len(reviews)} reviews...")
 
    semaphore = asyncio.Semaphore(MAX_CONCURRENT)
    tasks = [analyze_structured(semaphore, r, i) for i, r in enumerate(reviews)]
    results = await asyncio.gather(*tasks)
    sorted_results = [data for _, data in sorted(results)]
 
    success = sum(1 for r in sorted_results if r["status"] == "ok")
    output = {
        "total": len(reviews),
        "success": success,
        "errors": len(reviews) - success,
        "results": sorted_results,
    }
 
    Path(output_json).write_text(
        json.dumps(output, ensure_ascii=False, indent=2), encoding="utf-8"
    )
    print(f"Done: {success} succeeded, {len(reviews) - success} failed → {output_json}")
 
if __name__ == "__main__":
    asyncio.run(run_pipeline("reviews.csv", "analysis_results.json"))

The response_mime_type="application/json" parameter is worth calling out — it instructs the model to return strictly valid JSON, which dramatically reduces parse errors. It's not prominently featured in the official docs, but it's one of the most useful options for structured output pipelines.

Making the Most of Antigravity's AI Assist

Async Python code is harder to reason about than sync code — race conditions, incomplete error propagation, and deadlocks tend to be subtle. Antigravity's AI tools help a lot here.

Inline Chat (⌘I) for error handling: Select an async function and ask "Add proper error handling for google.api_core exceptions." Antigravity will suggest SDK-specific exception classes you might not know off the top of your head, like ResourceExhausted or DeadlineExceeded.

AI Chat (⌘L) for debugging: When you hit a confusing async error, paste the traceback into the chat and ask for diagnosis. Async stack traces can be misleading, and having an AI parse them in context saves a lot of head-scratching.

For more on production patterns with the Python SDK, see the Antigravity Python SDK Production Master Guide. If you're working with real-time streaming output rather than batch processing, Python Streaming API Real-Time Display Guide covers that side of things.

Start Small, Measure the Difference

Pick one synchronous loop in your existing code that calls an external API. Convert just that function to async, wrap the call site in asyncio.run(), and time both versions. Seeing the actual speedup — not a benchmark, but your own real workload — makes the complexity of async code feel worthwhile.

Once you've felt the difference on a 50-item batch, you won't want to go back.

Common Pitfalls When Going Async

A few issues come up often when developers first add asyncio to a project that was previously synchronous:

Forgetting to await coroutines. If you call an async function without await, Python won't raise an error — it'll silently return a coroutine object instead of the result. Antigravity's linting integration (via Pylance or Pyright) flags these, but it's easy to miss in a large refactor.

# ❌ Common mistake — returns a coroutine, not the result
result = analyze_single(review, 0)  # Missing await\!
 
# ✅ Correct
result = await analyze_single(review, 0)

Mixing sync and async code in the same call path. If you call asyncio.run() inside an already-running event loop (common in Jupyter notebooks or FastAPI endpoints), you'll get a RuntimeError: This event loop is already running. In those environments, use await directly or rely on the host framework's async runner.

Not handling partial failures in asyncio.gather(). By default, if one task raises an unhandled exception, gather() propagates it immediately and cancels pending tasks. If you want all tasks to complete regardless of individual failures, pass return_exceptions=True:

results = await asyncio.gather(*tasks, return_exceptions=True)
# Results may contain Exception objects — check each one
for i, result in enumerate(results):
    if isinstance(result, Exception):
        print(f"Task {i} failed: {result}")
    else:
        # Process normally
        _, text = result

The pipeline code earlier handles this at the function level (try/except inside each task), which I prefer over return_exceptions=True because it gives you more control over the error data structure.

Underestimating token costs at scale. Concurrent processing makes it easy to burn through API quota quickly. Add a simple counter to track tokens if you're running large batches:

import asyncio
from google.genai.types import GenerateContentResponse
 
total_tokens = 0
token_lock = asyncio.Lock()
 
async def analyze_and_track(semaphore, review, index):
    global total_tokens
    async with semaphore:
        response = await client.aio.models.generate_content(...)
        if response.usage_metadata:
            async with token_lock:
                total_tokens += response.usage_metadata.total_token_count
        return index, response.text.strip()

This pattern keeps token tracking thread-safe without blocking concurrent execution.