Production agents break when a single tool call fails. A fallback chain ensures your agent always returns useful results by cascading through primary search, a secondary source, and locally cached data. This tutorial builds a three-tier fallback chain using the Scavio Search API as the primary source, with a secondary re-rank pass and a local cache as the final safety net.
Prerequisites
- Python 3.11+ or Node.js 20+
- A Scavio API key from https://scavio.dev
- Basic understanding of try/catch error handling
- A local SQLite or JSON file for caching
Walkthrough
Step 1: Set up the primary search call
Make the initial search request to the Scavio API. If this succeeds, return results directly. Wrap the call in error handling so failures cascade to the next tier.
Python
import httpx
import json
from pathlib import Path
SCAVIO_API_KEY = "your-api-key"
CACHE_FILE = Path("search_cache.json")
async def primary_search(query: str) -> dict:
"""Tier 1: Live Scavio search."""
async with httpx.AsyncClient(timeout=10) as client:
resp = await client.post(
"https://api.scavio.dev/api/v1/search",
headers={"x-api-key": SCAVIO_API_KEY},
json={"query": query, "num_results": 10}
)
resp.raise_for_status()
return resp.json()Step 2: Add a secondary search with re-ranking
If the primary call returns fewer than three results, fire a broader query and re-rank locally. This catches edge cases where the original query was too narrow.
Python
async def secondary_search(query: str) -> dict:
"""Tier 2: Broader query with local re-rank."""
broad_query = " ".join(query.split()[:4]) # simplify query
async with httpx.AsyncClient(timeout=10) as client:
resp = await client.post(
"https://api.scavio.dev/api/v1/search",
headers={"x-api-key": SCAVIO_API_KEY},
json={"query": broad_query, "num_results": 20}
)
resp.raise_for_status()
data = resp.json()
# Re-rank: prefer results containing original query terms
terms = set(query.lower().split())
results = data.get("results", [])
scored = []
for r in results:
title_lower = r.get("title", "").lower()
score = sum(1 for t in terms if t in title_lower)
scored.append((score, r))
scored.sort(key=lambda x: x[0], reverse=True)
data["results"] = [r for _, r in scored[:10]]
return dataStep 3: Implement the local cache fallback
Read from a local JSON cache file as the last resort. Every successful search writes to this cache so stale data is always available when both API tiers fail.
Python
def write_cache(query: str, data: dict):
cache = json.loads(CACHE_FILE.read_text()) if CACHE_FILE.exists() else {}
cache[query] = data
CACHE_FILE.write_text(json.dumps(cache, indent=2))
def read_cache(query: str) -> dict | None:
if not CACHE_FILE.exists():
return None
cache = json.loads(CACHE_FILE.read_text())
return cache.get(query)Step 4: Wire the fallback chain together
Combine all three tiers into a single function. The agent calls this one function and always gets a result, with metadata indicating which tier served it.
Python
async def fallback_search(query: str) -> dict:
# Tier 1: primary search
try:
data = await primary_search(query)
if len(data.get("results", [])) >= 3:
write_cache(query, data)
return {"tier": "primary", **data}
except Exception as e:
print(f"Primary search failed: {e}")
# Tier 2: broader secondary search
try:
data = await secondary_search(query)
if data.get("results"):
write_cache(query, data)
return {"tier": "secondary", **data}
except Exception as e:
print(f"Secondary search failed: {e}")
# Tier 3: local cache
cached = read_cache(query)
if cached:
return {"tier": "cache", **cached}
return {"tier": "none", "results": [], "error": "All tiers exhausted"}Python Example
Python
import asyncio
import httpx
import json
from pathlib import Path
SCAVIO_API_KEY = "your-api-key"
CACHE_FILE = Path("search_cache.json")
async def primary_search(query: str) -> dict:
async with httpx.AsyncClient(timeout=10) as client:
resp = await client.post(
"https://api.scavio.dev/api/v1/search",
headers={"x-api-key": SCAVIO_API_KEY},
json={"query": query, "num_results": 10}
)
resp.raise_for_status()
return resp.json()
async def secondary_search(query: str) -> dict:
broad_query = " ".join(query.split()[:4])
async with httpx.AsyncClient(timeout=10) as client:
resp = await client.post(
"https://api.scavio.dev/api/v1/search",
headers={"x-api-key": SCAVIO_API_KEY},
json={"query": broad_query, "num_results": 20}
)
resp.raise_for_status()
data = resp.json()
terms = set(query.lower().split())
results = data.get("results", [])
scored = [(sum(1 for t in terms if t in r.get("title", "").lower()), r) for r in results]
scored.sort(key=lambda x: x[0], reverse=True)
data["results"] = [r for _, r in scored[:10]]
return data
def write_cache(query: str, data: dict):
cache = json.loads(CACHE_FILE.read_text()) if CACHE_FILE.exists() else {}
cache[query] = data
CACHE_FILE.write_text(json.dumps(cache, indent=2))
def read_cache(query: str) -> dict | None:
if not CACHE_FILE.exists():
return None
return json.loads(CACHE_FILE.read_text()).get(query)
async def fallback_search(query: str) -> dict:
try:
data = await primary_search(query)
if len(data.get("results", [])) >= 3:
write_cache(query, data)
return {"tier": "primary", **data}
except Exception:
pass
try:
data = await secondary_search(query)
if data.get("results"):
write_cache(query, data)
return {"tier": "secondary", **data}
except Exception:
pass
cached = read_cache(query)
if cached:
return {"tier": "cache", **cached}
return {"tier": "none", "results": []}
result = asyncio.run(fallback_search("best headless CMS 2026"))
print(f"Served from: {result['tier']}")
print(f"Results: {len(result.get('results', []))}")JavaScript Example
JavaScript
const SCAVIO_API_KEY = "your-api-key";
const fs = require("fs");
const CACHE_FILE = "search_cache.json";
async function primarySearch(query) {
const resp = await fetch("https://api.scavio.dev/api/v1/search", {
method: "POST",
headers: { "x-api-key": SCAVIO_API_KEY, "Content-Type": "application/json" },
body: JSON.stringify({ query, num_results: 10 })
});
if (!resp.ok) throw new Error("Primary search failed: " + resp.status);
return resp.json();
}
async function secondarySearch(query) {
const broad = query.split(" ").slice(0, 4).join(" ");
const resp = await fetch("https://api.scavio.dev/api/v1/search", {
method: "POST",
headers: { "x-api-key": SCAVIO_API_KEY, "Content-Type": "application/json" },
body: JSON.stringify({ query: broad, num_results: 20 })
});
if (!resp.ok) throw new Error("Secondary search failed");
const data = await resp.json();
const terms = new Set(query.toLowerCase().split(" "));
data.results = data.results
.map(r => ({ score: [...terms].filter(t => (r.title || "").toLowerCase().includes(t)).length, ...r }))
.sort((a, b) => b.score - a.score)
.slice(0, 10);
return data;
}
function writeCache(query, data) {
const cache = fs.existsSync(CACHE_FILE) ? JSON.parse(fs.readFileSync(CACHE_FILE)) : {};
cache[query] = data;
fs.writeFileSync(CACHE_FILE, JSON.stringify(cache, null, 2));
}
function readCache(query) {
if (!fs.existsSync(CACHE_FILE)) return null;
return JSON.parse(fs.readFileSync(CACHE_FILE))[query] || null;
}
async function fallbackSearch(query) {
try {
const data = await primarySearch(query);
if ((data.results || []).length >= 3) { writeCache(query, data); return { tier: "primary", ...data }; }
} catch {}
try {
const data = await secondarySearch(query);
if (data.results?.length) { writeCache(query, data); return { tier: "secondary", ...data }; }
} catch {}
const cached = readCache(query);
if (cached) return { tier: "cache", ...cached };
return { tier: "none", results: [] };
}
fallbackSearch("best headless CMS 2026").then(r => {
console.log("Served from:", r.tier);
console.log("Results:", (r.results || []).length);
});Expected Output
JSON
Served from: primary
Results: 10