Articles/Integrations

⬡ Integrations/2026-04-17Advanced

Google Antigravity Python SDK Production Masterguide: Multimodal, Agents, and RAG Pipelines from Design to Deployment

The complete guide to using the Google Antigravity Python SDK in production. Covers multimodal input, tool calling, RAG pipelines, streaming, cost optimization, and Cloud Run deployment with working code examples.

python²⁵ antigravity⁴²⁹ sdk⁵ rag⁸ agents¹²³ multimodal⁷ production⁷¹

✦ Premium Article

There's a surprisingly wide gap between "got it working" and "production-ready" when building with Antigravity. I discovered this the hard way when I reviewed the logs of my first chatbot: rate limit errors scattered throughout, conversation history wiped clean between sessions, and file uploads silently failing above a certain size. The code was running. But it wasn't usable.

I'm Masaki Hirokawa (Dolice), an indie developer who has shipped wallpaper and relaxation apps on the App Store and Google Play since 2014, with cumulative downloads now past 50 million. Most of those apps are AdMob-monetized, and bolting AI features ("send an image, get a personalized wallpaper suggestion") onto that AdMob-funded economics is where most of the production lessons in this guide were forged.

This guide covers what you actually need to build apps that hold up in production using the Antigravity Python SDK. I've focused on the specific places I got stuck, walking through multimodal input, agent design, RAG pipelines, and Cloud Run deployment with working code every step of the way.

Understanding the Python SDK Design Philosophy — Why Not Just Call the API Directly?

Let's start with why you'd use the Python SDK instead of calling the REST API directly. Understanding this changes how you read the documentation.

The greatest value the Antigravity Python SDK (google-generativeai package) provides isn't protocol abstraction — it's automated state management. With raw API requests, you have to implement multi-turn conversation history management, streaming response buffering, temporary URL management for file uploads, and tool-calling loop logic all yourself. The SDK handles all of this for you.

Another common source of confusion: there are two separate packages — google-generativeai and vertexai.

# Pattern 1: google-generativeai (for individuals and prototyping)
import google.generativeai as genai
genai.configure(api_key="YOUR_API_KEY")
 
# Pattern 2: vertexai (for Google Cloud and production)
import vertexai
from vertexai.generative_models import GenerativeModel
vertexai.init(project="YOUR_PROJECT_ID", location="us-central1")

Use google-generativeai for quick experimentation with a personal API key. Switch to vertexai when you need to tie things to a Google Cloud project and scale. For production, I recommend the latter — the integration with Cloud Run is seamless, and you get proper IAM-based access control.

This guide primarily uses google-generativeai, but I'll show the migration path to vertexai in the deployment section.

Getting Past Authentication Errors in Under an Hour

Auth is the first hurdle. Here are the patterns that bite most developers.

# Installation
pip install google-generativeai python-dotenv
 
# .env file (must be in .gitignore)
GOOGLE_API_KEY=AIza...your-key-here

import os
from dotenv import load_dotenv
import google.generativeai as genai
 
load_dotenv()
genai.configure(api_key=os.environ["GOOGLE_API_KEY"])
 
# Connectivity test
model = genai.GenerativeModel("gemini-2.0-flash")
response = model.generate_content("Say hello in one sentence.")
print(response.text)
# → "Hello\! I'm an AI assistant built by Google."

Common errors and fixes:

google.api_core.exceptions.PermissionDenied: 403 API key not valid → The API key almost always has a stray space or newline at the beginning or end. Wrap with .strip() or re-check your .env file.

google.api_core.exceptions.ResourceExhausted: 429 Quota exceeded → You've hit the free tier rate limit (requests per minute). The retry implementation in a later section is essential.

google.auth.exceptions.DefaultCredentialsError → Occurs with the vertexai package when ADC (Application Default Credentials) isn't configured. Run gcloud auth application-default login.

✦

Thank you for reading this far.

Continue Reading

What follows includes implementation code, benchmarks, and practical content we hope you'll find useful. This site runs without ads — server and development costs are supported entirely by members like you. If it's been helpful, we'd be truly grateful for your support.

WHAT YOU'LL LEARN

✦Token-budget math (from 50M-download indie dev experience) that keeps Antigravity cost under $0.07 per user per month

✦Production-grade AdMob × AI patterns: two-stage response, finish_reason logging, canary model comparison sheets

✦Three non-negotiable indie monitoring guardrails (100% sampling, Slack alerts on SAFETY, hard cost-threshold fallback)

Secure payment via Stripe · Cancel anytime

✦

Unlock This Article

Get full access to the rest of this article. Buy once, read anytime. This site is ad-free — your support goes directly toward keeping it running.

Unlock all articles with Membership →

Multimodal Input: Handling Text, Images, PDFs, and Video

One of Antigravity's most powerful features is accepting non-text media as input. But there are several input patterns, and using the wrong one leads to silent failures or cost explosions.

Image Input — Inline Data vs File API

import google.generativeai as genai
import time
 
