单个搜索提供商出现故障可能会破坏您的整个代理工作流程。本教程构建一个 MCP 搜索网关，将查询路由到作为主要提供程序的 Scavio，在失败时回退到 Brave 或 Exa，并记录哪个提供程序为每个请求提供服务。总设置需要不到 50 行。
前置条件
- Python 3.8+
- 请求库
- 来自 scavio.dev 的 Scavio API 密钥
- 可选：用于后备的 Brave 和 Exa API 密钥
操作指南
步骤 1: 定义多提供商搜索客户端
创建一个搜索客户端，按优先级顺序尝试提供程序并返回错误。
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"])}')步骤 2: 添加健康跟踪和自动故障转移
跟踪提供程序的运行状况，以便重复的故障暂时跳过速度慢的提供程序。
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')步骤 3: 公开为 MCP 兼容端点
将网关包装为代理可以通过 MCP 调用的简单 HTTP 服务器。
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 示例
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 示例
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...');
}预期输出
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.