MelekiandClaude Opus 4.6
53cee67bef
Remote-Verbindungen erkennen jetzt das OS des Clients und passen die Tool-Description dynamisch an (PowerShell für Windows, Bash für Linux). Neues remote.sh Script für Linux/macOS-Clients mit embedded Python WebSocket-Client. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
349 lines
12 KiB
Python
349 lines
12 KiB
Python
#!/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"
|
||
REMOTE_OS = os.environ.get("GOTT_REMOTE_OS", "").lower() # z.B. "windows", "linux"
|
||
|
||
TOOLS = []
|
||
|
||
# Remote-Execute Tool (nur wenn Remote-Verbindung vorhanden)
|
||
if HAS_REMOTE:
|
||
if "linux" in REMOTE_OS:
|
||
_desc = (
|
||
"Führt einen Bash-Befehl auf der verbundenen Remote-Linux-Workstation aus. "
|
||
"Gibt stdout und stderr zurück. Nutze dies für Diagnose und Administration des Remote-PCs "
|
||
"(z.B. Netzwerk, Services, Logs, Pakete). Timeout: 180 Sekunden."
|
||
)
|
||
_cmd_desc = "Bash-Befehl zum Ausführen auf dem Remote-PC"
|
||
else:
|
||
_desc = (
|
||
"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."
|
||
)
|
||
_cmd_desc = "PowerShell-Befehl zum Ausführen auf dem Remote-PC"
|
||
|
||
TOOLS.append({
|
||
"name": "remote_execute",
|
||
"description": _desc,
|
||
"inputSchema": {
|
||
"type": "object",
|
||
"properties": {
|
||
"command": {"type": "string", "description": _cmd_desc}
|
||
},
|
||
"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()
|