model = genai.GenerativeModel("gemini-2.0-flash")
 
# Method 1: Small images inline as base64 (recommended under 4MB)
with open("screenshot.png", "rb") as f:
    image_data = f.read()
 
response = model.generate_content([
    {"mime_type": "image/png", "data": image_data},
    "List three UI problems in this screenshot."
])
print(response.text)
 
# Method 2: Large files via File API (up to 2GB, valid for 48 hours)
sample_file = genai.upload_file(
    path="large_video.mp4",
    display_name="Analysis target video"
)
 
# Wait for video processing to complete
while sample_file.state.name == "PROCESSING":
    print(".", end="", flush=True)
    time.sleep(2)
    sample_file = genai.get_file(sample_file.name)
 
if sample_file.state.name == "FAILED":
    raise ValueError("File processing failed")
 
response = model.generate_content([
    sample_file,
    "Summarize the key points from this video."
])
print(response.text)
 
# ⚠️ Important: Delete File API files when done
# They auto-delete after 48 hours, but charges accumulate until then
genai.delete_file(sample_file.name)

The most common mistake: Trying to send video using Method 1 (inline) and getting an invalid_argument error. Video must always go through the File API. Same for PDFs larger than 20MB.

PDF Document Analysis

# Upload and analyze a PDF document
pdf_file = genai.upload_file(
    path="technical_spec.pdf",
    display_name="Technical specification"
)
 
# Wait for processing
while pdf_file.state.name == "PROCESSING":
    time.sleep(1)
    pdf_file = genai.get_file(pdf_file.name)
 
response = model.generate_content([
    pdf_file,
    "Extract all API endpoints that need implementation from this spec as JSON. "
    "Format: [{\"endpoint\": \"/api/xxx\", \"method\": \"POST\", \"description\": \"...\"}]"
])
 
import json
try:
    json_start = response.text.index("[")
    json_end = response.text.rindex("]") + 1
    endpoints = json.loads(response.text[json_start:json_end])
    for ep in endpoints:
        print(f"{ep['method']} {ep['endpoint']}: {ep['description']}")
except (ValueError, json.JSONDecodeError) as e:
    print(f"JSON parse error: {e}")
    print("Raw response:", response.text)

Production-Grade Agent Implementation Patterns

Function Calling is the core technology for building practical agents with Antigravity. But beyond tool definitions, production deployments require solid error recovery and state management.

Basic Tool-Calling Implementation

import google.generativeai as genai
import requests
import json
 
# Tool definition (accurate type information is critical)
tools = [
    {
        "function_declarations": [
            {
                "name": "get_weather",
                "description": "Get current weather information for a specified city",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "city": {
                            "type": "string",
                            "description": "City name (e.g., Tokyo, New York)"
                        },
                        "unit": {
                            "type": "string",
                            "enum": ["celsius", "fahrenheit"],
                            "description": "Temperature unit"
                        }
                    },
                    "required": ["city"]
                }
            }
        ]
    }
]
 
def get_weather(city: str, unit: str = "celsius") -> dict:
    """Actual weather API call with error handling"""
    try:
        response = requests.get(
            "https://api.openweathermap.org/data/2.5/weather",
            params={"q": city, "appid": "YOUR_OWM_KEY", "units": "metric"},
            timeout=5  # Timeout is mandatory
        )
        response.raise_for_status()
        data = response.json()
        temp = data["main"]["temp"]
        if unit == "fahrenheit":
            temp = temp * 9/5 + 32
        return {
            "city": city,
            "temperature": round(temp, 1),
            "unit": unit,
            "description": data["weather"][0]["description"],
            "humidity": data["main"]["humidity"]
        }
    except requests.Timeout:
        return {"error": "Weather API timed out. Please try again later."}
    except requests.HTTPError as e:
        return {"error": f"Failed to fetch weather: {e.response.status_code}"}
    except Exception as e:
        return {"error": f"Unexpected error: {str(e)}"}
 
 
def run_agent(user_message: str) -> str:
    """Agent that manages the tool-calling loop"""
    model = genai.GenerativeModel("gemini-2.0-flash", tools=tools)
    messages = [{"role": "user", "parts": [user_message]}]
    
    max_iterations = 5  # Prevent infinite loops
    for iteration in range(max_iterations):
        response = model.generate_content(messages)
        candidate = response.candidates[0]
        
        has_tool_call = False
        tool_results = []
        
        for part in candidate.content.parts:
            if hasattr(part, "function_call") and part.function_call:
                has_tool_call = True
                func_name = part.function_call.name
                func_args = dict(part.function_call.args)
                
                print(f"[Tool Call] {func_name}({func_args})")
                
                if func_name == "get_weather":
                    result = get_weather(**func_args)
                else:
                    result = {"error": f"Unknown tool: {func_name}"}
                
                print(f"[Tool Result] {result}")
                tool_results.append({
                    "function_response": {
                        "name": func_name,
                        "response": result
                    }
                })
        
        if not has_tool_call:
            return candidate.content.parts[0].text
        
        messages.append({"role": "model", "parts": candidate.content.parts})
        messages.append({"role": "user", "parts": tool_results})
    
    return "Maximum iterations reached. Processing halted."
 
