MCP Search Health Monitor

import requests
import json
import time
from pathlib import Path
from datetime import datetime

API_KEY = "your_scavio_api_key"
CREDIT_WARNING_THRESHOLD = 100  # Alert when below this many credits
MAX_RESPONSE_TIME_MS = 5000  # Alert if response takes longer

def check_search_endpoint() -> dict:
    start = time.time()
    try:
        res = requests.post(
            "https://api.scavio.dev/api/v1/search",
            headers={"x-api-key": API_KEY},
            json={"platform": "google", "query": "test health check"},
            timeout=15,
        )
        elapsed_ms = round((time.time() - start) * 1000)
        return {
            "status": "up",
            "status_code": res.status_code,
            "response_time_ms": elapsed_ms,
            "has_organic": bool(res.json().get("organic")),
            "valid_json": True,
        }
    except requests.RequestException as e:
        elapsed_ms = round((time.time() - start) * 1000)
        return {
            "status": "down",
            "error": str(e),
            "response_time_ms": elapsed_ms,
            "has_organic": False,
            "valid_json": False,
        }

def evaluate_health(check: dict, credits_remaining: int | None) -> dict:
    issues = []
    if check["status"] != "up":
        issues.append("Endpoint is down")
    if check.get("status_code") and check["status_code"] != 200:
        issues.append(f"Non-200 status: {check['status_code']}")
    if check["response_time_ms"] > MAX_RESPONSE_TIME_MS:
        issues.append(f"Slow response: {check['response_time_ms']}ms")
    if not check["has_organic"]:
        issues.append("No organic results in response")
    if credits_remaining is not None and credits_remaining < CREDIT_WARNING_THRESHOLD:
        issues.append(f"Low credits: {credits_remaining} remaining")

    status = "healthy" if not issues else "degraded" if check["status"] == "up" else "down"
    return {"status": status, "issues": issues}

def run():
    date = datetime.utcnow().strftime("%Y-%m-%d")

    check = check_search_endpoint()
    # Note: credit balance check depends on your Scavio dashboard/API
    credits = None  # Replace with actual credit balance API call

    health = evaluate_health(check, credits)

    report = {
        "date": date,
        "endpoint_check": check,
        "credits_remaining": credits,
        "health": health,
    }

    Path("mcp_health_latest.json").write_text(json.dumps(report, indent=2))
    print(f"MCP Health Check {date}: {health['status'].upper()}")
    print(f"  Response time: {check['response_time_ms']}ms")
    if health["issues"]:
        for issue in health["issues"]:
            print(f"  ALERT: {issue}")
    else:
        print("  All checks passed")

if __name__ == "__main__":
    run()

const API_KEY = "your_scavio_api_key";
const CREDIT_THRESHOLD = 100;

async function checkEndpoint() {
  const start = Date.now();
  try {
    const res = await fetch("https://api.scavio.dev/api/v1/search", {
      method: "POST",
      headers: { "x-api-key": API_KEY, "content-type": "application/json" },
      body: JSON.stringify({ platform: "google", query: "test health check" }),
    });
    const elapsed = Date.now() - start;
    const data = await res.json();
    return { status: "up", statusCode: res.status, responseMs: elapsed, hasOrganic: !!(data.organic?.length) };
  } catch (err) {
    return { status: "down", error: err.message, responseMs: Date.now() - start, hasOrganic: false };
  }
}

const check = await checkEndpoint();
const issues = [];
if (check.status !== "up") issues.push("Endpoint is down");
if (check.responseMs > 5000) issues.push(`Slow: ${check.responseMs}ms`);
if (!check.hasOrganic) issues.push("No organic results");

const health = issues.length === 0 ? "healthy" : check.status === "up" ? "degraded" : "down";
console.log(`MCP Health: ${health.toUpperCase()} (${check.responseMs}ms)`);
for (const issue of issues) console.log(`  ALERT: ${issue}`);