Gabriel Vidal
d4b2efb077
The companion app turns an absolute https `clickAction` into a bare ACTION_VIEW intent, which Android 12+ hands to the default browser — a Chrome-minted WebAPK never has verified links for its host, so every notify/ask tap opened a browser tab next to the installed app. New optional `pwa_package` / `pwa_origin` config: a notification URL on the PWA's origin is sent as an `intent:` URI pinned to the WebAPK's package (path + query kept), which the app launches by package. Other hosts stay plain links; empty package = previous behaviour. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
303 lines
13 KiB
Python
303 lines
13 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;
|
|
* optionally hands each **notify** to a *voice* leg as well — any HA service
|
|
named by ``phone_service`` (the homelab rings its AI desk phone that way),
|
|
so one webhook covers both the screen and the speaker;
|
|
* fires ``ai_agent_notify`` / ``ai_agent_ask`` on the HA event bus, so
|
|
automations can react to agent activity without touching the webhook.
|
|
|
|
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)
|
|
phone_service: script.ai_agent_phone_notify # optional voice leg
|
|
pwa_package: org.chromium.webapk.<hash>_v2 # optional: open the PWA
|
|
pwa_origin: https://ai-agent.lab.gabvdl.xyz # for URLs on this origin
|
|
|
|
Why ``pwa_package``: the companion app turns an absolute http(s) ``clickAction``
|
|
into a bare ``ACTION_VIEW`` intent, and since Android 12 that lands in the
|
|
default browser unless an app has *verified* links for the host — which a
|
|
Chrome-minted WebAPK never has. An ``intent:`` URI naming the WebAPK's package
|
|
resolves straight to the installed app, path and query intact.
|
|
"""
|
|
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_PHONE_SERVICE, CONF_PWA_ORIGIN, CONF_PWA_PACKAGE,
|
|
CONF_WEBHOOK_ID, DEFAULT_CALLBACK_URL,
|
|
DEFAULT_NOTIFY_TARGET, DEFAULT_PHONE_SERVICE,
|
|
DEFAULT_PWA_ORIGIN, DEFAULT_PWA_PACKAGE, DEFAULT_STYLE,
|
|
DEFAULT_WEBHOOK_ID, DOMAIN, EVENT_ASK, EVENT_NOTIFY,
|
|
PHONE_FIELDS, 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,
|
|
vol.Optional(CONF_PHONE_SERVICE,
|
|
default=DEFAULT_PHONE_SERVICE): cv.string,
|
|
vol.Optional(CONF_PWA_PACKAGE,
|
|
default=DEFAULT_PWA_PACKAGE): cv.string,
|
|
vol.Optional(CONF_PWA_ORIGIN,
|
|
default=DEFAULT_PWA_ORIGIN): 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("/")
|
|
phone_service = (conf.get(CONF_PHONE_SERVICE,
|
|
DEFAULT_PHONE_SERVICE) or "").strip()
|
|
pwa_package = (conf.get(CONF_PWA_PACKAGE,
|
|
DEFAULT_PWA_PACKAGE) or "").strip()
|
|
pwa_origin = (conf.get(CONF_PWA_ORIGIN,
|
|
DEFAULT_PWA_ORIGIN) or "").strip().rstrip("/")
|
|
|
|
hass.data[DOMAIN] = {"target": target, "callback_url": callback_url,
|
|
"phone_service": phone_service,
|
|
"pwa_package": pwa_package, "pwa_origin": pwa_origin}
|
|
|
|
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 phone=%s "
|
|
"pwa=%s", webhook_id, target, callback_url,
|
|
phone_service or "-", pwa_package or "-")
|
|
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 pwa_click_action(url: str, package: str, origin: str) -> str:
|
|
"""The companion-app ``clickAction`` for ``url``.
|
|
|
|
A URL on the installed PWA's ``origin`` becomes an Android ``intent:`` URI
|
|
pinned to the WebAPK's ``package`` (path + query kept), which the companion
|
|
app launches by package — so the tap opens the app, not a browser tab. Any
|
|
other URL, or no package configured, stays a plain link. A URL with its own
|
|
``#fragment`` is left alone: the intent syntax owns the fragment.
|
|
"""
|
|
if not package or not origin:
|
|
return url
|
|
if "#" in url:
|
|
return url
|
|
prefix = origin.rstrip("/") + "/"
|
|
if url != origin.rstrip("/") and not url.startswith(prefix):
|
|
return url
|
|
scheme, _, rest = url.partition("://")
|
|
if not rest:
|
|
return url
|
|
return f"intent://{rest}#Intent;scheme={scheme};package={package};end"
|
|
|
|
|
|
def _base_mobile_data(hass: HomeAssistant, 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:
|
|
conf = hass.data.get(DOMAIN) or {}
|
|
out["clickAction"] = pwa_click_action(
|
|
url, conf.get("pwa_package") or "", conf.get("pwa_origin") or "")
|
|
out["url"] = url
|
|
image = (data.get("image") or "").strip()
|
|
if image:
|
|
out["image"] = image
|
|
return out
|
|
|
|
|
|
async def _ring_phone(hass: HomeAssistant, data: dict) -> None:
|
|
"""Hand the notify to the voice leg — the AI desk phone, via HA.
|
|
|
|
`phone_service` is a plain ``<domain>.<service>`` (the homelab points it at
|
|
``script.ai_agent_phone_notify``, see the phone service's HA package), so
|
|
*when* and *how* the phone rings stays editable from Home Assistant. Called
|
|
non-blocking and best-effort: the screen notification is the contract here,
|
|
the call is a bonus, and the ai-agent hub forwards webhooks on a short
|
|
timeout that a ringing phone must never eat into.
|
|
"""
|
|
service = hass.data[DOMAIN].get("phone_service") or ""
|
|
domain, _, name = service.partition(".")
|
|
if not domain or not name:
|
|
if service:
|
|
_LOGGER.warning("ai_agent: phone_service %r is not "
|
|
"'<domain>.<service>' — skipping", service)
|
|
return
|
|
payload = {k: (data.get(k) or "") for k in PHONE_FIELDS}
|
|
try:
|
|
await hass.services.async_call(domain, name, payload, blocking=False)
|
|
except Exception:
|
|
_LOGGER.exception("ai_agent: phone service %s failed", service)
|
|
|
|
|
|
async def _send_notify(hass: HomeAssistant, data: dict) -> None:
|
|
hass.bus.async_fire(EVENT_NOTIFY, data)
|
|
await _ring_phone(hass, data)
|
|
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(hass, 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")
|
|
hass.bus.async_fire(EVENT_ASK, data)
|
|
# No phone leg for asks on purpose: answering one means tapping a button,
|
|
# which a handset can't offer.
|
|
mobile = _base_mobile_data(hass, 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_…";
|
|
# a plain push's extra button (notify send --action-cmd) is
|
|
# "<ntf_id>__<index>" and answers on /api/notify/{id}/action.
|
|
if "__" not in action or not action.startswith(("ask_", "ntf_")):
|
|
return
|
|
ask_id, _, idx_s = action.rpartition("__")
|
|
try:
|
|
index = int(idx_s)
|
|
except ValueError:
|
|
return
|
|
route = ("/api/ask/%s/answer" if action.startswith("ask_")
|
|
else "/api/notify/%s/action") % ask_id
|
|
callback = hass.data[DOMAIN]["callback_url"]
|
|
session = async_get_clientsession(hass)
|
|
try:
|
|
async with session.post(f"{callback}{route}",
|
|
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
|