tomasz-syn-grzegorza
359afb3a59
Infrastructure configs for GMKtec K11 (Docker, vLLM, LocalAI, ComfyUI, control-plane, gpu-fan agent, Server UI with CLI/file explorer/GPU fan curve). Co-authored-by: Cursor <cursoragent@cursor.com>
76 lines
2.6 KiB
Python
76 lines
2.6 KiB
Python
"""Proxy requests to the gpu-fan host agent."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
import urllib.error
|
|
import urllib.request
|
|
from typing import Any
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
|
|
class AgentProxyError(Exception):
|
|
"""Raised when the gpu-fan agent request fails."""
|
|
|
|
|
|
def check_agent_health(agent_url: str, agent_api_key: str) -> dict[str, Any]:
|
|
"""Return health info for the gpu-fan agent."""
|
|
try:
|
|
forward_request(agent_url, "GET", "status", agent_api_key=agent_api_key)
|
|
return {"ok": True, "agent_url": agent_url}
|
|
except AgentProxyError as exc:
|
|
return {"ok": False, "agent_url": agent_url, "error": str(exc)}
|
|
|
|
|
|
def resolve_agent_url(configured_url: str, agent_api_key: str) -> str:
|
|
"""Pick first reachable agent URL (18090 preferred, 8090 legacy fallback)."""
|
|
candidates: list[str] = []
|
|
if configured_url:
|
|
candidates.append(configured_url.rstrip("/"))
|
|
for url in ("http://127.0.0.1:18090", "http://127.0.0.1:8090"):
|
|
if url not in candidates:
|
|
candidates.append(url)
|
|
for url in candidates:
|
|
if check_agent_health(url, agent_api_key).get("ok"):
|
|
log.info("GPU fan agent reachable at %s", url)
|
|
return url
|
|
return candidates[0] if candidates else "http://127.0.0.1:18090"
|
|
|
|
|
|
def forward_request(
|
|
agent_url: str,
|
|
method: str,
|
|
path: str,
|
|
*,
|
|
body: bytes | None = None,
|
|
agent_api_key: str,
|
|
content_type: str = "application/json",
|
|
) -> tuple[int, dict[str, str], bytes]:
|
|
"""Forward a request to the gpu-fan agent API."""
|
|
path = path.lstrip("/")
|
|
url = f"{agent_url.rstrip('/')}/api/{path}"
|
|
headers = {"Accept": "application/json"}
|
|
if agent_api_key:
|
|
headers["X-API-Key"] = agent_api_key
|
|
if body is not None:
|
|
headers["Content-Type"] = content_type
|
|
|
|
req = urllib.request.Request(url, data=body, headers=headers, method=method.upper())
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=10) as resp:
|
|
return resp.status, dict(resp.headers), resp.read()
|
|
except urllib.error.HTTPError as exc:
|
|
payload = exc.read()
|
|
detail = payload.decode("utf-8", errors="replace")
|
|
try:
|
|
parsed = json.loads(detail)
|
|
if isinstance(parsed, dict) and "detail" in parsed:
|
|
detail = str(parsed["detail"])
|
|
except json.JSONDecodeError:
|
|
pass
|
|
raise AgentProxyError(detail or f"Agent HTTP {exc.code}") from exc
|
|
except urllib.error.URLError as exc:
|
|
raise AgentProxyError(f"Agent unreachable at {agent_url}: {exc.reason}") from exc
|