gott/scripts/mcp-remote-execute.py

#!/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", "")

TOOL_NAME = "remote_execute"
TOOL_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."
)


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": [
                {
                    "name": TOOL_NAME,
                    "description": TOOL_DESCRIPTION,
                    "inputSchema": {
                        "type": "object",
                        "properties": {
                            "command": {
                                "type": "string",
                                "description": "PowerShell-Befehl zum Ausführen auf dem Remote-PC"
                            }
                        },
                        "required": ["command"]
                    }
                }
            ]
        }
    }


def handle_tools_call(req_id, params):
    """Führt ein Tool aus."""
    tool_name = params.get("name", "")
    arguments = params.get("arguments", {})

    if tool_name != TOOL_NAME:
        return {
            "jsonrpc": "2.0",
            "id": req_id,
            "result": {
                "content": [{"type": "text", "text": f"Unbekanntes Tool: {tool_name}"}],
                "isError": True
            }
        }

    command = arguments.get("command", "")
    if not command:
        return {
            "jsonrpc": "2.0",
            "id": req_id,
            "result": {
                "content": [{"type": "text", "text": "Fehler: Kein Befehl angegeben"}],
                "isError": True
            }
        }

    # API-Call an gott
    try:
        url = f"{API_URL}/api/internal/remote-execute"
        payload = json.dumps({
            "command": command,
            "sessionId": SESSION_ID
        }).encode("utf-8")

        req = urllib.request.Request(url, data=payload, 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:
            result = json.loads(resp.read().decode("utf-8"))

        # Ergebnis formatieren
        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)"
        is_error = result.get("exitCode", 0) != 0

        return {
            "jsonrpc": "2.0",
            "id": req_id,
            "result": {
                "content": [{"type": "text", "text": text}],
                "isError": is_error
            }
        }

    except urllib.error.HTTPError as e:
        body = ""
        try:
            body = e.read().decode("utf-8")
        except Exception:
            pass
        return {
            "jsonrpc": "2.0",
            "id": req_id,
            "result": {
                "content": [{"type": "text", "text": f"API-Fehler {e.code}: {body}"}],
                "isError": True
            }
        }
    except Exception as e:
        return {
            "jsonrpc": "2.0",
            "id": req_id,
            "result": {
                "content": [{"type": "text", "text": f"Fehler: {str(e)}"}],
                "isError": 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()