result = run_agent("What's the weather in Tokyo in Celsius?")
print(result)

Persisting Conversation History

Session-spanning conversation history is non-negotiable for production apps. In-memory only means every server restart or load balancer routing change wipes the slate clean.

import json
import redis
import google.generativeai as genai
 
class PersistentConversation:
    """Conversation history persistence using Redis"""
    
    def __init__(self, session_id: str, redis_url: str = "redis://localhost:6379"):
        self.session_id = session_id
        self.redis = redis.from_url(redis_url, decode_responses=True)
        self.model = genai.GenerativeModel("gemini-2.0-flash")
        self.ttl = 24 * 60 * 60  # Auto-delete sessions after 24 hours
    
    def _load_history(self) -> list:
        raw = self.redis.get(f"chat:{self.session_id}")
        if raw:
            return json.loads(raw)
        return []
    
    def _save_history(self, history: list) -> None:
        self.redis.setex(
            f"chat:{self.session_id}",
            self.ttl,
            json.dumps(history, ensure_ascii=False)
        )
    
    def send_message(self, user_message: str) -> str:
        history = self._load_history()
        chat = self.model.start_chat(history=history)
        response = chat.send_message(user_message)
        
        updated_history = []
        for message in chat.history:
            updated_history.append({
                "role": message.role,
                "parts": [part.text for part in message.parts if hasattr(part, "text")]
            })
        self._save_history(updated_history)
        
        return response.text
    
    def clear(self) -> None:
        self.redis.delete(f"chat:{self.session_id}")
 
conv = PersistentConversation(session_id="user_12345")
print(conv.send_message("Hi\! My name is Masaki."))
print(conv.send_message("Do you remember my name?"))  # Should reply "Masaki"

Building a RAG Pipeline — From Vector Search to Response Generation

Retrieval-Augmented Generation (RAG) is the most practical way to integrate your own data with Antigravity. The key to getting good results is understanding exactly where precision degrades.

Embeddings and Vector Search

import google.generativeai as genai
import numpy as np
from typing import List, Tuple
 
def embed_text(text: str) -> list[float]:
    """Convert text to a vector embedding"""
    result = genai.embed_content(
        model="models/text-embedding-004",
        content=text,
        task_type="retrieval_document"  # For documents
    )
    return result["embedding"]
 
def embed_query(query: str) -> list[float]:
    """Convert a search query to a vector (different task_type than documents)"""
    result = genai.embed_content(
        model="models/text-embedding-004",
        content=query,
        task_type="retrieval_query"  # For queries — wrong type here kills accuracy
    )
    return result["embedding"]
 
def cosine_similarity(a: list, b: list) -> float:
    a_np, b_np = np.array(a), np.array(b)
    return float(np.dot(a_np, b_np) / (np.linalg.norm(a_np) * np.linalg.norm(b_np)))
 
class SimpleVectorStore:
    def __init__(self):
        self.documents: List[dict] = []
    
    def add(self, text: str, metadata: dict = None) -> None:
        embedding = embed_text(text)
        self.documents.append({
            "text": text,
            "embedding": embedding,
            "metadata": metadata or {}
        })
    
    def search(self, query: str, top_k: int = 3) -> List[Tuple[str, float, dict]]:
        query_embedding = embed_query(query)
        results = []
        for doc in self.documents:
            score = cosine_similarity(query_embedding, doc["embedding"])
            results.append((doc["text"], score, doc["metadata"]))
        results.sort(key=lambda x: x[1], reverse=True)
        return results[:top_k]
 
def rag_query(store: SimpleVectorStore, question: str) -> str:
    results = store.search(question, top_k=3)
    
    # Filter out low-quality matches (skipping this causes hallucinations)
    relevant = [(text, score) for text, score, _ in results if score > 0.6]
    
    if not relevant:
        return "No relevant information found. Try a more specific question."
    
    context_parts = []
    for i, (text, score) in enumerate(relevant, 1):
        context_parts.append(f"[Reference {i} (relevance: {score:.2f})]\n{text}")
    context = "\n\n".join(context_parts)
    
    model = genai.GenerativeModel("gemini-2.0-flash")
    prompt = f"""Answer the question based on the references below.
If the answer isn't in the references, say "This information isn't in the provided documents."
 
References:
{context}
 
Question: {question}
 
Answer:"""
    
    response = model.generate_content(prompt)
    return response.text
 
store = SimpleVectorStore()
store.add("Antigravity is an AI development environment released by Google in 2025.", {"source": "intro.md"})
store.add("Antigravity's premium plan costs $19.99 per month.", {"source": "pricing.md"})
store.add("Antigravity provides code completion powered by Gemini 2.0.", {"source": "features.md"})
 
