Spaces:
Sleeping
Sleeping
File size: 2,346 Bytes
652136d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 | """Divination AI — runtime client for the Modal backend."""
from __future__ import annotations
import os
import httpx
API_URL = os.environ.get(
"DIVINATION_AI_API_URL",
"https://rafalbogusdxc--divination-ai-api.modal.run",
).rstrip("/")
TIMEOUT_S = 900
class BackendError(RuntimeError):
pass
def health() -> dict:
try:
resp = httpx.get(f"{API_URL}/health", timeout=10, follow_redirects=True)
resp.raise_for_status()
return resp.json()
except Exception:
return {"status": "unreachable"}
def intake(messages: list[dict]) -> dict:
try:
resp = httpx.post(
f"{API_URL}/intake",
json={"messages": messages},
timeout=TIMEOUT_S,
follow_redirects=True,
)
resp.raise_for_status()
return resp.json()
except httpx.ConnectError as e:
raise BackendError("Cannot reach the Oracle backend — is the Modal app deployed?") from e
except httpx.ReadTimeout as e:
raise BackendError("The Oracle is taking too long to respond (cold start). Try again in ~1 minute.") from e
except httpx.HTTPStatusError as e:
raise BackendError(f"Backend error {e.response.status_code}: {e.response.text[:300]}") from e
def generate(messages: list[dict]) -> dict:
try:
resp = httpx.post(
f"{API_URL}/generate",
json={"messages": messages},
timeout=TIMEOUT_S,
follow_redirects=True,
)
resp.raise_for_status()
return resp.json()
except httpx.ConnectError as e:
raise BackendError("Cannot reach the Oracle backend — is the Modal app deployed?") from e
except httpx.ReadTimeout as e:
raise BackendError("The Oracle is taking too long (generation in progress). Please wait...") from e
except httpx.HTTPStatusError as e:
raise BackendError(f"Backend error {e.response.status_code}: {e.response.text[:300]}") from e
def share_trace(messages: list[dict]) -> dict:
try:
resp = httpx.post(
f"{API_URL}/share",
json={"messages": messages},
timeout=30,
follow_redirects=True,
)
resp.raise_for_status()
return resp.json()
except Exception:
return {"shared": False, "reason": "Network error"}
|