Alfredv2/hermes/MCP_home_assistant/home_assistant_mcp.py

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()