A single search provider going down can break your entire agent workflow. This tutorial builds an MCP search gateway that routes queries to Scavio as the primary provider, falls back to Brave or Exa on failure, and logs which provider served each request. Total setup takes under 50 lines.
Prerequisites
- Python 3.8+
- requests library
- A Scavio API key from scavio.dev
- Optional: Brave and Exa API keys for fallback
Walkthrough
Step 1: Define the multi-provider search client
Create a search client that tries providers in priority order and falls back on errors.
Python
import os, requests, time
SCAVIO_KEY = os.environ['SCAVIO_API_KEY']
BRAVE_KEY = os.environ.get('BRAVE_API_KEY', '')
EXA_KEY = os.environ.get('EXA_API_KEY', '')
def search_scavio(query):
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'}, 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_brave(query):
resp = requests.get('https://api.search.brave.com/res/v1/web/search',
headers={'X-Subscription-Token': BRAVE_KEY},
params={'q': query}, timeout=10)
resp.raise_for_status()
return [{'title': r['title'], 'url': r['url'], 'snippet': r.get('description', '')}
for r in resp.json().get('web', {}).get('results', [])]
def search_exa(query):
resp = requests.post('https://api.exa.ai/search',
headers={'x-api-key': EXA_KEY, 'Content-Type': 'application/json'},
json={'query': query, 'numResults': 10}, timeout=10)
resp.raise_for_status()
return [{'title': r['title'], 'url': r['url'], 'snippet': r.get('text', '')[:200]}
for r in resp.json().get('results', [])]
PROVIDERS = [
('scavio', search_scavio),
('brave', search_brave),
('exa', search_exa),
]
def gateway_search(query):
for name, fn in PROVIDERS:
try:
results = fn(query)
print(f' Served by: {name} ({len(results)} results)')
return {'provider': name, 'results': results}
except Exception as e:
print(f' {name} failed: {str(e)[:60]}')
return {'provider': 'none', 'results': []}
result = gateway_search('best search api for agents 2026')
print(f'Provider: {result["provider"]}, Results: {len(result["results"])}')Step 2: Add health tracking and automatic failover
Track provider health so repeated failures skip slow providers temporarily.
Python
from collections import defaultdict
health = defaultdict(lambda: {'failures': 0, 'last_fail': 0, 'cooldown': 30})
def is_healthy(name):
h = health[name]
if h['failures'] >= 3:
if time.time() - h['last_fail'] < h['cooldown']:
return False
h['failures'] = 0 # Reset after cooldown
return True
def smart_gateway(query):
for name, fn in PROVIDERS:
if not is_healthy(name):
print(f' Skipping {name} (cooldown)')
continue
try:
results = fn(query)
health[name]['failures'] = 0
print(f' Served by: {name} ({len(results)} results)')
return {'provider': name, 'results': results}
except Exception as e:
health[name]['failures'] += 1
health[name]['last_fail'] = time.time()
print(f' {name} failed ({health[name]["failures"]}x): {str(e)[:50]}')
return {'provider': 'none', 'results': []}
# Test gateway
for q in ['mcp search setup', 'agent tool calling', 'search api comparison']:
result = smart_gateway(q)
print(f' -> {result["provider"]}: {len(result["results"])} results\n')Step 3: Expose as MCP-compatible endpoint
Wrap the gateway as a simple HTTP server that agents can call via MCP.
Python
from http.server import HTTPServer, BaseHTTPRequestHandler
import json
class MCPSearchHandler(BaseHTTPRequestHandler):
def do_POST(self):
length = int(self.headers.get('Content-Length', 0))
body = json.loads(self.rfile.read(length)) if length else {}
query = body.get('query', '')
if not query:
self.send_response(400)
self.end_headers()
self.wfile.write(b'{"error": "query required"}')
return
result = smart_gateway(query)
self.send_response(200)
self.send_header('Content-Type', 'application/json')
self.end_headers()
self.wfile.write(json.dumps(result).encode())
def log_message(self, fmt, *args):
print(f' MCP Gateway: {args[0]}')
# Uncomment to run:
# server = HTTPServer(('localhost', 8900), MCPSearchHandler)
# print('MCP Search Gateway on :8900')
# server.serve_forever()
print('Gateway ready. POST {"query": "..."} to localhost:8900')
print('Primary: Scavio ($0.005/query). Fallback: Brave, Exa.')Python Example
Python
import os, requests
SH = {'x-api-key': os.environ['SCAVIO_API_KEY'], 'Content-Type': 'application/json'}
def search(query):
try:
resp = requests.post('https://api.scavio.dev/api/v1/search',
headers=SH, json={'query': query, 'country_code': 'us'}, timeout=10)
resp.raise_for_status()
return resp.json().get('organic_results', [])
except Exception:
return [] # Fallback to next provider
results = search('mcp search gateway')
print(f'Results: {len(results)}')JavaScript Example
JavaScript
const SH = { 'x-api-key': process.env.SCAVIO_API_KEY, 'Content-Type': 'application/json' };
try {
const resp = await fetch('https://api.scavio.dev/api/v1/search', {
method: 'POST', headers: SH,
body: JSON.stringify({ query: 'mcp search gateway', country_code: 'us' })
});
const data = await resp.json();
console.log(`Results: ${(data.organic_results || []).length}`);
} catch (e) {
console.log('Primary failed, trying fallback...');
}Expected Output
JSON
Served by: scavio (10 results)
Provider: scavio, Results: 10
Served by: scavio (10 results)
-> scavio: 10 results
scavio failed (1x): Connection timed out
Served by: brave (8 results)
-> brave: 8 results
Gateway ready. POST {"query": "..."} to localhost:8900
Primary: Scavio ($0.005/query). Fallback: Brave, Exa.