print(rag_query(store, "How much does Antigravity cost?"))

A critical design decision: Using retrieval_document for documents and retrieval_query for queries is mandatory. If you embed your queries with retrieval_document, search accuracy drops noticeably. This is in the official docs, but it's easy to miss.

For production deployments, choose between ChromaDB (lightweight, local), pgvector (PostgreSQL integration), or Vertex AI Search (fully managed GCP) based on your requirements.

Streaming Responses and Real-Time Processing

To deliver the experience of text appearing character by character in a chat UI, you need streaming. Here's a FastAPI + WebSocket implementation:

import asyncio
import google.generativeai as genai
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
 
app = FastAPI()
model = genai.GenerativeModel("gemini-2.0-flash")
 
@app.websocket("/ws/chat")
async def websocket_chat(websocket: WebSocket):
    await websocket.accept()
    
    try:
        while True:
            message = await websocket.receive_text()
            
            # Stream generation (run sync SDK in thread pool to avoid blocking)
            response_stream = await asyncio.to_thread(
                model.generate_content,
                message,
                stream=True
            )
            
            full_response = ""
            for chunk in response_stream:
                if chunk.text:
                    full_response += chunk.text
                    await websocket.send_json({
                        "type": "chunk",
                        "text": chunk.text
                    })
            
            await websocket.send_json({
                "type": "done",
                "full_text": full_response
            })
    
    except WebSocketDisconnect:
        print("Client disconnected")
    except Exception as e:
        await websocket.send_json({"type": "error", "message": str(e)})
        await websocket.close()

Rate Limiting and Exponential Backoff Retry

import time
import random
from functools import wraps
from google.api_core.exceptions import ResourceExhausted, ServiceUnavailable
 
