Production AI agents using MCP servers need observability into tool-level performance. When search quality degrades, you need to know whether the MCP server is down, the upstream provider is rate-limited, or the agent is making poor tool selections. This tutorial builds a health monitoring system that checks all MCP tools, logs performance metrics, and alerts on degradation.
Prerequisites
- Python 3.8+ installed
- requests library installed
- A Scavio API key from scavio.dev
- A Slack webhook URL (optional, for alerts)
Walkthrough
Step 1: Define the health check function
Check each search platform with a test query and measure latency and result count.
Python
import requests, os, time, json
# Scavio has one endpoint per platform - there is no dispatcher endpoint and no
# `platform` request param, so the selector lives in your code.
SCAVIO = "https://api.scavio.dev"
SCAVIO_ENDPOINTS = {
"google": "/api/v2/google",
"reddit": "/api/v1/reddit/search",
"youtube": "/api/v1/youtube/search",
"amazon": "/api/v1/amazon/search",
"walmart": "/api/v1/walmart/search",
}
SCAVIO_QUERY_KEY = {"youtube": "search"}
SCAVIO_RESULTS_KEY = {"google": "organic_results", "reddit": "results",
"youtube": "results", "amazon": "products", "walmart": "products"}
def scavio_url(platform):
return SCAVIO + SCAVIO_ENDPOINTS[platform or "google"]
def scavio_body(platform, query):
return {SCAVIO_QUERY_KEY.get(platform or "google", "query"): query}
def scavio_payload(payload, platform="google"):
"""Google v2 passes Google's response through as-is; every other endpoint
wraps its payload in `data`. Item fields differ per platform (see
https://scavio.dev/docs), so only the result list is normalised here."""
platform = platform or "google"
out = payload if platform == "google" else payload["data"]
return {**out, "results": out.get(SCAVIO_RESULTS_KEY[platform], [])}
H = {'Authorization': 'Bearer ' + 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(scavio_url(platform), headers=H, json=scavio_body(platform, 'test query'), timeout=15)
latency = round(time.time() - start, 3)
data = scavio_payload(resp.json(), platform)
result_count = len(data.get('results', []))
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 reportStep 2: Set alerting thresholds
Define what constitutes degraded performance for each metric.
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 alertsStep 3: Send alerts to Slack
Forward alerts to a Slack channel for on-call visibility.
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')Step 4: Log health data for trending
Append each health check to a JSONL file for historical analysis.
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 Example
Python
import requests, os, time
# Scavio has one endpoint per platform - there is no dispatcher endpoint and no
# `platform` request param, so the selector lives in your code.
SCAVIO = "https://api.scavio.dev"
SCAVIO_ENDPOINTS = {
"google": "/api/v2/google",
"reddit": "/api/v1/reddit/search",
"youtube": "/api/v1/youtube/search",
"amazon": "/api/v1/amazon/search",
"walmart": "/api/v1/walmart/search",
}
SCAVIO_QUERY_KEY = {"youtube": "search"}
SCAVIO_RESULTS_KEY = {"google": "organic_results", "reddit": "results",
"youtube": "results", "amazon": "products", "walmart": "products"}
def scavio_url(platform):
return SCAVIO + SCAVIO_ENDPOINTS[platform or "google"]
def scavio_body(platform, query):
return {SCAVIO_QUERY_KEY.get(platform or "google", "query"): query}
def scavio_payload(payload, platform="google"):
"""Google v2 passes Google's response through as-is; every other endpoint
wraps its payload in `data`. Item fields differ per platform (see
https://scavio.dev/docs), so only the result list is normalised here."""
platform = platform or "google"
out = payload if platform == "google" else payload["data"]
return {**out, "results": out.get(SCAVIO_RESULTS_KEY[platform], [])}
H = {'Authorization': 'Bearer ' + os.environ['SCAVIO_API_KEY']}
def quick_health():
for p in ['google', 'reddit', 'youtube', 'amazon', 'walmart']:
start = time.time()
try:
r = requests.post(scavio_url(p), headers=H, json=scavio_body(p, 'test'), timeout=10)
print(f'{p:10s} {r.status_code} {time.time()-start:.2f}s {len(scavio_payload(r.json(), p).get("results",[]))} results')
except Exception as e:
print(f'{p:10s} ERROR {e}')
quick_health()JavaScript Example
JavaScript
// Scavio has one endpoint per platform - there is no dispatcher endpoint and no
// `platform` request param, so the selector lives in your code.
const SCAVIO = "https://api.scavio.dev";
const SCAVIO_ENDPOINTS = {
google: "/api/v2/google",
reddit: "/api/v1/reddit/search",
youtube: "/api/v1/youtube/search",
amazon: "/api/v1/amazon/search",
walmart: "/api/v1/walmart/search",
};
const SCAVIO_QUERY_KEY = { youtube: "search" };
const SCAVIO_RESULTS_KEY = { google: "organic_results", reddit: "results",
youtube: "results", amazon: "products", walmart: "products" };
const scavioUrl = (platform) => SCAVIO + SCAVIO_ENDPOINTS[platform || "google"];
const scavioBody = (platform, query) =>
({ [SCAVIO_QUERY_KEY[platform || "google"] || "query"]: query });
// Google v2 passes Google's response through as-is; every other endpoint wraps
// its payload in `data`. Item fields differ per platform (see
// https://scavio.dev/docs), so only the result list is normalised here.
function scavioPayload(json, platform = "google") {
const p = platform || "google";
const out = p === "google" ? json : json.data;
return { ...out, results: out[SCAVIO_RESULTS_KEY[p]] || [] };
}
async function quickHealth() {
for (const p of ['google', 'reddit', 'youtube', 'amazon', 'walmart']) {
const start = Date.now();
try {
const r = await fetch(scavioUrl(p), {
method: 'POST', headers: {'Authorization': `Bearer ${process.env.SCAVIO_API_KEY}`, 'Content-Type': 'application/json'},
body: JSON.stringify(scavioBody(p, 'test'))
});
const data = scavioPayload(await r.json(), p);
console.log(`${p.padEnd(10)} ${r.status} ${Date.now()-start}ms ${(data.results||[]).length} results`);
} catch (e) { console.log(`${p.padEnd(10)} ERROR ${e.message}`); }
}
}Expected Output
JSON
A health monitoring system that checks MCP search tools every 5 minutes, alerts on degradation via Slack, and logs metrics for trending.