grotti/grotti/mcp_server.py

"""MCP-Server für Grotti — spricht nur die HTTP-API mit einem API-Token.

Umgebung:
  GROTTI_URL          Basis-URL (Standard http://127.0.0.1:8770)
  GROTTI_TOKEN        Bearer-Token (grt_…)
  GROTTI_MCP_READ_ONLY=1  nur Lese-Tools

Start:
  python -m grotti.mcp_server
"""

from __future__ import annotations

import json
import os
import sys
import urllib.error
import urllib.parse
import urllib.request

READ_ONLY = os.environ.get("GROTTI_MCP_READ_ONLY", "").strip() in ("1", "true", "yes")
BASE = (os.environ.get("GROTTI_URL") or "http://127.0.0.1:8770").rstrip("/")
TOKEN = (os.environ.get("GROTTI_TOKEN") or "").strip()


def _api(method: str, path: str, body: dict | None = None):
    if not TOKEN:
        raise RuntimeError("GROTTI_TOKEN fehlt.")
    url = BASE + "/api/v1" + path
    data = None
    headers = {
        "Authorization": "Bearer " + TOKEN,
        "Accept": "application/json",
    }
    if body is not None:
        data = json.dumps(body).encode("utf-8")
        headers["Content-Type"] = "application/json"
    req = urllib.request.Request(url, data=data, headers=headers, method=method)
    try:
        with urllib.request.urlopen(req, timeout=60) as resp:
            raw = resp.read()
            if not raw:
                return None
            return json.loads(raw.decode("utf-8"))
    except urllib.error.HTTPError as exc:
        detail = exc.read().decode("utf-8", errors="replace")
        try:
            payload = json.loads(detail)
            message = payload.get("error") or detail
        except json.JSONDecodeError:
            message = detail or str(exc)
        raise RuntimeError(message) from exc


def _write_guard():
    if READ_ONLY:
        raise RuntimeError("MCP läuft im Read-only-Modus (GROTTI_MCP_READ_ONLY).")


