sap-diagrams-mcp v0.1.0 (mirror of sap-architecture-diagrams/mcp-server)

README.md

@@ -0,0 +1,17 @@
+# sap-diagrams-mcp
+
+MCP server exposing SAP BTP solution-diagram generation (the sap-drawio app's
+backend) to any MCP client. Thin REST wrapper — the backend owns the icon DB,
+SAP service catalog, landscape renderer and QA scorer.
+
+Tools: `render_sketch` (deterministic, bring your own architecture),
+`generate_diagram` (full pipeline from prose), `convert_mermaid`,
+`get_diagram`, `list_diagrams`.
+
+Canonical source lives in the `sap-architecture-diagrams` repo (`mcp-server/`);
+this repo is the public install mirror for the MCP hub.
+
+```
+uv tool install git+https://git.alexzaw.dev/alexz/sap-diagrams-mcp
+SAP_DIAGRAMS_API=http://<backend>/api sap-diagrams-mcp
+```

pyproject.toml

@@ -0,0 +1,19 @@
+[project]
+name = "sap-diagrams-mcp"
+version = "0.1.0"
+description = "MCP server exposing SAP BTP solution-diagram generation (sap-drawio backend) to any MCP client"
+requires-python = ">=3.10"
+dependencies = [
+    "mcp>=1.2.0",
+    "httpx>=0.27",
+]
+
+[project.scripts]
+sap-diagrams-mcp = "sap_diagrams_mcp.server:main"
+
+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"
+
+[tool.hatch.build.targets.wheel]
+packages = ["src/sap_diagrams_mcp"]

src/sap_diagrams_mcp/__init__.py

@@ -0,0 +1 @@
+"""sap-diagrams-mcp — thin MCP wrapper over the sap-drawio backend API."""

src/sap_diagrams_mcp/server.py

