""" Home Assistant MCP tool — let XiaoZhi control HA devices directly via REST API. Bypasses OpenClaw for much lower latency (~200ms vs ~10s). """ from mcp.server.fastmcp import FastMCP import os import json import logging import urllib.request import urllib.error logging.basicConfig(level=logging.INFO) logger = logging.getLogger("ha_mcp") mcp = FastMCP("HomeAssistant Direct") HA_URL = os.getenv("HA_URL", "http://192.168.50.139:8123").rstrip("/") HA_TOKEN = os.getenv("HA_TOKEN", "YOUR_TOKEN_HERE").strip() def ha_request(method, path, data=None): """Make a request to Home Assistant REST API.""" url = f"{HA_URL}/api/{path}" headers = { "Authorization": f"Bearer {HA_TOKEN}", "Content-Type": "application/json", } body = json.dumps(data).encode() if data else None req = urllib.request.Request(url, data=body, headers=headers, method=method) try: with urllib.request.urlopen(req, timeout=10) as resp: return json.loads(resp.read().decode()) except urllib.error.HTTPError as e: error_body = e.read().decode() if e.fp else "" return {"error": f"HTTP {e.code}: {error_body}"} except Exception as e: return {"error": str(e)} @mcp.tool() def get_states() -> str: """ 获取所有设备的状态列表。 返回所有实体的 entity_id、state 和 friendly_name。 用于查看家里有哪些设备以及它们的当前状态。 """ result = ha_request("GET", "states") if isinstance(result, dict) and "error" in result: return f"错误: {result['error']}" summary = [] for entity in result: eid = entity.get("entity_id", "") state = entity.get("state", "") name = entity.get("attributes", {}).get("friendly_name", eid) summary.append(f"{name} ({eid}): {state}") return "\n".join(summary) @mcp.tool() def get_entity_state(entity_id: str) -> str: """ 获取单个设备的详细状态。 参数 entity_id: 设备ID，如 light.living_room, switch.bedroom_fan, climate.ac 返回设备的完整状态和属性信息。 """ result = ha_request("GET", f"states/{entity_id}") if isinstance(result, dict) and "error" in result: return f"错误: {result['error']}" state = result.get("state", "unknown") attrs = result.get("attributes", {}) name = attrs.get("friendly_name", entity_id) lines = [f"{name}: {state}"] for k, v in attrs.items(): if k != "friendly_name": lines.append(f" {k}: {v}") return "\n".join(lines) @mcp.tool() def call_service(domain: str, service: str, entity_id: str, extra_data: str = "") -> str: """ 调用 Home Assistant 服务来控制设备。 参数: - domain: 服务域，如 light, switch, climate, cover, media_player, fan, scene, script, automation - service: 服务名，如 turn_on, turn_off, toggle, set_temperature, set_hvac_mode - entity_id: 设备ID，如 light.living_room, switch.plug_1 - extra_data: 可选的额外参数JSON字符串，如 {"brightness": 128, "color_temp": 300} 常用示例: - 开灯: domain="light", service="turn_on", entity_id="light.living_room" - 关灯: domain="light", service="turn_off", entity_id="light.living_room" - 调亮度: domain="light", service="turn_on", entity_id="light.living_room", extra_data='{"brightness_pct": 50}' - 开开关: domain="switch", service="turn_on", entity_id="switch.plug_1" - 设空调温度: domain="climate", service="set_temperature", entity_id="climate.ac", extra_data='{"temperature": 26}' - 设空调模式: domain="climate", service="set_hvac_mode", entity_id="climate.ac", extra_data='{"hvac_mode": "cool"}' - 执行场景: domain="scene", service="turn_on", entity_id="scene.movie_time" """ payload = {"entity_id": entity_id} if extra_data: try: payload.update(json.loads(extra_data)) except json.JSONDecodeError: return f"错误: extra_data 不是有效的 JSON: {extra_data}" result = ha_request("POST", f"services/{domain}/{service}", payload) if isinstance(result, dict) and "error" in result: return f"错误: {result['error']}" return f"已执行: {domain}.{service} -> {entity_id}" @mcp.tool() def control_light(entity_id: str, action: str, brightness_pct: int = -1, color_temp: int = -1) -> str: """ 控制灯光的快捷工具。 参数: - entity_id: 灯的ID，如 light.living_room, light.bedroom - action: 动作，可选 on/off/toggle - brightness_pct: 亮度百分比 0-100，-1表示不设置 - color_temp: 色温(mireds)，-1表示不设置。越小越冷白(153)，越大越暖黄(500) """ data = {"entity_id": entity_id} if brightness_pct >= 0: data["brightness_pct"] = max(0, min(100, brightness_pct)) if color_temp >= 0: data["color_temp"] = color_temp service = {"on": "turn_on", "off": "turn_off", "toggle": "toggle"}.get(action, "turn_on") result = ha_request("POST", f"services/light/{service}", data) if isinstance(result, dict) and "error" in result: return f"错误: {result['error']}" return f"灯光已{'打开' if action == 'on' else '关闭' if action == 'off' else '切换'}: {entity_id}" @mcp.tool() def control_climate(entity_id: str, action: str = "", temperature: float = -1, hvac_mode: str = "") -> str: """ 控制空调/温控设备的快捷工具。 参数: - entity_id: 空调ID，如 climate.living_room_ac - action: on/off，开关空调 - temperature: 目标温度，-1表示不设置 - hvac_mode: 模式，可选 cool/heat/auto/dry/fan_only，空字符串表示不设置 """ if action == "off": result = ha_request("POST", "services/climate/turn_off", {"entity_id": entity_id}) if isinstance(result, dict) and "error" in result: return f"错误: {result['error']}" return f"空调已关闭: {entity_id}" if action == "on": ha_request("POST", "services/climate/turn_on", {"entity_id": entity_id}) if temperature > 0: data = {"entity_id": entity_id, "temperature": temperature} ha_request("POST", "services/climate/set_temperature", data) if hvac_mode: data = {"entity_id": entity_id, "hvac_mode": hvac_mode} ha_request("POST", "services/climate/set_hvac_mode", data) parts = [] if action == "on": parts.append("已开启") if temperature > 0: parts.append(f"温度设为{temperature}°C") if hvac_mode: parts.append(f"模式设为{hvac_mode}") return f"空调 {entity_id}: {', '.join(parts)}" if parts else f"空调 {entity_id}: 无操作" @mcp.tool() def control_switch(entity_id: str, action: str) -> str: """ 控制开关设备（插座、风扇开关等）。 参数: - entity_id: 开关ID，如 switch.bedroom_fan, switch.plug_1 - action: on/off/toggle """ service = {"on": "turn_on", "off": "turn_off", "toggle": "toggle"}.get(action, "toggle") result = ha_request("POST", f"services/switch/{service}", {"entity_id": entity_id}) if isinstance(result, dict) and "error" in result: return f"错误: {result['error']}" return f"开关已{'打开' if action == 'on' else '关闭' if action == 'off' else '切换'}: {entity_id}" @mcp.tool() def run_scene(entity_id: str) -> str: """ 执行一个场景。 参数: - entity_id: 场景ID，如 scene.movie_time, scene.good_night, scene.leave_home """ result = ha_request("POST", "services/scene/turn_on", {"entity_id": entity_id}) if isinstance(result, dict) and "error" in result: return f"错误: {result['error']}" return f"场景已执行: {entity_id}" @mcp.tool() def run_script(entity_id: str) -> str: """ 执行一个脚本/自动化。 参数: - entity_id: 脚本ID，如 script.morning_routine, automation.night_mode """ domain = entity_id.split(".")[0] if "." in entity_id else "script" service = "turn_on" if domain == "script" else "trigger" result = ha_request("POST", f"services/{domain}/{service}", {"entity_id": entity_id}) if isinstance(result, dict) and "error" in result: return f"错误: {result['error']}" return f"已执行: {entity_id}" if __name__ == "__main__": mcp.run(transport="stdio")