Feat: Hermes PC-Zugriff via MCP (executor + mcp_pc)
- mcp/mcp_pc.py: MCP-Server der PC-Tools an den Executor proxied (pc_shell, pc_screenshot, pc_type, pc_key, pc_open, pc_status) - client/hermes-pc/executor.py: FastAPI auf Port 7777 (Daemon-Reg entfernt) - client/hermes-pc/start.bat, install.bat, requirements.txt - client/hermes-voice/agent_client.py, main.py: agent_client integration Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Hitonabi
@@ -0,0 +1,132 @@
|
||||
"""
|
||||
Hermes PC Executor — läuft auf dem Windows PC.
|
||||
Empfängt Tool-Befehle vom MCP-Server der AI Box und führt sie lokal aus.
|
||||
"""
|
||||
import base64
|
||||
import io
|
||||
import os
|
||||
import socket
|
||||
import subprocess
|
||||
import webbrowser
|
||||
from urllib.parse import quote
|
||||
|
||||
import uvicorn
|
||||
from fastapi import FastAPI, HTTPException
|
||||
from fastapi.middleware.cors import CORSMiddleware
|
||||
|
||||
app = FastAPI(title="Hermes PC Executor")
|
||||
app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"])
|
||||
|
||||
MY_PORT = int(os.environ.get("HERMES_PC_PORT", "7777"))
|
||||
|
||||
|
||||
# ── Shell ──────────────────────────────────────────────────────────────────
|
||||
|
||||
@app.post("/shell")
|
||||
async def run_shell(req: dict):
|
||||
cmd = req.get("command", "")
|
||||
try:
|
||||
result = subprocess.run(
|
||||
["powershell", "-NoProfile", "-NonInteractive", "-Command", cmd],
|
||||
capture_output=True, text=True, timeout=60,
|
||||
encoding="utf-8", errors="replace",
|
||||
)
|
||||
return {
|
||||
"stdout": result.stdout.strip(),
|
||||
"stderr": result.stderr.strip(),
|
||||
"returncode": result.returncode,
|
||||
}
|
||||
except subprocess.TimeoutExpired:
|
||||
raise HTTPException(408, "Timeout nach 60s")
|
||||
except Exception as e:
|
||||
raise HTTPException(500, str(e))
|
||||
|
||||
|
||||
# ── Screen ─────────────────────────────────────────────────────────────────
|
||||
|
||||
@app.post("/screenshot")
|
||||
async def take_screenshot():
|
||||
try:
|
||||
import mss
|
||||
from PIL import Image
|
||||
|
||||
with mss.mss() as sct:
|
||||
monitor = sct.monitors[1]
|
||||
raw = sct.grab(monitor)
|
||||
img = Image.frombytes("RGB", raw.size, raw.bgra, "raw", "BGRX")
|
||||
img.thumbnail((1280, 720))
|
||||
buf = io.BytesIO()
|
||||
img.save(buf, format="JPEG", quality=75)
|
||||
return {"screenshot": base64.b64encode(buf.getvalue()).decode()}
|
||||
except Exception as e:
|
||||
raise HTTPException(500, str(e))
|
||||
|
||||
|
||||
# ── Input ──────────────────────────────────────────────────────────────────
|
||||
|
||||
@app.post("/type")
|
||||
async def type_text(req: dict):
|
||||
text = req.get("text", "")
|
||||
try:
|
||||
import pyautogui
|
||||
pyautogui.write(text, interval=0.03)
|
||||
return {"ok": True}
|
||||
except Exception as e:
|
||||
raise HTTPException(500, str(e))
|
||||
|
||||
|
||||
@app.post("/key")
|
||||
async def press_keys(req: dict):
|
||||
keys = req.get("keys", "")
|
||||
try:
|
||||
import pyautogui
|
||||
parts = [k.strip() for k in keys.split("+")]
|
||||
pyautogui.hotkey(*parts)
|
||||
return {"ok": True}
|
||||
except Exception as e:
|
||||
raise HTTPException(500, str(e))
|
||||
|
||||
|
||||
# ── Apps / Browser ─────────────────────────────────────────────────────────
|
||||
|
||||
@app.post("/open")
|
||||
async def open_target(req: dict):
|
||||
target = req.get("target", "")
|
||||
try:
|
||||
if target.startswith("http://") or target.startswith("https://"):
|
||||
webbrowser.open(target)
|
||||
else:
|
||||
os.startfile(target)
|
||||
return {"ok": True}
|
||||
except Exception as e:
|
||||
raise HTTPException(500, str(e))
|
||||
|
||||
|
||||
@app.post("/search")
|
||||
async def search_web(req: dict):
|
||||
query = req.get("query", "")
|
||||
url = f"https://www.google.com/search?q={quote(query)}"
|
||||
webbrowser.open(url)
|
||||
return {"ok": True, "url": url}
|
||||
|
||||
|
||||
# ── Health ─────────────────────────────────────────────────────────────────
|
||||
|
||||
@app.get("/health")
|
||||
async def health():
|
||||
return {"status": "ok", "host": socket.gethostname()}
|
||||
|
||||
|
||||
def _get_local_ip() -> str:
|
||||
try:
|
||||
with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as s:
|
||||
s.connect(("8.8.8.8", 80))
|
||||
return s.getsockname()[0]
|
||||
except Exception:
|
||||
return socket.gethostbyname(socket.gethostname())
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
my_ip = _get_local_ip()
|
||||
print(f"Hermes PC Executor läuft auf http://{my_ip}:{MY_PORT}")
|
||||
uvicorn.run(app, host="0.0.0.0", port=MY_PORT)
|
||||
@@ -0,0 +1,8 @@
|
||||
@echo off
|
||||
cd /d "%~dp0"
|
||||
echo === Hermes PC Executor Setup ===
|
||||
python -m venv .venv
|
||||
.venv\Scripts\pip install -r requirements.txt -q
|
||||
echo.
|
||||
echo Fertig. Starte mit start.bat
|
||||
pause
|
||||
@@ -0,0 +1,6 @@
|
||||
fastapi>=0.111.0
|
||||
uvicorn>=0.30.0
|
||||
httpx>=0.27.0
|
||||
mss>=9.0.1
|
||||
Pillow>=10.0.0
|
||||
pyautogui>=0.9.54
|
||||
@@ -0,0 +1,9 @@
|
||||
@echo off
|
||||
cd /d "%~dp0"
|
||||
echo === Hermes PC Executor ===
|
||||
echo Port: 7777
|
||||
echo Hermes (AI Box) kann jetzt auf diesen PC zugreifen.
|
||||
echo Fenster offen lassen solange Hermes PC-Zugriff braucht.
|
||||
echo.
|
||||
.venv\Scripts\python executor.py
|
||||
pause
|
||||
@@ -0,0 +1,31 @@
|
||||
"""Spricht mit dem Hermes Agent Daemon auf der AI Box."""
|
||||
import httpx
|
||||
|
||||
TIMEOUT = 120
|
||||
|
||||
|
||||
def ask_agent(text: str, screenshot_b64: str | None, settings: dict) -> tuple[str, list]:
|
||||
"""Schickt Nachricht an den Daemon, gibt (response, tool_calls) zurück."""
|
||||
daemon_url = settings.get("agent_daemon_url", "http://192.168.178.151:8765")
|
||||
try:
|
||||
with httpx.Client(timeout=TIMEOUT) as c:
|
||||
resp = c.post(f"{daemon_url}/chat", json={
|
||||
"message": text,
|
||||
"screenshot": screenshot_b64,
|
||||
})
|
||||
resp.raise_for_status()
|
||||
data = resp.json()
|
||||
return data.get("response", ""), data.get("tool_calls", [])
|
||||
except httpx.TimeoutException:
|
||||
return "Antwort hat zu lange gedauert.", []
|
||||
except Exception as e:
|
||||
return f"Agent nicht erreichbar: {e}", []
|
||||
|
||||
|
||||
def get_agent_status(settings: dict) -> dict:
|
||||
daemon_url = settings.get("agent_daemon_url", "http://192.168.178.151:8765")
|
||||
try:
|
||||
with httpx.Client(timeout=5) as c:
|
||||
return c.get(f"{daemon_url}/status").json()
|
||||
except Exception:
|
||||
return {"alive": False}
|
||||
@@ -12,7 +12,7 @@ import config_manager as cfg
|
||||
from audio_input import start_recording, stop_recording, transcribe
|
||||
from audio_output import speak, stop_speaking, play_ding
|
||||
from screen_capture import capture_screen
|
||||
from ai_client import ask
|
||||
from agent_client import ask_agent, get_agent_status
|
||||
from hotkey_listener import HotkeyListener, KeyCapturer, str_to_display
|
||||
|
||||
ctk.set_appearance_mode("dark")
|
||||
@@ -145,7 +145,12 @@ class HermesApp(ctk.CTk):
|
||||
ui_event("log_user", text)
|
||||
screenshot = (capture_screen()
|
||||
if self.settings["screenshot_enabled"] else None)
|
||||
response = ask(text, screenshot, self.settings)
|
||||
|
||||
ui_event("status", ("thinking", "💭 Denke…", "#3b82f6"))
|
||||
response, tool_calls = ask_agent(text, screenshot, self.settings)
|
||||
|
||||
for tc in tool_calls:
|
||||
ui_event("log_tool", tc)
|
||||
|
||||
ui_event("log_hermes", response)
|
||||
ui_event("status", ("speaking", "🔊 Spricht…", "#a855f7"))
|
||||
|
||||
+114
@@ -0,0 +1,114 @@
|
||||
#!/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("/")
|
||||
|
||||
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."}
|
||||
try:
|
||||
with httpx.Client(timeout=90) as c:
|
||||
if data is None:
|
||||
r = c.get(f"{PC_URL}{path}")
|
||||
else:
|
||||
r = c.post(f"{PC_URL}{path}", json=data)
|
||||
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_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()
|
||||
Reference in New Issue
Block a user