agents/cliproxyapi/health.py

"""TCP-probe the CLIProxyAPI port. Endpoint-agnostic — works regardless of
which protocol surfaces (Anthropic / OpenAI / Gemini) the proxy version
exposes."""

from __future__ import annotations

import socket
import time

from .endpoint import ProxyEndpoint


def is_ready(ep: ProxyEndpoint, timeout: float = 2.0) -> bool:
    try:
        with socket.create_connection((ep.host, ep.port), timeout=timeout):
            return True
    except OSError:
        return False


def wait_until_ready(ep: ProxyEndpoint, timeout: float = 30.0) -> None:
    deadline = time.monotonic() + timeout
    while time.monotonic() < deadline:
        if is_ready(ep):
            return
        time.sleep(0.5)
    raise SystemExit(
        f"CLIProxyAPI at {ep.base_url()} did not respond within {timeout:.0f}s.\n"
        f"Start it (e.g. `cliproxy --config ~/.cli-proxy-api/config.yaml`) "
        f"and confirm CLIPROXYAPI_HOST / CLIPROXYAPI_PORT."
    )