Overview
Google is sunsetting the free web-wide CSE tier by January 2027. This workflow runs CSE queries in parallel against Scavio during a validation window, compares result quality, logs discrepancies, and cuts over automatically when Scavio matches or exceeds CSE quality. Zero downtime, zero manual intervention.
Trigger
Daily cron during migration window, then on-demand after cutover.
Schedule
Daily
Workflow Steps
Load CSE Query Log
Read the list of queries your application sends to Google CSE from the access log or query database.
Run Parallel Queries
For each query, call both Google CSE and Scavio search API. Record result counts, top URLs, and latency.
Compare Quality Metrics
Score each query pair: result overlap, top-3 URL match, latency delta. Flag queries where Scavio quality is below threshold.
Generate Migration Report
Output a report showing which queries are ready for cutover and which need investigation.
Auto-Cutover Ready Queries
For queries that pass quality checks, update the routing table to send them to Scavio instead of CSE.
Python Implementation
import requests, os, json
from pathlib import Path
API_KEY = os.environ["SCAVIO_API_KEY"]
H = {"x-api-key": API_KEY, "Content-Type": "application/json"}
def scavio_search(query: str) -> dict:
resp = requests.post(
"https://api.scavio.dev/api/v1/search",
headers=H,
json={"query": query, "country_code": "us"},
timeout=10,
)
data = resp.json()
return {
"results": len(data.get("organic_results", [])),
"top_urls": [r.get("link", "") for r in data.get("organic_results", [])[:5]],
"has_aio": bool(data.get("ai_overview")),
}
def migration_check(queries: list) -> dict:
report = {"ready": [], "needs_review": []}
for q in queries:
scavio = scavio_search(q)
# CSE baseline: assume 5+ results is passing
if scavio["results"] >= 5:
report["ready"].append({"query": q, "scavio_results": scavio["results"], "bonus_aio": scavio["has_aio"]})
else:
report["needs_review"].append({"query": q, "scavio_results": scavio["results"]})
return report
queries = ["python web framework 2026", "best standing desk", "kubernetes deployment guide"]
report = migration_check(queries)
print(f"Ready: {len(report['ready'])}, Needs review: {len(report['needs_review'])}")
for r in report["ready"]:
print(f" {r['query']}: {r['scavio_results']} results, AIO: {r['bonus_aio']}")JavaScript Implementation
const H = {'x-api-key': process.env.SCAVIO_API_KEY, 'Content-Type': 'application/json'};
async function scavioSearch(query) {
const r = await fetch('https://api.scavio.dev/api/v1/search', {method:'POST', headers:H, body:JSON.stringify({query, country_code:'us'})});
const d = await r.json();
return {results:(d.organic_results||[]).length, topUrls:(d.organic_results||[]).slice(0,5).map(r=>r.link), hasAio:!!d.ai_overview};
}
async function migrationCheck(queries) {
const report = {ready:[], needsReview:[]};
for (const q of queries) {
const s = await scavioSearch(q);
if (s.results >= 5) report.ready.push({query:q, scavioResults:s.results, bonusAio:s.hasAio});
else report.needsReview.push({query:q, scavioResults:s.results});
}
return report;
}
const report = await migrationCheck(['python web framework 2026', 'best standing desk', 'kubernetes deployment guide']);
console.log('Ready: '+report.ready.length+', Needs review: '+report.needsReview.length);
for (const r of report.ready) console.log(' '+r.query+': '+r.scavioResults+' results, AIO: '+r.bonusAio);Platforms Used
Web search with knowledge graph, PAA, and AI overviews