def main():
    try:
        from mcp.server.fastmcp import FastMCP
    except ImportError:
        print(
            "Paket „mcp“ fehlt. Installieren: pip install mcp",
            file=sys.stderr,
        )
        sys.exit(1)

    server = FastMCP(
        "grotti",
        instructions=(
            "Grotti steuert die 3D-Drucker-Farm. Destruktive Aktionen brauchen "
            "confirm=true. Kein freier G-code, kein Firmware-Restart."
        ),
    )

    @server.tool()
    def list_printers(live: bool = True) -> str:
        """Druckerliste mit optionalem Live-Zustand."""
        q = "?live=1" if live else "?live=0"
        return json.dumps(_api("GET", "/drucker" + q), ensure_ascii=False, indent=2)

    @server.tool()
    def get_printer(host: str) -> str:
        """Einen Drucker anhand Host/IP aus der Live-Liste."""
        rows = _api("GET", "/drucker?live=1") or []
        for row in rows:
            if row.get("host") == host:
                return json.dumps(row, ensure_ascii=False, indent=2)
        raise RuntimeError("Drucker nicht gefunden: %s" % host)

    @server.tool()
    def list_running() -> str:
        """Laufende Drucke und Abläufe."""
        return json.dumps(_api("GET", "/laeuft"), ensure_ascii=False, indent=2)

    @server.tool()
    def list_queue(limit: int = 40) -> str:
        """Druckwarteschlange."""
        return json.dumps(
            _api("GET", "/druck-queue?limit=%d" % max(1, min(limit, 200))),
            ensure_ascii=False,
            indent=2,
        )

    @server.tool()
    def list_events(limit: int = 30) -> str:
        """Letzte Ereignisse / Aktivität."""
        return json.dumps(
            _api("GET", "/ereignisse?limit=%d" % max(1, min(limit, 200))),
            ensure_ascii=False,
            indent=2,
        )

    @server.tool()
    def list_files(limit: int = 50) -> str:
        """Dateien in der Bibliothek."""
        rows = _api("GET", "/dateien") or []
        return json.dumps(rows[: max(1, min(limit, 200))], ensure_ascii=False, indent=2)

    @server.tool()
    def list_spools() -> str:
        """Spulen / Materialbestand."""
        return json.dumps(_api("GET", "/spulen"), ensure_ascii=False, indent=2)

    @server.tool()
    def auftraege_lesen(limit: int = 30, status: str = "") -> str:
        """Aufträge (optional nach Status filtern)."""
        st = (status or "").strip().lower()
        q = "?status=" + urllib.parse.quote(st) if st else ""
        rows = _api("GET", "/auftraege" + q) or []
        cap = max(1, min(limit, 200))
        if len(rows) > cap:
            rows = rows[:cap]
        return json.dumps(rows, ensure_ascii=False, indent=2)

    @server.tool()
    def bedarf_lesen() -> str:
        """Filament-Bedarf (Queue vs. Spulenrest)."""
        return json.dumps(_api("GET", "/bedarf"), ensure_ascii=False, indent=2)

    @server.tool()
    def wartung_lesen() -> str:
        """Wartungsteile und Zählerstände."""
        return json.dumps(_api("GET", "/wartung"), ensure_ascii=False, indent=2)

    @server.tool()
    def get_version() -> str:
        """Grotti-Version."""
        return json.dumps(_api("GET", "/version"), ensure_ascii=False, indent=2)

    @server.tool()
    def pause_printer(host: str) -> str:
        """Druck pausieren."""
        _write_guard()
        return json.dumps(
            _api("POST", "/drucker/%s/pause" % urllib.parse.quote(host)),
            ensure_ascii=False,
        )

    @server.tool()
    def resume_printer(host: str) -> str:
        """Druck fortsetzen."""
        _write_guard()
        return json.dumps(
            _api("POST", "/drucker/%s/fortsetzen" % urllib.parse.quote(host)),
            ensure_ascii=False,
        )

    @server.tool()
    def cancel_printer(host: str, confirm: bool = False) -> str:
        """Druck abbrechen. confirm muss true sein."""
        _write_guard()
        if not confirm:
            raise RuntimeError("Abbruch braucht confirm=true.")
        return json.dumps(
            _api("POST", "/drucker/%s/abbrechen" % urllib.parse.quote(host)),
            ensure_ascii=False,
        )

    @server.tool()
    def clear_bed(host: str, confirm: bool = False) -> str:
        """Platte als geräumt quittieren (kein physisches Schieben)."""
        _write_guard()
        if not confirm:
            raise RuntimeError("Quittierung braucht confirm=true.")
        return json.dumps(
            _api(
                "POST",
                "/drucker/%s/abraeumen" % urllib.parse.quote(host),
                {"confirm": True},
            ),
            ensure_ascii=False,
        )

    @server.tool()
    def enqueue_when_free(
        file_id: int,
        host: str = "",
        confirm: bool = False,
        bed_leveling: bool = False,
    ) -> str:
        """Datei in die Warteschlange („wenn frei“). confirm muss true sein."""
        _write_guard()
        if not confirm:
            raise RuntimeError("Einreihen braucht confirm=true.")
        payload = {
            "file_id": file_id,
            "host": host,
            "confirm": True,
            "bed_leveling": bool(bed_leveling),
        }
        return json.dumps(_api("POST", "/druck-queue", payload), ensure_ascii=False, indent=2)

    @server.tool()
    def enqueue_print(
        file_id: int,
        host: str,
        confirm: bool = False,
        bed_leveling: bool = False,
    ) -> str:
        """Sofort senden und starten. confirm muss true sein."""
        _write_guard()
        if not confirm:
            raise RuntimeError("Start braucht confirm=true.")
        if not host:
            raise RuntimeError("host fehlt.")
        payload = {
            "file_id": file_id,
            "host": host,
            "confirm": True,
            "bed_leveling": bool(bed_leveling),
        }
        return json.dumps(_api("POST", "/senden", payload), ensure_ascii=False, indent=2)

    @server.tool()
    def steer_temp(host: str, nozzle: float | None = None, bed: float | None = None) -> str:
        """Zieltemperaturen setzen (Whitelist)."""
        _write_guard()
        out = []
        if nozzle is not None:
            out.append(_api("POST", "/drucker/%s/steuern" % urllib.parse.quote(host), {
                "action": "nozzle", "temp": nozzle,
            }))
        if bed is not None:
            out.append(_api("POST", "/drucker/%s/steuern" % urllib.parse.quote(host), {
                "action": "bed", "temp": bed,
            }))
        if not out:
            raise RuntimeError("nozzle und/oder bed angeben.")
        return json.dumps(out, ensure_ascii=False, indent=2)

    @server.tool()
    def steer_speed(host: str, speed: float) -> str:
        """Geschwindigkeitsfaktor in Prozent."""
        _write_guard()
        return json.dumps(
            _api("POST", "/drucker/%s/steuern" % urllib.parse.quote(host), {
                "action": "speed", "percent": speed,
            }),
            ensure_ascii=False,
        )

    @server.tool()
    def steer_fan(host: str, fan: float) -> str:
        """Teilekühlung 0–100."""
        _write_guard()
        return json.dumps(
            _api("POST", "/drucker/%s/steuern" % urllib.parse.quote(host), {
                "action": "fan", "percent": fan,
            }),
            ensure_ascii=False,
        )

    @server.resource("grotti://version")
    def resource_version() -> str:
        return json.dumps(_api("GET", "/version"), ensure_ascii=False, indent=2)

    server.run()


if __name__ == "__main__":
    main()