""" Validate Azure OpenAI connectivity using settings from .env. Checks performed (in order): 1. Required env vars are set and non-empty 2. Endpoint URL is reachable (HTTPS GET → should return 401/404, not timeout) 3. API key authenticates (list deployments via Management-style probe) 4. Target deployment exists and accepts a minimal chat completion Usage (from repo root, outside Docker): python -m scripts.validate_azure Usage (inside Docker container): docker-compose run --rm backend python -m scripts.validate_azure """ import json import os import pathlib import sys import urllib.request import urllib.error PASS = "✅" FAIL = "❌" WARN = "⚠️ " # ── Load .env without any third-party deps ──────────────────────────────────── def _load_dotenv() -> None: """Parse the nearest .env file and inject missing keys into os.environ.""" script_dir = pathlib.Path(__file__).resolve().parent candidates = [ script_dir, # backend/scripts/ script_dir.parent, # backend/ script_dir.parent.parent, # repo root ← .env lives here pathlib.Path.cwd(), # wherever python was invoked from ] for directory in candidates: env_file = directory / ".env" if env_file.exists(): print(f" Loading .env from: {env_file}") with open(env_file) as f: for line in f: line = line.strip() if not line or line.startswith("#") or "=" not in line: continue key, _, value = line.partition("=") key = key.strip() value = value.strip().strip('"').strip("'") if key and key not in os.environ: os.environ[key] = value return print(f" {WARN} No .env file found in any of: {[str(d) for d in candidates]}") class _Settings: def __init__(self) -> None: _load_dotenv() self.model_provider = os.environ.get("MODEL_PROVIDER", "openai") self.azure_openai_api_key = os.environ.get("AZURE_OPENAI_API_KEY", "") self.azure_openai_endpoint = os.environ.get("AZURE_OPENAI_ENDPOINT", "") self.azure_openai_deployment = os.environ.get("AZURE_OPENAI_DEPLOYMENT", "gpt-4o") self.azure_openai_api_version = os.environ.get("AZURE_OPENAI_API_VERSION", "2024-02-01") def get_settings() -> _Settings: return _Settings() def section(title: str) -> None: print(f"\n{'─' * 55}") print(f" {title}") print(f"{'─' * 55}") def ok(msg: str) -> None: print(f" {PASS} {msg}") def fail(msg: str) -> None: print(f" {FAIL} {msg}") def warn(msg: str) -> None: print(f" {WARN} {msg}") # ── 1. Env var presence ─────────────────────────────────────────────────────── def check_env(s) -> bool: section("1 · Environment variables") errors = [] fields = { "AZURE_OPENAI_API_KEY": s.azure_openai_api_key, "AZURE_OPENAI_ENDPOINT": s.azure_openai_endpoint, "AZURE_OPENAI_DEPLOYMENT": s.azure_openai_deployment, "AZURE_OPENAI_API_VERSION": s.azure_openai_api_version, } for name, value in fields.items(): if not value or value.startswith("<"): fail(f"{name} is not set or still contains a placeholder") errors.append(name) else: display = value if "KEY" not in name else value[:8] + "…" ok(f"{name} = {display}") endpoint = s.azure_openai_endpoint if endpoint and not endpoint.startswith("https://"): fail("AZURE_OPENAI_ENDPOINT must start with 'https://'") errors.append("endpoint-scheme") elif endpoint and not endpoint.rstrip("/").endswith(".openai.azure.com"): warn("AZURE_OPENAI_ENDPOINT does not end with '.openai.azure.com' — unusual but may be valid for sovereign clouds") return len(errors) == 0 # ── 2. Endpoint reachability ────────────────────────────────────────────────── def check_reachability(s) -> bool: section("2 · Endpoint reachability") # Probe the root of the Azure OpenAI resource — expect 401 or 404, not a network error probe_url = s.azure_openai_endpoint.rstrip("/") + "/" try: req = urllib.request.Request(probe_url, method="GET") urllib.request.urlopen(req, timeout=8) ok(f"Endpoint reachable: {probe_url}") return True except urllib.error.HTTPError as e: # 401 Unauthorized or 404 are fine — the host responded if e.code in (401, 403, 404): ok(f"Endpoint reachable (HTTP {e.code}): {probe_url}") return True fail(f"Unexpected HTTP {e.code} from endpoint: {probe_url}") return False except urllib.error.URLError as e: fail(f"Cannot reach endpoint: {probe_url}\n Reason: {e.reason}") print("\n Possible causes:") print(" • AZURE_OPENAI_ENDPOINT typo or wrong region") print(" • Outbound HTTPS blocked (firewall / proxy)") print(" • Running inside Docker without outbound internet access") return False # ── 3. List deployed models ─────────────────────────────────────────────────── def check_deployments(s) -> list[str]: """ Calls the Azure OpenAI REST API to list model deployments. Returns list of deployment names found, or empty list on failure. """ section("3 · Deployed models (REST list)") url = ( s.azure_openai_endpoint.rstrip("/") + f"/openai/deployments?api-version={s.azure_openai_api_version}" ) req = urllib.request.Request(url, headers={"api-key": s.azure_openai_api_key}) try: with urllib.request.urlopen(req, timeout=10) as resp: data = json.loads(resp.read()) deployments = [d["id"] for d in data.get("data", [])] if deployments: ok(f"Found {len(deployments)} deployment(s):") for name in deployments: print(f" • {name}") else: warn("No deployments returned — the resource may be empty") return deployments except urllib.error.HTTPError as e: body = e.read().decode(errors="replace") if e.code == 401: fail("Authentication failed (401) — AZURE_OPENAI_API_KEY is wrong or expired") elif e.code == 404: fail(f"404 listing deployments — endpoint or API version may be wrong\n URL tried: {url}") print("\n Common fixes:") print(" • Check AZURE_OPENAI_ENDPOINT matches your Azure portal resource URL") print(" • Try API version 2024-02-01, 2024-05-01-preview, or 2025-01-01-preview") else: fail(f"HTTP {e.code}: {body[:200]}") return [] except Exception as e: fail(f"Unexpected error: {e}") return [] # ── 4. Minimal chat completion against the configured deployment ────────────── def check_completion(s) -> bool: section(f"4 · Chat completion ({s.azure_openai_deployment})") url = ( s.azure_openai_endpoint.rstrip("/") + f"/openai/deployments/{s.azure_openai_deployment}" + f"/chat/completions?api-version={s.azure_openai_api_version}" ) payload = json.dumps({ "messages": [{"role": "user", "content": "Reply with the single word OK."}], "max_tokens": 5, }).encode() req = urllib.request.Request( url, data=payload, headers={"api-key": s.azure_openai_api_key, "Content-Type": "application/json"}, method="POST", ) try: with urllib.request.urlopen(req, timeout=20) as resp: data = json.loads(resp.read()) reply = data["choices"][0]["message"]["content"].strip() ok(f"Completion succeeded — model replied: {reply!r}") model_id = data.get("model", "unknown") ok(f"Serving model ID: {model_id}") return True except urllib.error.HTTPError as e: body = e.read().decode(errors="replace") if e.code == 404: fail( f"404 on deployment '{s.azure_openai_deployment}' — deployment not found\n" f" URL tried: {url}" ) print("\n Common fixes:") print(" • AZURE_OPENAI_DEPLOYMENT must match the deployment name in Azure portal") print(" (deployment name ≠ model name — check 'Deployments' tab in Azure AI Foundry)") print(f" • Current value: {s.azure_openai_deployment!r}") elif e.code == 401: fail("Authentication failed (401) — API key rejected for this deployment") else: fail(f"HTTP {e.code}: {body[:300]}") return False except Exception as e: fail(f"Unexpected error: {e}") return False # ── Main ────────────────────────────────────────────────────────────────────── def main() -> None: print("\n══════════════════════════════════════════════════════") print(" Azure OpenAI Connectivity Validator") print("══════════════════════════════════════════════════════") s = get_settings() if s.model_provider != "azure_openai": warn(f"MODEL_PROVIDER is currently '{s.model_provider}', not 'azure_openai'") warn("Validation will still run using the Azure credentials in .env") env_ok = check_env(s) if not env_ok: print("\n Stopping — fix the missing env vars above first.\n") sys.exit(1) reach_ok = check_reachability(s) if not reach_ok: print("\n Stopping — endpoint unreachable.\n") sys.exit(1) deployments = check_deployments(s) completion_ok = check_completion(s) # ── Summary ─────────────────────────────────────────────────────────────── section("Summary") if completion_ok: ok("Azure OpenAI is correctly configured and reachable.") ok(f"Set MODEL_PROVIDER=azure_openai in .env to use it.") else: fail("Chat completion failed — see details above.") if deployments and s.azure_openai_deployment not in deployments: print(f"\n {WARN} Deployment mismatch detected:") print(f" .env has: AZURE_OPENAI_DEPLOYMENT={s.azure_openai_deployment!r}") print(f" Azure has: {deployments}") print(f"\n Update AZURE_OPENAI_DEPLOYMENT to one of the names above.") sys.exit(1) print() if __name__ == "__main__": main()