check_endpoint.py

#!/usr/bin/env python3
"""
check_endpoint.py — diagnose a custom LLM endpoint before you wire it into a coding agent.

Tests both the OpenAI Chat Completions and Anthropic Messages protocols, reports
which one your endpoint actually speaks, measures TTFT, and tells you the exact
base_url string to paste into each agent.

Usage:
    python check_endpoint.py --url https://apimaster.ai --key sk-xxx
    python check_endpoint.py --url https://apimaster.ai --key sk-xxx --model claude-sonnet-4-6

No dependencies beyond the standard library.
"""

import argparse
import json
import ssl
import sys
import time
import urllib.error
import urllib.request

TIMEOUT = 30


def _post(url, headers, payload, stream=False):
    """POST JSON. Returns (status, body_or_reader, elapsed_to_first_byte)."""
    data = json.dumps(payload).encode()
    req = urllib.request.Request(url, data=data, headers=headers, method="POST")
    ctx = ssl.create_default_context()
    start = time.perf_counter()
    try:
        resp = urllib.request.urlopen(req, timeout=TIMEOUT, context=ctx)
        if stream:
            first = resp.readline()
            ttft = time.perf_counter() - start
            resp.read()
            return resp.status, first.decode("utf-8", "replace"), ttft
        body = resp.read().decode("utf-8", "replace")
        return resp.status, body, time.perf_counter() - start
    except urllib.error.HTTPError as e:
        return e.code, e.read().decode("utf-8", "replace"), time.perf_counter() - start
    except urllib.error.URLError as e:
        return None, f"connection failed: {e.reason}", time.perf_counter() - start
    except Exception as e:
        return None, f"{type(e).__name__}: {e}", time.perf_counter() - start


def _get(url, headers):
    req = urllib.request.Request(url, headers=headers)
    try:
        resp = urllib.request.urlopen(req, timeout=TIMEOUT, context=ssl.create_default_context())
        return resp.status, resp.read().decode("utf-8", "replace")
    except urllib.error.HTTPError as e:
        return e.code, e.read().decode("utf-8", "replace")
    except Exception as e:
        return None, f"{type(e).__name__}: {e}"


def check_openai(root, key, model):
    url = f"{root}/v1/chat/completions"
    headers = {"Authorization": f"Bearer {key}", "Content-Type": "application/json"}
    payload = {
        "model": model,
        "messages": [{"role": "user", "content": "Reply with the single word: ok"}],
        "max_tokens": 16,
    }
    status, body, elapsed = _post(url, headers, payload)
    return {"name": "OpenAI Chat Completions", "url": url, "status": status,
            "body": body, "elapsed": elapsed,
            "base_url_for_clients": f"{root}/v1"}


def check_anthropic(root, key, model):
    url = f"{root}/v1/messages"
    headers = {"x-api-key": key, "anthropic-version": "2023-06-01",
               "Content-Type": "application/json"}
    payload = {
        "model": model,
        "max_tokens": 16,
        "messages": [{"role": "user", "content": "Reply with the single word: ok"}],
    }
    status, body, elapsed = _post(url, headers, payload)
    return {"name": "Anthropic Messages", "url": url, "status": status,
            "body": body, "elapsed": elapsed,
            "base_url_for_clients": root}


def check_ttft(root, key, model):
    url = f"{root}/v1/chat/completions"
    headers = {"Authorization": f"Bearer {key}", "Content-Type": "application/json"}
    payload = {
        "model": model,
        "messages": [{"role": "user", "content": "Count from 1 to 30."}],
        "max_tokens": 200,
        "stream": True,
    }
    status, first, ttft = _post(url, headers, payload, stream=True)
    if status == 200:
        return ttft
    return None


def list_models(root, key):
    status, body = _get(f"{root}/v1/models", {"Authorization": f"Bearer {key}"})
    if status != 200:
        return None
    try:
        data = json.loads(body).get("data", [])
        return [m.get("id") for m in data if m.get("id")]
    except Exception:
        return None


def explain(status, body):
    if status == 200:
        return "OK"
    if status == 401:
        return "auth rejected — wrong key, trailing whitespace, or key belongs to a different host"
    if status == 403:
        return "authenticated but forbidden — key may lack permission for this model"
    if status == 404:
        return ("path not found — this protocol is probably not served here, "
                "or the base URL already contained the path segment")
    if status == 429:
        return "rate limited — key is valid"
    if status is None:
        return body
    snippet = body[:160].replace("\n", " ")
    return f"HTTP {status}: {snippet}"


def main():
    ap = argparse.ArgumentParser(description="Diagnose a custom LLM endpoint.")
    ap.add_argument("--url", required=True,
                    help="Site root, e.g. https://apimaster.ai (do NOT include /v1)")
    ap.add_argument("--key", required=True, help="Your API key")
    ap.add_argument("--model", default="claude-sonnet-4-6", help="Model id to test")
    ap.add_argument("--json", action="store_true", help="Machine-readable output")
    args = ap.parse_args()

    root = args.url.rstrip("/")
    for suffix in ("/v1/chat/completions", "/v1/messages", "/v1"):
        if root.endswith(suffix):
            root = root[: -len(suffix)]
            print(f"note: stripped '{suffix}' from --url; using site root {root}\n",
                  file=sys.stderr)
            break

    results = [check_openai(root, args.key, args.model),
               check_anthropic(root, args.key, args.model)]

    models = list_models(root, args.key)
    ttft = check_ttft(root, args.key, args.model) if results[0]["status"] == 200 else None

    if args.json:
        print(json.dumps({
            "root": root, "model": args.model,
            "protocols": [{k: v for k, v in r.items() if k != "body"} for r in results],
            "ttft_seconds": ttft, "models": models,
        }, indent=2))
        return 0

    print(f"\nEndpoint: {root}")
    print(f"Model:    {args.model}")
    print("=" * 62)

    working = []
    for r in results:
        mark = "PASS" if r["status"] == 200 else "FAIL"
        print(f"\n[{mark}] {r['name']}")
        print(f"       {r['url']}")
        print(f"       {explain(r['status'], r['body'])}")
        print(f"       round trip: {r['elapsed']:.2f}s")
        if r["status"] == 200:
            working.append(r)

    if ttft is not None:
        print(f"\nTTFT (streaming): {ttft:.2f}s")

    if models:
        print(f"\nModels advertised: {len(models)}")
        for m in models[:12]:
            print(f"  - {m}")
        if len(models) > 12:
            print(f"  ... and {len(models) - 12} more")
        if args.model not in models:
            print(f"\n  WARNING: '{args.model}' is not in the advertised list.")

    print("\n" + "=" * 62)
    if not working:
        print("No protocol responded successfully.")
        print("Check: key correctness, whether --url is the site root, network reachability.")
        return 1

    print("Config values to use:\n")
    names = [r["name"] for r in working]
    if "OpenAI Chat Completions" in names:
        print(f"  Cline / Kilo Code / OpenCode / Cursor / Aider / Codex:")
        print(f"      base_url = {root}/v1\n")
    if "Anthropic Messages" in names:
        print(f"  Claude Code (~/.claude/settings.json):")
        print(f"      ANTHROPIC_BASE_URL = {root}    <-- no /v1\n")
    else:
        print("  Claude Code: this endpoint does not serve the Anthropic Messages")
        print("      protocol. Use claude-code-router as a translation layer, with")
        print(f"      api_base_url = {root}/v1/chat/completions\n")

    print("Reminder: a successful response confirms the endpoint works, not that")
    print("the model behind it is what the id claims. Verify separately if it matters.")
    return 0


if __name__ == "__main__":
    sys.exit(main())