sap-architecture/mcp-server/server.py

#!/usr/bin/env python3
"""
sap-arch-mcp — MCP server exposing the sap-architecture skill pipeline.

Runs next to a clone of the sap-architecture repo and wraps its bundled
scripts (scaffold, validate, autofix, score, icons) as MCP tools so an
LLM agent (e.g. via OpenWebUI + mcpo) can produce SAP Architecture
Center-style .drawio diagrams with the full deterministic pipeline.

Usage:
    pip install "mcp[cli]"
    export SKILL_DIR=/opt/sap-architecture      # repo root (contains scripts/, assets/)
    export WORK_DIR=/tmp/sap-arch-work          # writable scratch dir
    python3 server.py                           # stdio transport

With mcpo for OpenWebUI:
    uvx mcpo --port 8600 -- python3 /opt/sap-architecture/mcp-server/server.py
"""

import json
import os
import subprocess
import tempfile
import uuid
from pathlib import Path

from mcp.server.fastmcp import FastMCP

SKILL_DIR = Path(os.environ.get("SKILL_DIR", Path(__file__).resolve().parent.parent))
WORK_DIR = Path(os.environ.get("WORK_DIR", "/tmp/sap-arch-work"))
WORK_DIR.mkdir(parents=True, exist_ok=True)

SCRIPTS = SKILL_DIR / "scripts"
TEMPLATES = SKILL_DIR / "assets" / "reference-examples"

mcp = FastMCP("sap-architecture")


def _run(args: list[str], timeout: int = 120) -> dict:
    """Run a bundled script and capture output."""
    proc = subprocess.run(
        ["python3", *args],
        capture_output=True, text=True, timeout=timeout, cwd=str(WORK_DIR),
    )
    return {
        "returncode": proc.returncode,
        "stdout": proc.stdout[-20000:],
        "stderr": proc.stderr[-5000:],
    }


def _tmpfile(xml: str, suffix: str = ".drawio") -> Path:
    p = WORK_DIR / f"{uuid.uuid4().hex[:12]}{suffix}"
    p.write_text(xml, encoding="utf-8")
    return p


@mcp.tool()
def list_templates(query: str = "") -> str:
    """List the bundled official SAP reference templates. Optional free-text
    filter matched against titles, aliases, domains and tags from
    template-metadata.json."""
    meta = json.loads((TEMPLATES / "template-metadata.json").read_text())
    out = []
    q = query.lower()
    for name, info in meta.get("templates", {}).items():
        hay = " ".join([
            name, info.get("title", ""), info.get("domain", ""),
            " ".join(info.get("aliases", [])), " ".join(info.get("tags", [])),
        ]).lower()
        if not q or all(tok in hay for tok in q.split()):
            out.append({
                "file": name,
                "title": info.get("title"),
                "domain": info.get("domain"),
                "level": info.get("level"),
                "primary": info.get("primary", False),
            })
    return json.dumps(out, indent=2)


@mcp.tool()
def scaffold(request: str, template: str = "", diagram_name: str = "") -> str:
    """MANDATORY first step for any new diagram. Ranks the 71 bundled SAP
    templates against the natural-language request, copies the best match,
    and returns the pristine template XML plus the SAP design recipe.
    Optionally pin a specific template filename."""
    out = WORK_DIR / f"scaffold-{uuid.uuid4().hex[:8]}.drawio"
    args = [str(SCRIPTS / "scaffold_diagram.py"), request,
            "--include-external-sap-references", "--out", str(out)]
    if template:
        args += ["--template", template]
    if diagram_name:
        args += ["--diagram-name", diagram_name]
    res = _run(args)
    xml = out.read_text(encoding="utf-8") if out.exists() else ""
    return json.dumps({"recipe_and_ranking": res["stdout"],
                       "errors": res["stderr"], "xml": xml})


@mcp.tool()
def extract_icon(service_name: str, x: int = 0, y: int = 0,
                 cell_id: str = "", parent: str = "1") -> str:
    """Look up an official SAP BTP service icon by fuzzy name and return a
    ready-to-paste <mxCell> snippet with the exact SVG data URI, snapped to
    the 10-px grid at 32x32."""
    args = [str(SCRIPTS / "extract_icon.py"), service_name,
            "--x", str(x), "--y", str(y), "--parent", parent]
    if cell_id:
        args += ["--id", cell_id]
    res = _run(args)
    return res["stdout"] or res["stderr"]


