#!/usr/bin/env python3 """Homelab MCP — Heizung, PV, Raumtemps, freie InfluxQL-Abfrage. Eine Tür pro Datenbereich: kapselt die Homelab-HTTP-API (iobroker-Influx) und das Raumtemp-Skript (sensors-Influx) als saubere Tools. Ersetzt das frühere curl/terminal-Gebastel im channel_prompt. Quellen: - HTTP-API: http://100.109.101.12:8765 (Heizung/PV/Status; Oelknoten via MQTT+sensors-Influx) - raumtemps: /root/.hermes/bin/homelab-temps (sensors-Influx) """ from __future__ import annotations import json import os import subprocess import sys import urllib.parse import urllib.request API_BASE = os.environ.get("HOMELAB_API_BASE", "http://100.109.101.12:8765") TEMPS_BIN = os.environ.get("HOMELAB_TEMPS_BIN", "/root/.hermes/bin/homelab-temps") HTTP_TIMEOUT = float(os.environ.get("HOMELAB_HTTP_TIMEOUT", "15")) def _api_get(path: str, params: dict | None = None) -> str: url = API_BASE + path if params: url += "?" + urllib.parse.urlencode(params) try: with urllib.request.urlopen(url, timeout=HTTP_TIMEOUT) as r: raw = r.read().decode("utf-8", "replace") except Exception as e: # noqa: BLE001 return f"Homelab-API nicht erreichbar ({url}): {e}" # Pretty-print falls JSON, sonst roh try: return json.dumps(json.loads(raw), ensure_ascii=False, indent=2) except Exception: return raw def _int_arg(args: dict, key: str, default: int) -> int: try: return int(args.get(key, default)) except (TypeError, ValueError): return default # --- Tool-Handler --------------------------------------------------------- def tool_heizung_now(_args: dict) -> str: return _api_get("/api/heizung/now") def tool_heizung_brenner(args: dict) -> str: return _api_get("/api/heizung/brenner", {"days": _int_arg(args, "days", 7)}) def tool_heizung_kessel(args: dict) -> str: return _api_get("/api/heizung/kessel", {"hours": _int_arg(args, "hours", 48)}) def tool_heizung_history(args: dict) -> str: p = {"hours": _int_arg(args, "hours", 24)} sensor = args.get("sensor") if sensor: p["sensor"] = str(sensor) return _api_get("/api/heizung/history", p) def tool_pv_now(_args: dict) -> str: return _api_get("/api/pv/now") def tool_pv_history(args: dict) -> str: return _api_get("/api/pv/history", {"days": _int_arg(args, "days", 14)}) def tool_status(_args: dict) -> str: return _api_get("/api/status") def tool_measurements(_args: dict) -> str: return _api_get("/api/measurements") def tool_query(args: dict) -> str: q = str(args.get("q", "")).strip() if not q: return "Fehlt: q (InfluxQL-Abfrage, nur SELECT/SHOW)." low = q.lstrip().lower() if not (low.startswith("select") or low.startswith("show")): return "Nur SELECT/SHOW erlaubt (read-only)." return _api_get("/api/query", {"q": q}) def tool_raumtemps(args: dict) -> str: cmd = [TEMPS_BIN] if args.get("timeline"): cmd.append("--timeline") try: out = subprocess.run( cmd, capture_output=True, text=True, timeout=30 ) return (out.stdout or out.stderr or "").strip() or "Keine Ausgabe." except Exception as e: # noqa: BLE001 return f"raumtemps-Skript Fehler: {e}" # --- MCP-Gerüst ----------------------------------------------------------- TOOLS = [ { "name": "heizung_now", "description": "Live-Heizung: Heizraum (Pumpen/Brenner), Oelknoten (4x DS18B20 Erdtrasse), Temperaturen. online=true wenn Heizraum ODER Oelknoten aktiv. IMMER dieses Tool, nicht curl/terminal.", "inputSchema": {"type": "object", "properties": {}, "required": []}, }, { "name": "heizung_brenner", "description": "Brenner Takt-Analyse (Starts/Laufzeit). Param days (default 7).", "inputSchema": {"type": "object", "properties": {"days": {"type": "number", "description": "Tage zurück (default 7)"}}, "required": []}, }, { "name": "heizung_kessel", "description": "Ölkessel Vorlauf-Verlauf. Param hours (default 48).", "inputSchema": {"type": "object", "properties": {"hours": {"type": "number", "description": "Stunden zurück (default 48)"}}, "required": []}, }, { "name": "heizung_history", "description": "Heizungs-Temperatur-Verlauf. Params hours (default 24), sensor (ROM-ID oder 'all').", "inputSchema": {"type": "object", "properties": {"hours": {"type": "number"}, "sensor": {"type": "string"}}, "required": []}, }, { "name": "pv_now", "description": "Live-PV: Ertrag heute, Netzbezug, Verbrauch. IMMER dieses Tool, nicht curl/terminal.", "inputSchema": {"type": "object", "properties": {}, "required": []}, }, { "name": "pv_history", "description": "PV-Tageserträge. Param days (default 14).", "inputSchema": {"type": "object", "properties": {"days": {"type": "number", "description": "Tage zurück (default 14)"}}, "required": []}, }, { "name": "homelab_status", "description": "System-Übersicht: online/offline aller Knoten.", "inputSchema": {"type": "object", "properties": {}, "required": []}, }, { "name": "homelab_measurements", "description": "Alle verfügbaren InfluxDB-Measurements + Zeitraum (für homelab_query-Planung).", "inputSchema": {"type": "object", "properties": {}, "required": []}, }, { "name": "homelab_query", "description": ( "FREIE read-only InfluxQL-Abfrage auf die iobroker-DB (Heizung/PV/Sensoren). " "Nur SELECT/SHOW. Für Sonderauswertungen ohne festes Tool. " "Ersetzt das frühere curl /api/query — NIEMALS terminal/curl dafür." ), "inputSchema": {"type": "object", "properties": {"q": {"type": "string", "description": "InfluxQL, z.B. SELECT count(value) FROM brennerstarts WHERE time > now()-7d"}}, "required": ["q"]}, }, { "name": "raumtemps", "description": ( "Raumtemperaturen Muldenstein (Außen, Garage, Wohnstube, Küche, ...) mit 24h-Tief/Hoch. " "Param timeline=true für Verlauf mit Zeitstempeln. Für Garage-KÜHLDECKE-Hydraulik " "(VL/RL/ΔT/kW) stattdessen garage_decke_live." ), "inputSchema": {"type": "object", "properties": {"timeline": {"type": "boolean", "description": "Verlauf mit Min/Max-Zeitstempeln"}}, "required": []}, }, ] HANDLERS = { "heizung_now": tool_heizung_now, "heizung_brenner": tool_heizung_brenner, "heizung_kessel": tool_heizung_kessel, "heizung_history": tool_heizung_history, "pv_now": tool_pv_now, "pv_history": tool_pv_history, "homelab_status": tool_status, "homelab_measurements": tool_measurements, "homelab_query": tool_query, "raumtemps": tool_raumtemps, } 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": "homelab", "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()