def with_retry(max_retries: int = 3, base_delay: float = 1.0):
    """Retry decorator with exponential backoff"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except (ResourceExhausted, ServiceUnavailable) as e:
                    if attempt == max_retries - 1:
                        raise
                    delay = base_delay * (2 ** attempt) + random.uniform(0, 1)
                    print(f"API error ({type(e).__name__}): retrying in {delay:.1f}s... ({attempt + 1}/{max_retries})")
                    time.sleep(delay)
        return wrapper
    return decorator
 
model = genai.GenerativeModel("gemini-2.0-flash")
 
@with_retry(max_retries=3)
def safe_generate(prompt: str) -> str:
    response = model.generate_content(prompt)
    return response.text

Cost Optimization and Monitoring

Controlling API costs for indie development requires usage visibility and intentional caching strategy.

Token Usage Tracking

from dataclasses import dataclass, field
import time
 
@dataclass
class TokenUsageTracker:
    input_tokens: int = 0
    output_tokens: int = 0
    total_requests: int = 0
    start_time: float = field(default_factory=time.time)
    
    # Gemini 2.0 Flash pricing (as of April 2026)
    INPUT_COST_PER_1M = 0.075   # USD / 1M tokens
    OUTPUT_COST_PER_1M = 0.30   # USD / 1M tokens
    
    def record(self, response) -> None:
        if hasattr(response, "usage_metadata"):
            self.input_tokens += response.usage_metadata.prompt_token_count
            self.output_tokens += response.usage_metadata.candidates_token_count
        self.total_requests += 1
    
    def estimated_cost_usd(self) -> float:
        input_cost = (self.input_tokens / 1_000_000) * self.INPUT_COST_PER_1M
        output_cost = (self.output_tokens / 1_000_000) * self.OUTPUT_COST_PER_1M
        return input_cost + output_cost
    
    def report(self) -> str:
        elapsed = time.time() - self.start_time
        return (
            f"=== Token Usage Report ===\n"
            f"Input tokens: {self.input_tokens:,}\n"
            f"Output tokens: {self.output_tokens:,}\n"
            f"Total requests: {self.total_requests}\n"
            f"Estimated cost: ${self.estimated_cost_usd():.4f} USD\n"
            f"Elapsed: {elapsed:.1f}s\n"
            f"Avg cost/request: ${self.estimated_cost_usd() / max(1, self.total_requests):.5f}"
        )
 
tracker = TokenUsageTracker()
response = model.generate_content("Explain Python list comprehensions.")
tracker.record(response)
print(tracker.report())

A common money sink: Including the same system prompt with every request wastes input tokens at scale. The Context Caching API lets you cache frequently-used system prompts and documents, dramatically reducing costs on subsequent requests.

Production Deployment — Moving to Cloud Run

Here's what actually matters when taking a working local app to Cloud Run.

Dockerfile and Requirements

FROM python:3.12-slim
 
WORKDIR /app
 
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
 
COPY . .
 
# Cloud Run expects port 8080
ENV PORT=8080
EXPOSE 8080
 
CMD ["gunicorn", "main:app", "-w", "2", "-k", "uvicorn.workers.UvicornWorker", "--bind", "0.0.0.0:8080"]

Managing Secrets with Secret Manager

# Register API key in Secret Manager
echo -n "YOUR_GOOGLE_API_KEY" | gcloud secrets create GOOGLE_API_KEY --data-file=-
 
# Deploy to Cloud Run with secret mounting
gcloud run deploy my-ai-app \
  --image gcr.io/YOUR_PROJECT_ID/my-ai-app \
  --region us-central1 \
  --set-secrets "GOOGLE_API_KEY=GOOGLE_API_KEY:latest" \
  --memory 1Gi \
  --cpu 1 \
  --concurrency 80 \
  --max-instances 10 \
  --allow-unauthenticated

# Reading the API key in Cloud Run (mounted as env var by Secret Manager)
import os
import google.generativeai as genai
 
api_key = os.environ.get("GOOGLE_API_KEY")
if not api_key:
    raise EnvironmentError("GOOGLE_API_KEY environment variable not set")
 
genai.configure(api_key=api_key)

Cost reality check: Cloud Run is serverless — zero cost when idle. For early-stage indie projects, monthly bills typically stay under a few dollars. Once you scale up, migrating to the vertexai package with IAM authentication eliminates the API key entirely, which is both more secure and more manageable at scale.

Common Mistakes and Pitfalls in Production

Here are the recurring error patterns I've encountered running the Antigravity Python SDK in production.

1. Ignoring the Safety Filter Response

# ❌ Catching all exceptions without handling safety specifically
try:
    response = model.generate_content(user_message)
    return response.text
except Exception:
    return "An error occurred"  # Root cause is hidden
 
# ✅ Handle finish_reason explicitly
from google.generativeai.types import StopCandidateException
 
try:
    response = model.generate_content(user_message)
    if response.candidates and response.candidates[0].finish_reason == "STOP":
        return response.text
    elif response.candidates and response.candidates[0].finish_reason == "SAFETY":
        return "This content could not be processed due to Antigravity's safety guidelines."
    else:
        finish_reason = response.candidates[0].finish_reason if response.candidates else "no candidates"
        return f"Unexpected finish reason: {finish_reason}"
except StopCandidateException:
    return "Processing interrupted by the content safety filter."

2. Blocking the Event Loop with Synchronous SDK Calls

# ❌ Calling the synchronous SDK directly inside an async FastAPI handler
@app.get("/generate")
async def generate_bad(prompt: str):
    response = model.generate_content(prompt)  # Blocks the event loop
    return {"text": response.text}
 
# ✅ Offload to thread pool with asyncio.to_thread
import asyncio
 
@app.get("/generate")
async def generate_good(prompt: str):
    response = await asyncio.to_thread(model.generate_content, prompt)
    return {"text": response.text}

3. Context Window Overflow with Long Documents

Gemini 2.0 Flash offers a 1M token context window, but stuffing massive amounts of documents doesn't linearly improve quality — the "lost in the middle" problem (where information in the middle of a long context gets overlooked) becomes more severe.

Always check token count before processing:

def count_tokens(text: str) -> int:
    response = model.count_tokens(text)
    return response.total_tokens
 
MAX_CONTEXT_TOKENS = 100_000  # Safe margin
document = "Very long document..."
token_count = count_tokens(document)
 
if token_count > MAX_CONTEXT_TOKENS:
    print(f"⚠️ Token count {token_count:,} exceeds limit. Processing with RAG.")
    # → Route to RAG pipeline
else:
    response = model.generate_content([document, "Summarize this document."])

Choosing the Right Vector Database for Production RAG

For small to medium projects, the in-memory store shown earlier works fine for prototyping. Once you hit production, you need a real vector database. Here is how I think about the choice.

ChromaDB is the right call when you want to get something deployed quickly without adding cloud infrastructure. It runs in-process or as a standalone server, and the Python API is nearly identical to the simple store above. The limitation is that you manage scaling yourself.

import chromadb
import google.generativeai as genai
 
# Initialize ChromaDB (persistent mode — survives restarts)
chroma_client = chromadb.PersistentClient(path="./chroma_data")
collection = chroma_client.get_or_create_collection(
    name="knowledge_base",
    metadata={"hnsw:space": "cosine"}
)
 
def add_documents_to_chroma(texts: list[str], ids: list[str]) -> None:
    """Add multiple documents to ChromaDB with Antigravity embeddings"""
    embeddings = []
    for text in texts:
        result = genai.embed_content(
            model="models/text-embedding-004",
            content=text,
            task_type="retrieval_document"
        )
        embeddings.append(result["embedding"])
    
    collection.add(
        documents=texts,
        embeddings=embeddings,
        ids=ids
    )
    print(f"Added {len(texts)} documents to ChromaDB")
 
def query_chroma(query: str, n_results: int = 3) -> list[dict]:
    """Search ChromaDB with a natural language query"""
    query_embedding = genai.embed_content(
        model="models/text-embedding-004",
        content=query,
        task_type="retrieval_query"
    )["embedding"]
    
    results = collection.query(
        query_embeddings=[query_embedding],
        n_results=n_results,
        include=["documents", "distances"]
    )
    
    output = []
    for doc, distance in zip(results["documents"][0], results["distances"][0]):
        # ChromaDB returns L2 distance by default, convert to similarity score
        similarity = 1 - (distance / 2)  # Rough conversion for cosine space
        output.append({"text": doc, "score": similarity})
    
    return output
 
# Example workflow
add_documents_to_chroma(
    texts=[
        "Antigravity supports Gemini 2.0 Flash and Gemini 2.0 Pro models.",
        "The Context Caching API reduces costs for repeated system prompts.",
        "File API supports uploads up to 2GB, valid for 48 hours."
    ],
    ids=["doc1", "doc2", "doc3"]
)
 
results = query_chroma("How long are uploaded files available?")
for r in results:
    print(f"[{r['score']:.2f}] {r['text']}")

pgvector is the right call when you are already on PostgreSQL. Adding vector search to an existing database keeps your infrastructure simple — one less service to operate. The asyncpg driver combined with pgvector makes it easy to integrate into async FastAPI applications.

Vertex AI Search is worth the setup cost when you need Google-managed scaling, built-in access control, and the ability to search across structured and unstructured data in a single query. For projects that already live on Google Cloud, it often saves more engineering time than it costs.

Implementing Structured Output with Response Schema

For applications that need Antigravity to return reliably parseable data rather than free-form text, the response schema feature is transformative. It eliminates the fragile "extract JSON from the response text" pattern.

import google.generativeai as genai
import json
 
# Define the expected output schema
response_schema = {
    "type": "object",
    "properties": {
        "summary": {
            "type": "string",
            "description": "A 2-3 sentence summary of the code"
        },
        "language": {
            "type": "string",
            "enum": ["python", "javascript", "typescript", "go", "rust", "other"]
        },
        "complexity": {
            "type": "string",
            "enum": ["low", "medium", "high"]
        },
        "issues": {
            "type": "array",
            "items": {
                "type": "object",
                "properties": {
                    "line": {"type": "integer"},
                    "severity": {"type": "string", "enum": ["error", "warning", "info"]},
                    "message": {"type": "string"}
                },
                "required": ["severity", "message"]
            }
        },
        "suggestions": {
            "type": "array",
            "items": {"type": "string"}
        }
    },
    "required": ["summary", "language", "complexity", "issues", "suggestions"]
}
 
def analyze_code(code: str) -> dict:
    """Analyze code and return structured results — no JSON parsing needed"""
    model = genai.GenerativeModel(
        "gemini-2.0-flash",
        generation_config=genai.GenerationConfig(
            response_mime_type="application/json",
            response_schema=response_schema
        )
    )
    
    prompt = f"""Analyze the following code for quality, issues, and improvement opportunities.

