From a17031dfbe434cbbbee2bb4287bf50a2af0358f8 Mon Sep 17 00:00:00 2001 From: Homelab Cursor Date: Thu, 16 Jul 2026 18:23:23 +0200 Subject: [PATCH] Add Hermes shelly MCP for LAN Shelly control. Gives CT151 native shelly_list/status/cover/pulse tools with safety gates for bad_unten motor and cron-blocked door/garage pulses. --- infra/hermes/README.md | 7 + infra/hermes/shelly_mcp.py | 510 +++++++++++++++++++++++++++++++++++++ 2 files changed, 517 insertions(+) create mode 100755 infra/hermes/shelly_mcp.py diff --git a/infra/hermes/README.md b/infra/hermes/README.md index 1ba3abe16..e24ea7904 100644 --- a/infra/hermes/README.md +++ b/infra/hermes/README.md @@ -37,3 +37,10 @@ Heizung/PV/Raumtemps MCP fuer Hermes (CT151). Deploy analog `garage_decke_mcp.py` nach `/root/.hermes/homelab_mcp.py`, dann `systemctl restart hermes-gateway`. Heizungs-HTTP-API: `infra/heizung-api/pruefstand_web.py` auf pve-mu-3 (`pruefstand-web.service`, Port 8765). + +## shelly_mcp.py (16.07.2026) + +Tools: `shelly_list`, `shelly_status`, `shelly_cover`, `shelly_pulse`. + +Safety: Bad unten Bewegung gesperrt; Tor/Tuersummer nur `confirm=true` und nie in Cron (`HERMES_CRON_SESSION`). +Deploy-Ziel CT151: `/root/.hermes/shelly_mcp.py` + `mcp_servers.shelly` in config.yaml. diff --git a/infra/hermes/shelly_mcp.py b/infra/hermes/shelly_mcp.py new file mode 100755 index 000000000..b43818695 --- /dev/null +++ b/infra/hermes/shelly_mcp.py @@ -0,0 +1,510 @@ +#!/usr/bin/env python3 +"""Shelly MCP — Status und Steuerung der Muldenstein-Shellys per LAN-HTTP. + +Inventar fest verdrahtet (CT999 projekte/shelly-inventar.md). +Sicherheit: + - Rollo Bad unten (.96): Bewegung blockiert (Motor defekt) + - Garagentor / Haustuer-Impuls: nur mit confirm=true, nie in Cron + - Cron-Erkennung: HERMES_CRON_SESSION=1 +""" +from __future__ import annotations + +import json +import os +import sys +import urllib.error +import urllib.parse +import urllib.request + +TIMEOUT = float(os.environ.get("SHELLY_HTTP_TIMEOUT", "5")) + +# alias -> device key +DEVICES: dict[str, dict] = { + "garagentor": { + "ip": "192.168.178.71", + "name": "Garagentor Oeffner", + "gen": 1, + "kind": "pulse", + "aliases": ["garagentor", "tor", "garage_tor", "garage-tor"], + }, + "haustuer": { + "ip": "192.168.178.52", + "name": "Haustuer Tuersummer", + "gen": 1, + "kind": "pulse", + "aliases": ["haustuer", "tuersummer", "summer", "tuer", "klingel"], + }, + "kueche": { + "ip": "192.168.178.72", + "name": "Rollo Kueche", + "gen": 1, + "kind": "cover", + "api": "roller", # Shelly 2.5 gen1 + "aliases": ["kueche", "rollo_kueche", "kitchen"], + }, + "computerraum": { + "ip": "192.168.178.26", + "name": "Rollo Computerraum", + "gen": 2, + "kind": "cover", + "api": "cover", + "aliases": ["computerraum", "pc", "buero", "büro"], + }, + "gaestezimmer": { + "ip": "192.168.178.38", + "name": "Rollo Gaestezimmer", + "gen": 2, + "kind": "cover", + "api": "cover", + "aliases": ["gaestezimmer", "gastezimmer", "gaeste", "guest"], + }, + "bad_unten": { + "ip": "192.168.178.96", + "name": "Rollo Bad unten", + "gen": 2, + "kind": "cover", + "api": "cover", + "motor_broken": True, + "aliases": ["bad_unten", "bad", "badunten", "bathroom"], + }, + "garage_decke_a": { + "ip": "192.168.178.86", + "name": "Garage Decke A", + "gen": 1, + "kind": "sensor", + "aliases": ["garage_decke_a", "decke_a", "garage-a"], + }, + "garage_decke_b": { + "ip": "192.168.178.78", + "name": "Garage Decke B", + "gen": 1, + "kind": "sensor", + "aliases": ["garage_decke_b", "decke_b", "garage-b"], + }, +} + +# flatten alias lookup +_ALIAS: dict[str, str] = {} +for key, d in DEVICES.items(): + _ALIAS[key.lower()] = key + _ALIAS[d["ip"]] = key + _ALIAS[d["name"].lower()] = key + for a in d.get("aliases", []): + _ALIAS[a.lower()] = key + + +def _is_cron() -> bool: + v = (os.environ.get("HERMES_CRON_SESSION") or os.environ.get("HERMES_CRON") or "").strip().lower() + return v in ("1", "true", "yes", "on") + + +def _http_get(url: str) -> dict | str: + req = urllib.request.Request(url, method="GET") + with urllib.request.urlopen(req, timeout=TIMEOUT) as r: + raw = r.read().decode("utf-8", "replace") + try: + return json.loads(raw) + except json.JSONDecodeError: + return raw + + +def _http_post_json(url: str, payload: dict) -> dict | str: + data = json.dumps(payload).encode() + req = urllib.request.Request( + url, data=data, headers={"Content-Type": "application/json"}, method="POST" + ) + with urllib.request.urlopen(req, timeout=TIMEOUT) as r: + raw = r.read().decode("utf-8", "replace") + try: + return json.loads(raw) + except json.JSONDecodeError: + return raw + + +def _resolve(target: str | None) -> tuple[str, dict] | str: + if not target or not str(target).strip(): + return "Ziel fehlt. Alias/IP/Name angeben (z.B. kueche, garagentor, 192.168.178.72)." + t = str(target).strip().lower() + # normalize umlauts lightly + t = ( + t.replace("ä", "ae") + .replace("ö", "oe") + .replace("ü", "ue") + .replace("ß", "ss") + .replace(" ", "_") + ) + key = _ALIAS.get(t) + if not key: + # partial match on name + for k, d in DEVICES.items(): + if t in k or t in d["name"].lower().replace(" ", "_"): + key = k + break + if not key: + known = ", ".join(sorted(DEVICES.keys())) + return f"Unbekanntes Ziel {target!r}. Bekannt: {known}" + return key, DEVICES[key] + + +def _truthy(v) -> bool: + if isinstance(v, bool): + return v + if v is None: + return False + return str(v).strip().lower() in ("1", "true", "yes", "on", "ja") + + +def tool_list(_args: dict) -> str: + lines = ["Shelly-Inventar Muldenstein (LAN):", ""] + for key, d in DEVICES.items(): + caps = d["kind"] + note = "" + if d.get("motor_broken"): + note = " [MOTOR DEFEKT — Bewegung gesperrt]" + if d["kind"] == "pulse": + note += " [Impuls: confirm=true, kein Cron]" + lines.append( + f"- {key}: {d['name']} @ {d['ip']} ({d['kind']}, Gen{d['gen']}){note}" + ) + lines.append(f" Aliase: {', '.join(d.get('aliases', [key]))}") + lines.append("") + lines.append("Tools: shelly_list, shelly_status, shelly_cover, shelly_pulse") + lines.append( + "Cron: Rollos OK; Tor/Tuersummer verboten; Bad unten nie bewegen." + ) + return "\n".join(lines) + + +def _status_gen1(d: dict) -> dict: + ip = d["ip"] + settings = _http_get(f"http://{ip}/settings") + status = _http_get(f"http://{ip}/status") + out: dict = { + "ip": ip, + "name": d["name"] if isinstance(settings, dict) and not settings.get("name") else ( + settings.get("name") if isinstance(settings, dict) else d["name"] + ), + "gen": 1, + "kind": d["kind"], + } + if not isinstance(status, dict): + out["error"] = str(status) + return out + if d["kind"] == "cover" and d.get("api") == "roller": + rollers = status.get("rollers") or [] + if rollers: + r0 = rollers[0] + out["cover"] = { + "state": r0.get("state"), + "current_pos": r0.get("current_pos"), + "last_direction": r0.get("last_direction"), + "power": r0.get("power"), + } + if d["kind"] == "pulse": + relays = status.get("relays") or [] + out["relay"] = { + "ison": relays[0].get("ison") if relays else None, + "has_timer": relays[0].get("has_timer") if relays else None, + } + if d["kind"] == "sensor": + out["ext_temperature"] = status.get("ext_temperature") + out["mqtt_connected"] = (status.get("mqtt") or {}).get("connected") + out["wifi_rssi"] = (status.get("wifi_sta") or {}).get("rssi") + out["uptime"] = status.get("uptime") + if d.get("motor_broken"): + out["warning"] = "Motor defekt — Bewegung gesperrt" + return out + + +def _status_gen2(d: dict) -> dict: + ip = d["ip"] + info = _http_get(f"http://{ip}/rpc/Shelly.GetDeviceInfo") + st = _http_get(f"http://{ip}/rpc/Shelly.GetStatus") + out: dict = { + "ip": ip, + "name": (info.get("name") if isinstance(info, dict) else None) or d["name"], + "gen": 2, + "kind": d["kind"], + "id": info.get("id") if isinstance(info, dict) else None, + } + if not isinstance(st, dict): + out["error"] = str(st) + return out + cover = st.get("cover:0") or {} + out["cover"] = { + "state": cover.get("state"), + "current_pos": cover.get("current_pos"), + "apower": cover.get("apower"), + "voltage": cover.get("voltage"), + "source": cover.get("source"), + } + wifi = st.get("wifi") or {} + out["wifi_rssi"] = wifi.get("rssi") + if d.get("motor_broken"): + out["warning"] = "Motor defekt — Bewegung gesperrt" + return out + + +def tool_status(args: dict) -> str: + resolved = _resolve(args.get("target") or args.get("device") or args.get("name")) + if isinstance(resolved, str): + return resolved + key, d = resolved + try: + data = _status_gen2(d) if d["gen"] == 2 else _status_gen1(d) + data["key"] = key + return json.dumps(data, ensure_ascii=False, indent=2) + except Exception as e: # noqa: BLE001 + return f"Fehler Status {d['name']} ({d['ip']}): {e}" + + +def _cover_gen1_roller(ip: str, action: str, position: int | None) -> str: + if action == "open": + url = f"http://{ip}/roller/0?go=open" + elif action == "close": + url = f"http://{ip}/roller/0?go=close" + elif action == "stop": + url = f"http://{ip}/roller/0?go=stop" + elif action == "position": + if position is None: + return "position erfordert Argument position (0-100)." + url = f"http://{ip}/roller/0?go=to_pos&roller_pos={int(position)}" + else: + return f"Unbekannte action {action!r}" + res = _http_get(url) + return json.dumps({"ok": True, "action": action, "result": res}, ensure_ascii=False) + + +def _cover_gen2(ip: str, action: str, position: int | None) -> str: + if action == "open": + method, params = "Cover.Open", {"id": 0} + elif action == "close": + method, params = "Cover.Close", {"id": 0} + elif action == "stop": + method, params = "Cover.Stop", {"id": 0} + elif action == "position": + if position is None: + return "position erfordert Argument position (0-100)." + method, params = "Cover.GoToPosition", {"id": 0, "pos": int(position)} + else: + return f"Unbekannte action {action!r}" + res = _http_post_json(f"http://{ip}/rpc", {"id": 1, "method": method, "params": params}) + return json.dumps({"ok": True, "action": action, "result": res}, ensure_ascii=False) + + +def tool_cover(args: dict) -> str: + resolved = _resolve(args.get("target") or args.get("device") or args.get("name")) + if isinstance(resolved, str): + return resolved + key, d = resolved + if d["kind"] != "cover": + return ( + f"{d['name']} ist kein Rollo (kind={d['kind']}). " + "Fuer Tor/Tuersummer: shelly_pulse. Fuer Sensoren: shelly_status." + ) + if d.get("motor_broken"): + return ( + f"BLOCKIERT: {d['name']} ({d['ip']}) — Motor defekt. " + "Keine open/close/stop/position. Nur Status via shelly_status." + ) + action = str(args.get("action") or "").strip().lower() + if action in ("auf", "oeffnen", "öffnen"): + action = "open" + elif action in ("zu", "schliessen", "schließen"): + action = "close" + elif action in ("halt", "stopp"): + action = "stop" + elif action in ("pos", "goto", "to_pos"): + action = "position" + if action not in ("open", "close", "stop", "position"): + return "action muss open|close|stop|position sein." + pos = args.get("position") + if pos is not None: + try: + pos = int(pos) + except (TypeError, ValueError): + return "position muss 0-100 sein." + if not 0 <= pos <= 100: + return "position muss 0-100 sein." + try: + if d["gen"] == 2: + return _cover_gen2(d["ip"], action, pos) + return _cover_gen1_roller(d["ip"], action, pos) + except Exception as e: # noqa: BLE001 + return f"Fehler shelly_cover {d['name']}: {e}" + + +def tool_pulse(args: dict) -> str: + if _is_cron(): + return ( + "BLOCKIERT: Garagentor/Haustuer-Impuls ist in Cron-Jobs verboten. " + "Nur interaktiv im Chat mit confirm=true." + ) + resolved = _resolve(args.get("target") or args.get("device") or args.get("name")) + if isinstance(resolved, str): + return resolved + key, d = resolved + if d["kind"] != "pulse": + return f"{d['name']} ist kein Impuls-Geraet. Nutze shelly_cover fuer Rollos." + if not _truthy(args.get("confirm")): + return ( + f"Sicherheit: Impuls fuer {d['name']} braucht confirm=true. " + "Ohne Bestaetigung keine Ausfuehrung." + ) + ip = d["ip"] + try: + # Gen1 Shelly 1: toggle or on — devices use auto_off=1s + res = _http_get(f"http://{ip}/relay/0?turn=on") + return json.dumps( + {"ok": True, "device": d["name"], "ip": ip, "action": "pulse_on", "result": res}, + ensure_ascii=False, + ) + except Exception as e: # noqa: BLE001 + return f"Fehler shelly_pulse {d['name']}: {e}" + + +TOOLS = [ + { + "name": "shelly_list", + "description": ( + "Listet alle bekannten Shellys (IP, Name, Typ, Aliase, Safety-Hinweise). " + "Kein Netzwerknoetig fuer die Inventarliste." + ), + "inputSchema": {"type": "object", "properties": {}, "required": []}, + }, + { + "name": "shelly_status", + "description": ( + "Live-Status eines Shellys (Rollo-Position, Relay, Sensor-Temps). " + "Ziel per Alias/IP/Name: kueche, computerraum, gaestezimmer, bad_unten, " + "garagentor, haustuer, garage_decke_a/b." + ), + "inputSchema": { + "type": "object", + "properties": { + "target": { + "type": "string", + "description": "Alias, Name oder IP des Geraets", + } + }, + "required": ["target"], + }, + }, + { + "name": "shelly_cover", + "description": ( + "Rollo steuern: action=open|close|stop|position. " + "Ziele: kueche, computerraum, gaestezimmer. " + "bad_unten ist BLOCKIERT (Motor defekt). " + "Cron-Jobs duerfen Rollos steuern." + ), + "inputSchema": { + "type": "object", + "properties": { + "target": {"type": "string", "description": "Rollo-Alias"}, + "action": { + "type": "string", + "description": "open|close|stop|position", + }, + "position": { + "type": "number", + "description": "0-100 bei action=position (0=zu, 100=auf)", + }, + }, + "required": ["target", "action"], + }, + }, + { + "name": "shelly_pulse", + "description": ( + "Impuls fuer Garagentor oder Haustuer-Tuersummer. " + "PFLICHT: confirm=true. In Cron-Jobs IMMER blockiert." + ), + "inputSchema": { + "type": "object", + "properties": { + "target": { + "type": "string", + "description": "garagentor oder haustuer", + }, + "confirm": { + "type": "boolean", + "description": "Muss true sein zur Ausfuehrung", + }, + }, + "required": ["target", "confirm"], + }, + }, +] + +HANDLERS = { + "shelly_list": tool_list, + "shelly_status": tool_status, + "shelly_cover": tool_cover, + "shelly_pulse": tool_pulse, +} + + +def handle(req): + method = req.get("method") + rid = req.get("id") + if method == "initialize": + return { + "jsonrpc": "2.0", + "id": rid, + "result": { + "protocolVersion": "2024-11-05", + "capabilities": {"tools": {}}, + "serverInfo": {"name": "shelly", "version": "1.0.0"}, + }, + } + if method in ("notifications/initialized", "initialized"): + return None + if method == "tools/list": + return {"jsonrpc": "2.0", "id": rid, "result": {"tools": TOOLS}} + if method == "tools/call": + name = req.get("params", {}).get("name") + args = req.get("params", {}).get("arguments", {}) or {} + handler = HANDLERS.get(name) + try: + text = handler(args) if handler else f"Unbekanntes Tool: {name}" + except Exception as e: # noqa: BLE001 + text = f"Fehler: {e}" + return { + "jsonrpc": "2.0", + "id": rid, + "result": {"content": [{"type": "text", "text": text}]}, + } + if method == "ping": + return {"jsonrpc": "2.0", "id": rid, "result": {}} + return { + "jsonrpc": "2.0", + "id": rid, + "error": {"code": -32601, "message": f"Unknown: {method}"}, + } + + +def main(): + for line in sys.stdin: + line = line.strip() + if not line: + continue + try: + req = json.loads(line) + except json.JSONDecodeError: + continue + try: + resp = handle(req) + except Exception as e: # noqa: BLE001 + resp = { + "jsonrpc": "2.0", + "id": req.get("id"), + "error": {"code": -32603, "message": str(e)}, + } + if resp is not None: + sys.stdout.write(json.dumps(resp, ensure_ascii=False) + "\n") + sys.stdout.flush() + + +if __name__ == "__main__": + main()