Piramyd API Compatibility Matrix and Feature Flags

import httpx
import sys

BASE = "https://api.piramyd.cloud/v1"


def check_capabilities(required: list[str]) -> dict:
    """
    Fetch the capabilities matrix and assert that every required
    feature is available. Raises RuntimeError if any are missing.
    """
    resp = httpx.get(f"{BASE}/capabilities")
    resp.raise_for_status()
    payload = resp.json()

    compat = payload["compatibility"]
    missing = [feat for feat in required if not compat.get(feat)]

    if missing:
        raise RuntimeError(
            f"Required capabilities not available: {missing}\n"
            f"Full matrix: {compat}"
        )

    return compat


# Guard at startup — fail fast if the gateway doesn't support streaming
# or tool-call usage reporting
try:
    caps = check_capabilities([
        "chat_completions",
        "streaming_sse",
        "stream_include_usage",
    ])
    print(f"Gateway ready — api_version={caps.get('api_version', 'unknown')}")
except RuntimeError as exc:
    print(f"Startup check failed: {exc}", file=sys.stderr)
    sys.exit(1)

const BASE = "https://api.piramyd.cloud/v1";

interface Capabilities {
  service: string;
  api_version: string;
  compatibility: Record<string, boolean>;
}

async function checkCapabilities(required: string[]): Promise<Capabilities["compatibility"]> {
  const resp = await fetch(`${BASE}/capabilities`);
  if (!resp.ok) throw new Error(`Capabilities check failed: ${resp.status}`);

  const payload: Capabilities = await resp.json();
  const compat = payload.compatibility;

  const missing = required.filter((feat) => !compat[feat]);
  if (missing.length > 0) {
    throw new Error(
      `Required capabilities not available: ${missing.join(", ")}`
    );
  }

  return compat;
}

// Guard at startup
const caps = await checkCapabilities([
  "chat_completions",
  "streaming_sse",
  "stream_include_usage",
]);
console.log("Gateway ready:", caps);

import httpx

BASE = "https://api.piramyd.cloud"

def is_gateway_healthy() -> bool:
    try:
        resp = httpx.get(f"{BASE}/health", timeout=5)
        return resp.status_code == 200
    except httpx.RequestError:
        return False

def get_model_health() -> dict:
    resp = httpx.get(f"{BASE}/v1/status/models", timeout=10)
    resp.raise_for_status()
    return {m["id"]: m["status"] for m in resp.json().get("models", [])}

# Quick startup sanity check
if not is_gateway_healthy():
    raise RuntimeError("Piramyd gateway is not reachable")

model_statuses = get_model_health()
healthy_models = [mid for mid, status in model_statuses.items() if status == "healthy"]
print(f"{len(healthy_models)} healthy model(s) available")

const BASE = "https://api.piramyd.cloud";

async function isGatewayHealthy(): Promise<boolean> {
  try {
    const resp = await fetch(`${BASE}/health`, { signal: AbortSignal.timeout(5000) });
    return resp.ok;
  } catch {
    return false;
  }
}

async function getHealthyModelIds(): Promise<string[]> {
  const resp = await fetch(`${BASE}/v1/status/models`);
  if (!resp.ok) throw new Error(`Status check failed: ${resp.status}`);
  const { models } = await resp.json();
  return (models as any[])
    .filter((m) => m.status === "healthy")
    .map((m) => m.id);
}

if (!(await isGatewayHealthy())) {
  throw new Error("Piramyd gateway is not reachable");
}

const healthyIds = await getHealthyModelIds();
console.log(`${healthyIds.length} healthy model(s):`, healthyIds);