92 lines
4.3 KiB
Python
92 lines
4.3 KiB
Python
"""Home Assistant tool — control and query HA entities via REST API."""
|
|
import json
|
|
import os
|
|
import urllib.request
|
|
import urllib.error
|
|
from strands import tool
|
|
|
|
HA_URL = "https://homeassistant.home.everyonce.com"
|
|
# Token stored in workspace or env; fallback to hardcoded for AgentCore runtime
|
|
HA_TOKEN = os.environ.get(
|
|
"HA_TOKEN",
|
|
"eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJpc3MiOiJlMDExN2YwNzhlM2Q0NjViODJhNjJiZWFiMzI1ZWU4MiIsImlhdCI6MTc3MTM1MjU0MiwiZXhwIjoyMDg2NzEyNTQyfQ.UySLD6JV4e_bdd1nQjdbZcimdCD6B3kBGDftcRz1H6Q"
|
|
)
|
|
|
|
|
|
def _ha_request(method: str, path: str, body: dict | None = None) -> dict | list:
|
|
url = f"{HA_URL}{path}"
|
|
headers = {
|
|
"Authorization": f"Bearer {HA_TOKEN}",
|
|
"Content-Type": "application/json",
|
|
}
|
|
data = json.dumps(body).encode() if body else None
|
|
req = urllib.request.Request(url, data=data, 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:
|
|
return {"error": f"HTTP {e.code}: {e.reason}", "body": e.read().decode()[:500]}
|
|
except Exception as e:
|
|
return {"error": str(e)}
|
|
|
|
|
|
@tool
|
|
def home_assistant(action: str, entity_id: str = "", domain: str = "", service: str = "",
|
|
service_data: dict | None = None) -> str:
|
|
"""Control and query your Home Assistant smart home.
|
|
|
|
Actions:
|
|
- "get_state": Get the current state of a specific entity (requires entity_id).
|
|
- "list_states": List all entity states (optionally filter by domain prefix like 'light', 'switch', 'climate', 'sensor').
|
|
- "call_service": Call a HA service (requires domain, service, and optional service_data with entity_id).
|
|
- "get_history": Not yet implemented.
|
|
|
|
Common service examples:
|
|
- Turn light on: domain="light", service="turn_on", service_data={"entity_id": "light.living_room"}
|
|
- Turn light off: domain="light", service="turn_off", service_data={"entity_id": "light.living_room"}
|
|
- Set brightness: domain="light", service="turn_on", service_data={"entity_id": "light.x", "brightness_pct": 50}
|
|
- Lock door: domain="lock", service="lock", service_data={"entity_id": "lock.front_door"}
|
|
- Set thermostat: domain="climate", service="set_temperature", service_data={"entity_id": "climate.x", "temperature": 72}
|
|
|
|
Args:
|
|
action: One of "get_state", "list_states", "call_service".
|
|
entity_id: Entity ID for get_state (e.g. "light.living_room").
|
|
domain: Service domain for call_service (e.g. "light", "switch", "lock", "climate").
|
|
service: Service name for call_service (e.g. "turn_on", "turn_off", "lock").
|
|
service_data: Dict of extra params for call_service (e.g. {"entity_id": "light.x", "brightness_pct": 80}).
|
|
|
|
Returns:
|
|
JSON string with the result.
|
|
"""
|
|
if action == "get_state":
|
|
if not entity_id:
|
|
return "entity_id is required for get_state"
|
|
result = _ha_request("GET", f"/api/states/{entity_id}")
|
|
if isinstance(result, dict) and "error" not in result:
|
|
return f"{entity_id}: {result.get('state')} (attrs: {json.dumps(result.get('attributes', {}))[:300]})"
|
|
return json.dumps(result)
|
|
|
|
elif action == "list_states":
|
|
result = _ha_request("GET", "/api/states")
|
|
if isinstance(result, list):
|
|
# Filter by domain prefix if entity_id used as filter
|
|
prefix = entity_id or domain
|
|
if prefix:
|
|
result = [s for s in result if s.get("entity_id", "").startswith(prefix)]
|
|
# Return concise summary
|
|
lines = [f"{s['entity_id']}: {s['state']}" for s in result[:50]]
|
|
return "\n".join(lines) + (f"\n... ({len(result)} total)" if len(result) > 50 else "")
|
|
return json.dumps(result)
|
|
|
|
elif action == "call_service":
|
|
if not domain or not service:
|
|
return "domain and service are required for call_service"
|
|
body = service_data or {}
|
|
if entity_id and "entity_id" not in body:
|
|
body["entity_id"] = entity_id
|
|
result = _ha_request("POST", f"/api/services/{domain}/{service}", body)
|
|
return f"Service {domain}.{service} called successfully" if isinstance(result, list) else json.dumps(result)
|
|
|
|
else:
|
|
return f"Unknown action: {action}. Use 'get_state', 'list_states', or 'call_service'."
|