# app/tools.py from __future__ import annotations from typing import Any, Dict, List, Optional import importlib from utils.config import get_settings # ------------------------------ # Backend service loader # ------------------------------ def _load_service(): """ Load a service module providing the following callables: - get_hours() -> dict - menu_lookup(filters: List[str]) -> List[dict] - create_reservation(name: str, phone: Optional[str], party_size: int, datetime_str: str) -> dict - create_order(items: List[dict]) -> dict Selection is controlled by `API_BACKEND` in .env: - "sim" -> use built-in simulated API (app.sim_api_bridge) - "mock" -> import app.mock_api.service or mock_api.service - "http" -> import app.http_api.service (you can implement later) """ s = get_settings() backend = (getattr(s, "API_BACKEND", None) or "sim").lower() module_candidates: List[str] = [] if backend == "sim": module_candidates = ["app.sim_api_bridge"] elif backend == "mock": module_candidates = ["app.mock_api.service", "mock_api.service"] elif backend == "http": module_candidates = ["app.http_api.service"] else: # unknown -> fall back to sim module_candidates = ["app.sim_api_bridge"] last_err = None for modname in module_candidates: try: return importlib.import_module(modname) except Exception as e: last_err = e # Final fallback to sim bridge even if env asked otherwise try: return importlib.import_module("app.sim_api_bridge") except Exception as e: raise RuntimeError(f"Could not load any service module ({module_candidates}): {last_err or e}") _service = None def _service_module(): global _service if _service is None: _service = _load_service() return _service # ------------------------------ # Input helpers # ------------------------------ def _as_int(x: Any, default: int) -> int: try: return int(x) except Exception: return default def _as_list(x: Any) -> List[Any]: if x is None: return [] if isinstance(x, list): return x return [x] def _ensure_items(items: Any) -> List[dict]: if items is None: return [] if isinstance(items, list): # keep only dict-like lines return [it for it in items if isinstance(it, dict)] return [] # ------------------------------ # Public dispatch # ------------------------------ def dispatch_tool(tool: str, args: Dict[str, Any]) -> Dict[str, Any]: svc = _service_module() try: if tool == "get_hours": return svc.get_hours() if tool == "menu_lookup": filters = _as_list(args.get("filters")) return {"items": svc.menu_lookup(filters)} if tool == "create_reservation": name = args.get("name") or "Guest" phone = args.get("phone") # accept either "party_size" or "partySize" party_size = _as_int(args.get("party_size") or args.get("partySize"), 2) # accept "datetime_str" or split date/time if your UI produces them separately datetime_str = args.get("datetime_str") or args.get("datetime") or "" if not datetime_str: # optional convenience: build from date + time if present date = (args.get("date") or "").strip() time_val = (args.get("time") or "").strip() if date or time_val: datetime_str = f"{date} {time_val}".strip() return svc.create_reservation( name=name, phone=phone, party_size=party_size, datetime_str=datetime_str, ) if tool == "create_order": items = _ensure_items(args.get("items")) if not items: return {"ok": False, "reason": "no_items", "message": "No order items were provided."} return svc.create_order(items) # Unknown tool return {"ok": False, "reason": "unknown_tool", "tool": tool} except Exception as e: # Never raise to the UI; always return a structured error return {"ok": False, "reason": "exception", "error": str(e), "tool": tool}