feat(config,mcps,policy,compose): komodo alerts with hold, vibegram and ha python_tools, vault rw home zones

mcps/homeassistant.py

@@ -0,0 +1,169 @@
+"""§4.4: Home Assistant как python_tool с перечислимыми действиями.
+
+Токен HA живёт в gateway и в контекст модели не попадает. Действия:
+найти сущности, посмотреть состояние, вызвать сервис, перечислить сервисы
+домена. Ничего про конфигурацию, пользователей и интеграции - только
+REST `/api/states` и `/api/services`.
+"""
+
+from __future__ import annotations
+
+import json
+from dataclasses import dataclass
+from typing import Any, Literal
+
+import aiohttp
+
+__all__ = ["Action", "HomeAssistant"]
+
+Action = Literal["search", "state", "call", "services"]
+
+SEARCH_LIMIT = 60
+ATTR_LIMIT = 25
+# Не влезает в ответ и не нужно модели: гигантские списки и бинарники.
+NOISY_ATTRS = frozenset({"entity_picture", "icon", "supported_color_modes", "options"})
+
+
+@dataclass(frozen=True, slots=True)
+class HomeAssistant:
+    url: str
+    token: str
+    timeout: float = 20.0
+    max_chars: int = 8_000
+
+    async def ha(
+        self,
+        action: Action,
+        *,
+        query: str | None = None,
+        entity: str | None = None,
+        domain: str | None = None,
+        service: str | None = None,
+        data: dict[str, Any] | None = None,
+    ) -> str:
+        """Home Assistant - умный дом: свет, кнопки ИК-пульта, сцены, скрипты, датчики.
+
+        action:
+        - search: сущности по подстроке в id или имени (query; пусто - все),
+          сгруппированы по домену. Начни отсюда: точные entity_id нужны везде.
+        - state: состояние и атрибуты одной сущности (entity).
+        - call: вызвать сервис (domain, service, entity опционально, data -
+          доп. поля: brightness, color_temp, …). Примеры: light/turn_on,
+          light/turn_off, script/turn_on, scene/turn_on, button/press.
+        - services: какие сервисы есть у домена (domain) и их поля.
+
+        Делай то, о чём попросили: «выключи свет» - это одна сущность или
+        группа, не весь дом.
+        """
+        match action:
+            case "search":
+                return await self._search(query or "")
+            case "state":
+                return await self._state(_need(entity, "entity"))
+            case "call":
+                return await self._call(
+                    _need(domain, "domain"), _need(service, "service"), entity, data
+                )
+            case "services":
+                return await self._services(_need(domain, "domain"))
+
+    async def _search(self, query: str) -> str:
+        states = await self._get("/api/states")
+        q = query.lower()
+        hits = [
+            s
+            for s in states
+            if q in s["entity_id"].lower()
+            or q in str(s.get("attributes", {}).get("friendly_name", "")).lower()
+        ]
+        if not hits:
+            return "ничего не нашлось"
+        by_domain: dict[str, list[dict[str, Any]]] = {}
+        for s in sorted(hits, key=lambda s: s["entity_id"]):
+            by_domain.setdefault(s["entity_id"].split(".", 1)[0], []).append(s)
+        lines = []
+        shown = 0
+        for dom, items in by_domain.items():
+            lines.append(f"{dom}:")
+            for s in items:
+                if shown >= SEARCH_LIMIT:
+                    lines.append(f"  … ещё {len(hits) - shown}, уточни query")
+                    return "\n".join(lines)
+                name = s.get("attributes", {}).get("friendly_name") or ""
+                lines.append(f"  {s['entity_id']:40} {s['state']:12} {name}")
+                shown += 1
+        return "\n".join(lines)
+
+    async def _state(self, entity: str) -> str:
+        s = await self._get(f"/api/states/{entity}")
+        attrs = {
+            k: v for k, v in (s.get("attributes") or {}).items() if k not in NOISY_ATTRS
+        }
+        lines = [f"{s['entity_id']}: {s['state']} (с {s.get('last_changed', '?')})"]
+        for k, v in list(attrs.items())[:ATTR_LIMIT]:
+            lines.append(f"  {k}: {json.dumps(v, ensure_ascii=False)[:200]}")
+        return self._clip("\n".join(lines))
+
+    async def _call(
+        self, domain: str, service: str, entity: str | None, data: dict[str, Any] | None
+    ) -> str:
+        body: dict[str, Any] = dict(data or {})
+        if entity:
+            body["entity_id"] = entity
+        changed = await self._post(f"/api/services/{domain}/{service}", body)
+        if not changed:
+            return f"{domain}.{service}: выполнено, состояния не менялись"
+        lines = [f"{domain}.{service}: выполнено, изменилось:"]
+        lines.extend(f"  {s['entity_id']}: {s['state']}" for s in changed[:20])
+        return "\n".join(lines)
+
+    async def _services(self, domain: str) -> str:
+        for block in await self._get("/api/services"):
+            if block.get("domain") == domain:
+                lines = [f"{domain}:"]
+                for name, spec in sorted((block.get("services") or {}).items()):
+                    fields = ", ".join((spec.get("fields") or {}).keys())
+                    lines.append(
+                        f"  {name}: {spec.get('description') or ''}".rstrip()
+                        + (f" [{fields}]" if fields else "")
+                    )
+                return self._clip("\n".join(lines))
+        return f"домена {domain} нет"
+
+    async def _get(self, path: str) -> Any:
+        return await self._request("GET", path, None)
+
+    async def _post(self, path: str, body: dict[str, Any]) -> Any:
+        return await self._request("POST", path, body)
+
+    async def _request(
+        self, method: str, path: str, body: dict[str, Any] | None
+    ) -> Any:
+        headers = {"Authorization": f"Bearer {self.token}"}
+        async with (
+            aiohttp.ClientSession() as http,
+            http.request(
+                method,
+                f"{self.url.rstrip('/')}{path}",
+                json=body,
+                headers=headers,
+                timeout=aiohttp.ClientTimeout(total=self.timeout),
+            ) as response,
+        ):
+            text = await response.text()
+            if response.status >= 400:
+                msg = f"Home Assistant {method} {path}: {response.status} {text[:300]}"
+                raise RuntimeError(msg)
+            return json.loads(text) if text else None
+
+    def _clip(self, text: str) -> str:
+        if len(text) <= self.max_chars:
+            return text
+        return text[: self.max_chars] + "\n…[обрезано]"
+
+
+def _need(value: str | None, name: str) -> str:
+    if not value:
+        msg = f"для этого действия нужен параметр {name}"
+        raise ValueError(msg)
+    return value