Add GrampsWeb MCP server with Phase 1 read tools (STI-104)

New MCP server (app/gramps_mcp_server/) that exposes the GrampsWeb REST API
as 12 MCP tools for genealogy data access: person_suchen, person_details,
familie_details, ereignis_details, ort_suchen, ort_details, quelle_suchen,
quelle_details, stammbaum_export, stammbaum_info, medien_liste, notiz_details.

Includes HTTP client with auto-login/token management and Docker compose
services for both prod and dev environments.

Co-Authored-By: Paperclip <noreply@paperclip.ing>

app/gramps_mcp_server/__main__.py

@@ -0,0 +1,4 @@
+"""Ermöglicht Start via: python -m gramps_mcp_server"""
+from gramps_mcp_server.server import mcp
+
+mcp.run(transport="stdio")

app/gramps_mcp_server/client.py

@@ -0,0 +1,116 @@
+"""
+HTTP-Client für die GrampsWeb REST API.
+
+Konfiguration über Umgebungsvariablen:
+  GRAMPS_URL       – Basis-URL (z.B. http://grampsweb:5000)
+  GRAMPS_USERNAME  – Benutzername für Login
+  GRAMPS_PASSWORD  – Passwort für Login
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+
+import requests
+
+logger = logging.getLogger("gramps_mcp_server.client")
+
+
+class GrampsWebClient:
+    """HTTP-Client für GrampsWeb mit automatischem Token-Management."""
+
+    def __init__(
+        self,
+        base_url: str | None = None,
+        username: str | None = None,
+        password: str | None = None,
+    ):
+        self.base_url = (base_url or os.environ.get("GRAMPS_URL", "http://grampsweb:5000")).rstrip("/")
+        self.username = username or os.environ.get("GRAMPS_USERNAME", "")
+        self.password = password or os.environ.get("GRAMPS_PASSWORD", "")
+        self._session = requests.Session()
+        self._token: str | None = None
+
+    def _ensure_auth(self) -> None:
+        """Login falls noch kein Token vorhanden."""
+        if self._token:
+            return
+        if not self.username or not self.password:
+            raise RuntimeError(
+                "GRAMPS_USERNAME und GRAMPS_PASSWORD müssen gesetzt sein."
+            )
+        self._login()
+
+    def _login(self) -> None:
+        """Authentifizierung bei GrampsWeb und Token-Speicherung."""
+        login_endpoints = [
+            ("/api/token/", "form"),
+            ("/api/login/", "json"),
+        ]
+        for path, mode in login_endpoints:
+            url = f"{self.base_url}{path}"
+            payload = {"username": self.username, "password": self.password}
+            try:
+                if mode == "json":
+                    r = self._session.post(url, json=payload, timeout=15)
+                else:
+                    r = self._session.post(url, data=payload, timeout=15)
+                if r.status_code in (200, 201):
+                    data = r.json()
+                    token = (
+                        data.get("access_token")
+                        or data.get("token")
+                        or data.get("access")
+                    )
+                    if token:
+                        self._token = token
+                        self._session.headers["Authorization"] = f"Bearer {token}"
+                        logger.info("GrampsWeb Login erfolgreich via %s", path)
+                        return
+            except Exception:
+                continue
+        raise RuntimeError(
+            f"GrampsWeb Login fehlgeschlagen. URL: {self.base_url}, User: {self.username}"
+        )
+
+    def _request(self, method: str, path: str, **kwargs) -> requests.Response:
+        """HTTP-Request mit automatischer Re-Auth bei 401."""
+        self._ensure_auth()
+        kwargs.setdefault("timeout", 30)
+        url = f"{self.base_url}{path}"
+        r = self._session.request(method, url, **kwargs)
+        if r.status_code == 401:
+            self._token = None
+            self._login()
+            r = self._session.request(method, url, **kwargs)
+        r.raise_for_status()
+        return r
+
+    def get(self, path: str, **kwargs) -> dict | list:
+        """GET-Request, gibt JSON zurück."""
+        return self._request("GET", path, **kwargs).json()
+
+    def get_raw(self, path: str, **kwargs) -> bytes:
+        """GET-Request, gibt Rohdaten zurück (z.B. für Datei-Downloads)."""
+        return self._request("GET", path, **kwargs).content
+
+    def post(self, path: str, **kwargs) -> dict | list:
+        """POST-Request, gibt JSON zurück."""
+        return self._request("POST", path, **kwargs).json()
+
+    def put(self, path: str, **kwargs) -> dict | list:
+        """PUT-Request, gibt JSON zurück."""
+        return self._request("PUT", path, **kwargs).json()
+
+
+# Singleton-Instanz
+_client: GrampsWebClient | None = None
+
+
+def get_client() -> GrampsWebClient:
+    """Gibt die (gecachte) Client-Instanz zurück."""
+    global _client
+    if _client is None:
+        _client = GrampsWebClient()
+    return _client

app/gramps_mcp_server/server.py

@@ -0,0 +1,90 @@
+"""
+MCP Server für GrampsWeb Ahnenforschung.
+
+Startmodus:
+  python -m gramps_mcp_server
+
+Konfiguration über Umgebungsvariablen:
+  GRAMPS_URL       – GrampsWeb Basis-URL (Standard: http://grampsweb:5000)
+  GRAMPS_USERNAME  – GrampsWeb Benutzername
+  GRAMPS_PASSWORD  – GrampsWeb Passwort
+"""
+
+import logging
+import os
+import sys
+
+logging.basicConfig(
+    level=logging.WARNING,
+    format="%(asctime)s %(levelname)s %(name)s: %(message)s",
+    stream=sys.stderr,
+)
+logger = logging.getLogger("gramps_mcp_server")
+
+# ──────────────────────────────────────────────────────────────────────────────
+# Startup-Check: GrampsWeb-Verbindung prüfen
+# ──────────────────────────────────────────────────────────────────────────────
+
+_gramps_url = os.environ.get("GRAMPS_URL", "http://grampsweb:5000")
+_gramps_user = os.environ.get("GRAMPS_USERNAME", "")
+
+if not _gramps_user:
+    logger.error("GRAMPS_USERNAME nicht gesetzt. Server kann nicht starten.")
+    sys.exit(1)
+
+logger.info("GrampsWeb MCP Server startet – URL: %s, User: %s", _gramps_url, _gramps_user)
+
+# ──────────────────────────────────────────────────────────────────────────────
+# MCP Server Initialisierung
+# ──────────────────────────────────────────────────────────────────────────────
+
+from mcp.server.fastmcp import FastMCP  # noqa: E402
+
+mcp = FastMCP(
+    "GrampsWeb Ahnenforschung",
+    instructions=(
+        "MCP-Server für die GrampsWeb-Genealogie-Datenbank der Stiftung. "
+        "Bietet Zugriff auf Personen, Familien, Ereignisse, Orte, Quellen, "
+        "Medien und Notizen des Stammbaums. "
+        f"Verbunden mit: {_gramps_url}"
+    ),
+)
+
+# ──────────────────────────────────────────────────────────────────────────────
+# Lese-Tools registrieren (Phase 1)
+# ──────────────────────────────────────────────────────────────────────────────
+
+from gramps_mcp_server.tools.lesen import (  # noqa: E402
+    ereignis_details,
+    familie_details,
+    medien_liste,
+    notiz_details,
+    ort_details,
+    ort_suchen,
+    person_details,
+    person_suchen,
+    quelle_details,
+    quelle_suchen,
+    stammbaum_export,
+    stammbaum_info,
+)
+
+mcp.tool()(person_suchen)
+mcp.tool()(person_details)
+mcp.tool()(familie_details)
+mcp.tool()(ereignis_details)
+mcp.tool()(ort_suchen)
+mcp.tool()(ort_details)
+mcp.tool()(quelle_suchen)
+mcp.tool()(quelle_details)
+mcp.tool()(stammbaum_export)
+mcp.tool()(medien_liste)
+mcp.tool()(notiz_details)
+mcp.tool()(stammbaum_info)
+
+# ──────────────────────────────────────────────────────────────────────────────
+# Server starten
+# ──────────────────────────────────────────────────────────────────────────────
+
+if __name__ == "__main__":
+    mcp.run(transport="stdio")

app/gramps_mcp_server/tools/lesen.py

@@ -0,0 +1,270 @@
+"""
+Lese-Tools für den GrampsWeb MCP Server (Phase 1).
+
+12 Tools für Lese-Zugriff auf die GrampsWeb REST API:
+  - person_suchen, person_details
+  - familie_details
+  - ereignis_details
+  - ort_suchen, ort_details
+  - quelle_suchen, quelle_details
+  - stammbaum_export, stammbaum_info
+  - medien_liste
+  - notiz_details
+"""
+
+from __future__ import annotations
+
+import json
+from base64 import b64encode
+
+
+def _fmt(data) -> str:
+    """Formatiert Daten als JSON-String."""
+    return json.dumps(data, ensure_ascii=False, indent=2, default=str)
+
+
+def _client():
+    from gramps_mcp_server.client import get_client
+    return get_client()
+
+
+# ──────────────────────────────────────────────────────────────────────────────
+# Personen
+# ──────────────────────────────────────────────────────────────────────────────
+
+def person_suchen(
+    suchbegriff: str = "",
+    seite: int = 1,
+    pro_seite: int = 20,
+) -> str:
+    """
+    Sucht Personen im Stammbaum nach Name.
+
+    Args:
+        suchbegriff: Name oder Suchbegriff (Vor-/Nachname)
+        seite: Seitennummer (ab 1)
+        pro_seite: Ergebnisse pro Seite (max. 100)
+    """
+    pro_seite = min(pro_seite, 100)
+    client = _client()
+
+    if suchbegriff:
+        results = client.get(
+            "/api/search/",
+            params={
+                "query": suchbegriff,
+                "page": seite,
+                "pagesize": pro_seite,
+            },
+        )
+    else:
+        results = client.get(
+            "/api/people/",
+            params={
+                "page": seite,
+                "pagesize": pro_seite,
+                "sort": "surname",
+            },
+        )
+
+    return _fmt({"anzahl": len(results) if isinstance(results, list) else 0, "ergebnisse": results})
+
+
+def person_details(handle: str) -> str:
+    """
+    Gibt vollständige Details einer Person zurück inkl. Ereignisse, Familien, Medien.
+
+    Args:
+        handle: GrampsWeb-Handle der Person (z.B. aus person_suchen)
+    """
+    client = _client()
+    person = client.get(f"/api/people/{handle}", params={"extend": "all", "profile": "all"})
+    return _fmt(person)
+
+
+# ──────────────────────────────────────────────────────────────────────────────
+# Familien
+# ──────────────────────────────────────────────────────────────────────────────
+
+def familie_details(handle: str) -> str:
+    """
+    Gibt Details einer Familie zurück (Eltern, Kinder, Ereignisse).
+
+    Args:
+        handle: GrampsWeb-Handle der Familie
+    """
+    client = _client()
+    family = client.get(f"/api/families/{handle}", params={"extend": "all", "profile": "all"})
+    return _fmt(family)
+
+
+# ──────────────────────────────────────────────────────────────────────────────
+# Ereignisse
+# ──────────────────────────────────────────────────────────────────────────────
+
+def ereignis_details(handle: str) -> str:
+    """
+    Gibt Details eines Ereignisses zurück (Geburt, Tod, Heirat, etc.).
+
+    Args:
+        handle: GrampsWeb-Handle des Ereignisses
+    """
+    client = _client()
+    event = client.get(f"/api/events/{handle}", params={"extend": "all", "profile": "all"})
+    return _fmt(event)
+
+
+# ──────────────────────────────────────────────────────────────────────────────
+# Orte
+# ──────────────────────────────────────────────────────────────────────────────
+
+def ort_suchen(
+    suchbegriff: str = "",
+    seite: int = 1,
+    pro_seite: int = 20,
+) -> str:
+    """
+    Sucht Orte im Stammbaum.
+
+    Args:
+        suchbegriff: Ortsname oder Suchbegriff
+        seite: Seitennummer (ab 1)
+        pro_seite: Ergebnisse pro Seite (max. 100)
+    """
+    pro_seite = min(pro_seite, 100)
+    client = _client()
+    params = {"page": seite, "pagesize": pro_seite}
+    if suchbegriff:
+        params["q"] = suchbegriff
+    results = client.get("/api/places/", params=params)
+    return _fmt({"anzahl": len(results) if isinstance(results, list) else 0, "orte": results})
+
+
+def ort_details(handle: str) -> str:
+    """
+    Gibt Details eines Ortes zurück.
+
+    Args:
+        handle: GrampsWeb-Handle des Ortes
+    """
+    client = _client()
+    place = client.get(f"/api/places/{handle}", params={"extend": "all", "profile": "all"})
+    return _fmt(place)
+
+
+# ──────────────────────────────────────────────────────────────────────────────
+# Quellen
+# ──────────────────────────────────────────────────────────────────────────────
+
+def quelle_suchen(
+    suchbegriff: str = "",
+    seite: int = 1,
+    pro_seite: int = 20,
+) -> str:
+    """
+    Sucht Quellen (Kirchenbücher, Urkunden, etc.) im Stammbaum.
+
+    Args:
+        suchbegriff: Quellenname oder Suchbegriff
+        seite: Seitennummer (ab 1)
+        pro_seite: Ergebnisse pro Seite (max. 100)
+    """
+    pro_seite = min(pro_seite, 100)
+    client = _client()
+    params = {"page": seite, "pagesize": pro_seite}
+    if suchbegriff:
+        params["q"] = suchbegriff
+    results = client.get("/api/sources/", params=params)
+    return _fmt({"anzahl": len(results) if isinstance(results, list) else 0, "quellen": results})
+
+
+def quelle_details(handle: str) -> str:
+    """
+    Gibt Details einer Quelle zurück inkl. Zitierungen.
+
+    Args:
+        handle: GrampsWeb-Handle der Quelle
+    """
+    client = _client()
+    source = client.get(f"/api/sources/{handle}", params={"extend": "all", "profile": "all"})
+    return _fmt(source)
+
+
+# ──────────────────────────────────────────────────────────────────────────────
+# Stammbaum-Export & Info
+# ──────────────────────────────────────────────────────────────────────────────
+
+def stammbaum_export(
+    format: str = "gedcom",
+) -> str:
+    """
+    Exportiert den Stammbaum als GEDCOM oder Gramps-XML.
+
+    Args:
+        format: Export-Format – 'gedcom' oder 'gramps' (Gramps-XML)
+
+    Gibt den Export als Base64-kodierten String zurück.
+    """
+    allowed = {"gedcom", "gramps"}
+    if format not in allowed:
+        return _fmt({"fehler": f"Ungültiges Format '{format}'. Erlaubt: {', '.join(allowed)}"})
+
+    client = _client()
+    # GrampsWeb exporters endpoint
+    ext = "ged" if format == "gedcom" else "gramps"
+    data = client.get_raw(f"/api/exporters/{ext}/file")
+    encoded = b64encode(data).decode("ascii")
+    return _fmt({
+        "format": format,
+        "dateiname": f"stammbaum.{ext}",
+        "groesse_bytes": len(data),
+        "inhalt_base64": encoded[:200] + "..." if len(encoded) > 200 else encoded,
+        "hinweis": "Vollständiger Export als Base64. Bei großen Dateien ggf. abgeschnitten in der Anzeige.",
+    })
+
+
+def stammbaum_info() -> str:
+    """
+    Gibt Metadaten und Statistiken des Stammbaums zurück
+    (Anzahl Personen, Familien, Orte, etc.).
+    """
+    client = _client()
+    metadata = client.get("/api/metadata/")
+    return _fmt(metadata)
+
+
+# ──────────────────────────────────────────────────────────────────────────────
+# Medien
+# ──────────────────────────────────────────────────────────────────────────────
+
+def medien_liste(
+    seite: int = 1,
+    pro_seite: int = 20,
+) -> str:
+    """
+    Listet Medienobjekte (Fotos, Dokumente, Scans) im Stammbaum auf.
+
+    Args:
+        seite: Seitennummer (ab 1)
+        pro_seite: Ergebnisse pro Seite (max. 50)
+    """
+    pro_seite = min(pro_seite, 50)
+    client = _client()
+    results = client.get("/api/media/", params={"page": seite, "pagesize": pro_seite})
+    return _fmt({"anzahl": len(results) if isinstance(results, list) else 0, "medien": results})
+
+
+# ──────────────────────────────────────────────────────────────────────────────
+# Notizen
+# ──────────────────────────────────────────────────────────────────────────────
+
+def notiz_details(handle: str) -> str:
+    """
+    Gibt den Inhalt einer Notiz zurück.
+
+    Args:
+        handle: GrampsWeb-Handle der Notiz
+    """
+    client = _client()
+    note = client.get(f"/api/notes/{handle}")
+    return _fmt(note)

compose.dev.yml

@@ -149,6 +149,19 @@ services:
       - ./app:/app
     command: ["python", "-m", "mcp_server"]
 
+  gramps-mcp:
+    build: ./app
+    depends_on:
+      - grampsweb
+    environment:
+      - GRAMPS_URL=http://grampsweb:5000
+      - GRAMPS_USERNAME=${GRAMPS_USERNAME:-admin@localhost}
+      - GRAMPS_PASSWORD=${GRAMPS_PASSWORD:-gramps_dev_password}
+    stdin_open: true
+    volumes:
+      - ./app:/app
+    command: ["python", "-m", "gramps_mcp_server"]
+
   ollama:
     image: ollama/ollama:latest
     # Kein externes Port-Mapping — nur über internes Docker-Netzwerk erreichbar

compose.yml

@@ -149,6 +149,22 @@ services:
     stdin_open: true
     command: ["python", "-m", "mcp_server"]
 
+  gramps-mcp:
+    build:
+      context: ./app
+      args:
+        APP_VERSION: ${APP_VERSION:-unknown}
+    depends_on:
+      - grampsweb
+    environment:
+      - GRAMPS_URL=http://grampsweb:5000
+      - GRAMPS_USERNAME=${GRAMPS_USERNAME}
+      - GRAMPS_PASSWORD=${GRAMPS_PASSWORD}
+    # Kein Port-Mapping – nur internes Netz
+    # Start via: docker compose run --rm gramps-mcp
+    stdin_open: true
+    command: ["python", "-m", "gramps_mcp_server"]
+
   ollama:
     image: ollama/ollama:latest
     # Kein externes Port-Mapping — nur über internes Docker-Netzwerk erreichbar