CaptchaAI Health Check and Monitoring Script

When your workflow depends on CaptchaAI, you need to know the moment something goes wrong — low balance, slow solve times, or API errors. A health check script catches problems before they cascade into failed jobs and lost data.

This tutorial builds a monitoring script that checks API status, measures solve latency, tracks balance, and sends alerts.

What the health check covers

What you need

pip install requests

Python health check script

import requests
import time
import json
from datetime import datetime

API_KEY = "YOUR_API_KEY"
BASE_URL = "https://ocr.captchaai.com"

# Thresholds
BALANCE_WARNING = 5.0       # Alert below $5
MAX_SOLVE_TIME = 60         # Alert if solve > 60s
ALERT_WEBHOOK = None        # Set to webhook URL for alerts


def check_balance():
    """Check account balance and return status."""
    try:
        resp = requests.get(f"{BASE_URL}/res.php", params={
            "key": API_KEY, "action": "getbalance", "json": 1
        }, timeout=10)
        data = resp.json()
        balance = float(data.get("request", 0))
        return {
            "check": "balance",
            "status": "warning" if balance < BALANCE_WARNING else "ok",
            "value": balance,
            "message": f"${balance:.2f}" + (" — LOW BALANCE" if balance < BALANCE_WARNING else "")
        }
    except Exception as e:
        return {"check": "balance", "status": "error", "value": None, "message": str(e)}


def check_api_reachability():
    """Verify the API endpoints respond."""
    results = {}
    for endpoint in ["/in.php", "/res.php"]:
        try:
            start = time.time()
            resp = requests.get(f"{BASE_URL}{endpoint}", params={
                "key": API_KEY, "json": 1
            }, timeout=10)
            latency_ms = round((time.time() - start) * 1000)
            results[endpoint] = {
                "status": "ok" if resp.status_code == 200 else "error",
                "latency_ms": latency_ms,
                "http_status": resp.status_code
            }
        except requests.Timeout:
            results[endpoint] = {"status": "error", "latency_ms": None, "http_status": "timeout"}
        except Exception as e:
            results[endpoint] = {"status": "error", "latency_ms": None, "http_status": str(e)}

    all_ok = all(r["status"] == "ok" for r in results.values())
    return {
        "check": "api_reachability",
        "status": "ok" if all_ok else "error",
        "endpoints": results,
        "message": "All endpoints responding" if all_ok else "Endpoint issues detected"
    }


def check_solve_latency(sitekey="6Le-wvkSAAAAAPBMRTvw0Q4Muexq9bi0DJwx_mJ-",
                         pageurl="https://www.google.com/recaptcha/api2/demo"):
    """Submit a test reCAPTCHA and measure solve time."""
    try:
        start = time.time()

        # Submit
        submit_resp = requests.post(f"{BASE_URL}/in.php", data={
            "key": API_KEY,
            "method": "userrecaptcha",
            "googlekey": sitekey,
            "pageurl": pageurl,
            "json": 1
        }, timeout=10).json()

        if submit_resp.get("status") != 1:
            return {
                "check": "solve_latency",
                "status": "error",
                "value": None,
                "message": f"Submit failed: {submit_resp.get('request')}"
            }

        task_id = submit_resp["request"]
        time.sleep(10)

        # Poll
        for _ in range(24):
            result = requests.get(f"{BASE_URL}/res.php", params={
                "key": API_KEY, "action": "get", "id": task_id, "json": 1
            }, timeout=10).json()

            if result.get("status") == 1:
                elapsed = round(time.time() - start, 1)
                return {
                    "check": "solve_latency",
                    "status": "warning" if elapsed > MAX_SOLVE_TIME else "ok",
                    "value": elapsed,
                    "message": f"{elapsed}s" + (" — SLOW" if elapsed > MAX_SOLVE_TIME else "")
                }
            if result.get("request") != "CAPCHA_NOT_READY":
                return {
                    "check": "solve_latency",
                    "status": "error",
                    "value": None,
                    "message": f"Solve error: {result.get('request')}"
                }
            time.sleep(5)

        return {"check": "solve_latency", "status": "error", "value": None, "message": "Timed out"}
    except Exception as e:
        return {"check": "solve_latency", "status": "error", "value": None, "message": str(e)}


def send_alert(report):
    """Send alert via webhook if any check failed."""
    issues = [r for r in report["checks"] if r["status"] != "ok"]
    if not issues or not ALERT_WEBHOOK:
        return

    payload = {
        "text": f"CaptchaAI Health Alert\n" +
                "\n".join(f"  {i['check']}: {i['message']}" for i in issues)
    }
    try:
        requests.post(ALERT_WEBHOOK, json=payload, timeout=10)
    except Exception:
        pass


