Init commit

hermes/MCP_home_assistant/home_assistant_mcp.py

@@ -0,0 +1,245 @@
+import os
+from pathlib import Path
+from typing import Any, Dict, Optional
+
+import requests
+from dotenv import load_dotenv
+from mcp.server.fastmcp import FastMCP
+
+
+BASE_DIR = Path(__file__).resolve().parent
+load_dotenv(BASE_DIR / ".env")
+
+HA_URL = os.getenv("HA_URL", "http://192.168.1.141:8123").rstrip("/")
+HA_TOKEN = os.getenv("HA_TOKEN")
+
+if not HA_TOKEN:
+    raise RuntimeError("HA_TOKEN manquant dans /opt/hermes/ha-mcp/.env")
+
+HEADERS = {
+    "Authorization": f"Bearer {HA_TOKEN}",
+    "Content-Type": "application/json",
+}
+
+mcp = FastMCP("home-assistant")
+
+
+def ha_get(path: str) -> Any:
+    response = requests.get(
+        f"{HA_URL}{path}",
+        headers=HEADERS,
+        timeout=10,
+    )
+    response.raise_for_status()
+    return response.json()
+
+
+def ha_post(path: str, payload: Optional[Dict[str, Any]] = None) -> Any:
+    response = requests.post(
+        f"{HA_URL}{path}",
+        headers=HEADERS,
+        json=payload or {},
+        timeout=10,
+    )
+    response.raise_for_status()
+
+    if response.text:
+        return response.json()
+
+    return {"status": "ok"}
+
+
+@mcp.tool()
+def ha_check_api() -> Any:
+    """Check if Home Assistant API is reachable."""
+    return ha_get("/api/")
+
+
+@mcp.tool()
+def ha_list_entities(domain: Optional[str] = None) -> Any:
+    """
+    List Home Assistant entities.
+
+    Optional domain examples:
+    - light
+    - switch
+    - sensor
+    - binary_sensor
+    - automation
+    - script
+    """
+    states = ha_get("/api/states")
+
+    if domain:
+        prefix = f"{domain}."
+        states = [
+            entity for entity in states
+            if entity.get("entity_id", "").startswith(prefix)
+        ]
+
+    return [
+        {
+            "entity_id": entity.get("entity_id"),
+            "state": entity.get("state"),
+            "friendly_name": entity.get("attributes", {}).get("friendly_name"),
+        }
+        for entity in states
+    ]
+
+
+@mcp.tool()
+def ha_search_entities(search: str) -> Any:
+    """
+    Search Home Assistant entities by entity_id or friendly name.
+    """
+    search_lower = search.lower()
+    states = ha_get("/api/states")
+
+    results = []
+
+    for entity in states:
+        entity_id = entity.get("entity_id", "")
+        friendly_name = entity.get("attributes", {}).get("friendly_name", "")
+
+        if (
+            search_lower in entity_id.lower()
+            or search_lower in friendly_name.lower()
+        ):
+            results.append(
+                {
+                    "entity_id": entity_id,
+                    "state": entity.get("state"),
+                    "friendly_name": friendly_name,
+                }
+            )
+
+    return results
+
+
+@mcp.tool()
+def ha_get_state(entity_id: str) -> Any:
+    """Get the current state of a Home Assistant entity."""
+    return ha_get(f"/api/states/{entity_id}")
+
+
+@mcp.tool()
+def ha_turn_on(entity_id: str) -> Any:
+    """
+    Turn on a Home Assistant entity.
+
+    Works with domains such as:
+    - light
+    - switch
+    - fan
+    - input_boolean
+    """
+    domain = entity_id.split(".")[0]
+    return ha_post(
+        f"/api/services/{domain}/turn_on",
+        {"entity_id": entity_id},
+    )
+
+
+@mcp.tool()
+def ha_turn_off(entity_id: str) -> Any:
+    """
+    Turn off a Home Assistant entity.
+
+    Works with domains such as:
+    - light
+    - switch
+    - fan
+    - input_boolean
+    """
+    domain = entity_id.split(".")[0]
+    return ha_post(
+        f"/api/services/{domain}/turn_off",
+        {"entity_id": entity_id},
+    )
+
+
+@mcp.tool()
+def ha_toggle(entity_id: str) -> Any:
+    """Toggle a Home Assistant entity."""
+    domain = entity_id.split(".")[0]
+    return ha_post(
+        f"/api/services/{domain}/toggle",
+        {"entity_id": entity_id},
+    )
+
+
+@mcp.tool()
+def ha_call_service(
+    domain: str,
+    service: str,
+    data: Dict[str, Any],
+) -> Any:
+    """
+    Call any Home Assistant service.
+
+    Example:
+    domain = "light"
+    service = "turn_on"
+    data = {
+        "entity_id": "light.salon",
+        "brightness_pct": 50
+    }
+    """
+    return ha_post(
+        f"/api/services/{domain}/{service}",
+        data,
+    )
+
+
+@mcp.tool()
+def ha_run_script(script_entity_id: str) -> Any:
+    """Run a Home Assistant script entity."""
+    return ha_post(
+        "/api/services/script/turn_on",
+        {"entity_id": script_entity_id},
+    )
+
+
+@mcp.tool()
+def ha_trigger_automation(
+    automation_entity_id: str,
+    skip_condition: bool = True,
+) -> Any:
+    """
+    Trigger a Home Assistant automation manually.
+
+    By default, conditions are skipped.
+    """
+    return ha_post(
+        "/api/services/automation/trigger",
+        {
+            "entity_id": automation_entity_id,
+            "skip_condition": skip_condition,
+        },
+    )
+
+
+@mcp.tool()
+def ha_restart_home_assistant() -> Any:
+    """
+    Restart Home Assistant.
+
+    Use carefully.
+    """
+    return ha_post("/api/services/homeassistant/restart")
+
+
+@mcp.tool()
+def ha_reload_automations() -> Any:
+    """Reload Home Assistant automations."""
+    return ha_post("/api/services/automation/reload")
+
+
+@mcp.tool()
+def ha_reload_scripts() -> Any:
+    """Reload Home Assistant scripts."""
+    return ha_post("/api/services/script/reload")
+
+
+if __name__ == "__main__":
+    mcp.run()