Python 88.3%
TypeScript 7.6%
Shell 4.1%
1#!/usr/bin/env python32"""Minimal, dependency-free live-API helper used by examples, tests and agents.34 from scripts.live import openai_request, anthropic_request, load_env5 st, body, headers = openai_request("GET", "/v1/models")6 st, body, headers = anthropic_request("POST", "/v1/messages", json={...}, beta="prompt-caching-2024-07-31")78- Loads .env from the repo root (never prints keys).9- Every call is appended (sanitized) to reports/live-requests.jsonl.10- Response bodies are returned as parsed JSON when possible, else bytes.11- `stream=True` returns an iterator of decoded lines (SSE) instead of a body.1213CLI: python3 scripts/live.py openai GET /v1/models14 python3 scripts/live.py anthropic POST /v1/messages '{"model":"claude-haiku-4-5-20251001","max_tokens":8,"messages":[{"role":"user","content":"Reply with OK."}]}'15"""16from __future__ import annotations1718import json19import os20import re21import sys22import time23import urllib.error24import urllib.request25from datetime import datetime, timezone26from pathlib import Path27from typing import Any, Iterator2829ROOT = Path(__file__).resolve().parent.parent30LIVE_LOG = ROOT / "reports" / "live-requests.jsonl"31KEY_RE = re.compile(r"sk-(?:ant-)?[A-Za-z0-9_-]{8,}|xai-[A-Za-z0-9_-]{16,}|AIza[A-Za-z0-9_-]{20,}|AQ\.[A-Za-z0-9_-]{20,}")32XAI_BASE = os.environ.get("XAI_BASE_URL", "https://api.x.ai")33GEMINI_BASE = os.environ.get("GEMINI_BASE_URL", "https://generativelanguage.googleapis.com")34OPENAI_BASE = os.environ.get("OPENAI_BASE_URL", "https://api.openai.com")35ANTHROPIC_BASE = os.environ.get("ANTHROPIC_BASE_URL", "https://api.anthropic.com")36ANTHROPIC_VERSION = os.environ.get("ANTHROPIC_VERSION", "2023-06-01")373839def load_env() -> None:40 env_file = ROOT / ".env"41 if env_file.exists():42 for line in env_file.read_text().splitlines():43 line = line.strip()44 if not line or line.startswith("#") or "=" not in line:45 continue46 k, v = line.split("=", 1)47 os.environ.setdefault(k.strip(), v.strip().strip('"').strip("'"))484950load_env()515253def mask(text: str) -> str:54 text = KEY_RE.sub("***REDACTED-KEY***", text)55 text = re.sub(r"([?&]key=)[A-Za-z0-9._-]{8,}", r"\1***REDACTED***", text)56 return re.sub(r"(?i)(bearer|x-api-key:?|x-goog-api-key:?)\s+[A-Za-z0-9._-]{8,}", r"\1 ***REDACTED***", text)575859def now() -> str:60 return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")616263def log_request(provider: str, method: str, path: str, status: int, est_cost_usd: float = 0.0, note: str = "") -> None:64 LIVE_LOG.parent.mkdir(parents=True, exist_ok=True)65 rec = {"ts": now(), "provider": provider, "method": method, "path": path, "status": status,66 "est_cost_usd": round(est_cost_usd, 6), "note": mask(note)[:300]}67 with LIVE_LOG.open("a") as f:68 f.write(json.dumps(rec) + "\n")697071def _request(url: str, method: str, headers: dict[str, str], data: bytes | None, timeout: float,72 stream: bool) -> tuple[int, Any, dict[str, str]]:73 req = urllib.request.Request(url, method=method, headers=headers, data=data)74 try:75 resp = urllib.request.urlopen(req, timeout=timeout)76 except urllib.error.HTTPError as e:77 body = e.read()78 try:79 parsed = json.loads(body)80 except Exception: # noqa: BLE00181 parsed = body82 return e.code, parsed, dict(e.headers.items())83 hdrs = dict(resp.headers.items())84 if stream:85 def gen() -> Iterator[str]:86 with resp:87 for raw in resp:88 yield raw.decode("utf-8", "replace").rstrip("\n")89 return resp.status, gen(), hdrs90 with resp:91 body = resp.read()92 ctype = hdrs.get("Content-Type", "")93 if "json" in ctype:94 try:95 return resp.status, json.loads(body), hdrs96 except Exception: # noqa: BLE00197 return resp.status, body, hdrs98 return resp.status, body, hdrs99100101def openai_request(method: str, path: str, json_body: Any = None, *, data: bytes | None = None,102 content_type: str | None = "application/json", extra_headers: dict[str, str] | None = None,103 timeout: float = 120, stream: bool = False, est_cost_usd: float = 0.0, note: str = "",104 admin: bool = False) -> tuple[int, Any, dict[str, str]]:105 key = os.environ.get("OPENAI_ADMIN_KEY") if admin else None106 key = key or os.environ["OPENAI_API_KEY"]107 headers = {"Authorization": f"Bearer {key}"}108 if content_type:109 headers["Content-Type"] = content_type110 if extra_headers:111 headers.update(extra_headers)112 if json_body is not None:113 data = json.dumps(json_body).encode()114 st, body, hdrs = _request(OPENAI_BASE + path, method, headers, data, timeout, stream)115 log_request("openai", method, path.split("?")[0], st, est_cost_usd, note)116 return st, body, hdrs117118119def anthropic_request(method: str, path: str, json_body: Any = None, *, data: bytes | None = None,120 content_type: str | None = "application/json", beta: str | None = None,121 extra_headers: dict[str, str] | None = None, timeout: float = 120, stream: bool = False,122 est_cost_usd: float = 0.0, note: str = "", admin: bool = False) -> tuple[int, Any, dict[str, str]]:123 key = os.environ.get("ANTHROPIC_ADMIN_KEY") if admin else None124 key = key or os.environ["ANTHROPIC_API_KEY"]125 headers = {"x-api-key": key, "anthropic-version": ANTHROPIC_VERSION}126 if content_type:127 headers["Content-Type"] = content_type128 if beta:129 headers["anthropic-beta"] = beta130 if extra_headers:131 headers.update(extra_headers)132 if json_body is not None:133 data = json.dumps(json_body).encode()134 st, body, hdrs = _request(ANTHROPIC_BASE + path, method, headers, data, timeout, stream)135 log_request("anthropic", method, path.split("?")[0], st, est_cost_usd, note)136 return st, body, hdrs137138139def xai_request(method: str, path: str, json_body: Any = None, *, data: bytes | None = None,140 content_type: str | None = "application/json", extra_headers: dict[str, str] | None = None,141 timeout: float = 120, stream: bool = False, est_cost_usd: float = 0.0, note: str = "",142 management: bool = False) -> tuple[int, Any, dict[str, str]]:143 """xAI REST API (OpenAI-compatible surface + xAI-specific endpoints). management=True targets https://management-api.x.ai144 with XAI_MANAGEMENT_KEY if set (Management API keys are distinct from inference keys)."""145 key = os.environ.get("XAI_MANAGEMENT_KEY") if management else None146 key = key or os.environ["XAI_API_KEY"]147 base = os.environ.get("XAI_MANAGEMENT_BASE_URL", "https://management-api.x.ai") if management else XAI_BASE148 headers = {"Authorization": f"Bearer {key}"}149 if content_type:150 headers["Content-Type"] = content_type151 if extra_headers:152 headers.update(extra_headers)153 if json_body is not None:154 data = json.dumps(json_body).encode()155 st, body, hdrs = _request(base + path, method, headers, data, timeout, stream)156 log_request("xai", method, path.split("?")[0], st, est_cost_usd, note)157 return st, body, hdrs158159160def gemini_request(method: str, path: str, json_body: Any = None, *, data: bytes | None = None,161 content_type: str | None = "application/json", extra_headers: dict[str, str] | None = None,162 timeout: float = 120, stream: bool = False, est_cost_usd: float = 0.0,163 note: str = "") -> tuple[int, Any, dict[str, str]]:164 """Gemini API (generativelanguage.googleapis.com). `path` includes the version, e.g.165 '/v1beta/models/gemini-2.5-flash:generateContent'. The key is sent as the x-goog-api-key header (never in the URL)."""166 headers = {"x-goog-api-key": os.environ["GEMINI_API_KEY"]}167 if content_type:168 headers["Content-Type"] = content_type169 if extra_headers:170 headers.update(extra_headers)171 if json_body is not None:172 data = json.dumps(json_body).encode()173 st, body, hdrs = _request(GEMINI_BASE + path, method, headers, data, timeout, stream)174 log_request("gemini", method, re.sub(r"models/[^:/?]+", "models/{model}", path.split("?")[0]), st, est_cost_usd, note)175 return st, body, hdrs176177178def interesting_headers(hdrs: dict[str, str]) -> dict[str, str]:179 """Rate-limit / request-id headers worth recording (never auth)."""180 keep = ("x-ratelimit", "anthropic-ratelimit", "x-goog", "server-timing", "alt-svc-", "x-xai", "retry-after", "x-request-id", "request-id",181 "openai-processing-ms", "openai-version", "openai-organization", "openai-project", "cf-ray",182 "anthropic-organization-id", "x-should-retry")183 return {k: v for k, v in hdrs.items() if k.lower().startswith(keep)}184185186def save_sanitized(obj: Any, dest: Path) -> None:187 dest.parent.mkdir(parents=True, exist_ok=True)188 text = obj if isinstance(obj, str) else json.dumps(obj, indent=2, ensure_ascii=False, default=str)189 dest.write_text(mask(text))190191192if __name__ == "__main__":193 if len(sys.argv) < 4:194 print(__doc__)195 sys.exit(1)196 provider, method, path = sys.argv[1], sys.argv[2].upper(), sys.argv[3]197 body = json.loads(sys.argv[4]) if len(sys.argv) > 4 else None198 beta = os.environ.get("ANTHROPIC_BETA")199 t0 = time.time()200 if provider == "openai":201 st, out, hdrs = openai_request(method, path, body)202 elif provider == "xai":203 st, out, hdrs = xai_request(method, path, body)204 elif provider == "gemini":205 st, out, hdrs = gemini_request(method, path, body)206 else:207 st, out, hdrs = anthropic_request(method, path, body, beta=beta)208 print(f"HTTP {st} ({time.time() - t0:.2f}s)", file=sys.stderr)209 print(mask(json.dumps(interesting_headers(hdrs), indent=1)), file=sys.stderr)210 print(mask(json.dumps(out, indent=2, ensure_ascii=False) if not isinstance(out, bytes) else f"<{len(out)} bytes>"))211