"""Demo „MCP-Entwicklung“: MCP-Server für ein Service-Desk-Ticketsystem, Leitfaden, Spezifikation in Kürze, Anbindung an Bifröst, API-Beschreibung, Review-Protokoll und ein Log der Tool-Aufrufe aus der Testumgebung. Bewusst eingebaut: `close_ticket` ist als readOnlyHint markiert (Bifröst würde es als lesend vorschlagen und ohne Freigabe zulassen), und `search_tickets` hat keine Obergrenze. Beides hat das Review nach dem Commit beanstandet; im Code ist es noch offen. Im Log sieht man die Folgen: große, langsame Suchergebnisse. """ import csv import io import random from datetime import datetime, timedelta from app.demo import Demo, day NAME = "Demo: MCP-Entwicklung" DESCRIPTION = ("Beispiel-Workspace für die Entwicklung eines MCP-Servers: Repository ticket-mcp (Anbindung des " "Service-Desk), interner MCP-Leitfaden, Anbindung an Bifröst, API-Beschreibung und ein Log der " "Tool-Aufrufe aus der Testumgebung.") EXAMPLES = [ "Welche Tools bietet ticket-mcp, und sind ihre Annotations nach unserem Leitfaden korrekt gesetzt?", "Passt der Code zu den Beschlüssen aus dem Review von ticket-mcp? Was ist noch offen?", "Welche Tools waren im Oktober am langsamsten, und wo häufen sich Fehler? Zeig es als Diagramm.", "Was muss ich beachten, damit Bifröst unseren MCP-Server anbinden kann?", "Setze die Review-Beschlüsse für close_ticket und search_tickets um und ergänze Tests.", "Entwirf ein neues Tool list_my_tickets nach unserem Leitfaden und der API-Beschreibung.", ] # --- Repository ------------------------------------------------------------------------------------------------- README = """# ticket-mcp MCP-Server für das Service-Desk: Tickets suchen, lesen, kommentieren und schließen. Läuft zentral als Streamable HTTP (`/mcp`), damit Bifröst und andere Clients ihn erreichen. ## Starten ```bash uv sync SERVICEDESK_URL=https://servicedesk.intern/api SERVICEDESK_TOKEN=... uv run ticket-mcp ``` ## Tools | Tool | Zweck | |---|---| | `search_tickets` | Tickets nach Text und Status suchen | | `get_ticket` | Ein Ticket mit Verlauf | | `add_comment` | Kommentar an ein Ticket | | `close_ticket` | Ticket schließen | """ AGENTS = """# Projektregeln ticket-mcp - Jedes Tool hat eine Beschreibung, die sagt, wann man es nutzt, und Annotations nach dem MCP-Leitfaden (`readOnlyHint`, `destructiveHint`, `idempotentHint`, `openWorldHint`). Im Zweifel nicht als lesend markieren. - Parameter mit `Annotated[..., Field(description=...)]`; Grenzen (z. B. `limit`) im Schema, nicht nur im Code. - Listen höchstens 50 Einträge je Aufruf, weitere Seiten über `cursor`. - Fehler als `ToolError` mit einer Meldung, die dem Modell sagt, was es anders machen kann. - Keine Tokens oder personenbezogenen Daten ins Log. - Zu jedem Tool ein Test in `tests/` mit dem Fake-Client. """ PYPROJECT = """[project] name = "ticket-mcp" version = "0.4.0" requires-python = ">=3.12" dependencies = ["mcp>=1.9", "httpx>=0.27"] [project.scripts] ticket-mcp = "ticket_mcp.server:main" [tool.pytest.ini_options] testpaths = ["tests"] """ CLIENT = '''"""Schlanker Client für die Service-Desk-API (siehe Projektunterlagen: Service-Desk-API).""" import os import httpx class ServiceDesk: def __init__(self, url: str | None = None, token: str | None = None): self.http = httpx.Client(base_url=url or os.environ["SERVICEDESK_URL"], timeout=10, headers={"Authorization": f"Bearer {token or os.environ['SERVICEDESK_TOKEN']}"}) def search(self, query: str, status: str | None, limit: int) -> list[dict]: params = {"q": query, "limit": limit} | ({"status": status} if status else {}) response = self.http.get("/tickets", params=params) response.raise_for_status() return response.json()["items"] def get(self, ticket_id: str) -> dict | None: response = self.http.get(f"/tickets/{ticket_id}") if response.status_code == 404: return None response.raise_for_status() return response.json() def comment(self, ticket_id: str, text: str) -> dict: response = self.http.post(f"/tickets/{ticket_id}/comments", json={"text": text}) response.raise_for_status() return response.json() def close(self, ticket_id: str, resolution: str) -> dict: response = self.http.post(f"/tickets/{ticket_id}/close", json={"resolution": resolution}) response.raise_for_status() return response.json() ''' SERVER_START = '''"""MCP-Server ticket-mcp: Service-Desk-Tickets für KI-Assistenten.""" from typing import Annotated from mcp.server.fastmcp import FastMCP from mcp.server.fastmcp.exceptions import ToolError from mcp.types import ToolAnnotations from pydantic import Field from ticket_mcp.client import ServiceDesk mcp = FastMCP("ticket-mcp", instructions=( "Tickets des Service-Desk. Erst mit search_tickets suchen, dann mit get_ticket lesen. " "Ticketnummern haben die Form SD-12345.")) desk = ServiceDesk def client() -> ServiceDesk: return desk() @mcp.tool(annotations=ToolAnnotations(title="Tickets suchen", readOnlyHint=True, openWorldHint=True)) def search_tickets(query: Annotated[str, Field(description="Suchbegriffe, z. B. 'VPN Ausfall'")], status: Annotated[str | None, Field(description="open, in_progress oder closed")] = None, limit: int = 20) -> list[dict]: """Tickets nach Text und Status suchen; liefert Nummer, Titel, Status und letzte Änderung.""" return [{k: t[k] for k in ("id", "title", "status", "updated")} for t in client().search(query, status, limit)] @mcp.tool(annotations=ToolAnnotations(title="Ticket lesen", readOnlyHint=True, openWorldHint=True)) def get_ticket(ticket_id: Annotated[str, Field(description="Ticketnummer, z. B. SD-12345")]) -> dict: """Ein Ticket mit Beschreibung und Verlauf.""" ticket = client().get(ticket_id) if ticket is None: raise ToolError(f"Ticket {ticket_id} gibt es nicht. Nummer prüfen oder mit search_tickets suchen.") return ticket def main() -> None: mcp.run(transport="streamable-http") ''' ADD_COMMENT = '''@mcp.tool(annotations=ToolAnnotations(title="Kommentar schreiben", readOnlyHint=False, destructiveHint=False, openWorldHint=True)) def add_comment(ticket_id: Annotated[str, Field(description="Ticketnummer")], text: Annotated[str, Field(description="Kommentar, sichtbar für alle Bearbeitenden", max_length=4000)] ) -> dict: """Kommentar an ein Ticket schreiben.""" return client().comment(ticket_id, text) ''' SERVER_WRITE = SERVER_START.replace("def main() -> None:", ADD_COMMENT + "def main() -> None:") CLOSE = '''@mcp.tool(annotations=ToolAnnotations(title="Ticket schließen", readOnlyHint=True, openWorldHint=True)) def close_ticket(ticket_id: Annotated[str, Field(description="Ticketnummer")], resolution: Annotated[str, Field(description="Lösung in einem Satz")]) -> dict: """Ticket mit einer Lösung schließen.""" return client().close(ticket_id, resolution) ''' SERVER = SERVER_WRITE.replace("def main() -> None:", CLOSE + "def main() -> None:") FAKE = '''"""Fake-Client für Tests: Tickets im Speicher.""" class FakeDesk: def __init__(self): self.tickets = {f"SD-{n}": {"id": f"SD-{n}", "title": f"Ticket {n}", "status": "open", "updated": "2026-10-01", "comments": []} for n in range(1, 121)} def search(self, query, status, limit): found = [t for t in self.tickets.values() if not status or t["status"] == status] return found[:limit] def get(self, ticket_id): return self.tickets.get(ticket_id) def comment(self, ticket_id, text): self.tickets[ticket_id]["comments"].append(text) return {"id": ticket_id, "comments": len(self.tickets[ticket_id]["comments"])} def close(self, ticket_id, resolution): self.tickets[ticket_id]["status"] = "closed" return {"id": ticket_id, "status": "closed"} ''' TESTS = '''import pytest from mcp.server.fastmcp.exceptions import ToolError from tests.fake import FakeDesk from ticket_mcp import server @pytest.fixture(autouse=True) def fake(monkeypatch): desk = FakeDesk() monkeypatch.setattr(server, "desk", lambda: desk) return desk def test_suche_liefert_kurzform(): assert server.search_tickets("Ticket", limit=2) == [ {"id": "SD-1", "title": "Ticket 1", "status": "open", "updated": "2026-10-01"}, {"id": "SD-2", "title": "Ticket 2", "status": "open", "updated": "2026-10-01"}] def test_unbekanntes_ticket(): with pytest.raises(ToolError, match="gibt es nicht"): server.get_ticket("SD-999") def test_kommentar(fake): server.add_comment("SD-1", "Erledigt?") assert fake.tickets["SD-1"]["comments"] == ["Erledigt?"] ''' TESTS_CLOSE = TESTS + ''' def test_schliessen(fake): server.close_ticket("SD-1", "Neustart hat geholfen") assert fake.tickets["SD-1"]["status"] == "closed" ''' DOCKERFILE = """FROM python:3.12-slim WORKDIR /app COPY . . RUN pip install --no-cache-dir . EXPOSE 8000 CMD ["ticket-mcp"] """ ATTACHMENT = '''"""Anhänge (Entwurf): Datei als Base64 liefern, damit die Dokumenten-Auslese sie lesen kann.""" import base64 from ticket_mcp.server import client, mcp from mcp.types import ToolAnnotations @mcp.tool(annotations=ToolAnnotations(title="Anhang lesen", readOnlyHint=True, openWorldHint=True)) def get_attachment(ticket_id: str, attachment_id: str) -> dict: """Anhang eines Tickets als Base64 (höchstens 10 MB).""" data = client().http.get(f"/tickets/{ticket_id}/attachments/{attachment_id}").content return {"name": attachment_id, "content": base64.b64encode(data).decode()} ''' COMMITS = [ (20, "entwicklung", "Grundgerüst ticket-mcp mit search_tickets und get_ticket", { "README.md": README, "pyproject.toml": PYPROJECT, "ticket_mcp/__init__.py": "", "ticket_mcp/client.py": CLIENT, "ticket_mcp/server.py": SERVER_START}), (17, "entwicklung", "Tests mit Fake-Client", { "tests/__init__.py": "", "tests/fake.py": FAKE, "tests/test_tools.py": TESTS.replace(''' def test_kommentar(fake): server.add_comment("SD-1", "Erledigt?") assert fake.tickets["SD-1"]["comments"] == ["Erledigt?"] ''', "")}), (14, "fachbereich", "add_comment: Kommentare an Tickets", { "ticket_mcp/server.py": SERVER_WRITE, "tests/test_tools.py": TESTS}), (9, "admin", "Projektregeln in AGENTS.md", {"AGENTS.md": AGENTS}), (5, "entwicklung", "close_ticket: Tickets schließen", { "ticket_mcp/server.py": SERVER, "tests/test_tools.py": TESTS_CLOSE}), (2, "fachbereich", "Dockerfile für den Betrieb", {"Dockerfile": DOCKERFILE}), ] BRANCH = ("feature/anhaenge", 1, "entwicklung", "get_attachment: Anhänge als Base64 (Entwurf)", {"ticket_mcp/attachments.py": ATTACHMENT}) # --- Unterlagen ------------------------------------------------------------------------------------------------- # Review nach dem Commit von close_ticket, vor Dockerfile und Anhängen REVIEW = day(3) DOCUMENTS = { "MCP-Leitfaden intern.md": """# MCP-Leitfaden (intern) Regeln für MCP-Server, die wir selbst entwickeln und an Bifröst anbinden. ## Transport und Betrieb - Nur **Streamable HTTP** (ein Endpunkt, z. B. `/mcp`). Bifröst läuft zentral und erreicht keine lokalen stdio-Server; Dienste laufen deshalb als eigener Container. - Anmeldung mit `Authorization: Bearer `; der Token gehört in den Secret-Store, nie ins Repository. - Health-Check unter `/healthz`, strukturierte Logs ohne Tokens und ohne personenbezogene Daten. ## Tools - **Name:** `verb_objekt` in snake_case (`search_tickets`, `close_ticket`). - **Beschreibung:** sagt, *wann* man das Tool nutzt und was zurückkommt. Das Modell wählt Tools fast nur nach der Beschreibung. - **Parameter:** jeder mit Beschreibung; Grenzen (Länge, Bereich, Aufzählung) im Schema. - **Annotations** (Pflicht): - `readOnlyHint: true` nur, wenn das Tool **nichts** verändert, auch nicht im Fremdsystem. - `destructiveHint: true`, wenn es löscht oder schwer umkehrbar ist. - `idempotentHint: true`, wenn ein wiederholter Aufruf nichts zusätzlich bewirkt. - `openWorldHint: true`, wenn es mit einem externen System spricht. - **Listen:** höchstens 50 Einträge je Aufruf, weitere Seiten über einen `cursor`. Große Antworten kosten Tokens und werden in Bifröst bei 20.000 Zeichen abgeschnitten. - **Fehler:** als Tool-Fehler (`isError`) mit einer Meldung, die dem Modell sagt, was es anders machen kann („Ticket gibt es nicht, mit search_tickets suchen“), nicht als Stacktrace. - **Tabellen:** Liefert ein Tool `structuredContent` mit `columns` (Liste von Namen) und `rows` (Liste von Zeilen), zeigt Bifröst das Ergebnis als Tabelle mit CSV-Download und, wo es passt, als Diagramm. ## Anleitung des Servers Das Feld `instructions` beschreibt in wenigen Sätzen, wie die Tools zusammen zu nutzen sind. Bifröst gibt es dem Modell mit, sobald Tools des Servers angeboten werden. Keine Regeln hineinschreiben, die Bifröst-Regeln aushebeln sollen; das Modell behandelt die Anleitung als nachrangig. """, "MCP-Spezifikation Kurzfassung.md": """# MCP in Kürze Das Model Context Protocol verbindet KI-Anwendungen (Clients) mit Diensten (Servern) über JSON-RPC 2.0. ## Bausteine eines Servers - **Tools:** Funktionen, die das Modell aufruft (`tools/list`, `tools/call`). Jedes Tool hat Name, Beschreibung, `inputSchema` (JSON Schema) und optional `annotations` und ein `outputSchema`. - **Resources:** lesbare Inhalte mit URI (`resources/list`, `resources/read`), z. B. Dateien oder Datensätze. - **Prompts:** vorbereitete Vorlagen, die Nutzer auswählen (`prompts/list`, `prompts/get`). ## Ablauf 1. `initialize`: Client und Server tauschen Protokollversion und Fähigkeiten (capabilities) aus; der Server kann `instructions` mitgeben. 2. `notifications/initialized`, danach normale Anfragen. 3. Ein Tool-Ergebnis enthält `content` (Text, Bild, eingebettete Ressource …), optional `structuredContent` und `isError: true` bei einem Fehler, den das Modell sehen soll. ## Tool-Annotations `title`, `readOnlyHint`, `destructiveHint`, `idempotentHint`, `openWorldHint`. Es sind **Hinweise**: Ein Client darf ihnen nicht blind vertrauen, wenn der Server nicht vertrauenswürdig ist. ## Transporte - **stdio:** der Client startet den Server als Prozess (lokal). - **Streamable HTTP:** ein HTTP-Endpunkt für POST (Anfragen) und optional GET (Server-Nachrichten per SSE); eine Sitzung wird über den Header `Mcp-Session-Id` gehalten. """, "Anbindung an Bifröst.md": """# Einen MCP-Server an Bifröst anbinden 1. **Erreichbarkeit:** Der Host muss in der Freigabeliste des Betriebs stehen (`AIW_MCP_ALLOWED_HOSTS`). 2. **Registrieren:** Ein Plattform-Admin legt den Server unter *Administration → MCP-Server* an (URL, Token) und klickt *Tools abrufen*. Alternativ legt ein Workspace-Admin eine Verbindung im Workspace an; zu einer neuen Adresse wartet sie auf Freigabe durch einen Plattform-Admin. 3. **Einstufen:** Jedes Tool bekommt eine Risikoklasse. Bifröst schlägt sie aus den Annotations vor: - `readOnlyHint: true` → `read` (standardmäßig ohne Freigabe erlaubt), - `destructiveHint: true` → `destructive` (für die KI gesperrt), - sonst `write_external` (standardmäßig nur mit Freigabe durch einen Menschen). Der Vorschlag ist nur ein Vorschlag; entschieden wird von Hand. **Ein falsches `readOnlyHint` führt also leicht dazu, dass ein schreibendes Tool ohne Freigabe läuft.** 4. **Freischalten:** Workspace-Admins schalten den Server für ihren Workspace frei. 5. **Änderungen:** Ändern sich Name, Beschreibung oder Schema eines Tools, erkennt Bifröst das am Fingerabdruck; das Tool bleibt gesperrt, bis es neu eingestuft ist. Grenzen: Zeitlimit je Aufruf 30 Sekunden (`AIW_MCP_TIMEOUT_SECONDS`), Ausgabe höchstens 20.000 Zeichen. Tool-Ergebnisse gehen als Fremdinhalt markiert an das Modell. """, "Service-Desk-API.md": """# Service-Desk-API (Auszug) Basis-URL `https://servicedesk.intern/api`, Anmeldung mit Bearer-Token eines technischen Kontos. | Methode | Pfad | Zweck | |---|---|---| | GET | `/tickets?q=&status=&assignee=&limit=&cursor=` | Suche; `limit` höchstens 100, Antwort mit `items` und `next_cursor` | | GET | `/tickets/{id}` | Ticket mit Beschreibung, Verlauf, Anhängen (nur Metadaten) | | GET | `/tickets/{id}/attachments/{aid}` | Anhang als Datei | | POST | `/tickets/{id}/comments` | Kommentar (`{"text": ...}`), 4000 Zeichen | | POST | `/tickets/{id}/close` | Schließen (`{"resolution": ...}`); erneutes Schließen ändert nichts | | GET | `/me/tickets?status=` | Tickets, die dem technischen Konto bzw. per `X-On-Behalf-Of` einer Person zugewiesen sind | Status: `open`, `in_progress`, `closed`. Ticketnummern: `SD-` und fünf Ziffern. Fehler: 404 (nicht gefunden), 409 (Ticket bereits geschlossen beim Kommentieren), 429 (zu viele Anfragen, `Retry-After` beachten). """, f"Protokolle/{REVIEW:%Y-%m-%d} Review ticket-mcp.md": f"""# Review ticket-mcp, {REVIEW:%d.%m.%Y} Teilnehmende: Entwicklung, Betrieb, Plattform-Team Bifröst ## Ergebnisse 1. **close_ticket** verändert das Fremdsystem und darf nicht als lesend markiert sein: `readOnlyHint: false`, `destructiveHint: false` (Tickets lassen sich wieder öffnen), `idempotentHint: true`. In Bifröst wird es damit `write_external` und braucht eine Freigabe. 2. **search_tickets:** `limit` im Schema auf 1 bis 50 begrenzen (Standard 20) und `cursor` für weitere Seiten ergänzen. In der Testumgebung kamen Antworten mit über 200 KB vor. 3. **get_ticket:** Fehlertext ist gut; so beibehalten. 4. Neues Tool **list_my_tickets** für „meine offenen Tickets“ über `/me/tickets`, lesend. ## Offen - Anhänge (geplant als `feature/anhaenge`): Größe begrenzen, Inhalt über die Dokumenten-Auslese statt Base64 ans Modell. """, } # --- Log der Tool-Aufrufe --------------------------------------------------------------------------------------- TOOLS = { # Tool: (Anteil, mittlere Dauer ms, Fehlerquote, mittlere Antwort KB) "search_tickets": (45, 900, 0.03, 40), "get_ticket": (35, 180, 0.08, 6), "add_comment": (12, 250, 0.02, 1), "close_ticket": (8, 300, 0.05, 1)} ERRORS = {"search_tickets": ["Timeout nach 30 s", "429 Too Many Requests"], "get_ticket": ["Ticket gibt es nicht", "Ticket gibt es nicht", "429 Too Many Requests"], "add_comment": ["409 Ticket bereits geschlossen"], "close_ticket": ["409 Ticket bereits geschlossen"]} CLIENTS = ["bifroest", "bifroest", "bifroest", "test-skript"] def calls_csv(rows: int = 2000) -> bytes: """Tool-Aufrufe der Testumgebung im Oktober 2026, reproduzierbar. Große Suchen (limit > 50) sind langsam.""" rng = random.Random(1008) out = io.StringIO() writer = csv.writer(out) writer.writerow(["zeit", "tool", "client", "limit", "dauer_ms", "status", "fehler", "antwort_kb"]) start = datetime(2026, 10, 1, 7, 0) for _ in range(rows): when = start + timedelta(minutes=rng.randrange(0, 24 * 60 * 24)) tool = rng.choices(list(TOOLS), weights=[t[0] for t in TOOLS.values()])[0] _, duration, error_rate, size = TOOLS[tool] limit = rng.choice([10, 20, 20, 50, 100, 200]) if tool == "search_tickets" else "" factor = (limit / 20) if limit else 1 ms = int(rng.lognormvariate(0, 0.4) * duration * max(1, factor ** 0.8)) kb = round(rng.lognormvariate(0, 0.3) * size * factor, 1) status, error = "ok", "" if ms > 30000 or (limit and limit >= 200 and rng.random() < 0.15): status, error, ms = "timeout", "Timeout nach 30 s", 30000 elif rng.random() < error_rate: status, error = "error", rng.choice(ERRORS[tool]) writer.writerow([when.isoformat(timespec="minutes"), tool, rng.choice(CLIENTS), limit, ms, status, error, kb]) return out.getvalue().encode() DEMO = Demo(key="mcp", name=NAME, description=DESCRIPTION, examples=EXAMPLES, repo="ticket-mcp", commits=COMMITS, branch=BRANCH, upload=("tool_aufrufe_2026_10.csv", calls_csv), folder="MCP-Unterlagen", documents=DOCUMENTS)