{code}

    
    response = model.generate_content(prompt)
    
    # Response is guaranteed to be valid JSON matching the schema
    result = json.loads(response.text)
    return result
 
# Example usage
sample_code = """
def calculate_discount(price, discount):
    return price - price * discount / 100
"""
 
analysis = analyze_code(sample_code)
print(f"Language: {analysis['language']}")
print(f"Complexity: {analysis['complexity']}")
print(f"Summary: {analysis['summary']}")
if analysis['issues']:
    print("Issues found:")
    for issue in analysis['issues']:
        print(f"  [{issue['severity'].upper()}] {issue['message']}")
print("Suggestions:")
for s in analysis['suggestions']:
    print(f"  - {s}")

This pattern is particularly powerful for building developer tools and data pipelines. When Antigravity knows it must conform to a schema, it structures its reasoning to produce valid output — and you eliminate an entire category of runtime errors from your application.

Batch Processing with Concurrent Requests

When you need to process hundreds or thousands of items, sequential API calls are too slow. The Antigravity API supports concurrent requests, and the Python asyncio library makes it straightforward to saturate your quota safely.

import asyncio
import google.generativeai as genai
from typing import List, Tuple
import time
 
model = genai.GenerativeModel("gemini-2.0-flash")
 
async def process_single(item_id: str, text: str, semaphore: asyncio.Semaphore) -> Tuple[str, str]:
    """Process a single item with concurrency control"""
    async with semaphore:
        try:
            response = await asyncio.to_thread(
                model.generate_content,
                f"Classify the sentiment of this text as positive, negative, or neutral. "
                f"Reply with only one word.
 
Text: {text}"
            )
            return item_id, response.text.strip().lower()
        except Exception as e:
            return item_id, f"error: {str(e)}"
 
async def batch_process(items: List[Tuple[str, str]], max_concurrent: int = 5) -> dict:
    """Process multiple items concurrently with rate limit protection"""
    semaphore = asyncio.Semaphore(max_concurrent)
    
    tasks = [
        process_single(item_id, text, semaphore)
        for item_id, text in items
    ]
    
    start = time.time()
    results = await asyncio.gather(*tasks)
    elapsed = time.time() - start
    
    print(f"Processed {len(items)} items in {elapsed:.1f}s "
          f"({len(items)/elapsed:.1f} items/sec)")
    
    return dict(results)
 