def run_health_check(include_solve_test=True):
    """Run all health checks and return a report."""
    report = {
        "timestamp": datetime.now().isoformat(),
        "checks": []
    }

    report["checks"].append(check_balance())
    report["checks"].append(check_api_reachability())

    if include_solve_test:
        report["checks"].append(check_solve_latency())

    # Overall status
    statuses = [c["status"] for c in report["checks"]]
    if "error" in statuses:
        report["overall"] = "error"
    elif "warning" in statuses:
        report["overall"] = "warning"
    else:
        report["overall"] = "ok"

    send_alert(report)
    return report


if __name__ == "__main__":
    report = run_health_check(include_solve_test=True)
    print(json.dumps(report, indent=2))

Expected output:

{
  "timestamp": "2025-07-15T10:30:00.000000",
  "overall": "ok",
  "checks": [
    {
      "check": "balance",
      "status": "ok",
      "value": 24.50,
      "message": "$24.50"
    },
    {
      "check": "api_reachability",
      "status": "ok",
      "message": "All endpoints responding"
    },
    {
      "check": "solve_latency",
      "status": "ok",
      "value": 18.3,
      "message": "18.3s"
    }
  ]
}

Node.js health check

const https = require("https");

const API_KEY = "YOUR_API_KEY";
const BASE_URL = "https://ocr.captchaai.com";
const BALANCE_WARNING = 5.0;

async function fetchJSON(url) {
  return new Promise((resolve, reject) => {
    https.get(url, (res) => {
      let data = "";
      res.on("data", (chunk) => (data += chunk));
      res.on("end", () => {
        try { resolve(JSON.parse(data)); }
        catch (e) { reject(new Error(`Parse error: ${data}`)); }
      });
    }).on("error", reject);
  });
}

async function checkBalance() {
  try {
    const data = await fetchJSON(
      `${BASE_URL}/res.php?key=${API_KEY}&action=getbalance&json=1`
    );
    const balance = parseFloat(data.request || 0);
    return {
      check: "balance",
      status: balance < BALANCE_WARNING ? "warning" : "ok",
      value: balance,
      message: `$${balance.toFixed(2)}${balance < BALANCE_WARNING ? " — LOW" : ""}`
    };
  } catch (e) {
    return { check: "balance", status: "error", value: null, message: e.message };
  }
}

async function checkReachability() {
  try {
    const start = Date.now();
    await fetchJSON(`${BASE_URL}/res.php?key=${API_KEY}&json=1`);
    const latency = Date.now() - start;
    return {
      check: "api_reachability",
      status: "ok",
      latency_ms: latency,
      message: `Responding in ${latency}ms`
    };
  } catch (e) {
    return { check: "api_reachability", status: "error", message: e.message };
  }
}

async function runHealthCheck() {
  const checks = await Promise.all([checkBalance(), checkReachability()]);
  const overall = checks.some((c) => c.status === "error")
    ? "error"
    : checks.some((c) => c.status === "warning")
    ? "warning"
    : "ok";

  const report = {
    timestamp: new Date().toISOString(),
    overall,
    checks
  };

  console.log(JSON.stringify(report, null, 2));
  return report;
}

runHealthCheck();

Scheduling

Run the health check on an interval:

# Every 10 minutes via cron
*/10 * * * * cd /path/to/project && python health_check.py >> /var/log/captchaai-health.log 2>&1

Troubleshooting

FAQ

How often should I run health checks?

Every 5–15 minutes for production workflows. For critical systems, every 1–2 minutes with the solve test disabled (balance + reachability only) to avoid burning credits.

Does the solve test cost credits?

Yes. Each solve test submits a real CAPTCHA task. Disable include_solve_test for frequent checks and only run solve tests every 30–60 minutes.

What webhook services work for alerts?

Slack incoming webhooks, Discord webhooks, PagerDuty, or any HTTP endpoint that accepts JSON POST requests.

Get your CaptchaAI API key

Monitor your CAPTCHA-solving infrastructure at captchaai.com. Catch issues before they impact your workflows.

Related guides

• CaptchaAI Balance Check and Auto-Refill
• CaptchaAI Error Codes Reference
• Monitoring Solve Rates with Prometheus and Grafana
• Implementing Retry Logic with CaptchaAI