Eyob-Sol's picture
Upload 41 files
ac1f51b verified
# 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}