Gabriel Vidal
e62df7e6c3
Custom component under integrations/home-assistant/: registers the
/api/webhook/ai_agent webhook, renders notify events with the per-type
channel map, and renders asks on a dedicated 'Claude · Ask' channel with
the '. .. .._' vibration pattern as persistent tappable notifications —
the tapped answer is POSTed back to /api/ask/{id}/answer and the
notification cleared. install.sh copies it into the HA config and wires
configuration.yaml.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
214 lines
8.3 KiB
Python
214 lines
8.3 KiB
Python
"""AI Agent notifications — Home Assistant side of the ai-agent notify hub.
|
|
|
|
The ai-agent backend (the homelab's notification source of truth) forwards
|
|
every push and ask to a webhook this integration registers. This component:
|
|
|
|
* renders **notify** events as mobile notifications on the configured
|
|
``notify.<target>`` service, applying the per-type channel / icon / color /
|
|
vibration defaults (payload fields win);
|
|
* renders **ask** events as persistent, tappable notifications on a dedicated
|
|
``Claude · Ask`` channel with the ". .. .._" vibration pattern, then listens
|
|
for ``mobile_app_notification_action`` and POSTs the tapped answer back to
|
|
the ai-agent backend (``/api/ask/{id}/answer``), clearing the notification.
|
|
|
|
Configuration (``configuration.yaml``)::
|
|
|
|
ai_agent:
|
|
webhook_id: ai_agent # -> /api/webhook/ai_agent
|
|
notify_target: mobile_app_pixel_9 # notify.<target>
|
|
callback_url: http://127.0.0.1:8096 # ai-agent backend (host port)
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
|
|
import voluptuous as vol
|
|
from aiohttp.web import Response
|
|
|
|
from homeassistant.components import webhook
|
|
from homeassistant.core import Event, HomeAssistant
|
|
from homeassistant.helpers import config_validation as cv
|
|
from homeassistant.helpers.aiohttp_client import async_get_clientsession
|
|
from homeassistant.helpers.typing import ConfigType
|
|
|
|
from .const import (ASK_CHANNEL, ASK_COLOR, ASK_ICON, ASK_IMPORTANCE,
|
|
ASK_VIBRATION, CONF_CALLBACK_URL, CONF_NOTIFY_TARGET,
|
|
CONF_WEBHOOK_ID, DEFAULT_CALLBACK_URL,
|
|
DEFAULT_NOTIFY_TARGET, DEFAULT_STYLE, DEFAULT_WEBHOOK_ID,
|
|
DOMAIN, TYPE_STYLES)
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
CONFIG_SCHEMA = vol.Schema(
|
|
{
|
|
DOMAIN: vol.Schema(
|
|
{
|
|
vol.Optional(CONF_WEBHOOK_ID,
|
|
default=DEFAULT_WEBHOOK_ID): cv.string,
|
|
vol.Optional(CONF_NOTIFY_TARGET,
|
|
default=DEFAULT_NOTIFY_TARGET): cv.string,
|
|
vol.Optional(CONF_CALLBACK_URL,
|
|
default=DEFAULT_CALLBACK_URL): cv.string,
|
|
}
|
|
)
|
|
},
|
|
extra=vol.ALLOW_EXTRA,
|
|
)
|
|
|
|
MOBILE_ACTION_EVENT = "mobile_app_notification_action"
|
|
|
|
|
|
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
|
|
conf = config.get(DOMAIN) or {}
|
|
webhook_id = conf.get(CONF_WEBHOOK_ID, DEFAULT_WEBHOOK_ID)
|
|
target = conf.get(CONF_NOTIFY_TARGET, DEFAULT_NOTIFY_TARGET)
|
|
callback_url = (conf.get(CONF_CALLBACK_URL,
|
|
DEFAULT_CALLBACK_URL)).rstrip("/")
|
|
|
|
hass.data[DOMAIN] = {"target": target, "callback_url": callback_url}
|
|
|
|
webhook.async_register(hass, DOMAIN, "AI Agent notifications", webhook_id,
|
|
_handle_webhook, local_only=True,
|
|
allowed_methods=["POST"])
|
|
hass.bus.async_listen(MOBILE_ACTION_EVENT, _make_action_listener(hass))
|
|
_LOGGER.info("ai_agent ready: webhook_id=%s target=%s callback=%s",
|
|
webhook_id, target, callback_url)
|
|
return True
|
|
|
|
|
|
async def _handle_webhook(hass: HomeAssistant, webhook_id: str, request):
|
|
"""Dispatch an ai-agent webhook payload to the phone."""
|
|
try:
|
|
data = await request.json()
|
|
except ValueError:
|
|
return Response(status=400, text="invalid json")
|
|
event = (data.get("event") or "").strip()
|
|
try:
|
|
if event == "notify":
|
|
await _send_notify(hass, data)
|
|
elif event == "ask":
|
|
await _send_ask(hass, data)
|
|
else:
|
|
return Response(status=400, text=f"unknown event {event!r}")
|
|
except Exception: # never let a bad payload 500 the webhook endpoint
|
|
_LOGGER.exception("ai_agent: failed to handle %s payload", event)
|
|
return Response(status=500, text="failed")
|
|
return Response(status=200, text="ok")
|
|
|
|
|
|
def _style_for(data: dict) -> tuple[str, str, str, str, str]:
|
|
"""(icon, color, channel, importance, vibration) for the payload's type,
|
|
each overridden by an explicit payload field when present."""
|
|
icon, color, channel, importance, vibration = TYPE_STYLES.get(
|
|
(data.get("type") or "").strip(), DEFAULT_STYLE)
|
|
return (data.get("icon") or icon,
|
|
data.get("color") or color,
|
|
data.get("channel") or channel,
|
|
data.get("importance") or importance,
|
|
data.get("vibrationPattern") or vibration)
|
|
|
|
|
|
def _base_mobile_data(data: dict) -> dict:
|
|
"""The mobile-app `data` block shared by pushes and asks."""
|
|
icon, color, channel, importance, vibration = _style_for(data)
|
|
out = {
|
|
"notification_icon": icon,
|
|
"color": color,
|
|
"ledColor": color,
|
|
"channel": channel,
|
|
"importance": importance,
|
|
"vibrationPattern": vibration,
|
|
}
|
|
url = (data.get("url") or "").strip()
|
|
if url:
|
|
out["clickAction"] = url
|
|
out["url"] = url
|
|
image = (data.get("image") or "").strip()
|
|
if image:
|
|
out["image"] = image
|
|
return out
|
|
|
|
|
|
async def _send_notify(hass: HomeAssistant, data: dict) -> None:
|
|
message = data.get("message") or ""
|
|
url = (data.get("url") or "").strip()
|
|
if url and url not in message:
|
|
# clickAction alone is invisible — surface the URL in the body too.
|
|
message = f"{message}\n{url}"
|
|
mobile = _base_mobile_data(data)
|
|
if data.get("tag"):
|
|
mobile["tag"] = data["tag"]
|
|
if data.get("persistent"):
|
|
mobile["persistent"] = True
|
|
if data.get("actions"):
|
|
mobile["actions"] = data["actions"]
|
|
await hass.services.async_call(
|
|
"notify", hass.data[DOMAIN]["target"],
|
|
{"title": data.get("title") or "Claude Code", "message": message,
|
|
"data": mobile}, blocking=True)
|
|
|
|
|
|
async def _send_ask(hass: HomeAssistant, data: dict) -> None:
|
|
ask_id = data.get("id") or ""
|
|
options = [str(o) for o in (data.get("options") or [])][:3]
|
|
if not ask_id or len(options) < 2:
|
|
raise ValueError("ask needs an id and 2-3 options")
|
|
mobile = _base_mobile_data(data)
|
|
# Asks always ride their own channel + morse vibration (". .. .._"),
|
|
# regardless of the payload's type styling.
|
|
mobile.update({
|
|
"notification_icon": data.get("icon") or ASK_ICON,
|
|
"color": data.get("color") or ASK_COLOR,
|
|
"ledColor": data.get("color") or ASK_COLOR,
|
|
"channel": ASK_CHANNEL,
|
|
"importance": ASK_IMPORTANCE,
|
|
"vibrationPattern": ASK_VIBRATION,
|
|
"tag": ask_id,
|
|
"persistent": True,
|
|
"sticky": True,
|
|
"actions": [{"action": f"{ask_id}__{i}", "title": t}
|
|
for i, t in enumerate(options)],
|
|
})
|
|
await hass.services.async_call(
|
|
"notify", hass.data[DOMAIN]["target"],
|
|
{"title": data.get("title") or "Claude needs a choice",
|
|
"message": data.get("question") or "", "data": mobile},
|
|
blocking=True)
|
|
|
|
|
|
def _make_action_listener(hass: HomeAssistant):
|
|
async def _on_action(event: Event) -> None:
|
|
action = str(event.data.get("action") or "")
|
|
# Ask buttons are "<ask_id>__<index>" with ask ids minted as "ask_…".
|
|
if not action.startswith("ask_") or "__" not in action:
|
|
return
|
|
ask_id, _, idx_s = action.rpartition("__")
|
|
try:
|
|
index = int(idx_s)
|
|
except ValueError:
|
|
return
|
|
callback = hass.data[DOMAIN]["callback_url"]
|
|
session = async_get_clientsession(hass)
|
|
try:
|
|
async with session.post(f"{callback}/api/ask/{ask_id}/answer",
|
|
json={"index": index}, timeout=10) as resp:
|
|
if resp.status == 404:
|
|
# Not one of ours (e.g. a legacy ask.sh nonce) — leave it
|
|
# to whatever host-side listener minted it.
|
|
return
|
|
resp.raise_for_status()
|
|
except Exception:
|
|
_LOGGER.exception("ai_agent: answer callback failed for %s",
|
|
ask_id)
|
|
return
|
|
# Answer recorded — clear the persistent notification off the phone.
|
|
try:
|
|
await hass.services.async_call(
|
|
"notify", hass.data[DOMAIN]["target"],
|
|
{"message": "clear_notification", "data": {"tag": ask_id}},
|
|
blocking=False)
|
|
except Exception:
|
|
_LOGGER.debug("ai_agent: could not clear notification %s", ask_id)
|
|
|
|
return _on_action
|