La caída de un único proveedor de búsqueda puede interrumpir todo el flujo de trabajo de su agente. Este tutorial crea una puerta de enlace de búsqueda MCP que enruta consultas a Scavio como proveedor principal, recurre a Brave o Exa en caso de falla y registra qué proveedor atendió cada solicitud. La configuración total requiere menos de 50 líneas.
Requisitos previos
- Python 3.8+
- solicita biblioteca
- Una clave API de Scavio de scavio.dev
- Opcional: claves API Brave y Exa para respaldo
Guia paso a paso
Paso 1: Definir el cliente de búsqueda multiproveedor
Cree un cliente de búsqueda que pruebe los proveedores en orden de prioridad y recurra a los errores.
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"])}')Paso 2: Agregar seguimiento de estado y conmutación por error automática
Realice un seguimiento del estado del proveedor para que las fallas repetidas omitan temporalmente los proveedores lentos.
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')Paso 3: Exponer como punto final compatible con MCP
Envuelva la puerta de enlace como un servidor HTTP simple al que los agentes pueden llamar a través de 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.')Ejemplo en 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)}')Ejemplo en 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...');
}Salida esperada
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.