Les agents IA en production utilisant des serveurs MCP ont besoin d'observabilité sur les performances au niveau des outils. Lorsque la qualité de recherche se dégrade, vous devez savoir si le serveur MCP est en panne, si le fournisseur en amont est limité en débit, ou si l'agent fait de mauvais choix d'outils. Ce tutoriel construit un système de surveillance de santé qui vérifie tous les outils MCP, enregistre les métriques de performance et alerte en cas de dégradation.
Prérequis
- Python 3.8+ installé
- bibliothèque requests installée
- Une clé API Scavio de scavio.dev
- Une URL de webhook Slack (optionnelle, pour les alertes)
Parcours
Étape 1: Définir la fonction de vérification de santé
Vérifier chaque plateforme de recherche avec une requête test et mesurer la latence et le nombre de résultats.
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Étape 2: Définir les seuils d'alerte
Définir ce qui constitue une performance dégradée pour chaque métrique.
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Étape 3: Envoyer des alertes à Slack
Transférer les alertes vers un canal Slack pour la visibilité des astreintes.
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')Étape 4: Enregistrer les données de santé pour les tendances
Ajouter chaque vérification de santé à un fichier JSONL pour l'analyse historique.
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)Exemple 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()Exemple 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}`); }
}
}Sortie attendue
JSON
A health monitoring system that checks MCP search tools every 5 minutes, alerts on degradation via Slack, and logs metrics for trending.