""" Model Registry settings — providers, hosts, models, and role assignments. Routes: GET /settings/models → settings page (canonical) GET /settings/local → redirect to /settings/models POST /settings/local/host → save/create a local host POST /settings/local/host/{id}/remove → remove a host (and its models) POST /settings/local/google-account → save/create a Google account POST /settings/local/google-account/{id}/remove → remove a Google account POST /settings/local/models/add → add a model (any provider) POST /settings/local/models/{id}/edit → edit an existing model entry POST /settings/local/models/{id}/remove → remove a model POST /api/models/role → AJAX: set a role assignment GET /api/local-llm/fetch-models → proxy to host /api/models (JSON) """ import json as _json import logging from pathlib import Path import httpx import jwt from fastapi import APIRouter, Form, Request from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse from auth_utils import COOKIE_NAME, decode_token from config import settings as app_settings import model_registry as reg from tools import TOOL_CATEGORIES logger = logging.getLogger(__name__) router = APIRouter() _STATIC = Path(__file__).parent.parent / "static" # ── Auth helper ─────────────────────────────────────────────────────────────── def _get_user(request: Request) -> str | None: token = request.cookies.get(COOKIE_NAME) if not token: return None try: return decode_token(token) except jwt.InvalidTokenError: return None # ── Page renderer ───────────────────────────────────────────────────────────── def _render(username: str, success: str = "", error: str = "") -> str: registry = reg.get_registry(username) hosts = registry.get("hosts", []) models = registry.get("models", []) roles = registry.get("roles", {}) builtins = reg._builtins() host_by_id = {h["id"]: h for h in hosts} goog_accts = registry.get("providers", {}).get("google", {}).get("accounts", []) # ── Google account rows ─────────────────────────────────────────────────── google_account_rows = "" for a in goog_accts: hint = (a.get("api_key") or "")[:10] + "…" if a.get("api_key") else "no key" google_account_rows += f'''
No accounts configured yet.
' # ── Local host rows ─────────────────────────────────────────────────────── host_rows = "" for h in hosts: key_hint = f"…{h['api_key'][-4:]}" if h.get("api_key") else "not set" ht = h.get("host_type", "openwebui") ow = ' selected' if ht == "openwebui" else '' ai = ' selected' if ht == "openai" else '' host_rows += f'''No hosts configured yet. Add one below.
' host_options = "".join( f'' for h in hosts ) # ── Model rows (all providers) ──────────────────────────────────────────── _PROVIDER_BADGE = { "claude_cli": ('Anthropic', "Claude CLI"), "gemini_api": ('Google', ""), "local_openai": ('Local', ""), } model_rows = "" for m in models: resolved = reg._resolve_model(registry, m["id"]) if not resolved: continue mtype = m.get("type", "local_openai") badge, default_secondary = _PROVIDER_BADGE.get(mtype, ("", "")) if mtype == "local_openai": h = host_by_id.get(m.get("host_id", ""), {}) secondary = h.get("label") or h.get("api_url", "") elif mtype == "gemini_api": acct = next((a for a in goog_accts if a["id"] == m.get("account_id")), None) secondary = acct["label"] if acct else "" else: secondary = default_secondary ctx = f'{m.get("context_k",0)}k' if m.get("context_k") else "" no_tools = '' if m.get("tools", True) else 'no tools' tags_html = " ".join(f'{t}' for t in (m.get("tags") or [])) sec = f'{secondary}' if secondary else "" # ── Inline edit form fields (type-specific) ─────────────────────────── if mtype == "local_openai": host_opts = "".join( f'' for h in hosts ) mid = m["id"] extra_fields = ( f'No models added yet.
' # ── Role assignment rows ────────────────────────────────────────────────── model_opts = '\n' model_opts += '\n' if models: model_opts += '\n' role_rows = "" for role in app_settings.get_defined_roles(): role_cfg = roles.get(role, {}) role_rows += ( f'{success}
') if error: html = html.replace("", f'{error}
') return html # ── Routes ──────────────────────────────────────────────────────────────────── @router.get("/settings/models", include_in_schema=False) async def models_page_canonical(request: Request): username = _get_user(request) if not username: return RedirectResponse("/login", status_code=302) return HTMLResponse(_render(username)) @router.get("/settings/local", include_in_schema=False) async def models_page_legacy(request: Request): return RedirectResponse("/settings/models", status_code=301) @router.post("/settings/local/google-account", include_in_schema=False) async def save_google_account( request: Request, account_id: str = Form(""), label: str = Form(""), api_key: str = Form(""), ): username = _get_user(request) if not username: return RedirectResponse("/login", status_code=302) if not api_key.strip() and not account_id.strip(): return HTMLResponse(_render(username, error="API key is required.")) reg.save_google_account(username, account_id or None, label, api_key) return HTMLResponse(_render(username, success="Google account saved.")) @router.post("/settings/local/google-account/{account_id}/remove", include_in_schema=False) async def remove_google_account(request: Request, account_id: str): username = _get_user(request) if not username: return RedirectResponse("/login", status_code=302) reg.remove_google_account(username, account_id) return HTMLResponse(_render(username, success="Google account removed.")) @router.post("/settings/local/host", include_in_schema=False) async def save_host( request: Request, host_id: str = Form(""), label: str = Form(""), api_url: str = Form(""), api_key: str = Form(""), host_type: str = Form("openwebui"), max_concurrent: int = Form(3), ): username = _get_user(request) if not username: return RedirectResponse("/login", status_code=302) if not api_url.strip(): return HTMLResponse(_render(username, error="API URL is required.")) reg.save_host(username, host_id or None, label, api_url, api_key, host_type, max_concurrent) return HTMLResponse(_render(username, success="Host saved.")) @router.post("/settings/local/host/{host_id}/remove", include_in_schema=False) async def remove_host(request: Request, host_id: str): username = _get_user(request) if not username: return RedirectResponse("/login", status_code=302) reg.remove_host(username, host_id) return HTMLResponse(_render(username, success="Host removed.")) @router.post("/settings/local/models/add", include_in_schema=False) async def add_model( request: Request, provider: str = Form("local"), label: str = Form(""), context_k: int = Form(0), max_rounds: int = Form(0), tools: int = Form(1), tags: str = Form(""), # local-only fields host_id: str = Form(""), model_name: str = Form(""), # cloud-only fields cloud_model_name: str = Form(""), account_id: str = Form(""), credential_id: str = Form("cli"), ): username = _get_user(request) if not username: return RedirectResponse("/login", status_code=302) tag_list = [t.strip() for t in tags.split(",") if t.strip()] max_rounds_ = max_rounds or None tools_bool = tools != 0 if provider == "local": if not model_name.strip(): return HTMLResponse(_render(username, error="Model name is required.")) if not host_id.strip(): return HTMLResponse(_render(username, error="Select a host.")) reg.save_model(username, None, host_id, label, model_name, context_k, tag_list, max_rounds=max_rounds_, tools=tools_bool) display = label or model_name elif provider in ("google", "anthropic"): if not cloud_model_name.strip(): return HTMLResponse(_render(username, error="Select a model from the catalog.")) if provider == "google" and not account_id.strip(): return HTMLResponse(_render(username, error="Select a Google account.")) reg.save_cloud_model( username, None, provider, cloud_model_name, label, account_id=account_id or None, credential_id=credential_id or None, context_k=context_k, tags=tag_list, max_rounds=max_rounds_, tools=tools_bool, ) display = label or cloud_model_name else: return HTMLResponse(_render(username, error=f"Unknown provider: {provider}")) logger.info("model added: %s / %s (%s)", username, display, provider) return HTMLResponse(_render(username, success=f'Model "{display}" added.')) @router.post("/settings/local/models/{model_id}/edit", include_in_schema=False) async def edit_model( request: Request, model_id: str, mtype: str = Form(""), label: str = Form(""), model_name: str = Form(""), context_k: int = Form(0), max_rounds: int = Form(0), tools: int = Form(1), tags: str = Form(""), host_id: str = Form(""), account_id: str = Form(""), credential_id: str = Form("cli"), ): username = _get_user(request) if not username: return RedirectResponse("/login", status_code=302) if not model_name.strip(): return HTMLResponse(_render(username, error="Model name is required.")) tag_list = [t.strip() for t in tags.split(",") if t.strip()] max_rounds_ = max_rounds or None tools_bool = tools != 0 if mtype == "local_openai": if not host_id.strip(): return HTMLResponse(_render(username, error="Select a host for this model.")) reg.save_model(username, model_id, host_id, label, model_name, context_k, tag_list, max_rounds=max_rounds_, tools=tools_bool) elif mtype == "gemini_api": reg.save_cloud_model(username, model_id, "google", model_name, label, account_id=account_id or None, context_k=context_k, tags=tag_list, max_rounds=max_rounds_, tools=tools_bool) elif mtype == "claude_cli": reg.save_cloud_model(username, model_id, "anthropic", model_name, label, credential_id=credential_id or "cli", context_k=context_k, tags=tag_list, max_rounds=max_rounds_, tools=tools_bool) else: return HTMLResponse(_render(username, error=f"Unknown model type: {mtype}")) display = label.strip() or model_name.strip() logger.info("model edited: %s / %s (%s)", username, display, mtype) return HTMLResponse(_render(username, success=f'Model "{display}" updated.')) @router.post("/settings/local/models/{model_id}/remove", include_in_schema=False) async def remove_model(request: Request, model_id: str): username = _get_user(request) if not username: return RedirectResponse("/login", status_code=302) reg.remove_model(username, model_id) return HTMLResponse(_render(username, success="Model removed.")) @router.post("/api/models/role") async def set_role(request: Request) -> JSONResponse: """AJAX: assign a model to a role priority slot. Body: {"role": "chat", "slot": "primary", "model_id": "abc123" | ""} """ username = _get_user(request) if not username: return JSONResponse({"error": "Not authenticated"}, status_code=401) try: body = await request.json() except Exception: return JSONResponse({"error": "Invalid JSON"}, status_code=400) role = body.get("role", "").strip() slot = body.get("slot", "").strip() model_id = body.get("model_id", "").strip() or None if not role or not slot: return JSONResponse({"error": "role and slot are required"}, status_code=400) ok = reg.set_role(username, role, slot, model_id) if not ok: return JSONResponse({"error": "Invalid slot or model_id not found"}, status_code=400) logger.info("role set: %s %s.%s = %s", username, role, slot, model_id) return JSONResponse({"ok": True}) @router.post("/api/models/role-config") async def set_role_config(request: Request) -> JSONResponse: """AJAX: save system_append, tool allow-list, and inject_datetime flag for a role. Body: {"role": "coder", "system_append": "...", "tools": [...] | null, "inject_datetime": true} tools=null clears the allow-list (role uses all accessible tools). inject_datetime=false suppresses the date/time header for pure processing roles. """ username = _get_user(request) if not username: return JSONResponse({"error": "Not authenticated"}, status_code=401) try: body = await request.json() except Exception: return JSONResponse({"error": "Invalid JSON"}, status_code=400) role = body.get("role", "").strip() system_append = body.get("system_append", "") tools = body.get("tools") # list[str] or None inject_datetime = body.get("inject_datetime", True) inject_mode = body.get("inject_mode", True) if not role: return JSONResponse({"error": "role is required"}, status_code=400) if tools is not None and not isinstance(tools, list): return JSONResponse({"error": "tools must be a list or null"}, status_code=400) reg.set_role_config(username, role, system_append, tools, inject_datetime=bool(inject_datetime), inject_mode=bool(inject_mode)) logger.info("role config saved: %s %s (tools=%s inject_datetime=%s inject_mode=%s)", username, role, len(tools) if tools is not None else "all", inject_datetime, inject_mode) return JSONResponse({"ok": True}) @router.get("/api/local-llm/fetch-models") async def fetch_models(request: Request, host_id: str = "") -> JSONResponse: """Proxy to the host's models endpoint. host_id selects which host.""" username = _get_user(request) if not username: return JSONResponse({"error": "Not authenticated"}, status_code=401) registry = reg.get_registry(username) hosts = registry.get("hosts", []) host = next((h for h in hosts if h["id"] == host_id), None) if host_id else (hosts[0] if hosts else None) if host: api_url, api_key, host_type = host.get("api_url",""), host.get("api_key",""), host.get("host_type","openwebui") else: api_url, api_key, host_type = app_settings.local_api_url, app_settings.local_api_key, "openwebui" if not api_url: return JSONResponse({"error": "No host configured."}, status_code=400) models_path = "/models" if host_type == "openai" else "/api/models" url = api_url.rstrip("/") + models_path headers = {"Authorization": f"Bearer {api_key}"} if api_key else {} try: async with httpx.AsyncClient(timeout=8) as client: resp = await client.get(url, headers=headers) resp.raise_for_status() data = resp.json() models = sorted( [{"id": m["id"], "name": m.get("name") or m["id"]} for m in data.get("data", [])], key=lambda m: m["name"].lower(), ) return JSONResponse({"models": models}) except httpx.HTTPStatusError as e: return JSONResponse({"error": f"Host returned {e.response.status_code}"}, status_code=502) except Exception as e: return JSONResponse({"error": str(e)}, status_code=502)