# Example: sentiment analysis on product reviews
reviews = [
    ("review_001", "This product exceeded my expectations. Highly recommended\!"),
    ("review_002", "Terrible quality. Broke after two days of use."),
    ("review_003", "Average product. Does what it says, nothing more."),
    ("review_004", "Fantastic value for the price. Will buy again."),
    ("review_005", "Packaging was damaged on arrival but product works fine."),
]
 
results = asyncio.run(batch_process(reviews, max_concurrent=3))
for review_id, sentiment in results.items():
    print(f"{review_id}: {sentiment}")

Start with max_concurrent=3 and increase carefully while monitoring for 429 errors. The sweet spot depends on your API tier — free tier typically handles 2-3 concurrent requests safely, while paid tiers allow significantly higher throughput.

Migrating from google-generativeai to vertexai for Production

When your project outgrows the API-key model, migrating to the vertexai package gives you service account authentication, better cost attribution, and first-class support for enterprise features like VPC Service Controls.

The migration is more mechanical than complex, but a few pieces require attention.

# Before (google-generativeai)
import google.generativeai as genai
genai.configure(api_key=os.environ["GOOGLE_API_KEY"])
model = genai.GenerativeModel("gemini-2.0-flash")
response = model.generate_content("Hello")
print(response.text)
 
# After (vertexai)
import vertexai
from vertexai.generative_models import GenerativeModel
 
vertexai.init(
    project=os.environ["GOOGLE_CLOUD_PROJECT"],
    location="us-central1"
    # No API key needed — uses ADC or service account automatically
)
model = GenerativeModel("gemini-2.0-flash-001")  # Note: version suffix required
response = model.generate_content("Hello")
print(response.text)

The three things that catch developers off guard during migration:

First, model identifiers include a version suffix in the Vertex AI API. gemini-2.0-flash becomes gemini-2.0-flash-001. Keeping a config file with model names avoids hunting for this across your codebase.

Second, the embedding call changes signature:

# google-generativeai
result = genai.embed_content(
    model="models/text-embedding-004",
    content=text,
    task_type="retrieval_document"
)
embedding = result["embedding"]
 
# vertexai
from vertexai.language_models import TextEmbeddingModel
embed_model = TextEmbeddingModel.from_pretrained("text-embedding-004")
embeddings = embed_model.get_embeddings([text])
embedding = embeddings[0].values

Third, the File API for multimodal uploads is not available in the Vertex AI SDK in the same form. For production multimodal pipelines on Vertex AI, you upload files to Google Cloud Storage first, then reference them by GCS URI:

from vertexai.generative_models import GenerativeModel, Part
 
model = GenerativeModel("gemini-2.0-flash-001")
 
# Reference a file in GCS instead of uploading via File API
response = model.generate_content([
    Part.from_uri(
        uri="gs://your-bucket/your-video.mp4",
        mime_type="video/mp4"
    ),
    "Summarize the key points from this video."
])
print(response.text)

The benefit of this pattern is that your GCS bucket persists independently of any 48-hour expiry window, and you can reuse the same file across multiple model calls without re-uploading.

Lessons From the Trenches — What the Docs Won't Tell You

Every pattern I've shown so far is something I run in production. But there are operational gotchas you simply don't see by reading the docs. Coming from a decade-plus of indie development on App Store and Google Play, with AdMob as the primary revenue stream, here's what I always check before flipping the switch on a new AI feature.

1. Compute Per-User Cost Before You Ship

API cost only makes sense per user, not in aggregate. My wallpaper apps sit at around 50 million cumulative downloads and serve a few million monthly actives. Drop unmetered Antigravity API calls into that pipeline and the AdMob eCPM can't possibly cover it.

The rule of thumb I use is: keep AI feature cost under $0.07 per user per month. At Gemini 2.0 Flash pricing (input $0.075 / 1M tokens, output $0.30 / 1M tokens), the math looks like this:

# Budget per user per month
BUDGET_USD_PER_USER_MONTH = 0.067
INPUT_COST_PER_TOKEN = 0.075 / 1_000_000
OUTPUT_COST_PER_TOKEN = 0.30 / 1_000_000
 
def budget_tokens(budget_usd: float) -> dict:
    # Assume 3:1 input/output ratio
    denom = INPUT_COST_PER_TOKEN + OUTPUT_COST_PER_TOKEN / 3
    input_tokens = budget_usd / denom
    output_tokens = input_tokens / 3
    return {
        "input_tokens": int(input_tokens),
        "output_tokens": int(output_tokens),
        "approx_requests_at_2k_each": int(input_tokens / 2_000),
    }
 
print(budget_tokens(BUDGET_USD_PER_USER_MONTH))
# → {'input_tokens': 446666, 'output_tokens': 148888, 'approx_requests_at_2k_each': 223}

