AI agents that depend on a single search provider are fragile. Rate limits hit, Cloudflare blocks trigger, and outages happen. A fallback chain tries multiple providers in priority order so your agent always gets data. This tutorial builds a production fallback chain with Scavio as primary, configurable retries, latency tracking, and automatic provider rotation. Each Scavio request costs $0.005 per credit with 6 platforms available from one endpoint.
Prerequisites
- Python 3.9+ installed
- requests library installed
- A Scavio API key from scavio.dev
- Optional: backup provider keys for full fallback coverage
Walkthrough
Step 1: Define the provider interface
Create a base class that every search provider must implement. This ensures uniform result format regardless of which provider responds.
Python
import os, time, requests
from dataclasses import dataclass
from typing import Optional
@dataclass
class SearchResult:
title: str
url: str
snippet: str
provider: str
class SearchProvider:
name: str
def search(self, query: str, num: int = 5) -> list[SearchResult]:
raise NotImplementedError
class ScavioProvider(SearchProvider):
name = 'scavio'
def __init__(self):
self.key = os.environ['SCAVIO_API_KEY']
def search(self, query: str, num: int = 5) -> list[SearchResult]:
resp = requests.post('https://api.scavio.dev/api/v1/search',
headers={'x-api-key': self.key, 'Content-Type': 'application/json'},
json={'query': query, 'country_code': 'us', 'num_results': num},
timeout=10)
resp.raise_for_status()
return [SearchResult(title=r['title'], url=r['link'],
snippet=r.get('snippet', ''), provider='scavio')
for r in resp.json().get('organic_results', [])]
print('Provider interface ready')Step 2: Build the fallback chain
The chain tries each provider in order. If one fails or returns empty results, it moves to the next. It tracks which provider succeeded and the latency.
Python
class FallbackChain:
def __init__(self, providers: list[SearchProvider]):
self.providers = providers
self.stats = {p.name: {'success': 0, 'fail': 0, 'total_ms': 0} for p in providers}
def search(self, query: str, num: int = 5) -> tuple[list[SearchResult], str]:
for provider in self.providers:
try:
start = time.time()
results = provider.search(query, num)
elapsed = (time.time() - start) * 1000
if results:
self.stats[provider.name]['success'] += 1
self.stats[provider.name]['total_ms'] += elapsed
return results, provider.name
except Exception as e:
self.stats[provider.name]['fail'] += 1
print(f'[fallback] {provider.name} failed: {e}')
continue
return [], 'none'
def report(self) -> str:
lines = ['Provider Stats:']
for name, s in self.stats.items():
total = s['success'] + s['fail']
avg_ms = s['total_ms'] / s['success'] if s['success'] else 0
lines.append(f" {name}: {s['success']}/{total} ok, {avg_ms:.0f}ms avg")
return '\n'.join(lines)
chain = FallbackChain([ScavioProvider()])
results, used = chain.search('python web framework 2026')
print(f'Got {len(results)} results from {used}')Step 3: Add retry logic with exponential backoff
Wrap each provider call with retry logic. Transient failures like timeouts should be retried before falling through to the next provider.
Python
class RetryProvider:
def __init__(self, provider: SearchProvider, max_retries: int = 2, base_delay: float = 0.5):
self.provider = provider
self.name = provider.name
self.max_retries = max_retries
self.base_delay = base_delay
def search(self, query: str, num: int = 5) -> list[SearchResult]:
last_error = None
for attempt in range(self.max_retries + 1):
try:
return self.provider.search(query, num)
except requests.exceptions.Timeout:
last_error = 'timeout'
delay = self.base_delay * (2 ** attempt)
print(f'[retry] {self.name} timeout, waiting {delay:.1f}s')
time.sleep(delay)
except requests.exceptions.HTTPError as e:
if e.response and e.response.status_code == 429:
delay = self.base_delay * (2 ** attempt)
print(f'[retry] {self.name} rate limited, waiting {delay:.1f}s')
time.sleep(delay)
last_error = 'rate_limit'
else:
raise
raise Exception(f'{self.name} failed after {self.max_retries} retries: {last_error}')
chain = FallbackChain([RetryProvider(ScavioProvider())])
results, used = chain.search('AI agent frameworks')
print(f'{len(results)} results via {used}')Step 4: Add health checking and automatic rotation
Track provider health and automatically deprioritize unhealthy providers. Periodically recheck them in case they recover.
Python
class HealthAwareFallback(FallbackChain):
def __init__(self, providers, unhealthy_threshold=3, recheck_interval=60):
super().__init__(providers)
self.unhealthy_threshold = unhealthy_threshold
self.recheck_interval = recheck_interval
self.consecutive_fails = {p.name: 0 for p in providers}
self.last_recheck = {p.name: 0 for p in providers}
def search(self, query: str, num: int = 5) -> tuple[list[SearchResult], str]:
now = time.time()
for provider in self.providers:
name = provider.name if hasattr(provider, 'name') else str(provider)
# Skip unhealthy providers unless recheck interval passed
if self.consecutive_fails.get(name, 0) >= self.unhealthy_threshold:
if now - self.last_recheck.get(name, 0) < self.recheck_interval:
continue
self.last_recheck[name] = now
print(f'[health] rechecking {name}')
try:
start = time.time()
results = provider.search(query, num)
elapsed = (time.time() - start) * 1000
if results:
self.consecutive_fails[name] = 0
self.stats[name]['success'] += 1
self.stats[name]['total_ms'] += elapsed
return results, name
except Exception as e:
self.consecutive_fails[name] = self.consecutive_fails.get(name, 0) + 1
self.stats[name]['fail'] += 1
print(f'[health] {name} fail #{self.consecutive_fails[name]}: {e}')
return [], 'none'
chain = HealthAwareFallback([RetryProvider(ScavioProvider())])
results, used = chain.search('search api comparison')
print(f'{len(results)} results via {used}')
print(chain.report())Python Example
Python
import os, time, requests
SCAVIO_KEY = os.environ['SCAVIO_API_KEY']
def scavio_search(query, num=5):
resp = requests.post('https://api.scavio.dev/api/v1/search',
headers={'x-api-key': SCAVIO_KEY, 'Content-Type': 'application/json'},
json={'query': query, 'country_code': 'us', 'num_results': num}, timeout=10)
resp.raise_for_status()
return [{'title': r['title'], 'url': r['link'], 'snippet': r.get('snippet', '')}
for r in resp.json().get('organic_results', [])]
def search_with_fallback(query, num=5, retries=2):
for attempt in range(retries + 1):
try:
return scavio_search(query, num)
except Exception as e:
if attempt < retries:
time.sleep(0.5 * (2 ** attempt))
else:
raise
results = search_with_fallback('AI agent search tools 2026')
for r in results:
print(f"{r['title']}\n {r['url']}")JavaScript Example
JavaScript
const SCAVIO_KEY = process.env.SCAVIO_API_KEY;
async function search(query, num = 5, retries = 2) {
for (let i = 0; i <= retries; i++) {
try {
const resp = await fetch('https://api.scavio.dev/api/v1/search', {
method: 'POST',
headers: { 'x-api-key': SCAVIO_KEY, 'Content-Type': 'application/json' },
body: JSON.stringify({ query, country_code: 'us', num_results: num }),
signal: AbortSignal.timeout(10000)
});
if (!resp.ok) throw new Error(`HTTP ${resp.status}`);
const data = await resp.json();
return (data.organic_results || []).map(r => ({ title: r.title, url: r.link, snippet: r.snippet || '' }));
} catch (e) {
if (i < retries) await new Promise(r => setTimeout(r, 500 * 2 ** i));
else throw e;
}
}
}
search('AI agent search tools 2026').then(r => r.forEach(x => console.log(x.title)));Expected Output
JSON
Got 5 results from scavio
Provider Stats:
scavio: 1/1 ok, 320ms avg
AI Agent Frameworks Comparison 2026
https://example.com/ai-agent-frameworks
Building Reliable AI Agents with Search
https://blog.example.com/reliable-agents