Timothy
298 lines
11 KiB
Python
298 lines
11 KiB
Python
"""Validate an LLM API key without consuming tokens.
|
|
|
|
Usage:
|
|
python scripts/check_llm_key.py <provider_id> <api_key> [api_base] [model]
|
|
|
|
Exit codes:
|
|
0 = valid key
|
|
1 = invalid key
|
|
2 = inconclusive (timeout, network error)
|
|
|
|
Output: single JSON line {"valid": bool, "message": str}
|
|
"""
|
|
|
|
import json
|
|
import sys
|
|
|
|
import httpx
|
|
|
|
from framework.config import HIVE_LLM_ENDPOINT
|
|
|
|
TIMEOUT = 10.0
|
|
|
|
|
|
def _extract_error_message(response: httpx.Response) -> str:
|
|
"""Best-effort extraction of a provider error message."""
|
|
try:
|
|
payload = response.json()
|
|
except Exception:
|
|
text = (response.text or "").strip()
|
|
return text[:240] if text else ""
|
|
|
|
if isinstance(payload, dict):
|
|
error_value = payload.get("error")
|
|
if isinstance(error_value, dict):
|
|
message = error_value.get("message")
|
|
if isinstance(message, str) and message.strip():
|
|
return message.strip()
|
|
if isinstance(error_value, str) and error_value.strip():
|
|
return error_value.strip()
|
|
message = payload.get("message")
|
|
if isinstance(message, str) and message.strip():
|
|
return message.strip()
|
|
|
|
return ""
|
|
|
|
|
|
def check_anthropic(api_key: str, **_: str) -> dict:
|
|
"""Send empty messages to trigger 400 without consuming tokens."""
|
|
with httpx.Client(timeout=TIMEOUT) as client:
|
|
r = client.post(
|
|
"https://api.anthropic.com/v1/messages",
|
|
headers={
|
|
"x-api-key": api_key,
|
|
"anthropic-version": "2023-06-01",
|
|
"Content-Type": "application/json",
|
|
},
|
|
json={"model": "claude-sonnet-4-20250514", "max_tokens": 1, "messages": []},
|
|
)
|
|
if r.status_code in (200, 400, 429):
|
|
return {"valid": True, "message": "API key valid"}
|
|
if r.status_code == 401:
|
|
return {"valid": False, "message": "Invalid API key"}
|
|
if r.status_code == 403:
|
|
return {"valid": False, "message": "API key lacks permissions"}
|
|
return {"valid": False, "message": f"Unexpected status {r.status_code}"}
|
|
|
|
|
|
def check_openai_compatible(api_key: str, endpoint: str, name: str) -> dict:
|
|
"""GET /models on any OpenAI-compatible API."""
|
|
with httpx.Client(timeout=TIMEOUT) as client:
|
|
r = client.get(
|
|
endpoint,
|
|
headers={"Authorization": f"Bearer {api_key}"},
|
|
)
|
|
if r.status_code in (200, 429):
|
|
return {"valid": True, "message": f"{name} API key valid"}
|
|
if r.status_code == 401:
|
|
return {"valid": False, "message": f"Invalid {name} API key"}
|
|
if r.status_code == 403:
|
|
return {"valid": False, "message": f"{name} API key lacks permissions"}
|
|
return {"valid": False, "message": f"{name} API returned status {r.status_code}"}
|
|
|
|
|
|
def check_openrouter(
|
|
api_key: str, api_base: str = "https://openrouter.ai/api/v1", **_: str
|
|
) -> dict:
|
|
"""Validate OpenRouter key against GET /models."""
|
|
endpoint = f"{api_base.rstrip('/')}/models"
|
|
with httpx.Client(timeout=TIMEOUT) as client:
|
|
r = client.get(endpoint, headers={"Authorization": f"Bearer {api_key}"})
|
|
if r.status_code in (200, 429):
|
|
return {"valid": True, "message": "OpenRouter API key valid"}
|
|
if r.status_code == 401:
|
|
return {"valid": False, "message": "Invalid OpenRouter API key"}
|
|
if r.status_code == 403:
|
|
return {"valid": False, "message": "OpenRouter API key lacks permissions"}
|
|
return {"valid": False, "message": f"OpenRouter API returned status {r.status_code}"}
|
|
|
|
|
|
def check_openrouter_model(
|
|
api_key: str,
|
|
model: str,
|
|
api_base: str = "https://openrouter.ai/api/v1",
|
|
**_: str,
|
|
) -> dict:
|
|
"""Validate that an OpenRouter model ID is routable with this key."""
|
|
endpoint = f"{api_base.rstrip('/')}/chat/completions"
|
|
with httpx.Client(timeout=TIMEOUT) as client:
|
|
r = client.post(
|
|
endpoint,
|
|
headers={
|
|
"Authorization": f"Bearer {api_key}",
|
|
"Content-Type": "application/json",
|
|
},
|
|
json={
|
|
"model": model,
|
|
"messages": [{"role": "user", "content": "Hello, are you working?"}],
|
|
"max_tokens": 1,
|
|
},
|
|
)
|
|
if r.status_code == 200:
|
|
return {"valid": True, "message": f"OpenRouter model is available: {model}"}
|
|
if r.status_code == 429:
|
|
return {
|
|
"valid": True,
|
|
"message": "OpenRouter model check rate-limited; assuming model is reachable",
|
|
}
|
|
if r.status_code == 401:
|
|
return {"valid": False, "message": "Invalid OpenRouter API key"}
|
|
if r.status_code == 403:
|
|
return {"valid": False, "message": "OpenRouter API key lacks permissions"}
|
|
|
|
detail = _extract_error_message(r)
|
|
if r.status_code in (400, 404, 422):
|
|
base = f"OpenRouter model is not available: {model}"
|
|
return {"valid": False, "message": f"{base}. {detail}" if detail else base}
|
|
|
|
suffix = f": {detail}" if detail else ""
|
|
return {
|
|
"valid": False,
|
|
"message": f"OpenRouter model check returned status {r.status_code}{suffix}",
|
|
}
|
|
|
|
|
|
def check_minimax(
|
|
api_key: str, api_base: str = "https://api.minimax.io/v1", **_: str
|
|
) -> dict:
|
|
"""Validate via chatcompletion_v2 endpoint with empty messages.
|
|
|
|
MiniMax doesn't support GET /models; their native endpoint is
|
|
/v1/text/chatcompletion_v2.
|
|
"""
|
|
with httpx.Client(timeout=TIMEOUT) as client:
|
|
r = client.post(
|
|
f"{api_base.rstrip('/')}/text/chatcompletion_v2",
|
|
headers={
|
|
"Authorization": f"Bearer {api_key}",
|
|
"Content-Type": "application/json",
|
|
},
|
|
json={"model": "MiniMax-M2.5", "messages": []},
|
|
)
|
|
if r.status_code in (200, 400, 422, 429):
|
|
return {"valid": True, "message": "MiniMax API key valid"}
|
|
if r.status_code == 401:
|
|
return {"valid": False, "message": "Invalid MiniMax API key"}
|
|
if r.status_code == 403:
|
|
return {"valid": False, "message": "MiniMax API key lacks permissions"}
|
|
return {"valid": False, "message": f"MiniMax API returned status {r.status_code}"}
|
|
|
|
|
|
def check_anthropic_compatible(api_key: str, endpoint: str, name: str) -> dict:
|
|
"""POST empty messages to an Anthropic-compatible endpoint to validate key."""
|
|
with httpx.Client(timeout=TIMEOUT) as client:
|
|
r = client.post(
|
|
endpoint,
|
|
headers={
|
|
"x-api-key": api_key,
|
|
"anthropic-version": "2023-06-01",
|
|
"Content-Type": "application/json",
|
|
},
|
|
json={"model": "kimi-k2.5", "max_tokens": 1, "messages": []},
|
|
)
|
|
if r.status_code in (200, 400, 429):
|
|
return {"valid": True, "message": f"{name} API key valid"}
|
|
if r.status_code == 401:
|
|
return {"valid": False, "message": f"Invalid {name} API key"}
|
|
if r.status_code == 403:
|
|
return {"valid": False, "message": f"{name} API key lacks permissions"}
|
|
return {"valid": False, "message": f"{name} API returned status {r.status_code}"}
|
|
|
|
|
|
def check_gemini(api_key: str, **_: str) -> dict:
|
|
"""List models with query param auth."""
|
|
with httpx.Client(timeout=TIMEOUT) as client:
|
|
r = client.get(
|
|
"https://generativelanguage.googleapis.com/v1beta/models",
|
|
params={"key": api_key},
|
|
)
|
|
if r.status_code in (200, 429):
|
|
return {"valid": True, "message": "Gemini API key valid"}
|
|
if r.status_code in (400, 401, 403):
|
|
return {"valid": False, "message": "Invalid Gemini API key"}
|
|
return {"valid": False, "message": f"Gemini API returned status {r.status_code}"}
|
|
|
|
|
|
PROVIDERS = {
|
|
"anthropic": lambda key, **kw: check_anthropic(key),
|
|
"openai": lambda key, **kw: check_openai_compatible(
|
|
key, "https://api.openai.com/v1/models", "OpenAI"
|
|
),
|
|
"gemini": lambda key, **kw: check_gemini(key),
|
|
"groq": lambda key, **kw: check_openai_compatible(
|
|
key, "https://api.groq.com/openai/v1/models", "Groq"
|
|
),
|
|
"cerebras": lambda key, **kw: check_openai_compatible(
|
|
key, "https://api.cerebras.ai/v1/models", "Cerebras"
|
|
),
|
|
"openrouter": lambda key, **kw: check_openrouter(key, **kw),
|
|
"minimax": lambda key, **kw: check_minimax(key),
|
|
# Kimi For Coding uses an Anthropic-compatible endpoint; check via /v1/messages
|
|
# with empty messages (same as check_anthropic, triggers 400 not 401).
|
|
"kimi": lambda key, **kw: check_anthropic_compatible(
|
|
key, "https://api.kimi.com/coding/v1/messages", "Kimi"
|
|
),
|
|
# Hive LLM uses an Anthropic-compatible endpoint
|
|
"hive": lambda key, **kw: check_anthropic_compatible(
|
|
key, f"{HIVE_LLM_ENDPOINT}/v1/messages", "Hive"
|
|
),
|
|
}
|
|
|
|
|
|
def main() -> None:
|
|
if len(sys.argv) < 3:
|
|
print(
|
|
json.dumps(
|
|
{
|
|
"valid": False,
|
|
"message": "Usage: check_llm_key.py <provider> <key> [api_base] [model]",
|
|
}
|
|
)
|
|
)
|
|
sys.exit(2)
|
|
|
|
provider_id = sys.argv[1]
|
|
api_key = sys.argv[2]
|
|
api_base = sys.argv[3] if len(sys.argv) > 3 else ""
|
|
model = sys.argv[4] if len(sys.argv) > 4 else ""
|
|
|
|
try:
|
|
if provider_id == "openrouter" and model:
|
|
result = check_openrouter_model(
|
|
api_key,
|
|
model=model,
|
|
api_base=(api_base or "https://openrouter.ai/api/v1"),
|
|
)
|
|
elif api_base and provider_id == "minimax":
|
|
result = check_minimax(api_key, api_base)
|
|
elif api_base and provider_id == "openrouter":
|
|
result = check_openrouter(api_key, api_base)
|
|
elif api_base and provider_id == "kimi":
|
|
# Kimi uses an Anthropic-compatible endpoint; check via /v1/messages
|
|
result = check_anthropic_compatible(
|
|
api_key, api_base.rstrip("/") + "/v1/messages", "Kimi"
|
|
)
|
|
elif api_base and provider_id == "hive":
|
|
result = check_anthropic_compatible(
|
|
api_key, api_base.rstrip("/") + "/v1/messages", "Hive"
|
|
)
|
|
elif api_base:
|
|
# Custom API base (ZAI or other OpenAI-compatible)
|
|
endpoint = api_base.rstrip("/") + "/models"
|
|
name = {"zai": "ZAI"}.get(provider_id, "Custom provider")
|
|
result = check_openai_compatible(api_key, endpoint, name)
|
|
elif provider_id in PROVIDERS:
|
|
result = PROVIDERS[provider_id](api_key)
|
|
else:
|
|
result = {"valid": True, "message": f"No health check for {provider_id}"}
|
|
print(json.dumps(result))
|
|
sys.exit(0)
|
|
|
|
print(json.dumps(result))
|
|
sys.exit(0 if result["valid"] else 1)
|
|
|
|
except httpx.TimeoutException:
|
|
print(json.dumps({"valid": None, "message": "Request timed out"}))
|
|
sys.exit(2)
|
|
except httpx.RequestError as e:
|
|
msg = str(e)
|
|
# Redact key from error messages
|
|
if api_key in msg:
|
|
msg = msg.replace(api_key, "***")
|
|
print(json.dumps({"valid": None, "message": f"Connection failed: {msg}"}))
|
|
sys.exit(2)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|