Grotti 2.2.0: Auth, API-Tokens, MCP und Doku-Trennung.
This commit is contained in:
Dominic Gundel committed
1 parent
b20f8a7779
commit
5a3716b26a
65 files changed
+2212
-198
No files matched your search
@@ -0,0 +1,263 @@
|
||||
"""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 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", "value": nozzle,
|
||||
}))
|
||||
if bed is not None:
|
||||
out.append(_api("POST", "/drucker/%s/steuern" % urllib.parse.quote(host), {
|
||||
"action": "bed", "value": 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", "value": 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", "value": 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()
|
||||
Reference in new issue
Block a user