La panne d'un seul fournisseur de recherche peut paralyser l'intégralité de votre workflow d'agent. Ce tutoriel construit une passerelle de recherche MCP qui achemine les requêtes vers Scavio comme fournisseur principal, bascule vers Brave ou Exa en cas d'échec, et enregistre quel fournisseur a servi chaque requête. La configuration totale prend moins de 50 lignes.
Prérequis
- Python 3.8+
- bibliothèque requests
- Une clé API Scavio depuis scavio.dev
- Optionnel : clés API Brave et Exa pour le basculement
Parcours
Étape 1: Définir le client de recherche multi-fournisseur
Créez un client de recherche qui tente les fournisseurs par ordre de priorité et bascule en cas d'erreur.
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"])}')Étape 2: Ajouter le suivi de santé et le basculement automatique
Suivez l'état de santé des fournisseurs afin que les échecs répétés contournent temporairement les fournisseurs lents.
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')Étape 3: Exposer comme point de terminaison compatible MCP
Enveloppez la passerelle en un simple serveur HTTP que les agents peuvent appeler 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.')Exemple Python
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)}')Exemple JavaScript
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...');
}Sortie attendue
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.