Files
mission-control-v2/mcp/mcp_orchestrator.py
T
HitonabiandClaude Opus 5.5 deaf35e953 lucy: Werkzeuge für den Orchestrator (MCP homelab-orchestrator)
User 25.09.: „Lucy ist hier der komplette Systemadmin + Helfer“ – „ja, bau das
mit Lucy und dem Orchestrator“. Lucy bekommt die Knöpfe der Oberfläche als
MCP-Werkzeuge: lesen (lage, updates, speicher, protokoll, radar) und handeln
(update_starten, alle_aktualisieren, nach_updates_suchen, dienst_neustarten,
kernel_aufraeumen, host_neustarten, radar_testen, radar_suchen) – über dieselben
MC2-Routen mit Snapshot, Prüfung, Rückweg und Sperren. Jede Aktion braucht einen
Grund (die Bitte des Commanders) und wird protokolliert; die Beschreibungen
verbieten Aktionen auf eigene Initiative (Reparatur-Neustart ausgenommen).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-09-25 13:27:35 +02:00

91 lines
2.7 KiB
Python

#!/usr/bin/env python3
"""
Homelab Orchestrator — MCP-Server für Lucy (Hermes), seit 25.09.2026.
Stdio-MCP-Server, von Hermes Gateway als Unterprozess gestartet (auf der KI-Box, mit dem venv von MC2). Gibt Lucy die
Knöpfe des Orchestrators: Lage lesen, Updates starten, „Alle aktualisieren“, Dienste neu starten, Kernel aufräumen,
Host neu starten, Modell-Radar. Die Logik steht in orchestrator_werkzeuge.py (ohne MCP testbar); alles geht über die
REST-API von MC2 mit denselben Leitplanken wie die Knöpfe der Oberfläche.
Die Beschreibungen sind bewusst als Leitplanken formuliert (Lesen jederzeit, Handeln nur auf Bitte des Commanders),
damit das Modell nicht auf eigene Faust Updates oder Neustarts auslöst — der Grund jeder Aktion wird protokolliert.
FastMCP liest die Beschreibung beim Anmelden, deshalb steht sie in description= (orchestrator_werkzeuge.BESCHREIBUNG).
Einrichten auf der Box (einmal): hermes mcp add homelab-orchestrator --command <venv>/bin/python --args <dieses Skript>
"""
import orchestrator_werkzeuge as w
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("homelab-orchestrator")
B = w.BESCHREIBUNG
@mcp.tool(description=B["lage"])
def lage() -> str:
return w.lage()
@mcp.tool(description=B["updates"])
def updates() -> str:
return w.updates()
@mcp.tool(description=B["speicher"])
def speicher() -> str:
return w.speicher()
@mcp.tool(description=B["protokoll"])
def protokoll(anzahl: int = 15) -> str:
return w.protokoll(anzahl)
@mcp.tool(description=B["radar"])
def radar() -> str:
return w.radar()
@mcp.tool(description=B["update_starten"])
def update_starten(ziel: str, baustein: str, grund: str) -> str:
return w.update_starten(ziel, baustein, grund)
@mcp.tool(description=B["alle_aktualisieren"])
def alle_aktualisieren(bereich: str, grund: str) -> str:
return w.alle_aktualisieren(bereich, grund)
@mcp.tool(description=B["nach_updates_suchen"])
def nach_updates_suchen(grund: str) -> str:
return w.nach_updates_suchen(grund)
@mcp.tool(description=B["dienst_neustarten"])
def dienst_neustarten(dienst: str, grund: str) -> str:
return w.dienst_neustarten(dienst, grund)
@mcp.tool(description=B["kernel_aufraeumen"])
def kernel_aufraeumen(grund: str) -> str:
return w.kernel_aufraeumen(grund)
@mcp.tool(description=B["host_neustarten"])
def host_neustarten(grund: str) -> str:
return w.host_neustarten(grund)
@mcp.tool(description=B["radar_testen"])
def radar_testen(kandidat: str, grund: str) -> str:
return w.radar_testen(kandidat, grund)
@mcp.tool(description=B["radar_suchen"])
def radar_suchen(grund: str) -> str:
return w.radar_suchen(grund)
if __name__ == "__main__":
mcp.run()