@@ -0,0 +1,203 @@
+"""MCP server for SAP BTP solution-diagram generation.
+
+Thin client over the sap-drawio backend REST API — no pipeline logic, no LLM
+calls of its own. The backend owns the icon DB, service catalog, landscape
+renderer, QA scorer, and conversion history; this server wraps its job-based
+endpoints as MCP tools and compacts the results (full draw.io XML / SVG are
+only returned by get_diagram on explicit request — they run ~300 KB because
+every icon is an embedded data URI).
+
+Env:
+  SAP_DIAGRAMS_API         backend API base (default http://192.168.0.150:8016/api)
+  SAP_DIAGRAMS_PUBLIC_URL  user-facing app URL for links (default https://sap-drawio.alexzaw.dev)
+"""
+import asyncio
+import os
+from typing import Literal, Optional
+
+import httpx
+from mcp.server.fastmcp import FastMCP
+from pydantic import BaseModel, Field
+
+API = os.environ.get("SAP_DIAGRAMS_API", "http://192.168.0.150:8016/api").rstrip("/")
+PUBLIC_URL = os.environ.get("SAP_DIAGRAMS_PUBLIC_URL", "https://sap-drawio.alexzaw.dev").rstrip("/")
+POLL_INTERVAL_S = 3
+POLL_CAP_S = 360  # describe path can run up to 3 refine rounds with a vision judge
+
+mcp = FastMCP("sap-diagrams")
+
+
+class Zone(BaseModel):
+    """Ownership boundary drawn as a bordered area."""
+    id: str
+    name: str
+    kind: Literal["btp", "s4", "non-sap"]
+
+
+class Group(BaseModel):
+    """Sub-frame inside a zone (e.g. 'Financial', 'Via CPI'). Needs >= 2 members."""
+    id: str
+    name: str
+    zone: str = Field(description="id of the zone this group lives in")
+
+
+class Component(BaseModel):
+    id: str
+    label: str = Field(description="display name, e.g. 'SAP Integration Suite' or 'Salesforce CRM'")
+    raw_type: str = Field(
+        default="service",
+        description="actor | mobile | ui | erp | service | db — actors/devices get the users lane")
+    zone: Optional[str] = Field(default=None, description="zone id; omit to let the backend assign heuristically")
+    group: Optional[str] = Field(default=None, description="group id within the same zone")
+    sublabel: Optional[str] = Field(default=None, description="API/product code shown under the label, e.g. 'SAP_COM_0002 · SOAP'")
+    description: Optional[str] = None
+
+
+class Connection(BaseModel):
+    from_id: str
+    to_id: str
+    label: str = Field(description="protocol label, e.g. 'OData POST', 'SFTP ISO 20022', 'IDoc SOAP'")
+    style: Literal["solid", "dashed"] = Field(
+        default="solid", description="dashed for file/batch/async transfers")
+
+
+class Sketch(BaseModel):
+    """Architecture sketch, schema v2 — the same shape the app's own pipeline uses."""
+    title: str
+    zones: list[Zone] = Field(default_factory=list)
+    groups: list[Group] = Field(default_factory=list)
+    components: list[Component]
+    connections: list[Connection] = Field(default_factory=list)
+
+
+def _summary(doc: dict) -> dict:
+    qa = (doc.get("validation") or {}).get("qa") or {}
+    return {
+        "conversion_id": doc["id"],
+        "title": doc["title"],
+        "qa_score": qa.get("score"),
+        "qa_errors": qa.get("errors") or [],
+        "review_warnings": (doc.get("validation") or {}).get("warnings") or [],
+        "components": [
+            {"label": c.get("label"), "sap_service": c.get("sap_service_name"),
+             "zone": c.get("zone"), "icon": c.get("icon_id")}
+            for c in doc.get("components") or []
+        ],
+        "connection_count": len(doc.get("connections") or []),
+        "open_in_app": f"{PUBLIC_URL} (History tab, id {doc['id']})",
+        "next_steps": "Call get_diagram(conversion_id, format='xml') for the draw.io file "
+                      "or format='svg' for a preview.",
+    }
+
+
+async def _post_and_poll(path: str, payload: dict, params: dict | None = None) -> dict:
+    async with httpx.AsyncClient(timeout=60) as client:
+        try:
+            resp = await client.post(f"{API}{path}", json=payload, params=params or {})
+        except httpx.HTTPError as e:
+            raise RuntimeError(f"diagram backend unavailable: {e.__class__.__name__}")
+        if resp.status_code == 422:
+            raise RuntimeError(f"rejected: {resp.json().get('detail', resp.text[:300])}")
+        if resp.status_code != 200:
+            raise RuntimeError(f"diagram backend error HTTP {resp.status_code}")
+        job_id = resp.json()["job_id"]
+
+        waited = 0
+        while waited < POLL_CAP_S:
+            await asyncio.sleep(POLL_INTERVAL_S)
+            waited += POLL_INTERVAL_S
+            job = (await client.get(f"{API}/jobs/{job_id}")).json()
+            if job.get("status") == "done":
+                return job["result"]
+            if job.get("status") == "error":
+                raise RuntimeError(job.get("error") or "conversion failed")
+        raise RuntimeError(
+            f"still processing after {POLL_CAP_S}s — the diagram may finish shortly; "
+            f"call list_diagrams to find it once complete (job {job_id})")
+
+
+@mcp.tool()
+async def render_sketch(sketch: Sketch, title: str = "") -> dict:
+    """Render an architecture sketch you have already designed into a polished SAP BTP
+    solution diagram (draw.io XML + SVG, official SAP style: zones as bordered areas,
+    services on grey circle icon tiles with real SAP/brand icons, orthogonal
+    protocol-labeled connectors) with a structural QA score.
+
+    Deterministic and fast (~2 s): the backend maps labels to its SAP service catalog
+    and icon database and renders — no AI runs server-side, so YOU are responsible for
+    the architecture quality. Give every component a zone, label every connection with
+    its protocol, use style='dashed' for file/batch/async flows. Prefer this tool when
+    you can design the architecture yourself; use generate_diagram to delegate instead.
+    The result is saved to the app's History and can be refined there later."""
+    payload = {"sketch": sketch.model_dump(exclude_none=True), "title": title}
+    return _summary(await _post_and_poll("/render-sketch", payload))
+
+
+@mcp.tool()
+async def generate_diagram(description: str, title: str = "", use_flagship: bool = False) -> dict:
+    """Generate a complete SAP BTP solution diagram from a plain-text description of a
+    landscape or integration scenario. The backend's own pipeline extracts components,
+    zones and protocols from the text, maps them to SAP services and icons, renders,
+    and iteratively refines against a QA gate (up to ~3 rounds; typically 1-4 minutes).
+
+    Use when you have prose, not a structured design. Set use_flagship=true for the
+    highest-quality extraction on complex scenarios. Mention every system, the
+    integration middleware, protocols per flow, and the target SAP system explicitly —
+    the pipeline never invents components that are not in the text."""
+    return _summary(await _post_and_poll(
+        "/convert-describe", {"description": description, "title": title},
+        params={"useFlagship": int(use_flagship)}))
+
+
+@mcp.tool()
+async def convert_mermaid(mermaid: str, title: str = "", use_flagship: bool = False) -> dict:
+    """Convert a Mermaid flowchart/graph definition into a polished SAP BTP solution
+    diagram. The Mermaid text is parsed structurally (nodes, edges, subgraphs, edge
+    labels); the backend pipeline then maps, renders and QA-refines it like any other
+    conversion. Edge labels become protocol labels — include them."""
+    return _summary(await _post_and_poll(
+        "/convert-mermaid", {"mermaid": mermaid, "title": title},
+        params={"useFlagship": int(use_flagship)}))
+
+
+@mcp.tool()
+async def get_diagram(conversion_id: str, format: Literal["summary", "xml", "svg"] = "summary") -> dict:
+    """Fetch a previously created diagram by conversion_id. format='summary' returns the
+    compact metadata; format='xml' returns the full draw.io file content and 'svg' the
+    inline preview — both are LARGE (~100-300 KB, icons embedded as data URIs), request
+    them only when you actually need the file content."""
+    async with httpx.AsyncClient(timeout=30) as client:
+        resp = await client.get(f"{API}/conversions/{conversion_id}")
+        if resp.status_code == 404:
+            raise RuntimeError(f"no conversion with id {conversion_id}")
+        resp.raise_for_status()
+        doc = resp.json()
+    if format == "xml":
+        return {"conversion_id": doc["id"], "filename": doc.get("filename"), "xml": doc["xml"]}
+    if format == "svg":
+        return {"conversion_id": doc["id"], "svg": doc["svg"]}
+    return _summary(doc)
+
+
+@mcp.tool()
+async def list_diagrams(limit: int = 10) -> list[dict]:
+    """List recent diagrams (newest first): conversion_id, title, source type, creation
+    time and QA score. Use to find an existing diagram to fetch with get_diagram."""
+    async with httpx.AsyncClient(timeout=30) as client:
+        resp = await client.get(f"{API}/conversions")
+        resp.raise_for_status()
+        rows = resp.json()
+    return [
+        {"conversion_id": r["id"], "title": r.get("title"),
+         "source_type": r.get("source_type"), "created_at": r.get("created_at"),
+         "qa_score": ((r.get("validation") or {}).get("qa") or {}).get("score")}
+        for r in rows[:max(1, min(limit, 50))]
+    ]
+
+
+def main() -> None:
+    mcp.run()
+
+
+if __name__ == "__main__":
+    main()