An r/PiCodingAgent post asked about reliable search for coding agents. The problem: single-provider search fails silently. This tutorial builds a router that tries multiple providers in priority order and falls back automatically. Useful for any agent runtime, not just Pi.
Prerequisites
- Scavio API key
- Python 3.8+
- Optional: Tavily API key, SearxNG instance
Walkthrough
Step 1: Define provider configuration
List providers with endpoints and priority.
providers = [
{
'name': 'scavio',
'url': 'https://api.scavio.dev/api/v1/search',
'headers': lambda: {'x-api-key': os.environ['SCAVIO_API_KEY']},
'body': lambda q: {'platform': 'google', 'query': q},
'parse': lambda r: r.get('organic_results', []),
'cost': 0.005,
'priority': 1
},
{
'name': 'searxng',
'url': 'http://localhost:8080/search',
'headers': lambda: {},
'body': lambda q: {'q': q, 'format': 'json'},
'parse': lambda r: r.get('results', []),
'cost': 0,
'priority': 2
},
{
'name': 'tavily',
'url': 'https://api.tavily.com/search',
'headers': lambda: {},
'body': lambda q: {'api_key': os.environ.get('TAVILY_API_KEY', ''), 'query': q},
'parse': lambda r: r.get('results', []),
'cost': 0.008,
'priority': 3
},
]Step 2: Build the router with fallback
Try each provider in priority order, fall back on failure.
import requests, os, time
def search_with_fallback(query, providers_list=providers):
sorted_providers = sorted(providers_list, key=lambda p: p['priority'])
errors = []
for provider in sorted_providers:
try:
resp = requests.post(provider['url'],
headers=provider['headers'](),
json=provider['body'](query),
timeout=10)
resp.raise_for_status()
results = provider['parse'](resp.json())
if results:
return {'provider': provider['name'], 'results': results, 'cost': provider['cost']}
errors.append(f"{provider['name']}: empty results")
except Exception as e:
errors.append(f"{provider['name']}: {str(e)}")
return {'provider': None, 'results': [], 'errors': errors}Step 3: Add latency-aware routing
Track response times and prefer faster providers.
provider_stats = {}
def update_stats(provider_name, latency, success):
if provider_name not in provider_stats:
provider_stats[provider_name] = {'total': 0, 'success': 0, 'avg_latency': 0}
stats = provider_stats[provider_name]
stats['total'] += 1
if success: stats['success'] += 1
stats['avg_latency'] = (stats['avg_latency'] * (stats['total'] - 1) + latency) / stats['total']
def smart_search(query):
start = time.time()
result = search_with_fallback(query)
latency = time.time() - start
if result['provider']:
update_stats(result['provider'], latency, True)
return resultStep 4: Log provider usage for cost tracking
Track which provider serves each query.
import json, datetime
def log_search(query, result):
with open('search_router_log.jsonl', 'a') as f:
f.write(json.dumps({
'ts': datetime.datetime.now().isoformat(),
'query': query,
'provider': result['provider'],
'cost': result.get('cost', 0),
'result_count': len(result.get('results', []))
}) + '\n')Python Example
import os, requests
def search(query):
for provider in ['scavio', 'searxng', 'tavily']:
try:
if provider == 'scavio':
r = requests.post('https://api.scavio.dev/api/v1/search',
headers={'x-api-key': os.environ['SCAVIO_API_KEY']},
json={'platform': 'google', 'query': query}, timeout=10)
results = r.json().get('organic_results', [])
if results: return {'provider': 'scavio', 'results': results}
except: continue
return {'provider': None, 'results': []}JavaScript Example
async function search(query) {
try {
const res = await fetch('https://api.scavio.dev/api/v1/search', {
method: 'POST',
headers: {'x-api-key': process.env.SCAVIO_API_KEY, 'Content-Type': 'application/json'},
body: JSON.stringify({platform: 'google', query})
});
const data = await res.json();
if (data.organic_results?.length) return {provider: 'scavio', results: data.organic_results};
} catch(e) { /* fallback */ }
}Expected Output
Multi-provider search router with automatic fallback. Priority: Scavio ($0.005) > SearxNG (free/self-hosted) > Tavily ($0.008). Latency tracking and JSONL cost log.