使用 MCP 服务器的生产 AI 代理需要可观察工具级性能。当搜索质量下降时，您需要了解 MCP 服务器是否已关闭、上游提供商是否受到速率限制，或者代理是否选择了错误的工具。本教程构建了一个健康监控系统，用于检查所有 MCP 工具、记录性能指标并发出降级警报。
前置条件
- 已安装 Python 3.8+
- 请求已安装库
- 来自 scavio.dev 的 Scavio API 密钥
- Slack webhook URL（可选，用于警报）
操作指南
步骤 1: 定义健康检查功能
使用测试查询检查每个搜索平台并测量延迟和结果计数。
Python
import requests, os, time, json
H = {'x-api-key': os.environ['SCAVIO_API_KEY']}
PLATFORMS = ['google', 'reddit', 'youtube', 'amazon', 'walmart']
def health_check_all() -> dict:
report = {'timestamp': time.strftime('%Y-%m-%dT%H:%M:%SZ')}
for platform in PLATFORMS:
start = time.time()
try:
resp = requests.post('https://api.scavio.dev/api/v1/search', headers=H,
json={'platform': platform, 'query': 'test query'}, timeout=15)
latency = round(time.time() - start, 3)
data = resp.json()
result_count = len(data.get('organic', []))
report[platform] = {'status': 'ok', 'latency_s': latency,
'results': result_count, 'http_code': resp.status_code}
except requests.Timeout:
report[platform] = {'status': 'timeout', 'latency_s': round(time.time() - start, 3)}
except Exception as e:
report[platform] = {'status': 'error', 'error': str(e)}
return report步骤 2: 设置警报阈值
定义每个指标的性能下降的原因。
Python
THRESHOLDS = {
'max_latency_s': 5.0,
'min_results': 1,
'alert_on_error': True,
}
def check_alerts(report: dict) -> list:
alerts = []
for platform in PLATFORMS:
data = report.get(platform, {})
if data.get('status') == 'error' or data.get('status') == 'timeout':
alerts.append(f'{platform}: {data.get("status")} - {data.get("error", "timeout")}')
elif data.get('latency_s', 0) > THRESHOLDS['max_latency_s']:
alerts.append(f'{platform}: high latency {data["latency_s"]}s')
elif data.get('results', 0) < THRESHOLDS['min_results']:
alerts.append(f'{platform}: low results ({data["results"]})')
return alerts步骤 3: 向 Slack 发送警报
将警报转发到 Slack 通道以获取待命可见性。
Python
SLACK_WEBHOOK = os.environ.get('SLACK_WEBHOOK_URL', '')
def send_slack_alert(alerts: list):
if not alerts or not SLACK_WEBHOOK:
return
text = 'MCP Search Health Alert:\n' + '\n'.join(f'- {a}' for a in alerts)
requests.post(SLACK_WEBHOOK, json={'text': text}, timeout=5)
# Run and alert:
report = health_check_all()
alerts = check_alerts(report)
if alerts:
send_slack_alert(alerts)
print(f'ALERTS: {alerts}')
else:
print('All platforms healthy')步骤 4: 记录健康数据以进行趋势分析
将每个运行状况检查附加到 JSONL 文件以进行历史分析。
Python
def log_health(report: dict, filepath: str = 'mcp_health.jsonl'):
with open(filepath, 'a') as f:
f.write(json.dumps(report) + '\n')
# Run as cron: */5 * * * * python mcp_health_check.py
report = health_check_all()
log_health(report)
alerts = check_alerts(report)
if alerts:
send_slack_alert(alerts)Python 示例
Python
import requests, os, time
H = {'x-api-key': os.environ['SCAVIO_API_KEY']}
def quick_health():
for p in ['google', 'reddit', 'youtube', 'amazon', 'walmart']:
start = time.time()
try:
r = requests.post('https://api.scavio.dev/api/v1/search', headers=H,
json={'platform': p, 'query': 'test'}, timeout=10)
print(f'{p:10s} {r.status_code} {time.time()-start:.2f}s {len(r.json().get("organic",[]))} results')
except Exception as e:
print(f'{p:10s} ERROR {e}')
quick_health()JavaScript 示例
JavaScript
async function quickHealth() {
for (const p of ['google', 'reddit', 'youtube', 'amazon', 'walmart']) {
const start = Date.now();
try {
const r = 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: p, query: 'test'})
});
const data = await r.json();
console.log(`${p.padEnd(10)} ${r.status} ${Date.now()-start}ms ${(data.organic||[]).length} results`);
} catch (e) { console.log(`${p.padEnd(10)} ERROR ${e.message}`); }
}
}预期输出
JSON
A health monitoring system that checks MCP search tools every 5 minutes, alerts on degradation via Slack, and logs metrics for trending.