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.
This commit is contained in:
Homelab Cursor
parent
03fd7952af
commit
a17031dfbe
2 changed files with 517 additions and 0 deletions
|
|
@ -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.
|
||||
|
|
|
|||
510
infra/hermes/shelly_mcp.py
Executable file
510
infra/hermes/shelly_mcp.py
Executable file
|
|
@ -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()
|
||||
Loading…
Add table
Reference in a new issue