That's about 223 requests per user per month, or 7 per day. Cross that line and AdMob revenue can't carry the cost — you'll need a membership tier or another revenue model. Doing this math before you ship saves you from having to disable the feature later.

2. Eliminate "Silent Failures" From Your Logs

The most painful thing in my first prototype was that safety filters, timeouts, and token overruns would frequently return empty strings instead of raising exceptions. To the user it looked like "nothing happened," and the App Store reviews said exactly that. For an AdMob-funded app, one-star reviews wreck your per-user economics.

In production I never accept an empty response without logging the finish reason:

import logging
 
logger = logging.getLogger("antigravity")
 
def generate_with_finish_log(model, prompt, user_id: str):
    """Never let a non-STOP finish slip past unlogged."""
    response = model.generate_content(prompt)
    candidates = response.candidates or []
    if not candidates:
        logger.warning(
            "empty_candidates user=%s prompt_hash=%s",
            user_id, hash(prompt) % 10_000_000,
        )
        return None, "no_candidates"
    finish = str(candidates[0].finish_reason)
    if finish != "STOP":
        logger.warning(
            "non_stop_finish user=%s finish=%s prompt_hash=%s",
            user_id, finish, hash(prompt) % 10_000_000,
        )
    text = response.text if finish == "STOP" else None
    return text, finish

I pipe these finish_reason events to BigQuery and track the daily rate of SAFETY versus STOP. My personal rule is that anything above 0.5% SAFETY means the prompt design is wrong, but more often it means client-side validation should reject the user input (low-resolution image, etc.) before it ever hits Antigravity.

3. Cloud Run Cold Starts Clash With AdMob Timing

Cloud Run's "zero cost when idle" pricing is appealing, but the 3–5 second cold start is brutal when paired with AdMob interstitials. The user closes the ad expecting your AI response, and instead sees a loading spinner. Bounce rates spike.

My production pattern is a two-stage response: a sub-250ms "intent hint" using a lightweight classifier, then the real Antigravity response streamed via Server-Sent Events.

# Server: split fast acknowledgment from the slow real response
@app.post("/chat")
async def chat(req: ChatRequest):
    # Return within 250ms with a lightweight intent classification
    quick_hint = generate_quick_hint(req.message)
    return StreamingResponse(
        stream_full_response(req.message, quick_hint),
        media_type="text/event-stream",
    )

Client-side, the moment quick_hint arrives, the UI updates with something like "Looks like you're asking about X — generating a response…" Perceived latency drops dramatically, and the AdMob impression window stays clean. It's not strictly the SDK's job, but in a production indie-app context you have to design for it.

4. Externalize Model Choice With Env Vars + Flags

The Gemini model lineup turns over every six months or so. Hard-coding a model name into production code means you'll redeploy every time you want to evaluate a new one. I always wrap model selection:

import os
 
MODEL_NAME = os.environ.get("ANTIGRAVITY_MODEL", "gemini-2.0-flash")
MODEL_FALLBACK = os.environ.get("ANTIGRAVITY_MODEL_FALLBACK", "gemini-1.5-flash")
 
def get_model(name: str | None = None):
    name = name or MODEL_NAME
    try:
        return genai.GenerativeModel(name)
    except Exception as e:
        logger.warning("model_init_failed name=%s err=%s; falling back", name, e)
        return genai.GenerativeModel(MODEL_FALLBACK)

When a new model lands, I shift 1–5% of traffic to it via the env var and compare finish_reason distribution, p95 latency, and cost per request in parallel. I keep a "canary comparison" sheet in Google Sheets and watch for at least 48 hours before rolling out broadly.

5. Indie Monitoring Floor — Three Non-Negotiables

This is more ops than code, but as an indie developer you can't afford to find out about a production issue from a user. Three things I always wire up before launch:

Cloud Logging at 100% sampling initially. Reduce only when traffic grows.
Force a Slack notification (5-minute aggregation) on SAFETY and RESOURCE_EXHAUSTED.
Hard threshold on hourly estimated cost — when crossed, fall back to a cheaper Flash model automatically.

Only after these three are in place do I consider the SDK "production-ready" in the indie-developer sense. The fancy code patterns matter, but it's these guardrails that let you sleep at night.

Your Next Step

The walls that stop most developers using the Antigravity Python SDK in production are authentication and rate limits. Once you clear those, the combination of function calling and streaming opens up remarkably expressive application possibilities.

What I'd suggest trying today: pick one concrete use case — maybe RAG question-answering over your own documents — and combine the code from this guide to get it running locally. Once it's working, start recording token usage from day one. Costs scale faster than you expect when you start testing in earnest.

If you're thinking about stepping up to multi-agent systems, the Antigravity Multi-Agent Orchestration Advanced Guide is a practical starting point for that next step.

Thank You for Reading

Antigravity Lab is ad-free, supported entirely by members like you. We publish practical guides daily with implementation code, benchmarks, and production-ready patterns. If you've found it useful, we'd love to have you on board.