92c87222e8
- tools/home_assistant.py: remove hardcoded URL/token; read from per-user
config injected via set_ha_config() at invocation time; return helpful
enrollment prompt when HA not configured
- main.py: inject HA config from user_profile.services at startup; add
manage_service tool (enroll/remove/list) that persists to DynamoDB;
show enrolled services in user context; add USERS_TABLE_NAME env var
- agent-runner/handler.py: pass services dict from DDB user record in
user_profile payload; initialize services={} for new users
- cdk/lib/agent-claw-stack.ts: grant usersTable read/write to runtime1Role
so manage_service tool can update user records
- agentclaw/agentcore/agentcore.json: add USERS_TABLE_NAME env var
97 lines
4.4 KiB
Python
97 lines
4.4 KiB
Python
"""Home Assistant tool — control and query HA entities via REST API (per-user config)."""
|
|
import json
|
|
import urllib.request
|
|
import urllib.error
|
|
from strands import tool
|
|
|
|
# Per-invocation config — set by main.py before agent runs
|
|
_ha_url: str = ''
|
|
_ha_token: str = ''
|
|
|
|
|
|
def set_ha_config(url: str, token: str) -> None:
|
|
global _ha_url, _ha_token
|
|
_ha_url = url
|
|
_ha_token = token
|
|
|
|
|
|
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).
|
|
|
|
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 not _ha_url or not _ha_token:
|
|
return ("Home Assistant is not configured for your account. "
|
|
"Use the manage_service tool to enroll it: "
|
|
"manage_service(action='enroll', service='home_assistant', "
|
|
"config={'url': 'https://your-ha-url', 'token': 'your-long-lived-token'})")
|
|
|
|
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):
|
|
prefix = entity_id or domain
|
|
if prefix:
|
|
result = [s for s in result if s.get("entity_id", "").startswith(prefix)]
|
|
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'."
|