mission-control-v2/mcp/mcp_pc.py

#!/usr/bin/env python3
"""
Mission Control 2.0 — PC Control MCP Server.

Stdio-MCP-Server. Läuft als Subprocess von Hermes Gateway (AI Box).
Proxied Befehle an den Hermes PC Executor (Windows PC, Port 7777).

Env: PC_EXECUTOR_URL (default http://192.168.178.XXX:7777)
"""

import os

import httpx
from mcp.server.fastmcp import FastMCP

PC_URL = os.environ.get("PC_EXECUTOR_URL", "").rstrip("/")
# Shared Secret — muss identisch zum HERMES_PC_TOKEN des Executors auf dem Windows-PC sein.
PC_TOKEN = os.environ.get("PC_EXECUTOR_TOKEN", "").strip()

mcp = FastMCP("hermes-pc-control")


def _pc(path: str, data: dict | None = None) -> dict:
    if not PC_URL:
        return {"error": "PC_EXECUTOR_URL nicht gesetzt. Setze die Env-Variable mit der IP des Windows PCs."}
    headers = {"Authorization": f"Bearer {PC_TOKEN}"} if PC_TOKEN else {}
    try:
        with httpx.Client(timeout=90) as c:
            if data is None:
                r = c.get(f"{PC_URL}{path}", headers=headers)
            else:
                r = c.post(f"{PC_URL}{path}", json=data, headers=headers)
            r.raise_for_status()
            return r.json()
    except httpx.ConnectError:
        return {"error": f"PC Executor nicht erreichbar ({PC_URL}). Läuft executor.py auf dem Windows PC?"}
    except Exception as e:
        return {"error": str(e)}


@mcp.tool()
def pc_shell(command: str) -> str:
    """Führt einen PowerShell-Befehl auf dem Windows-PC des Nutzers aus.
    Nutze dies um Dateien zu lesen/schreiben, Programme zu starten oder Systeminfos abzufragen.
    Beispiele: 'Get-Date', 'ls C:\\Users', 'notepad.exe test.txt'
    Gibt stdout, stderr und returncode zurück."""
    result = _pc("/shell", {"command": command})
    if "error" in result:
        return f"FEHLER: {result['error']}"
    parts = []
    if result.get("stdout"):
        parts.append(f"stdout:\n{result['stdout']}")
    if result.get("stderr"):
        parts.append(f"stderr:\n{result['stderr']}")
    parts.append(f"returncode: {result.get('returncode', '?')}")
    return "\n".join(parts)


@mcp.tool()
def pc_screenshot() -> str:
    """Macht einen Screenshot des Windows-PC-Bildschirms.
    Nutze dies um zu sehen was der Nutzer gerade auf dem Bildschirm hat.
    Gibt einen base64-kodierten JPEG-String zurück."""
    result = _pc("/screenshot", {})
    if "error" in result:
        return f"FEHLER: {result['error']}"
    b64 = result.get("screenshot", "")
    if b64:
        return f"[Screenshot aufgenommen, {len(b64)} Zeichen base64-JPEG]"
    return "Kein Screenshot-Daten erhalten."


@mcp.tool()
def pc_type(text: str) -> str:
    """Tippt Text in das aktive Fenster auf dem Windows-PC.
    Der Nutzer muss das Zielfenster vorher aktiviert haben.
    Nützlich um Text in Editoren, Terminals oder Formulare einzugeben."""
    result = _pc("/type", {"text": text})
    if "error" in result:
        return f"FEHLER: {result['error']}"
    return "Text getippt." if result.get("ok") else str(result)


@mcp.tool()
def pc_key(keys: str) -> str:
    """Drückt eine Tastenkombination auf dem Windows-PC.
    Format: Tasten mit '+' verbinden. Beispiele: 'ctrl+s', 'alt+F4', 'ctrl+shift+esc', 'win+d'
    Nützlich um Tastenkürzel auszuführen."""
    result = _pc("/key", {"keys": keys})
    if "error" in result:
        return f"FEHLER: {result['error']}"
    return f"Tastenkombination '{keys}' gedrückt." if result.get("ok") else str(result)


@mcp.tool()
def pc_media(action: str) -> str:
    """Steuert die Medien-Wiedergabe auf dem Windows-PC — wirkt auf den gerade aktiven Player
    (Spotify, YouTube, VLC …), wie die Tasten einer Multimedia-Tastatur.
    action: play_pause (Wiedergabe/Pause umschalten) | next (nächster Titel) |
    prev (voriger Titel) | stop."""
    result = _pc("/media", {"action": action})
    if "error" in result:
        return f"FEHLER: {result['error']}"
    return f"Medien-Aktion '{action}' ausgeführt." if result.get("ok") else str(result)


@mcp.tool()
def pc_volume(action: str, steps: int = 5) -> str:
    """Ändert die System-Lautstärke des Windows-PCs.
    action: up (lauter) | down (leiser) | mute (Stumm umschalten).
    steps = Stärke 1–25 (je Schritt ≈ 2 %); 5 ≈ 10 % — für „etwas lauter/leiser" passt der Default,
    für „viel" nimm 10–15."""
    result = _pc("/volume", {"action": action, "steps": steps})
    if "error" in result:
        return f"FEHLER: {result['error']}"
    return f"Lautstärke: {action} (x{result.get('steps', steps)})." if result.get("ok") else str(result)


@mcp.tool()
def pc_open(target: str) -> str:
    """Öffnet eine URL im Browser oder eine Datei/Programm auf dem Windows-PC.
    Für URLs: 'https://...' → öffnet im Standard-Browser.
    Für Dateien/Programme: 'C:\\Pfad\\zur\\Datei.txt' oder 'notepad.exe'"""
    result = _pc("/open", {"target": target})
    if "error" in result:
        return f"FEHLER: {result['error']}"
    return f"'{target}' geöffnet." if result.get("ok") else str(result)


@mcp.tool()
def pc_status() -> str:
    """Prüft ob der Windows-PC Executor erreichbar ist und gibt Hostname zurück.
    Rufe dies auf um zu prüfen ob PC-Steuerung möglich ist, bevor du andere pc_* Tools nutzt."""
    result = _pc("/health")
    if "error" in result:
        return f"PC Executor NICHT erreichbar: {result['error']}"
    return f"PC Executor erreichbar. Hostname: {result.get('host', '?')}, Status: {result.get('status', '?')}"


if __name__ == "__main__":
    mcp.run()