Create app.py

app.py

@@ -0,0 +1,218 @@
+import os
+from pathlib import Path
+from datetime import datetime, timedelta
+import logging
+import requests
+
+import gradio as gr
+
+# -------------------------------------------------------------
+# Configure logging
+# -------------------------------------------------------------
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+# -------------------------------------------------------------
+# Minimal Mindbody API wrapper
+# -------------------------------------------------------------
+class MindbodyClient:
+    """Thin wrapper around the MINDBODY Public API (v6).
+
+    Only the endpoints needed for this demo are implemented. Extend freely!
+    """
+
+    def __init__(self, api_key: str, site_id: str, source_name: str | None = None):
+        self.base_url = "https://api.mindbodyonline.com/public/v6"
+        self.headers = {
+            "API-Key": api_key,
+            "SiteId": site_id,
+            "Content-Type": "application/json",
+        }
+        if source_name:
+            self.headers["SourceName"] = source_name
+
+    # --------------------------- internal helpers ---------------------------
+    def _get(self, path: str, params: dict | None = None):
+        url = f"{self.base_url}{path}"
+        logger.debug("GET %s", url)
+        resp = requests.get(url, headers=self.headers, params=params, timeout=30)
+        resp.raise_for_status()
+        return resp.json()
+
+    def _post(self, path: str, payload: dict):
+        url = f"{self.base_url}{path}"
+        logger.debug("POST %s", url)
+        resp = requests.post(url, headers=self.headers, json=payload, timeout=30)
+        resp.raise_for_status()
+        return resp.json()
+
+    # --------------------------- public endpoints --------------------------
+    def get_class_descriptions(self, location_id: str | int):
+        """Return **types of training** offered at a location."""
+        params = {"LocationIds": location_id}
+        return self._get("/class/classdescriptions", params)
+
+    def get_classes(self, location_id: str | int, start_iso: str, end_iso: str):
+        """Return **scheduled classes** for a date range."""
+        params = {
+            "LocationIds": location_id,
+            "StartDateTime": start_iso,
+            "EndDateTime": end_iso,
+        }
+        return self._get("/class/classes", params)
+
+    def book_class(self, class_id: int, client_id: str, cross_regional: bool = False):
+        """Book a client into a class *after* verifying availability."""
+        # 1) confirm availability
+        info = self._get("/class/classes", {"ClassIds": class_id})
+        if not info.get("Classes"):
+            raise ValueError("Class ID not found")
+
+        cls = info["Classes"][0]
+        if not cls.get("IsAvailable", False):
+            raise ValueError("Class is not available for online booking")
+        if cls.get("TotalBooked", 0) >= cls.get("MaxCapacity", 0):
+            raise ValueError("Class is already full")
+
+        # 2) book
+        payload = {
+            "ClientId": client_id,
+            "ClassId": class_id,
+            "CrossRegionalBooking": cross_regional,
+        }
+        return self._post("/class/addclienttoclass", payload)
+
+
+# -------------------------------------------------------------
+# Global state (kept super-simple for a single-user MCP demo)
+# -------------------------------------------------------------
+mb_client: MindbodyClient | None = None
+
+
+# -------------------------------------------------------------
+# Helper functions wired to Gradio components
+# -------------------------------------------------------------
+
+def initialize_mindbody_client():
+    """Instantiate the global MindbodyClient reading credentials from ./config/.env"""
+    global mb_client
+    api_key = os.getenv("MINDBODY_API_KEY")
+    site_id = os.getenv("MINDBODY_SITE_ID")
+    source_name = os.getenv("MINDBODY_SOURCE_NAME")  # optional
+
+    if not api_key or not site_id:
+        return "❌ MINDBODY_API_KEY or MINDBODY_SITE_ID missing in .env"
+
+    try:
+        mb_client = MindbodyClient(api_key, site_id, source_name)
+        return "✅ Mindbody client initialised. Ready to query!"
+    except Exception as exc:
+        logger.exception("Failed initialising Mindbody client")
+        return f"❌ Initialisation error: {exc}"
+
+
+def get_training_types(location_id: str | int):
+    """Fetch human-readable list of class description names."""
+    if not mb_client:
+        return "❌ Authenticate first."
+
+    try:
+        res = mb_client.get_class_descriptions(location_id)
+        descs = res.get("ClassDescriptions", [])
+        if not descs:
+            return "⚠️ No training types found for this location."
+
+        lines = [f"- {d.get('Name')} (ID: {d.get('Id')})" for d in descs]
+        return "🏷️ Training Types\n" + "\n".join(lines)
+    except Exception as exc:
+        logger.exception("Error while fetching class descriptions")
+        return f"❌ Error: {exc}"
+
+
+def get_schedule(location_id: str | int, days: int):
+    """Return upcoming classes in a human-friendly Markdown list."""
+    if not mb_client:
+        return "❌ Authenticate first."
+
+    try:
+        start_iso = datetime.now().strftime("%Y-%m-%dT00:00:00")
+        end_iso = (datetime.now() + timedelta(days=int(days))).strftime("%Y-%m-%dT23:59:59")
+        res = mb_client.get_classes(location_id, start_iso, end_iso)
+        classes = res.get("Classes", [])
+        if not classes:
+            return "⚠️ No classes in the selected window."
+
+        formatted = []
+        for c in classes:
+            dt = c.get("StartDateTime", "?")
+            name = c.get("ClassDescription", {}).get("Name", "🆔" + str(c.get("ClassDescriptionId")))
+            ok = "✅" if c.get("IsAvailable") else "❌"
+            cid = c.get("Id")
+            formatted.append(f"{dt} | {name} | {ok} | ClassId: {cid}")
+
+        return "📅 Schedule\n" + "\n".join(formatted)
+    except Exception as exc:
+        logger.exception("Error while fetching schedule")
+        return f"❌ Error: {exc}"
+
+
+def book_class_gr(class_id: int, client_id: str, cross_regional: bool):
+    """Book a class through the UI."""
+    if not mb_client:
+        return "❌ Authenticate first."
+
+    try:
+        res = mb_client.book_class(class_id, client_id, cross_regional)
+        visit = res.get("Visit", {})
+        status = visit.get("AppointmentStatus", "OK")
+        return f"🎉 Booked! Status: {status}\n\nFull response:\n{res}"
+    except Exception as exc:
+        logger.exception("Booking failed")
+        return f"❌ Could not book: {exc}"
+
+
+# -------------------------------------------------------------
+# Gradio UI definition
+# -------------------------------------------------------------
+with gr.Blocks(title="Mindbody Class Booker") as demo:
+    gr.Markdown("# 🏋️‍♀️ MINDBODY Class Explorer & Booker")
+    gr.Markdown("Authenticate, browse classes by location, and secure your spot – all in one place.")
+
+    # ---------- Auth ----------
+    with gr.Row():
+        auth_out = gr.Textbox(label="Auth Status", interactive=False)
+        auth_btn = gr.Button("Authenticate")
+
+    # ---------- Training types ----------
+    with gr.Row():
+        loc_in = gr.Textbox(label="Location ID", placeholder="e.g. -99")
+        types_out = gr.Textbox(label="Training Types", lines=10, interactive=False)
+        types_btn = gr.Button("Get Training Types")
+
+    # ---------- Schedule ----------
+    with gr.Row():
+        days_in = gr.Number(value=7, label="Days Ahead")
+        sched_out = gr.Textbox(label="Class Schedule", lines=12, interactive=False)
+        sched_btn = gr.Button("Get Schedule")
+
+    # ---------- Booking ----------
+    gr.Markdown("## Book a Class")
+    with gr.Row():
+        class_id_in = gr.Number(label="Class ID")
+        client_id_in = gr.Textbox(label="Client ID")
+        cross_in = gr.Checkbox(label="Cross-regional booking", value=False)
+        book_out = gr.Textbox(label="Booking Result", lines=10, interactive=False)
+        book_btn = gr.Button("Book Class")
+
+    # ---------- Wiring ----------
+    auth_btn.click(fn=initialize_mindbody_client, outputs=auth_out)
+    types_btn.click(fn=get_training_types, inputs=loc_in, outputs=types_out)
+    sched_btn.click(fn=get_schedule, inputs=[loc_in, days_in], outputs=sched_out)
+    book_btn.click(fn=book_class_gr, inputs=[class_id_in, client_id_in, cross_in], outputs=book_out)
+
+# -------------------------------------------------------------
+# Launch
+# -------------------------------------------------------------
+if __name__ == "__main__":
+    # On HuggingFace or other MCP hosts, `mcp_server=True` makes Gradio bind
+    demo.launch(mcp_server=True)