Coverage for custom_components/supernotify/llm.py: 97%
286 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-11-08 17:42 +0000
1"""LLM tools for Assist conversation agents and Home Assistant's MCP server - beta.
3Home Assistant's `llm` integration merges the tools from every integration's `llm` platform into
4its built-in Assist API, which is also what the MCP server offers by default. Nothing here is
5offered until switched on in the Supernotify options, where action tools (send, snooze) and
6diagnostic tools (recent notifications, dry run, snoozes, documentation) are switched on separately.
8The tools deliberately leave out Supernotify's more open-ended fields - `custom_target` (any
9e-mail address, phone number etc), `actions`, and media URLs - so an agent can only reach the
10recipients, deliveries and scenarios already configured.
11"""
13from __future__ import annotations
15import datetime as dt
16import logging
17import re
18from dataclasses import dataclass
19from typing import TYPE_CHECKING, Any, override
21import aiohttp
22from homeassistant.components.llm import LLMTools # type: ignore[unresolved-import,unused-ignore] # HA < 2026.x on py3.13
23from homeassistant.core import HomeAssistant, callback
24from homeassistant.helpers.aiohttp_client import async_get_clientsession
25from homeassistant.helpers.llm import LLM_API_ASSIST, LLMContext, Tool, ToolInput
26from homeassistant.util import dt as dt_util
27from homeassistant.util.hass_dict import HassKey
29from . import DOMAIN
30from .archive import summarize_notification
31from .compat import vol
32from .const import (
33 ATTR_DELIVERY,
34 ATTR_DRY_RUN,
35 ATTR_PRIORITY,
36 ATTR_SCENARIOS_APPLY,
37 CONF_LLM_ACTION_TOOLS,
38 CONF_LLM_DIAGNOSTIC_TOOLS,
39 CONF_LLM_TOOLS,
40 DRY_RUN_SIMULATE,
41 PRIORITY_VALUES,
42)
43from .model import CommandType, GlobalTargetType, QualifiedTargetType, RecipientType, TargetType
44from .snoozer import SNOOZE_SCOPES, snooze_name_error
46if TYPE_CHECKING:
47 from homeassistant.util.json import JsonObjectType
49 from .engine import SupernotifyEngine
50 from .people import Recipient
52_LOGGER = logging.getLogger(__name__)
54EVERYONE = "everyone"
55MAX_HOURS = 7 * 24
56MAX_NOTIFICATIONS = 50
58# The documentation site publishes every page as one markdown file for LLMs, plus an index with links
59DOCS_SITE = "https://supernotify.rhizomatics.org.uk/"
60DOCS_URL = f"{DOCS_SITE}llms-full.txt"
61DOCS_INDEX_URL = f"{DOCS_SITE}llms.txt"
62DOCS_CACHE_TIME = dt.timedelta(days=1)
63# the line under each page title giving its link, from docgen/llms_preprocess.py
64DOCS_SOURCE_PREFIX = "Source: "
65DOCS_SECTIONS_RETURNED = 3
66DOCS_SECTION_MAX_CHARS = 4000
67DOCS_STOP_WORDS = frozenset([
68 "a",
69 "an",
70 "and",
71 "are",
72 "can",
73 "do",
74 "does",
75 "for",
76 "from",
77 "how",
78 "i",
79 "in",
80 "is",
81 "it",
82 "my",
83 "of",
84 "on",
85 "or",
86 "the",
87 "to",
88 "use",
89 "using",
90 "what",
91 "when",
92 "where",
93 "which",
94 "who",
95 "why",
96 "with",
97 # words in almost every page of these docs
98 "notification",
99 "notifications",
100 "notify",
101 "supernotify",
102 "send",
103])
106@dataclass
107class DocsPage:
108 title: str
109 url: str | None
110 text: str
113@dataclass
114class _DocsCache:
115 fetched: dt.datetime
116 pages: list[DocsPage]
119DOCS_CACHE: HassKey[_DocsCache] = HassKey(f"{DOMAIN}_llm_docs")
121SNOOZE_ACTIONS: dict[str, CommandType | None] = {
122 "snooze": CommandType.SNOOZE,
123 "silence": CommandType.SILENCE,
124 "unsnooze": CommandType.NORMAL,
125 "clear_all": None,
126}
129@callback
130def async_get_tools(hass: HomeAssistant, llm_context: LLMContext, api_id: str) -> LLMTools | None:
131 """Return the Supernotify tools switched on in the options, rebuilt each time so the
132 choices offered for deliveries, scenarios and recipients are always the current ones."""
133 if api_id != LLM_API_ASSIST:
134 return None
135 entries = hass.config_entries.async_loaded_entries(DOMAIN)
136 if not entries:
137 return None
138 engine: SupernotifyEngine = entries[0].runtime_data
139 options: dict[str, Any] = entries[0].options.get(CONF_LLM_TOOLS, {})
140 tools: list[Tool] = []
141 if options.get(CONF_LLM_ACTION_TOOLS):
142 tools.extend([NotifyTool(engine), SnoozeTool(engine)])
143 if options.get(CONF_LLM_DIAGNOSTIC_TOOLS):
144 tools.extend([RecentNotificationsTool(engine), DryRunTool(engine), SnoozesTool(engine), HelpTool(engine)])
145 if not tools:
146 return None
147 return LLMTools(tools=tools, prompt=_prompt(engine))
150def _prompt(engine: SupernotifyEngine) -> str:
151 context = engine.context
152 lines = ["Supernotify sends notifications to the household by phone, e-mail, speakers and other ways."]
153 if deliveries := [_labelled(name, d.alias) for name, d in context.delivery_registry.choosable_deliveries.items()]:
154 lines.append(f"Deliveries: {', '.join(deliveries)}.")
155 if scenarios := [_labelled(name, s.alias) for name, s in context.scenario_registry.scenarios.items()]:
156 lines.append(f"Scenarios: {', '.join(scenarios)}.")
157 if recipients := _recipient_names(engine):
158 lines.append(f"Recipients: {', '.join(recipients)}.")
159 return "\n".join(lines)
162def _labelled(name: str, alias: str | None) -> str:
163 return f"{name} ({alias})" if alias and alias != name else name
166def _display_name(recipient: Recipient) -> str:
167 return recipient.alias or recipient.name
170def _recipient_names(engine: SupernotifyEngine) -> list[str]:
171 return sorted(_display_name(r) for r in engine.context.people_registry.enabled_recipients())
174def _person_id(engine: SupernotifyEngine, name: str) -> str | None:
175 return engine.context.people_registry.person_id_for_name(name)
178def _notification_fields(engine: SupernotifyEngine, message_required: bool) -> dict[vol.Marker, Any]:
179 """The notification fields an agent may use - never custom_target, actions or media URLs"""
180 context = engine.context
181 fields: dict[vol.Marker, Any] = {
182 (vol.Required if message_required else vol.Optional)("message", description="The notification text"): str,
183 vol.Optional("title", description="A short title, used by deliveries that show one"): str,
184 vol.Optional("priority", description="How urgent the notification is, which decides which deliveries are used"): vol.In(
185 list(PRIORITY_VALUES)
186 ),
187 }
188 if recipients := _recipient_names(engine):
189 fields[vol.Optional("recipients", description="Who to notify. Leave out to use the usual recipients")] = [
190 vol.In(recipients)
191 ]
192 if deliveries := list(context.delivery_registry.choosable_deliveries):
193 fields[vol.Optional("deliveries", description="Use only these deliveries. Leave out to choose automatically")] = [
194 vol.In(deliveries)
195 ]
196 if scenarios := list(context.scenario_registry.scenarios):
197 fields[vol.Optional("scenarios", description="Scenarios to apply, as if their conditions were met")] = [
198 vol.In(scenarios)
199 ]
200 return fields
203def _notification_call(engine: SupernotifyEngine, args: dict[str, Any]) -> tuple[list[str] | None, dict[str, Any], list[str]]:
204 """Turn tool arguments into a target list and action data, plus any recipient names not known"""
205 person_ids: list[str] = []
206 unknown: list[str] = []
207 for name in args.get("recipients", []):
208 if person_id := _person_id(engine, name):
209 person_ids.append(person_id)
210 else:
211 unknown.append(name)
212 data: dict[str, Any] = {}
213 if "priority" in args:
214 data[ATTR_PRIORITY] = args["priority"]
215 if args.get("deliveries"):
216 data[ATTR_DELIVERY] = args["deliveries"]
217 if args.get("scenarios"):
218 data[ATTR_SCENARIOS_APPLY] = args["scenarios"]
219 return person_ids or None, data, unknown
222class SupernotifyTool(Tool):
223 def __init__(self, engine: SupernotifyEngine) -> None:
224 self.engine = engine
227class NotifyTool(SupernotifyTool):
228 name = "supernotify__notify"
229 description = (
230 "Send a notification to people in the household, e.g. 'tell everyone dinner is ready' or "
231 "'send Alice an urgent message'. Supernotify chooses how to reach each person from the priority, "
232 "the active scenarios and who is home, unless deliveries are given."
233 )
235 def __init__(self, engine: SupernotifyEngine) -> None:
236 super().__init__(engine)
237 self.parameters = vol.Schema(_notification_fields(engine, message_required=True))
239 @override
240 async def async_call(self, hass: HomeAssistant, tool_input: ToolInput, llm_context: LLMContext) -> JsonObjectType:
241 args = self.parameters(tool_input.tool_args)
242 target, data, unknown = _notification_call(self.engine, args)
243 if unknown:
244 return {"success": False, "error": f"Unknown recipients: {', '.join(unknown)}"}
245 notification = await self.engine.async_send_message(
246 args["message"], title=args.get("title"), target=target, data=data, context=llm_context.context
247 )
248 if notification is None:
249 return {"success": False, "error": "The notification could not be created"}
250 return {
251 "success": notification.delivered > 0,
252 "result": summarize_notification(self.engine, notification.contents()),
253 }
256class SnoozeTool(SupernotifyTool):
257 name = "supernotify__snooze"
258 description = (
259 "Snooze or silence notifications, or turn them back on, e.g. 'mute the doorbell camera alerts for an hour' "
260 "or 'stop non-urgent notifications until I say'. 'snooze' lasts for the given minutes, 'silence' until "
261 "undone, 'unsnooze' undoes one snooze, and 'clear_all' removes every snooze for everyone."
262 )
264 def __init__(self, engine: SupernotifyEngine) -> None:
265 super().__init__(engine)
266 self.parameters = vol.Schema({
267 vol.Required("action"): vol.In(list(SNOOZE_ACTIONS)),
268 vol.Optional(
269 "scope",
270 default="noncritical",
271 description="What kind of notifications to snooze. 'everything' holds back critical alerts too, "
272 "so use it only when that is asked for",
273 ): vol.In(list(SNOOZE_SCOPES)),
274 vol.Optional(
275 "name",
276 description="The delivery, transport, priority or camera entity_id to snooze, when scope is one of "
277 "those, or for 'tag' a scenario name, or an entity_id, name or alias of the entity a notification is about",
278 ): str,
279 vol.Optional(
280 "recipient",
281 description="Whose notifications to snooze. Leave out for the person asking, or everyone if unknown",
282 ): vol.In([*_recipient_names(engine), EVERYONE]),
283 vol.Optional("minutes", description="How long to snooze for"): vol.All(
284 vol.Coerce(int), vol.Range(min=1, max=MAX_HOURS * 60)
285 ),
286 })
288 @override
289 async def async_call(self, hass: HomeAssistant, tool_input: ToolInput, llm_context: LLMContext) -> JsonObjectType:
290 args = self.parameters(tool_input.tool_args)
291 snoozer = self.engine.context.snoozer
292 cmd: CommandType | None = SNOOZE_ACTIONS[args["action"]]
293 if cmd is None:
294 return {"success": True, "result": {"cleared": self.engine.clear_snoozes()}}
296 scope: TargetType = SNOOZE_SCOPES[args["scope"]]
297 name: str | None = args.get("name")
298 if isinstance(scope, QualifiedTargetType):
299 if error := await self._check_name(scope, name, cmd):
300 return {"success": False, "error": error}
301 else:
302 name = None
304 recipient: str | None = None
305 recipient_name: str | None = args.get("recipient")
306 if recipient_name is None:
307 recipient = self._requesting_person(llm_context)
308 elif recipient_name != EVERYONE:
309 recipient = _person_id(self.engine, recipient_name)
310 recipient_type = RecipientType.USER if recipient else RecipientType.EVERYONE
312 minutes: int | None = args.get("minutes")
313 snooze_for = dt.timedelta(minutes=minutes) if minutes else snoozer.snooze_period
314 snoozer.register_snooze(cmd, scope, name, recipient_type, recipient, snooze_for, reason="Assistant")
315 if cmd == CommandType.NORMAL and isinstance(scope, GlobalTargetType):
316 # turning notifications back on undoes a snooze of everything as well as one of noncritical
317 for other in GlobalTargetType:
318 if other != scope:
319 snoozer.register_snooze(cmd, other, None, recipient_type, recipient, None)
320 snoozes: dict[str, Any] = {"snoozes": self.engine.enquire_snoozes()}
321 return {"success": True, "result": snoozes}
323 async def _check_name(self, scope: QualifiedTargetType, name: str | None, cmd: CommandType) -> str | None:
324 """The same message, in the Home Assistant language, that supernotify.snooze refuses the name with"""
325 if problem := snooze_name_error(self.engine, scope, name, cmd):
326 strings: dict[str, str] = await self.engine.context.hass_api.translations("exceptions")
327 return strings[f"component.{DOMAIN}.exceptions.{problem[0]}.message"].format(**problem[1])
328 return None
330 def _requesting_person(self, llm_context: LLMContext) -> str | None:
331 user_id: str | None = llm_context.context.user_id if llm_context.context else None
332 return self.engine.context.people_registry.person_id_for_user_id(user_id)
335class RecentNotificationsTool(SupernotifyTool):
336 name = "supernotify__recent_notifications"
337 description = (
338 "List recent notifications and what happened to each one: which deliveries sent, were skipped or failed, "
339 "and why. Use this for questions like 'why didn't I get the doorbell alert?' or 'what notifications went "
340 "out today?'. 'occupancy' shows who was home at the time, which decides some deliveries."
341 )
343 def __init__(self, engine: SupernotifyEngine) -> None:
344 super().__init__(engine)
345 fields: dict[vol.Marker, Any] = {
346 vol.Optional("hours", default=24, description="How many hours back to look"): vol.All(
347 vol.Coerce(int), vol.Range(min=1, max=MAX_HOURS)
348 ),
349 vol.Optional("limit", default=10, description="The most notifications to return, newest first"): vol.All(
350 vol.Coerce(int), vol.Range(min=1, max=MAX_NOTIFICATIONS)
351 ),
352 }
353 if recipients := _recipient_names(engine):
354 fields[vol.Optional("recipient", description="Only notifications that reached this person")] = vol.In(recipients)
355 self.parameters = vol.Schema(fields)
357 @override
358 async def async_call(self, hass: HomeAssistant, tool_input: ToolInput, llm_context: LLMContext) -> JsonObjectType:
359 args = self.parameters(tool_input.tool_args)
360 since: dt.datetime = dt_util.now() - dt.timedelta(hours=args["hours"])
361 archive = self.engine.context.archive
362 result: dict[str, Any] = {}
363 if archive.archive_directory and archive.archive_directory.enabled:
364 archived = await archive.archive_directory.list_entries(limit=MAX_NOTIFICATIONS, after=since)
365 else:
366 last = self.engine.last_notification
367 archived = [last.contents()] if last and last.created >= since else []
368 result["note"] = (
369 "The notification archive is off, so only the latest notification since Home Assistant started "
370 "is known. It can be turned on in the Supernotify options."
371 )
372 summaries = [summarize_notification(self.engine, contents) for contents in archived]
373 if recipient := args.get("recipient"):
374 summaries = [s for s in summaries if any(recipient in d.get("recipients", []) for d in s["deliveries"].values())]
375 result["notifications"] = summaries[: args["limit"]]
376 return {"success": True, "result": result}
379class DryRunTool(SupernotifyTool):
380 name = "supernotify__dry_run"
381 description = (
382 "Work out who a notification would reach right now, and by which deliveries, without sending it. "
383 "Use this for questions like 'if the alarm goes off, who gets told and how?'. 'occupancy' shows who is "
384 "home, which decides some deliveries."
385 )
387 def __init__(self, engine: SupernotifyEngine) -> None:
388 super().__init__(engine)
389 self.parameters = vol.Schema(_notification_fields(engine, message_required=False))
391 @override
392 async def async_call(self, hass: HomeAssistant, tool_input: ToolInput, llm_context: LLMContext) -> JsonObjectType:
393 args = self.parameters(tool_input.tool_args)
394 target, data, unknown = _notification_call(self.engine, args)
395 if unknown:
396 return {"success": False, "error": f"Unknown recipients: {', '.join(unknown)}"}
397 data[ATTR_DRY_RUN] = DRY_RUN_SIMULATE
398 notification = await self.engine.async_send_message(
399 args.get("message", ""), title=args.get("title"), target=target, data=data, context=llm_context.context
400 )
401 if notification is None:
402 return {"success": False, "error": "The notification could not be created"}
403 return {"success": True, "result": summarize_notification(self.engine, notification.contents())}
406class SnoozesTool(SupernotifyTool):
407 name = "supernotify__snoozes"
408 description = "List the notification snoozes and silences in place, and who they apply to."
410 @override
411 async def async_call(self, hass: HomeAssistant, tool_input: ToolInput, llm_context: LLMContext) -> JsonObjectType:
412 snoozes: dict[str, Any] = {"snoozes": self.engine.enquire_snoozes()}
413 return {"success": True, "result": snoozes}
416def split_docs(text: str, index: str = "") -> list[DocsPage]:
417 """Split the combined documentation into pages, each starting at a top level heading with a
418 blank line either side - which a comment in a code example rarely has. Each page's link is its
419 own "Source:" line, published since v2.10.0, else taken from the index where the titles match"""
420 urls: dict[str, str] = {
421 title.casefold(): url.removesuffix("index.md")
422 for title, url in re.findall(r"^- \[([^\]]+)\]\(([^)]+)\)", index, flags=re.MULTILINE)
423 }
424 pages: list[DocsPage] = []
425 title: str | None = None
426 lines: list[str] = text.splitlines()
427 body: list[str] = []
429 def url_for(page_title: str) -> str | None:
430 # recipe pages are titled "Recipe - X", where the index often has a longer "X ..."
431 wanted = page_title.casefold().removeprefix("recipe - ")
432 return urls.get(wanted) or next((u for t, u in urls.items() if t.startswith(wanted)), None)
434 def finish() -> None:
435 if title and (page_text := "\n".join(body).strip()):
436 first_line, _, rest = page_text.partition("\n")
437 if first_line.startswith(DOCS_SOURCE_PREFIX):
438 pages.append(DocsPage(title, first_line.removeprefix(DOCS_SOURCE_PREFIX).strip(), rest.strip()))
439 else:
440 pages.append(DocsPage(title, url_for(title), page_text))
442 for i, line in enumerate(lines):
443 if line.startswith("# ") and (i == 0 or not lines[i - 1].strip()) and (i + 1 == len(lines) or not lines[i + 1].strip()):
444 finish()
445 title, body = line[2:].strip(), []
446 else:
447 body.append(line)
448 finish()
449 return pages
452def search_docs(pages: list[DocsPage], question: str) -> list[DocsPage]:
453 """The pages that best match the question - most of all in the title, then by how many of its
454 words they contain, then by how densely, so a long page doesn't win just by being long.
455 Words are matched by their stem, so 'snooze' finds 'Snoozing'."""
456 stems = {
457 w[: max(4, len(w) - 3)]
458 for w in re.findall(r"[a-z0-9_]+", question.casefold())
459 if len(w) > 2 and w not in DOCS_STOP_WORDS
460 }
461 scored: list[tuple[float, int, DocsPage]] = []
462 for position, page in enumerate(pages):
463 title, text = page.title.casefold(), page.text.casefold()
464 score = 0.0
465 for stem in stems:
466 if count := text.count(stem):
467 score += 2 + min(count * 1000 / len(text), 3)
468 if stem in title:
469 score += 6
470 if score:
471 scored.append((-score, position, page))
472 return [page for _score, _position, page in sorted(scored)[:DOCS_SECTIONS_RETURNED]]
475async def _docs(hass: HomeAssistant) -> list[DocsPage]:
476 """The documentation pages, fetched on first use and then at most once a day"""
477 cached: _DocsCache | None = hass.data.get(DOCS_CACHE)
478 if cached and dt_util.utcnow() - cached.fetched < DOCS_CACHE_TIME:
479 return cached.pages
480 session = async_get_clientsession(hass)
481 timeout = aiohttp.ClientTimeout(total=15)
482 async with session.get(DOCS_URL, timeout=timeout) as response:
483 response.raise_for_status()
484 text = await response.text()
485 index = ""
486 try:
487 async with session.get(DOCS_INDEX_URL, timeout=timeout) as response:
488 response.raise_for_status()
489 index = await response.text()
490 except (aiohttp.ClientError, TimeoutError) as e:
491 _LOGGER.debug("SUPERNOTIFY Documentation index not fetched, pages will have no links: %s", e)
492 pages = split_docs(text, index)
493 hass.data[DOCS_CACHE] = _DocsCache(dt_util.utcnow(), pages)
494 return pages
497class HelpTool(SupernotifyTool):
498 name = "supernotify__help"
499 description = (
500 "Look up the Supernotify documentation, for how-to questions such as 'how do I e-mail a camera snapshot?' "
501 "or 'what does inclusion fallback mean?'. Returns the best matching pages, with configuration examples "
502 "and recipes, to answer from. The documentation is for the latest release, in English."
503 )
505 def __init__(self, engine: SupernotifyEngine) -> None:
506 super().__init__(engine)
507 self.parameters = vol.Schema({
508 vol.Required("question", description="What to look up, in a few words or a whole question"): str
509 })
511 @override
512 async def async_call(self, hass: HomeAssistant, tool_input: ToolInput, llm_context: LLMContext) -> JsonObjectType:
513 args = self.parameters(tool_input.tool_args)
514 try:
515 pages = await _docs(hass)
516 except (aiohttp.ClientError, TimeoutError) as e:
517 _LOGGER.warning("SUPERNOTIFY Unable to fetch documentation from %s: %s", DOCS_URL, e)
518 return {"success": False, "error": f"The documentation site could not be reached ({e})"}
519 found: dict[str, Any] = {
520 "site": DOCS_SITE,
521 "pages": [
522 {"title": page.title, "url": page.url, "text": page.text[:DOCS_SECTION_MAX_CHARS]}
523 for page in search_docs(pages, args["question"])
524 ],
525 }
526 if not found["pages"]:
527 found["note"] = "Nothing matched, try other words"
528 return {"success": True, "result": found}