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() def get_class_descriptions(self, location_id: int | str): params = {"locationIds": location_id} return self._get("/class/classdescriptions", params) def get_classes( self, *, location_id: int | str | None = None, start_iso: str | None = None, end_iso: str | None = None, class_ids: list[int] | None = None, ): """Return scheduled classes. Accepts *either* a location or explicit classIds.""" params: dict[str, str | list[int]] = {} if location_id is not None: params["locationIds"] = location_id if class_ids: params["classIds"] = class_ids if start_iso: params["startDateTime"] = start_iso if end_iso: params["endDateTime"] = end_iso return self._get("/class/classes", params) def book_class( self, *, class_id: int, client_id: str, cross_regional: bool = False, require_payment: bool = False, send_email: bool = False, ): """Book a client into a class *after* verifying availability.""" # 1) quick availability check info = self.get_classes(class_ids=[class_id]) cls = (info.get("Classes") or [None])[0] if not cls: raise ValueError(f"Class {class_id} not found") if not cls.get("IsAvailable", False): raise ValueError("Class is not open for online booking") if cls.get("TotalBooked", 0) >= cls.get("MaxCapacity", 0): raise ValueError("Class is already full") # 2) book it payload = { "ClientId": client_id, "ClassId": class_id, "CrossRegionalBooking": cross_regional, "RequirePayment": require_payment, "SendEmail": send_email, } 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: float): """Return upcoming classes in a human-friendly Markdown list.""" if not mb_client: return "❌ Authenticate first." try: now = datetime.utcnow() start_iso = now.replace(hour=0, minute=0, second=0, microsecond=0).isoformat() end_iso = (now + timedelta(days=int(days))).replace( hour=23, minute=59, second=59, microsecond=0 ).isoformat() res = mb_client.get_classes( location_id=location_id, start_iso=start_iso, end_iso=end_iso ) classes = res.get("Classes", []) if not classes: return "⚠️ No classes in the selected window." lines: list[str] = [] for c in classes: when = c["StartDateTime"][:16].replace("T", " ") # YYYY-MM-DD HH:MM name = c.get("ClassDescription", {}).get("Name") or f"ID {c['ClassDescriptionId']}" ok = "✅" if c["IsAvailable"] else "❌" cid = c["Id"] lines.append(f"{when} | {name} | {ok} | ClassId: {cid}") return "### 📅 Schedule\n" + "\n".join(lines) 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): if not mb_client: return "❌ Authenticate first." try: res = mb_client.book_class( class_id=class_id, client_id=client_id, cross_regional=cross_regional, ) visit = res.get("Visit", {}) status = visit.get("AppointmentStatus", "OK") readable = visit.get("StartDateTime", "")[:16].replace("T", " ") return f"🎉 **Booked {readable} – status {status}**\n\n```json\n{res}\n```" 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)