hms-licht-ton/backend/app/mcp_server.py

"""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
mcp = FastMCP("hms-cms", streamable_http_path="/", stateless_http=True)

# ---------------------------------------------------------------------------#
# Bearer token authentication wrapper
# ---------------------------------------------------------------------------#
_MCP_TOKEN = os.environ.get("MCP_AUTH_TOKEN", "")


class MCPAuthMiddleware:
    """ASGI middleware: validate Authorization: Bearer <token> 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)