Hitonabi
181db38b49
mcp_mc.py: 7 -> 23 Tools (hf_search/quants, install/delete/load/unload_model, set_model_role, list_jobs, cancel_job, list_services, backup_now, list_backups, token_stats, check_updates, apply_update, service_logs). Hermes kann den Stack jetzt vollständig steuern (alles was die MC2-UI kann). executor.py: _ensure_streams() — unter pythonw (kein Konsolenfenster, für Scheduled-Task-Autostart) sind sys.stdout/stderr=None und uvicorn-Logging crasht beim Start. Jetzt Umleitung auf %LOCALAPPDATA%\HermesPCExecutor\executor.log. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
150 lines
4.9 KiB
Python
150 lines
4.9 KiB
Python
"""
|
|
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())
|
|
|
|
|
|
def _ensure_streams() -> None:
|
|
"""Unter pythonw (kein Konsolenfenster) sind sys.stdout/stderr = None →
|
|
uvicorns Logging crasht beim Start. Dann auf eine Logdatei umbiegen."""
|
|
import sys
|
|
if sys.stdout is not None and sys.stderr is not None:
|
|
return
|
|
logdir = os.path.join(
|
|
os.environ.get("LOCALAPPDATA", os.path.dirname(os.path.abspath(__file__))),
|
|
"HermesPCExecutor",
|
|
)
|
|
os.makedirs(logdir, exist_ok=True)
|
|
f = open(os.path.join(logdir, "executor.log"), "a", buffering=1, encoding="utf-8")
|
|
sys.stdout = f
|
|
sys.stderr = f
|
|
|
|
|
|
if __name__ == "__main__":
|
|
_ensure_streams()
|
|
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)
|