@mcp.tool()
def extract_asset(query: str, kind: str = "generic-icon",
                  x: int = 0, y: int = 0, cell_id: str = "") -> str:
    """Fetch any SAP starter-kit asset (connector presets, area shapes,
    number markers, brand-name text, generic icons...) as an mxCell snippet.
    kinds: connector, generic-icon, sap-brand-name, area-shape, number,
    text-element, annotation."""
    args = [str(SCRIPTS / "extract_asset.py"), query, "--kind", kind,
            "--x", str(x), "--y", str(y)]
    if cell_id:
        args += ["--id", cell_id]
    res = _run(args)
    return res["stdout"] or res["stderr"]


@mcp.tool()
def autofix(xml: str) -> str:
    """Apply mechanical SAP-style fixes (grid snapping, hex case,
    absoluteArcSize, strokeWidth, fontFamily) and return the fixed XML."""
    p = _tmpfile(xml)
    res = _run([str(SCRIPTS / "autofix.py"), "--write", str(p)])
    fixed = p.read_text(encoding="utf-8")
    p.unlink(missing_ok=True)
    return json.dumps({"log": res["stdout"] + res["stderr"], "xml": fixed})


@mcp.tool()
def validate(xml: str) -> str:
    """Run the full SAP-style validator: XML well-formedness, duplicate ids,
    bent arrows, label overflow, palette deviations, icon overlap, dark
    background, novelty pill verbs, orphaned edges. Fix every ERROR before
    delivering."""
    p = _tmpfile(xml)
    res = _run([str(SCRIPTS / "validate.py"), str(p)])
    p.unlink(missing_ok=True)
    return res["stdout"] + ("\n" + res["stderr"] if res["stderr"] else "")


@mcp.tool()
def score(xml: str, min_score: int = 90, sap_like: bool = False) -> str:
    """Score the candidate against the SAP reference corpus. Use corpus
    similarity for template-derived diagrams (default) or SAP-likeness
    (sap_like=true) for semantic-fallback diagrams. PASS >= min_score."""
    p = _tmpfile(xml)
    flag = "--min-sap-like" if sap_like else "--min-score"
    res = _run([str(SCRIPTS / "score_corpus.py"), flag, str(min_score), str(p)])
    p.unlink(missing_ok=True)
    return res["stdout"] + ("\n" + res["stderr"] if res["stderr"] else "")


@mcp.tool()
def compare(xml: str, template: str) -> str:
    """Pairwise fingerprint diff of the candidate against one named bundled
    template — shows exactly which dimensions drifted (zones, palette,
    pills, edge anchors, icon sizes)."""
    p = _tmpfile(xml)
    res = _run([str(SCRIPTS / "compare.py"), str(TEMPLATES / template), str(p)])
    p.unlink(missing_ok=True)
    return res["stdout"] + ("\n" + res["stderr"] if res["stderr"] else "")


@mcp.tool()
def render_png(xml: str, scale: int = 2) -> str:
    """Render the diagram to PNG via headless draw.io (requires drawio +
    xvfb installed, DRAWIO_CLI env set). Returns the output file path in
    WORK_DIR, or an error if the renderer is unavailable."""
    p = _tmpfile(xml)
    out = p.with_suffix(".png")
    res = _run([str(SCRIPTS / "render.py"), str(p), "--format", "png",
                "--scale", str(scale)], timeout=300)
    p.unlink(missing_ok=True)
    if out.exists():
        return json.dumps({"png_path": str(out), "log": res["stdout"]})
    return json.dumps({"error": "renderer unavailable or failed",
                       "log": res["stdout"] + res["stderr"]})


@mcp.tool()
def get_reference_doc(name: str) -> str:
    """Read a skill reference doc by name, e.g. 'drawio-gotchas',
    'palette-and-typography', 'shapes-and-edges', 'layout', 'levels',
    'do-and-dont', 'nudge-workflow'."""
    p = SKILL_DIR / "references" / f"{name.removesuffix('.md')}.md"
    if not p.exists():
        available = [f.stem for f in (SKILL_DIR / "references").glob("*.md")]
        return f"Not found. Available: {', '.join(available)}"
    return p.read_text(encoding="utf-8")


if __name__ == "__main__":
    mcp.run()