Handle Gemini 429 rate limit and 503 service unavailable errors by routing grounding queries to a search API as a fallback, then auto-recovering when Gemini becomes available again. Gemini's grounding feature is useful but unreliable under load: rate limits hit at unpredictable thresholds and 503s can last minutes. A search API fallback ensures your application never loses grounding capability. This tutorial wraps the Gemini call in error detection, routes failures to Scavio, and implements a health check loop that restores Gemini when it recovers.
Prerequisites
- Python 3.8+ installed
- requests library installed
- A Scavio API key from scavio.dev
- Google Gemini API access (or any LLM with grounding)
Walkthrough
Step 1: Wrap the Gemini grounding call
Create a wrapper around your Gemini API call that catches 429 and 503 errors instead of crashing.
import os, requests, time
GEMINI_KEY = os.environ.get('GEMINI_API_KEY', '')
SCAVIO_KEY = os.environ['SCAVIO_API_KEY']
gemini_healthy = True
gemini_last_check = 0
def gemini_ground(query: str) -> dict:
"""Call Gemini with grounding. Returns results or raises on failure."""
try:
resp = requests.post(
f'https://generativelanguage.googleapis.com/v1beta/models/gemini-pro:generateContent?key={GEMINI_KEY}',
json={'contents': [{'parts': [{'text': query}]}]},
timeout=15)
if resp.status_code in (429, 503):
raise requests.exceptions.HTTPError(f'Gemini {resp.status_code}')
resp.raise_for_status()
return {'source': 'gemini', 'data': resp.json()}
except requests.exceptions.RequestException as e:
return {'source': 'gemini', 'error': str(e)}
print('Gemini wrapper ready')Step 2: Detect errors and route to Scavio
When Gemini returns an error, automatically route the grounding query to Scavio's search API.
def scavio_ground(query: str) -> dict:
resp = requests.post('https://api.scavio.dev/api/v1/search',
headers={'x-api-key': SCAVIO_KEY},
json={'platform': 'google', 'query': query}, timeout=10)
resp.raise_for_status()
results = resp.json().get('organic_results', [])[:5]
context = '\n'.join(f"{r['title']}: {r.get('snippet', '')}" for r in results)
return {'source': 'scavio', 'context': context, 'results': results}
def grounded_query(query: str) -> dict:
global gemini_healthy
if gemini_healthy:
result = gemini_ground(query)
if 'error' not in result:
return result
print(f"Gemini failed: {result['error']}. Falling back to Scavio.")
gemini_healthy = False
return scavio_ground(query)
result = grounded_query('best project management tools 2026')
print(f"Grounded via: {result['source']}")Step 3: Route to Scavio fallback
Build the search-based grounding that replaces Gemini's grounding feature with equivalent context from web results.
def format_grounding_context(search_results: list) -> str:
"""Format search results to match the context format Gemini grounding provides."""
if not search_results:
return ''
parts = []
for r in search_results:
title = r.get('title', '')
snippet = r.get('snippet', '')
url = r.get('link', '')
parts.append(f'[{title}]({url}): {snippet}')
return '\n'.join(parts)
def grounded_prompt(query: str) -> str:
result = grounded_query(query)
if result['source'] == 'scavio':
context = format_grounding_context(result.get('results', []))
return f'Use the following web context to answer:\n{context}\n\nQuestion: {query}'
return f'Question: {query}'
print(grounded_prompt('best CRM for startups 2026'))Step 4: Auto-recover when Gemini comes back
Periodically test Gemini's health and restore it as the primary grounding source when it recovers.
RECOVERY_INTERVAL = 60 # seconds between recovery checks
def check_gemini_health() -> bool:
global gemini_healthy, gemini_last_check
now = time.time()
if now - gemini_last_check < RECOVERY_INTERVAL:
return gemini_healthy
gemini_last_check = now
result = gemini_ground('test query')
if 'error' not in result:
gemini_healthy = True
print('Gemini recovered. Restoring as primary.')
else:
print(f'Gemini still down: {result["error"]}')
return gemini_healthy
def smart_ground(query: str) -> dict:
if not gemini_healthy:
check_gemini_health()
return grounded_query(query)
result = smart_ground('latest python release')
print(f"Source: {result['source']}")Python Example
import requests, os
H = {'x-api-key': os.environ['SCAVIO_API_KEY']}
def fallback_ground(query):
# Try Gemini first, fall back to Scavio
try:
# gemini_call(query) # Your Gemini grounding call
raise Exception('Simulated Gemini 429')
except Exception:
data = requests.post('https://api.scavio.dev/api/v1/search', headers=H,
json={'platform': 'google', 'query': query}).json()
results = data.get('organic_results', [])[:3]
return '\n'.join(f"{r['title']}: {r.get('snippet', '')}" for r in results)
print(fallback_ground('best CRM 2026'))JavaScript Example
const H = {'x-api-key': process.env.SCAVIO_API_KEY, 'Content-Type': 'application/json'};
async function fallbackGround(query) {
try {
// await geminiGround(query); // Your Gemini call
throw new Error('Simulated 429');
} catch {
const r = await fetch('https://api.scavio.dev/api/v1/search', {
method: 'POST', headers: H, body: JSON.stringify({platform: 'google', query})
});
const results = (await r.json()).organic_results || [];
return results.slice(0, 3).map(r => `${r.title}: ${r.snippet}`).join('\n');
}
}
fallbackGround('best CRM 2026').then(console.log);Expected Output
A grounding system that uses Gemini when available, automatically falls back to Scavio search on 429/503 errors, and auto-recovers when Gemini is healthy again.