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).")


# Wird beim Connect an den Client gegeben (Claude, Cursor, …).
INSTRUCTIONS = """\
Grotti steuert eine 3D-Drucker-Farm (Anycubic Kobra, Klipper/Moonraker).
Antworte dem Nutzer auf Deutsch. Handle nur über diese Tools.

Identität
- Drucker über das Feld host (IP). Den Namen nur zur Anzeige nutzen. Hosts nie raten: erst list_printers oder get_printer.
- Dateien über die numerische id aus list_files (Feld id), nicht über den Dateinamen allein.
- Vor jedem Schreibzugriff den aktuellen Stand lesen.

Lesen
- Farm: list_printers (live=true). Ein Drucker: get_printer(host).
- Läuft: list_running. Warteschlange: list_queue. Ereignisse: list_events.
- Bibliothek: list_files. Spulen: list_spools. Filament-Fehlmenge: bedarf_lesen.
- Kundenaufträge: auftraege_lesen. Verschleißteile: wartung_lesen.

Zustand (Feld state)
- printing: druckt. paused: Pause. standby: frei. complete: fertig, Platte oft noch belegt.
- error: Fehler, nicht einfach neu starten. offline: nicht erreichbar.
- progress ist 0…1. filament_empty oder pause_reason bei Pause heißt Filament leer.
  Dann dem Nutzer sagen, die neue Spule in Grotti dem Drucker zuzuweisen, und erst danach resume_printer.
  Eine Filament-Pause nicht blind fortsetzen.

Schreiben
confirm=true nur setzen, wenn der Nutzer diese Aktion in diesem Zug klar will. Nicht vorauseilend bestätigen.
- enqueue_when_free: Datei in die Warteschlange. host leer = Grotti wählt den nächsten Drucker mit passender Sorte. host gesetzt = dieser Drucker, sobald er frei ist und die Platte klar ist. Das ist der normale Weg.
- enqueue_print: sofort senden und starten. Nur wenn der Nutzer einen Sofortstart auf einem genannten freien Drucker will (state standby, Platte frei). Große Dateien brauchen oft etwa eine Minute bis zum Start.
- Den Material-Match macht Grotti (PLA Meta ist nicht Matte und nicht Silk). Eine falsche Maschine nicht erzwingen. Eine Ablehnung wörtlich weitergeben.
- bed_leveling=true nur auf ausdrücklichen Wunsch (Bettnetz / Auto-Level).
- pause_printer und resume_printer während des Drucks, ohne confirm.
- cancel_printer bricht den Druck ab und braucht confirm=true. Bei Unklarheit erst nachfragen.
- clear_bed ist nur die Quittung „Platte ist leer“ und schiebt nichts physisch. Erst aufrufen, wenn der Nutzer sagt, das Teil ist herunter. Ohne diese Quittung startet der nächste „wenn frei“-Job nicht.
- steer_temp (Düse und Bett in °C), steer_speed (Prozent), steer_fan (0–100): nur diese Werte. Kein freier G-code.

Grenzen
- Keine G-code-Konsole, kein Firmware-Restart, kein E-Stop, keine Benutzer- oder Token-Verwaltung. Diese Lücken nicht umgehen.
- Im Read-only-Modus (GROTTI_MCP_READ_ONLY) schlagen Schreib-Tools fehl. Das sagen und nicht wiederholen.
- Gesperrte Drucker und ausgeschaltete Automatik starten keine Warteschlangen-Jobs. Den Fehler nennen.

Antworten
- Kurz und konkret: Druckername, host, Datei, Zustand, was du getan hast.
- Den Fehlertext von Grotti zitieren. Keinen Workaround außerhalb der Tools erfinden.
"""


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=INSTRUCTIONS)

    @server.tool()
    def list_printers(live: bool = True) -> str:
        """Alle Drucker. live=true liefert Zustand, Temps, Fortschritt, Dateiname. Host ist die IP."""
        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. Host vorher über list_printers holen."""
        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. Die numerische id ist die file_id fürs Einreihen und Starten."""
        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:
        """Quittung, dass der Nutzer die Platte geleert hat. Schiebt nichts. confirm muss true sein. Danach kann die Warteschlange den nächsten Job starten."""
        _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. host leer: Grotti wählt den nächsten Drucker mit passender Sorte. host gesetzt: dieser Drucker, sobald frei und Platte klar. confirm muss true sein. Normaler Weg zum Drucken."""
        _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 an diesen host senden und starten. Nur bei freiem Drucker (standby, Platte frei) und klarem Sofortstart-Wunsch. 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()