multi-tenant phase 3: per-user Home Assistant + enrolled services

- 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
This commit is contained in:
daniel
2026-05-07 09:10:39 -05:00
parent 4f551ce069
commit 92c87222e8
13 changed files with 369 additions and 54 deletions

View File

@@ -1,22 +1,24 @@
"""Home Assistant tool — control and query HA entities via REST API."""
"""Home Assistant tool — control and query HA entities via REST API (per-user config)."""
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"
)
# 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}"
url = f"{_ha_url}{path}"
headers = {
"Authorization": f"Bearer {HA_TOKEN}",
"Authorization": f"Bearer {_ha_token}",
"Content-Type": "application/json",
}
data = json.dumps(body).encode() if body else None
@@ -39,7 +41,6 @@ def home_assistant(action: str, entity_id: str = "", domain: str = "", service:
- "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"}
@@ -58,6 +59,12 @@ def home_assistant(action: str, entity_id: str = "", domain: str = "", service:
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"
@@ -69,11 +76,9 @@ def home_assistant(action: str, entity_id: str = "", domain: str = "", service:
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)