Kayos
130f96a34f
LAN-only Flask API that consumes Mealie (source of truth for recipes / plans
/ lists) and clawdforge (centralized claude -p runner) to do AI work.
v0.1 surface:
GET /healthz liveness + clawdforge upstream
GET /api/recipes proxy Mealie recipe list
POST /api/sterilize/preview/<slug> dry-run AI parse, return proposals
POST /api/sterilize/apply/<slug> write parses back to Mealie
Why sterilizer first: Mealie's CRF parser is mediocre and Cobb's hand-typed
recipes have lots of free-form ingredient strings ("about 2 cups cooked
white rice", "a pinch of salt") that don't aggregate cleanly into a
shopping list. We batch all ingredients of one recipe into a single Sonnet
call via clawdforge, get back parallel structured parses, then on apply
link each to Mealie food/unit records (creating missing by name) and PUT
the recipe back. Preview is non-destructive.
No UI in v0.1 — bearer-auth API only. Frontend + Authentik OIDC + Abby's
swamp/meadow/forest palette arrives in v0.2.
Auth: simple shared bearer in env (ADMIN_BEARER) until OIDC lands. LAN-only
deploy means the bearer is the only gate; no public exposure.
Stack: python:3.12-slim + Flask 3 + gunicorn + requests. No DB in v0.1.
64 lines
2 KiB
Python
64 lines
2 KiB
Python
"""Thin HTTP client for clawdforge — we're a consumer."""
|
|
import requests
|
|
|
|
|
|
class ForgeError(RuntimeError):
|
|
pass
|
|
|
|
|
|
class Forge:
|
|
def __init__(self, *, base_url: str, token: str, default_model: str, default_timeout: int):
|
|
self.base_url = base_url.rstrip("/")
|
|
self.token = token
|
|
self.default_model = default_model
|
|
self.default_timeout = default_timeout
|
|
|
|
def _headers(self) -> dict:
|
|
return {"Authorization": f"Bearer {self.token}"}
|
|
|
|
def healthz(self) -> dict:
|
|
r = requests.get(f"{self.base_url}/healthz", headers=self._headers(), timeout=10)
|
|
r.raise_for_status()
|
|
return r.json()
|
|
|
|
def run(
|
|
self,
|
|
prompt: str,
|
|
*,
|
|
model: str | None = None,
|
|
system: str | None = None,
|
|
files: list[str] | None = None,
|
|
timeout_secs: int | None = None,
|
|
) -> dict:
|
|
"""POST /run. Returns parsed result dict on success.
|
|
|
|
Raises ForgeError on transport or upstream failure. The 'result' field
|
|
in the return is whatever clawdforge parsed out of `claude -p` — usually
|
|
a dict (when the prompt asked for JSON), occasionally a string.
|
|
"""
|
|
body = {"prompt": prompt, "model": model or self.default_model}
|
|
if system:
|
|
body["system"] = system
|
|
if files:
|
|
body["files"] = files
|
|
if timeout_secs:
|
|
body["timeout_secs"] = timeout_secs
|
|
|
|
# HTTP timeout = subprocess timeout + a 30s margin so we don't bail
|
|
# while clawdforge is still doing work for us.
|
|
http_timeout = (timeout_secs or self.default_timeout) + 30
|
|
|
|
try:
|
|
r = requests.post(
|
|
f"{self.base_url}/run",
|
|
headers=self._headers(),
|
|
json=body,
|
|
timeout=http_timeout,
|
|
)
|
|
except requests.RequestException as e:
|
|
raise ForgeError(f"transport: {e}") from e
|
|
|
|
if r.status_code >= 400:
|
|
raise ForgeError(f"upstream {r.status_code}: {r.text[:500]}")
|
|
|
|
return r.json()
|