#!/usr/bin/env python3 """ MCP-Server (Model Context Protocol) für remote_execute. Kommuniziert über stdin/stdout (JSON-RPC 2.0) mit der Claude CLI. Ruft die gott-interne HTTP-API auf um Befehle auf dem Remote-PC auszuführen. Umgebungsvariablen: GOTT_API_URL - URL der gott-API (z.B. http://localhost:19000) GOTT_SESSION_ID - Session-ID für die Remote-Verbindungszuordnung GOTT_AUTH_TOKEN - Auth-Token für die gott-API (optional, für nicht-localhost) """ import json import sys import urllib.request import urllib.error import os API_URL = os.environ.get("GOTT_API_URL", "http://localhost:19000") SESSION_ID = os.environ.get("GOTT_SESSION_ID", "") AUTH_TOKEN = os.environ.get("GOTT_AUTH_TOKEN", "") AGENT_ID = os.environ.get("GOTT_AGENT_ID", "") HAS_REMOTE = os.environ.get("GOTT_HAS_REMOTE", "false") == "true" TOOLS = [] # Remote-Execute Tool (nur wenn Remote-Verbindung vorhanden) if HAS_REMOTE: TOOLS.append({ "name": "remote_execute", "description": ( "Führt einen PowerShell-Befehl auf der verbundenen Remote-Windows-Workstation aus. " "Gibt stdout und stderr zurück. Nutze dies für Diagnose und Administration des Remote-PCs " "(z.B. Druckerprobleme, Netzwerk, Services, Registry). Timeout: 180 Sekunden." ), "inputSchema": { "type": "object", "properties": { "command": {"type": "string", "description": "PowerShell-Befehl zum Ausführen auf dem Remote-PC"} }, "required": ["command"] } }) # ask_user_choice Tool (immer verfügbar) TOOLS.append({ "name": "ask_user_choice", "description": ( "Stellt dem User eine Frage mit Auswahlmöglichkeiten (2-5 Optionen). " "Der User wählt eine Option und du arbeitest mit seiner Antwort weiter. " "Nutze dies wenn du unsicher bist oder es mehrere sinnvolle Wege gibt." ), "inputSchema": { "type": "object", "properties": { "question": {"type": "string", "description": "Die Frage an den User"}, "choices": { "type": "array", "items": {"type": "string"}, "description": "Liste der Auswahlmöglichkeiten (2-5 Optionen)", "minItems": 2, "maxItems": 5 } }, "required": ["question", "choices"] } }) # Memory-Tools (immer verfügbar wenn agentId gesetzt) if AGENT_ID: TOOLS.append({ "name": "memory_read", "description": "Liest eine Memory-Datei des Agents. Nutze dies um gespeicherte Erinnerungen, Notizen oder Kontextinformationen abzurufen.", "inputSchema": { "type": "object", "properties": { "path": {"type": "string", "description": "Relativer Pfad zur Memory-Datei (z.B. 'MEMORY.md' oder 'memory/YYYY-MM-DD.md')"} }, "required": ["path"] } }) TOOLS.append({ "name": "memory_write", "description": "Schreibt oder überschreibt eine Memory-Datei. Nutze dies um wichtige Informationen, Entscheidungen oder Präferenzen dauerhaft zu speichern.", "inputSchema": { "type": "object", "properties": { "path": {"type": "string", "description": "Relativer Pfad zur Memory-Datei (z.B. 'MEMORY.md' oder 'memory/YYYY-MM-DD.md')"}, "content": {"type": "string", "description": "Inhalt der Datei"}, "append": {"type": "boolean", "description": "Wenn true, wird der Inhalt angehängt statt überschrieben (Default: false)"} }, "required": ["path", "content"] } }) TOOLS.append({ "name": "memory_list", "description": "Listet alle Memory-Dateien des Agents auf.", "inputSchema": { "type": "object", "properties": { "path": {"type": "string", "description": "Optionaler Unterpfad zum Auflisten (leer = Root)"} }, "required": [] } }) def send_response(response): """Sendet eine JSON-RPC Response über stdout.""" data = json.dumps(response) sys.stdout.write(data + "\n") sys.stdout.flush() def handle_initialize(req_id, params): """MCP initialize handshake.""" return { "jsonrpc": "2.0", "id": req_id, "result": { "protocolVersion": "2024-11-05", "capabilities": { "tools": {} }, "serverInfo": { "name": "gott-remote", "version": "1.0.0" } } } def handle_tools_list(req_id): """Gibt die verfügbaren Tools zurück.""" return { "jsonrpc": "2.0", "id": req_id, "result": { "tools": TOOLS } } def api_call(endpoint, payload): """Macht einen API-Call an gott und gibt das Ergebnis zurück.""" url = f"{API_URL}/api/internal/{endpoint}" data = json.dumps(payload).encode("utf-8") req = urllib.request.Request(url, data=data, method="POST") req.add_header("Content-Type", "application/json") if AUTH_TOKEN: req.add_header("X-Internal-Token", AUTH_TOKEN) with urllib.request.urlopen(req, timeout=200) as resp: return json.loads(resp.read().decode("utf-8")) def make_result(req_id, text, is_error=False): """Erstellt eine MCP Tool-Result Response.""" return { "jsonrpc": "2.0", "id": req_id, "result": { "content": [{"type": "text", "text": text}], "isError": is_error } } def handle_remote_execute(req_id, arguments): """Führt remote_execute aus.""" command = arguments.get("command", "") if not command: return make_result(req_id, "Fehler: Kein Befehl angegeben", True) result = api_call("remote-execute", {"command": command, "sessionId": SESSION_ID}) output_parts = [] if result.get("stdout"): output_parts.append(result["stdout"]) if result.get("stderr"): output_parts.append(f"[stderr] {result['stderr']}") if result.get("exitCode", 0) != 0: output_parts.append(f"[Exit Code: {result['exitCode']}]") text = "\n".join(output_parts) if output_parts else "(keine Ausgabe)" return make_result(req_id, text, result.get("exitCode", 0) != 0) def handle_memory_read(req_id, arguments): """Liest eine Memory-Datei.""" path = arguments.get("path", "") if not path: return make_result(req_id, "Fehler: path erforderlich", True) result = api_call("memory-read", {"agentId": AGENT_ID, "path": path}) return make_result(req_id, result.get("content", "")) def handle_memory_write(req_id, arguments): """Schreibt eine Memory-Datei.""" path = arguments.get("path", "") content = arguments.get("content", "") append = arguments.get("append", False) if not path: return make_result(req_id, "Fehler: path erforderlich", True) result = api_call("memory-write", {"agentId": AGENT_ID, "path": path, "content": content, "append": append}) return make_result(req_id, result.get("message", "OK")) def handle_memory_list(req_id, arguments): """Listet Memory-Dateien auf.""" path = arguments.get("path", "") result = api_call("memory-list", {"agentId": AGENT_ID, "path": path}) files = result.get("files", []) return make_result(req_id, "\n".join(files) if files else "(keine Memory-Dateien vorhanden)") def handle_ask_user_choice(req_id, arguments): """Stellt dem User eine Frage und wartet auf Antwort (short-poll).""" import time question = arguments.get("question", "") choices = arguments.get("choices", []) if not question: return make_result(req_id, "Fehler: question erforderlich", True) if not choices or len(choices) < 2: return make_result(req_id, "Fehler: Mindestens 2 choices erforderlich", True) # Short-Poll: Jede Sekunde beim Backend nachfragen ob der User geantwortet hat for i in range(300): # max 5 Minuten try: result = api_call("ask-user-choice", {"sessionId": SESSION_ID}) # 200 = Antwort vorhanden if "choice" in result: return make_result(req_id, f"User hat gewählt: {result['choice']}") except urllib.error.HTTPError as e: if e.code == 202: # 202 = noch ausstehend, weiter pollen pass else: body = "" try: body = e.read().decode("utf-8") except Exception: pass return make_result(req_id, f"API-Fehler {e.code}: {body}", True) except Exception as e: return make_result(req_id, f"Fehler: {str(e)}", True) time.sleep(1) return make_result(req_id, "Timeout – keine Auswahl getroffen (5 Minuten)", True) TOOL_HANDLERS = { "remote_execute": handle_remote_execute, "ask_user_choice": handle_ask_user_choice, "memory_read": handle_memory_read, "memory_write": handle_memory_write, "memory_list": handle_memory_list, } def handle_tools_call(req_id, params): """Führt ein Tool aus.""" tool_name = params.get("name", "") arguments = params.get("arguments", {}) handler = TOOL_HANDLERS.get(tool_name) if not handler: return make_result(req_id, f"Unbekanntes Tool: {tool_name}", True) # Prüfen ob das Tool auch in TOOLS registriert ist if not any(t["name"] == tool_name for t in TOOLS): return make_result(req_id, f"Tool nicht verfügbar: {tool_name}", True) try: return handler(req_id, arguments) except urllib.error.HTTPError as e: body = "" try: body = e.read().decode("utf-8") except Exception: pass return make_result(req_id, f"API-Fehler {e.code}: {body}", True) except Exception as e: return make_result(req_id, f"Fehler: {str(e)}", True) def handle_notifications_initialized(params): """Handle notifications/initialized - no response needed.""" pass def main(): """Hauptloop: Liest JSON-RPC Requests von stdin, verarbeitet sie.""" for line in sys.stdin: line = line.strip() if not line: continue try: request = json.loads(line) except json.JSONDecodeError: continue method = request.get("method", "") req_id = request.get("id") params = request.get("params", {}) # Notifications (kein id) brauchen keine Response if req_id is None: if method == "notifications/initialized": handle_notifications_initialized(params) continue if method == "initialize": response = handle_initialize(req_id, params) elif method == "tools/list": response = handle_tools_list(req_id) elif method == "tools/call": response = handle_tools_call(req_id, params) else: response = { "jsonrpc": "2.0", "id": req_id, "error": { "code": -32601, "message": f"Method not found: {method}" } } send_response(response) if __name__ == "__main__": main()