commit 761180f1056d16c8a6beb3b91877a565fdfbc7fd Author: alexz Date: Fri Jul 24 15:15:07 2026 -0700 sap-diagrams-mcp v0.1.0 (mirror of sap-architecture-diagrams/mcp-server) diff --git a/README.md b/README.md new file mode 100644 index 0000000..9e9fdca --- /dev/null +++ b/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:///api sap-diagrams-mcp +``` diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..f275e8c --- /dev/null +++ b/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"] diff --git a/src/sap_diagrams_mcp/__init__.py b/src/sap_diagrams_mcp/__init__.py new file mode 100644 index 0000000..e79197e --- /dev/null +++ b/src/sap_diagrams_mcp/__init__.py @@ -0,0 +1 @@ +"""sap-diagrams-mcp — thin MCP wrapper over the sap-drawio backend API.""" diff --git a/src/sap_diagrams_mcp/__pycache__/__init__.cpython-314.pyc b/src/sap_diagrams_mcp/__pycache__/__init__.cpython-314.pyc new file mode 100644 index 0000000..1cbc23e Binary files /dev/null and b/src/sap_diagrams_mcp/__pycache__/__init__.cpython-314.pyc differ diff --git a/src/sap_diagrams_mcp/__pycache__/server.cpython-314.pyc b/src/sap_diagrams_mcp/__pycache__/server.cpython-314.pyc new file mode 100644 index 0000000..7706c74 Binary files /dev/null and b/src/sap_diagrams_mcp/__pycache__/server.cpython-314.pyc differ diff --git a/src/sap_diagrams_mcp/server.py b/src/sap_diagrams_mcp/server.py new file mode 100644 index 0000000..37ee5aa --- /dev/null +++ b/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()