"""MCP Server for HMS CMS – provides resources and tools for AI agents. Resources (read-only): - design-rules: HMS design system as JSON - page-structure: React pages and components map - sync-status: current Rentman sync status Tools (execute): - trigger_sync: run Rentman equipment sync - get_sync_log: fetch sync history - set_rentman_token: update .env token + restart backend - read_file: read frontend source file - write_file: write frontend source file - create_page: create React page with design validation - deploy: rebuild + redeploy frontend container - git_commit: commit all changes - git_status: show working tree status """ import json import os import re import subprocess import asyncio import logging from pathlib import Path from typing import Any from mcp.server.fastmcp import FastMCP from app.database import async_session from app.services.sync_service import SyncService from app.config import get_settings logger = logging.getLogger(__name__) settings = get_settings() # ---------------------------------------------------------------------------# # Paths (mapped via docker-compose volumes) # ---------------------------------------------------------------------------# REPO_ROOT = Path("/data/repo") FRONTEND_ROOT = REPO_ROOT / "frontend" FRONTEND_SRC = FRONTEND_ROOT / "src" ENV_FILE = REPO_ROOT / ".env" # ---------------------------------------------------------------------------# # Static design rules (derived from frontend/src/index.css) # ---------------------------------------------------------------------------# DESIGN_RULES: dict[str, Any] = { "colors": { "accent": "#FA5C01", "bg": "#131313", "panel": "#1a1a1a", "surface": "#212121", "text": "#ffffff", "text_muted": "#d4d4d4", "border": "#444444a8", "secondary": "#656565", }, "fonts": {"primary": "Inter", "weights": [400, 500, 600, 700, 800]}, "components": { "hms-btn": "button class with variants: primary, secondary, ghost", "hms-card": "card container with border, bg, hover effect", "hms-badge": "badge with variants: primary, gray, success, error", "hms-input": "input field with focus state", "hms-chip": "filter chip with active state", "hms-skeleton": "loading skeleton with shimmer animation", "hms-speaker-grid": "responsive grid for equipment items", "hms-gallery": "responsive gallery grid", "hms-hero": "hero section with bg image and overlay", }, "layout": { "max_width": "1280px", "header_height": "4rem", "radius": {"sm": 2, "md": 3, "lg": 4, "xl": 4, "full": 9999}, }, "rules": [ "Use Inter font for all text", "Use #FA5C01 for all accent elements", "Use hms-* CSS classes, do not invent new ones", "All text in German", "Use CSS variables (var(--text), var(--bg), etc.) not hardcoded colors", "Use Tailwind utility classes for layout, hms-* classes for components", "Images: use object-cover, lazy loading", "Buttons: use hms-btn with variant classes", "Cards: use hms-card class", "Responsive: sm: breakpoint 640px, lg: breakpoint 1024px", ], } # Static page/component info (used when frontend source is not mounted) _STATIC_PAGES = [ {"path": "/", "file": "src/pages/Index.tsx", "components": ["SpeakerIcon", "ServiceCard"]}, {"path": "/mietkatalog", "file": "src/pages/Mietkatalog.tsx", "components": ["EquipmentCard", "LoadingSkeleton", "ErrorState", "EmptyState"]}, {"path": "/mietkatalog/:id", "file": "src/pages/EquipmentDetail.tsx", "components": ["ErrorState"]}, {"path": "/kontakt", "file": "src/pages/Kontakt.tsx", "components": []}, {"path": "/referenzen", "file": "src/pages/Referenzen.tsx", "components": ["ErrorState", "EmptyState"]}, {"path": "/warenkorb", "file": "src/pages/Warenkorb.tsx", "components": ["EmptyState"]}, {"path": "/admin", "file": "src/pages/Admin.tsx", "components": ["Logo"]}, {"path": "/agb", "file": "src/pages/AGB.tsx", "components": ["LegalContentPage"]}, {"path": "/datenschutz", "file": "src/pages/Datenschutz.tsx", "components": ["LegalContentPage"]}, {"path": "/impressum", "file": "src/pages/Impressum.tsx", "components": ["LegalContentPage"]}, {"path": "/*", "file": "src/pages/ErrorPage.tsx", "components": ["Header", "Footer"]}, ] _STATIC_COMPONENTS = [ {"name": "Header", "file": "src/components/Header.tsx", "props": []}, {"name": "Footer", "file": "src/components/Footer.tsx", "props": []}, {"name": "Logo", "file": "src/components/Logo.tsx", "props": ["size"]}, {"name": "EquipmentCard", "file": "src/components/EquipmentCard.tsx", "props": ["item", "onClick", "onAddToCart"]}, {"name": "LoadingSkeleton", "file": "src/components/LoadingSkeleton.tsx", "props": ["count"]}, {"name": "ErrorState", "file": "src/components/ErrorState.tsx", "props": ["title", "message", "onRetry"]}, {"name": "EmptyState", "file": "src/components/EmptyState.tsx", "props": ["icon", "title", "message", "actionLabel", "onAction"]}, {"name": "ServiceCard", "file": "src/components/ServiceCard.tsx", "props": ["icon", "title", "description"]}, {"name": "SpeakerIcon", "file": "src/components/SpeakerIcon.tsx", "props": ["type"]}, {"name": "LegalContentPage", "file": "src/components/LegalContentPage.tsx", "props": ["title", "content", "loading"]}, {"name": "Layout", "file": "src/components/Layout.tsx", "props": []}, ] # ---------------------------------------------------------------------------# # FastMCP server instance # ---------------------------------------------------------------------------# # streamable_http_path="/" so that when mounted at /mcp the endpoint is /mcp/ # stateless_http=True: each request is independent (no session management) # Disable DNS rebinding protection for Docker deployment try: from mcp.server.fastmcp.settings import TransportSecuritySettings _ts = TransportSecuritySettings(enable_dns_rebinding_protection=False) except Exception: _ts = None _mcp_kwargs: dict[str, Any] = { "streamable_http_path": "/", "stateless_http": True, } if _ts is not None: _mcp_kwargs["transport_security"] = _ts mcp = FastMCP("hms-cms", **_mcp_kwargs) # ---------------------------------------------------------------------------# # Bearer token authentication wrapper # ---------------------------------------------------------------------------# _MCP_TOKEN = os.environ.get("MCP_AUTH_TOKEN", "") class MCPAuthMiddleware: """ASGI middleware: validate Authorization: Bearer for /mcp endpoint.""" def __init__(self, app): self.app = app async def __call__(self, scope, receive, send): if scope["type"] != "http": return await self.app(scope, receive, send) # If no token configured, allow all (dev mode) if not _MCP_TOKEN: return await self.app(scope, receive, send) # Extract Authorization header headers = dict(scope.get("headers", [])) auth_header = headers.get(b"authorization", b"").decode() if auth_header.startswith("Bearer "): token = auth_header[7:] if token == _MCP_TOKEN: return await self.app(scope, receive, send) # Reject await send({ "type": "http.response.start", "status": 401, "headers": [[b"content-type", b"application/json"], [b"www-authenticate", b"Bearer"]], }) await send({ "type": "http.response.body", "body": b'{"error": "Unauthorized: invalid or missing Bearer token"}', }) def get_mcp_app(): """Return the MCP ASGI app wrapped with auth middleware.""" return MCPAuthMiddleware(_mcp_streamable_app) # Expose session manager for lifespan integration in main FastAPI app. # FastAPI mount() does not call sub-app lifespans, so we must # manually start/stop the session manager in the main app lifespan. # Call streamable_http_app() first to lazily initialize the session manager. _mcp_streamable_app = mcp.streamable_http_app() mcp_session_manager = mcp._session_manager # ---------------------------------------------------------------------------# # Resources (read-only) # ---------------------------------------------------------------------------# @mcp.resource("hms://design-rules") async def design_rules() -> str: """Return the HMS design rules as structured JSON.""" return json.dumps(DESIGN_RULES, indent=2, ensure_ascii=False) @mcp.resource("hms://page-structure") async def page_structure() -> str: """List all React pages and their component usage.""" pages = list(_STATIC_PAGES) components = list(_STATIC_COMPONENTS) # Dynamic scan if frontend source is accessible if FRONTEND_SRC.exists(): pages_dir = FRONTEND_SRC / "pages" comp_dir = FRONTEND_SRC / "components" if pages_dir.exists(): dyn_pages: list[dict] = [] for f in sorted(pages_dir.glob("*.tsx")): content = f.read_text() imports = re.findall(r"import\s+(\w+)\s+from\s+['\"]\.\.?/", content) comp_imports = [imp for imp in imports if imp[0].isupper()] stem = f.stem if stem == "Index": path = "/" elif stem == "ErrorPage": path = "/*" elif stem == "EquipmentDetail": path = "/mietkatalog/:id" else: path = f"/{stem.lower()}" dyn_pages.append({ "path": path, "file": f"src/pages/{f.name}", "components": comp_imports, }) if dyn_pages: pages = dyn_pages if comp_dir.exists(): dyn_comps: list[dict] = [] for f in sorted(comp_dir.glob("*.tsx")): content = f.read_text() props: list[str] = [] # Match interface Props { ... } or interface XxxProps { ... } prop_match = re.search(r"interface\s+\w+\s*\{([^}]*)\}", content) if prop_match: props = re.findall(r"(\w+)\??\s*:", prop_match.group(1)) dyn_comps.append({ "name": f.stem, "file": f"src/components/{f.name}", "props": props, }) if dyn_comps: components = dyn_comps return json.dumps({"pages": pages, "components": components}, indent=2, ensure_ascii=False) @mcp.resource("hms://sync-status") async def sync_status() -> str: """Return the current sync status (last sync, items, errors).""" try: async with async_session() as db: svc = SyncService(db) result = await svc.get_last_sync() return json.dumps(result, indent=2, default=str, ensure_ascii=False) except Exception as exc: return json.dumps({"error": str(exc)}, indent=2) # ---------------------------------------------------------------------------# # Tools (execute) # ---------------------------------------------------------------------------# @mcp.tool() async def trigger_sync() -> str: """Trigger a Rentman equipment sync (like POST /api/admin/sync).""" try: async with async_session() as db: svc = SyncService(db) result = await svc.run_sync() return json.dumps(result, indent=2) except Exception as exc: return json.dumps({"error": str(exc)}, indent=2) @mcp.tool() async def get_sync_log(page: int = 1, page_size: int = 20) -> str: """Return the sync history log. Args: page: Page number (1-based) page_size: Items per page (max 100) """ try: async with async_session() as db: svc = SyncService(db) result = await svc.get_sync_log_paginated(page=page, page_size=page_size) items = [] for log in result["items"]: items.append({ "id": log.id, "sync_type": log.sync_type, "status": log.status, "started_at": log.started_at.isoformat() if log.started_at else None, "completed_at": log.completed_at.isoformat() if log.completed_at else None, "items_processed": log.items_processed, "items_failed": log.items_failed, "error_message": log.error_message, }) return json.dumps({ "items": items, "total": result["total"], "page": result["page"], "page_size": result["page_size"], "total_pages": result["total_pages"], }, indent=2, default=str) except Exception as exc: return json.dumps({"error": str(exc)}, indent=2) @mcp.tool() async def set_rentman_token(token: str) -> str: """Update the Rentman API Token in .env and restart the backend. Args: token: New Rentman API token string """ try: if not ENV_FILE.exists(): return json.dumps({"error": f".env file not found at {ENV_FILE}"}) content = ENV_FILE.read_text() if "RENTMAN_API_TOKEN=" in content: content = re.sub( r"RENTMAN_API_TOKEN=.*", f"RENTMAN_API_TOKEN={token}", content, ) else: content += f"\nRENTMAN_API_TOKEN={token}\n" ENV_FILE.write_text(content) # Restart backend with a short delay so the response can be sent first async def _restart() -> None: await asyncio.sleep(2) try: subprocess.run( ["docker", "compose", "restart", "backend"], cwd=str(REPO_ROOT), capture_output=True, text=True, timeout=60, ) except Exception as exc: logger.error("Backend restart failed: %s", exc) asyncio.create_task(_restart()) return json.dumps({ "status": "success", "message": "Token updated. Backend restart initiated (2s delay).", }) except Exception as exc: return json.dumps({"error": str(exc)}, indent=2) @mcp.tool() async def read_file(path: str) -> str: """Read a file from the frontend source. Args: path: Relative path to the file (relative to frontend/) """ try: full_path = FRONTEND_ROOT / path if not full_path.exists() or not full_path.is_file(): return json.dumps({"error": f"File not found: {path}"}) # Security: ensure path stays within frontend dir full_path = full_path.resolve() if not str(full_path).startswith(str(FRONTEND_ROOT.resolve())): return json.dumps({"error": "Path traversal blocked"}) return full_path.read_text() except Exception as exc: return json.dumps({"error": str(exc)}, indent=2) @mcp.tool() async def write_file(path: str, content: str) -> str: """Write a file in the frontend source. Args: path: Relative path (relative to frontend/src/) content: File content to write """ try: full_path = (FRONTEND_SRC / path).resolve() if not str(full_path).startswith(str(FRONTEND_SRC.resolve())): return json.dumps({"error": "Path traversal blocked"}) full_path.parent.mkdir(parents=True, exist_ok=True) full_path.write_text(content) return json.dumps({ "status": "success", "path": str(full_path), "bytes": len(content), }) except Exception as exc: return json.dumps({"error": str(exc)}, indent=2) @mcp.tool() async def create_page(name: str, content: str) -> str: """Create a new React page with design rule validation. Args: name: Page name (e.g. 'UeberUns' creates src/pages/UeberUns.tsx) content: Full TSX content of the page """ try: issues: list[str] = [] # Validate: check for non-hms component classes custom_classes = re.findall(r'className="([^"]*)"', content) for cls_str in custom_classes: for word in cls_str.split(): if word.startswith("btn") and not word.startswith("hms-btn"): issues.append(f"Found '{word}' – use 'hms-btn' instead") if word.startswith("card") and not word.startswith("hms-card"): issues.append(f"Found '{word}' – use 'hms-card' instead") if word.startswith("badge") and not word.startswith("hms-badge"): issues.append(f"Found '{word}' – use 'hms-badge' instead") # Validate: hardcoded hex colors (excluding #FA5C01 accent) hex_colors = re.findall(r"#([0-9A-Fa-f]{6})\b", content) for hc in hex_colors: if hc.upper() not in ("FA5C01", "FFFFFF", "000000"): issues.append(f"Hardcoded color #{hc} – use CSS variable instead") # Validate: non-Inter font-family font_match = re.search(r'font-family:\s*["\']?(?!Inter)["\']?', content) if font_match: issues.append("Non-Inter font detected – use Inter for all text") # Write the file safe_name = re.sub(r"[^a-zA-Z0-9_-]", "", name) if not safe_name: return json.dumps({"error": "Invalid page name"}) filename = f"{safe_name}.tsx" file_path = FRONTEND_SRC / "pages" / filename file_path.parent.mkdir(parents=True, exist_ok=True) file_path.write_text(content) return json.dumps({ "status": "success", "file": f"src/pages/{filename}", "path": str(file_path), "validation": { "issues": issues, "passed": len(issues) == 0, }, }, indent=2, ensure_ascii=False) except Exception as exc: return json.dumps({"error": str(exc)}, indent=2) @mcp.tool() async def deploy() -> str: """Build and deploy the frontend. Runs: docker compose build --no-cache frontend docker compose up -d --force-recreate frontend docker compose restart frontend """ try: commands = [ ["docker", "compose", "build", "--no-cache", "frontend"], ["docker", "compose", "up", "-d", "--force-recreate", "frontend"], ["docker", "compose", "restart", "frontend"], ] results = [] for cmd in commands: r = subprocess.run( cmd, cwd=str(REPO_ROOT), capture_output=True, text=True, timeout=300, ) results.append({ "command": " ".join(cmd), "returncode": r.returncode, "stdout": r.stdout[-500:] if r.stdout else "", "stderr": r.stderr[-500:] if r.stderr else "", }) if r.returncode != 0: return json.dumps({ "status": "failed", "step": cmd[2], "results": results, }, indent=2) return json.dumps({"status": "success", "results": results}, indent=2) except Exception as exc: return json.dumps({"error": str(exc)}, indent=2) @mcp.tool() async def git_commit(message: str) -> str: """Commit all changes with a message. Args: message: Git commit message """ try: r = subprocess.run( ["git", "add", "-A"], cwd=str(REPO_ROOT), capture_output=True, text=True, timeout=30, ) if r.returncode != 0: return json.dumps({"error": "git add failed", "stderr": r.stderr}) r = subprocess.run( ["git", "commit", "-m", message], cwd=str(REPO_ROOT), capture_output=True, text=True, timeout=30, ) if r.returncode != 0: if "nothing to commit" in r.stdout: return json.dumps({"status": "noop", "message": "Nothing to commit"}) return json.dumps({"error": "git commit failed", "stderr": r.stderr}) return json.dumps({ "status": "success", "message": message, "output": r.stdout.strip(), }) except Exception as exc: return json.dumps({"error": str(exc)}, indent=2) @mcp.tool() async def git_status() -> str: """Return the git working tree status.""" try: r = subprocess.run( ["git", "status", "--porcelain"], cwd=str(REPO_ROOT), capture_output=True, text=True, timeout=30, ) if r.returncode != 0: return json.dumps({"error": r.stderr}) files = [] for line in r.stdout.strip().split("\n"): if line: files.append({ "status": line[:2].strip(), "file": line[3:].strip(), }) return json.dumps({"files": files, "clean": len(files) == 0}, indent=2) except Exception as exc: return json.dumps({"error": str(exc)}, indent=2)