Spaces:
Sleeping
Sleeping
| # 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} |