meleki/gott
1
0 Fork 0
Code Activity
Files
3e0a69e7a403fb2536af06068fd9cd53df9033a1
gott/scripts/mcp-remote-execute.py
T
MelekiandClaude Opus 4.6 3e0a69e7a4 feat: MCP-Tools ask_user_choice + memory_read/write/list, Text-Marker entfernt 
MCP-Server erweitert um ask_user_choice (Long-Poll), memory_read, memory_write
und memory_list Tools. MCP-Config wird jetzt immer generiert (nicht nur bei
Remote-Verbindung). InternalEndpoint um PendingChoices + choice-respond Endpoint
ergänzt. Frontend differenziert CLI-Modus (choice-respond API) vs. API-Modus.
Gesamter :::ASK_USER_CHOICE::: Text-Marker-Mechanismus aus ClaudeCliService,
AnthropicService, markdown.js und stream-ui.js entfernt.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-03-18 09:54:30 +01:00

330 lines
11 KiB
Python
Raw Blame History

#!/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 (long-poll)."""
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)
# Long-Poll: Blockiert bis der User antwortet (max 5 Min)
# Der Backend-Endpoint /api/internal/ask-user-choice blockiert selbst
result = api_call_long("ask-user-choice", {"sessionId": SESSION_ID})
choice = result.get("choice", "")
return make_result(req_id, f"User hat gewählt: {choice}")
def api_call_long(endpoint, payload):
"""API-Call mit langem Timeout (für Long-Poll Endpoints)."""
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=330) as resp:
return json.loads(resp.read().decode("utf-8"))
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()
View Git Blame Copy Permalink