Coverage for custom_components/supernotify/llm.py: 97%
321 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-09-25 14:29 +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
22import voluptuous as vol
23from homeassistant.components.llm import LLMTools # type: ignore[import-not-found,unused-ignore] # HA < 2026.x on py3.13
24from homeassistant.core import HomeAssistant, callback
25from homeassistant.helpers.aiohttp_client import async_get_clientsession
26from homeassistant.helpers.llm import LLM_API_ASSIST, LLMContext, Tool, ToolInput
27from homeassistant.util import dt as dt_util
28from homeassistant.util.hass_dict import HassKey
30from . import DOMAIN
31from .const import (
32 ATTR_DELIVERY,
33 ATTR_PRIORITY,
34 ATTR_SCENARIOS_APPLY,
35 CONF_LLM_ACTION_TOOLS,
36 CONF_LLM_DIAGNOSTIC_TOOLS,
37 CONF_LLM_TOOLS,
38 PRIORITY_VALUES,
39)
40from .model import CommandType, GlobalTargetType, QualifiedTargetType, RecipientType, TargetType
42if TYPE_CHECKING:
43 from homeassistant.util.json import JsonObjectType
45 from .engine import SupernotifyEngine
46 from .people import Recipient
48_LOGGER = logging.getLogger(__name__)
50EVERYONE = "everyone"
51MAX_HOURS = 7 * 24
52MAX_NOTIFICATIONS = 50
54# The documentation site publishes every page as one markdown file for LLMs, plus an index with links
55DOCS_SITE = "https://supernotify.rhizomatics.org.uk/"
56DOCS_URL = f"{DOCS_SITE}llms-full.txt"
57DOCS_INDEX_URL = f"{DOCS_SITE}llms.txt"
58DOCS_CACHE_TIME = dt.timedelta(days=1)
59# the line under each page title giving its link, from docgen/llms_preprocess.py
60DOCS_SOURCE_PREFIX = "Source: "
61DOCS_SECTIONS_RETURNED = 3
62DOCS_SECTION_MAX_CHARS = 4000
63DOCS_STOP_WORDS = frozenset([
64 "a",
65 "an",
66 "and",
67 "are",
68 "can",
69 "do",
70 "does",
71 "for",
72 "from",
73 "how",
74 "i",
75 "in",
76 "is",
77 "it",
78 "my",
79 "of",
80 "on",
81 "or",
82 "the",
83 "to",
84 "use",
85 "using",
86 "what",
87 "when",
88 "where",
89 "which",
90 "who",
91 "why",
92 "with",
93 # words in almost every page of these docs
94 "notification",
95 "notifications",
96 "notify",
97 "supernotify",
98 "send",
99])
102@dataclass
103class DocsPage:
104 title: str
105 url: str | None
106 text: str
109@dataclass
110class _DocsCache:
111 fetched: dt.datetime
112 pages: list[DocsPage]
115DOCS_CACHE: HassKey[_DocsCache] = HassKey(f"{DOMAIN}_llm_docs")
117SNOOZE_ACTIONS: dict[str, CommandType | None] = {
118 "snooze": CommandType.SNOOZE,
119 "silence": CommandType.SILENCE,
120 "unsnooze": CommandType.NORMAL,
121 "clear_all": None,
122}
123SNOOZE_SCOPES: dict[str, TargetType] = {
124 "everything": GlobalTargetType.EVERYTHING,
125 "noncritical": GlobalTargetType.NONCRITICAL,
126 "delivery": QualifiedTargetType.DELIVERY,
127 "transport": QualifiedTargetType.TRANSPORT,
128 "priority": QualifiedTargetType.PRIORITY,
129 "camera": QualifiedTargetType.CAMERA,
130}
133@callback
134def async_get_tools(hass: HomeAssistant, llm_context: LLMContext, api_id: str) -> LLMTools | None:
135 """Return the Supernotify tools switched on in the options, rebuilt each time so the
136 choices offered for deliveries, scenarios and recipients are always the current ones."""
137 if api_id != LLM_API_ASSIST:
138 return None
139 entries = hass.config_entries.async_loaded_entries(DOMAIN)
140 if not entries:
141 return None
142 engine: SupernotifyEngine = entries[0].runtime_data
143 options: dict[str, Any] = entries[0].options.get(CONF_LLM_TOOLS, {})
144 tools: list[Tool] = []
145 if options.get(CONF_LLM_ACTION_TOOLS):
146 tools.extend([NotifyTool(engine), SnoozeTool(engine)])
147 if options.get(CONF_LLM_DIAGNOSTIC_TOOLS):
148 tools.extend([RecentNotificationsTool(engine), DryRunTool(engine), SnoozesTool(engine), HelpTool(engine)])
149 if not tools:
150 return None
151 return LLMTools(tools=tools, prompt=_prompt(engine))
154def _prompt(engine: SupernotifyEngine) -> str:
155 context = engine.context
156 lines = ["Supernotify sends notifications to the household by phone, e-mail, speakers and other ways."]
157 if deliveries := [_labelled(name, d.alias) for name, d in context.delivery_registry.choosable_deliveries.items()]:
158 lines.append(f"Deliveries: {', '.join(deliveries)}.")
159 if scenarios := [_labelled(name, s.alias) for name, s in context.scenario_registry.scenarios.items()]:
160 lines.append(f"Scenarios: {', '.join(scenarios)}.")
161 if recipients := _recipient_names(engine):
162 lines.append(f"Recipients: {', '.join(recipients)}.")
163 return "\n".join(lines)
166def _labelled(name: str, alias: str | None) -> str:
167 return f"{name} ({alias})" if alias and alias != name else name
170def _display_name(recipient: Recipient) -> str:
171 return recipient.alias or recipient.name
174def _recipient_names(engine: SupernotifyEngine) -> list[str]:
175 return sorted(_display_name(r) for r in engine.context.people_registry.enabled_recipients())
178def _person_id(engine: SupernotifyEngine, name: str) -> str | None:
179 return engine.context.people_registry.person_id_for_name(name)
182def _names_for(engine: SupernotifyEngine, person_ids: list[str]) -> list[str]:
183 people = engine.context.people_registry.people
184 return sorted(_display_name(people[p]) if p in people else p for p in person_ids)
187def _notification_fields(engine: SupernotifyEngine, message_required: bool) -> dict[vol.Marker, Any]:
188 """The notification fields an agent may use - never custom_target, actions or media URLs"""
189 context = engine.context
190 fields: dict[vol.Marker, Any] = {
191 (vol.Required if message_required else vol.Optional)("message", description="The notification text"): str,
192 vol.Optional("title", description="A short title, used by deliveries that show one"): str,
193 vol.Optional("priority", description="How urgent the notification is, which decides which deliveries are used"): vol.In(
194 list(PRIORITY_VALUES)
195 ),
196 }
197 if recipients := _recipient_names(engine):
198 fields[vol.Optional("recipients", description="Who to notify. Leave out to use the usual recipients")] = [
199 vol.In(recipients)
200 ]
201 if deliveries := list(context.delivery_registry.choosable_deliveries):
202 fields[vol.Optional("deliveries", description="Use only these deliveries. Leave out to choose automatically")] = [
203 vol.In(deliveries)
204 ]
205 if scenarios := list(context.scenario_registry.scenarios):
206 fields[vol.Optional("scenarios", description="Scenarios to apply, as if their conditions were met")] = [
207 vol.In(scenarios)
208 ]
209 return fields
212def _notification_call(engine: SupernotifyEngine, args: dict[str, Any]) -> tuple[list[str] | None, dict[str, Any], list[str]]:
213 """Turn tool arguments into a target list and action data, plus any recipient names not known"""
214 person_ids: list[str] = []
215 unknown: list[str] = []
216 for name in args.get("recipients", []):
217 if person_id := _person_id(engine, name):
218 person_ids.append(person_id)
219 else:
220 unknown.append(name)
221 data: dict[str, Any] = {}
222 if "priority" in args:
223 data[ATTR_PRIORITY] = args["priority"]
224 if args.get("deliveries"):
225 data[ATTR_DELIVERY] = args["deliveries"]
226 if args.get("scenarios"):
227 data[ATTR_SCENARIOS_APPLY] = args["scenarios"]
228 return person_ids or None, data, unknown
231def summarize_notification(engine: SupernotifyEngine, contents: dict[str, Any]) -> dict[str, Any]:
232 """Cut an archived or live notification down to what explains what happened to it"""
233 deliveries: dict[str, dict[str, Any]] = {}
234 for name, outcomes in (contents.get("deliveries") or {}).items():
235 if skipped := outcomes.get("skipped"):
236 deliveries[name] = {"skipped": skipped.get("suppression_reason")}
237 continue
238 summary: dict[str, Any] = {}
239 recipients: set[str] = set()
240 for outcome in ("success", "suppressed", "error"):
241 envelopes: list[dict[str, Any]] = outcomes.get(outcome) or []
242 if not envelopes:
243 continue
244 summary[outcome] = len(envelopes)
245 for envelope in envelopes:
246 recipients.update(((envelope.get("target") or {}).get("person_id")) or [])
247 if outcome == "suppressed" and envelope.get("skip_reason"):
248 summary.setdefault("reasons", []).append(envelope["skip_reason"])
249 if outcome == "error":
250 summary.setdefault("errors", []).extend(
251 call.get("exception") for call in envelope.get("failed_calls") or [] if call.get("exception")
252 )
253 if recipients:
254 summary["recipients"] = _names_for(engine, sorted(recipients))
255 deliveries[name] = summary
256 condition_variables: dict[str, Any] = contents.get("condition_variables") or {}
257 result: dict[str, Any] = {
258 "id": contents.get("id"),
259 "created": contents.get("created"),
260 "outcome": contents.get("outcome"),
261 "message": contents.get("message"),
262 "title": condition_variables.get("notification_title"),
263 "priority": contents.get("priority"),
264 "scenarios": contents.get("enabled_scenarios") or [],
265 "occupancy": {
266 state: _names_for(engine, [p.get("person") for p in people if p.get("person")])
267 for state, people in (contents.get("occupancy") or {}).items()
268 },
269 "deliveries": deliveries,
270 "delivery_provenance": contents.get("delivery_provenance") or {},
271 }
272 if contents.get("unknown_names"):
273 result["unknown_names"] = contents["unknown_names"]
274 if contents.get("_suppression_reason"):
275 result["suppressed"] = contents["_suppression_reason"]
276 if requester := (contents.get("original_context") or {}).get("user"):
277 result["requested_by"] = requester
278 return result
281class SupernotifyTool(Tool):
282 def __init__(self, engine: SupernotifyEngine) -> None:
283 self.engine = engine
286class NotifyTool(SupernotifyTool):
287 name = "supernotify__notify"
288 description = (
289 "Send a notification to people in the household, e.g. 'tell everyone dinner is ready' or "
290 "'send Alice an urgent message'. Supernotify chooses how to reach each person from the priority, "
291 "the active scenarios and who is home, unless deliveries are given."
292 )
294 def __init__(self, engine: SupernotifyEngine) -> None:
295 super().__init__(engine)
296 self.parameters = vol.Schema(_notification_fields(engine, message_required=True))
298 @override
299 async def async_call(self, hass: HomeAssistant, tool_input: ToolInput, llm_context: LLMContext) -> JsonObjectType:
300 args = self.parameters(tool_input.tool_args)
301 target, data, unknown = _notification_call(self.engine, args)
302 if unknown:
303 return {"success": False, "error": f"Unknown recipients: {', '.join(unknown)}"}
304 notification = await self.engine.async_send_message(
305 args["message"], title=args.get("title"), target=target, data=data, context=llm_context.context
306 )
307 if notification is None:
308 return {"success": False, "error": "The notification could not be created"}
309 return {
310 "success": notification.delivered > 0,
311 "result": summarize_notification(self.engine, notification.contents()),
312 }
315class SnoozeTool(SupernotifyTool):
316 name = "supernotify__snooze"
317 description = (
318 "Snooze or silence notifications, or turn them back on, e.g. 'mute the doorbell camera alerts for an hour' "
319 "or 'stop non-urgent notifications until I say'. 'snooze' lasts for the given minutes, 'silence' until "
320 "undone, 'unsnooze' undoes one snooze, and 'clear_all' removes every snooze for everyone."
321 )
323 def __init__(self, engine: SupernotifyEngine) -> None:
324 super().__init__(engine)
325 self.parameters = vol.Schema({
326 vol.Required("action"): vol.In(list(SNOOZE_ACTIONS)),
327 vol.Optional("scope", default="everything", description="What kind of notifications to snooze"): vol.In(
328 list(SNOOZE_SCOPES)
329 ),
330 vol.Optional(
331 "name",
332 description="The delivery, transport, priority or camera entity_id to snooze, when scope is one of those",
333 ): str,
334 vol.Optional(
335 "recipient",
336 description="Whose notifications to snooze. Leave out for the person asking, or everyone if unknown",
337 ): vol.In([*_recipient_names(engine), EVERYONE]),
338 vol.Optional("minutes", description="How long to snooze for"): vol.All(
339 vol.Coerce(int), vol.Range(min=1, max=MAX_HOURS * 60)
340 ),
341 })
343 @override
344 async def async_call(self, hass: HomeAssistant, tool_input: ToolInput, llm_context: LLMContext) -> JsonObjectType:
345 args = self.parameters(tool_input.tool_args)
346 snoozer = self.engine.context.snoozer
347 cmd: CommandType | None = SNOOZE_ACTIONS[args["action"]]
348 if cmd is None:
349 return {"success": True, "result": {"cleared": self.engine.clear_snoozes()}}
351 scope: TargetType = SNOOZE_SCOPES[args["scope"]]
352 name: str | None = args.get("name")
353 if isinstance(scope, QualifiedTargetType):
354 if error := self._check_name(scope, name):
355 return {"success": False, "error": error}
356 else:
357 name = None
359 recipient: str | None = None
360 recipient_name: str | None = args.get("recipient")
361 if recipient_name is None:
362 recipient = self._requesting_person(llm_context)
363 elif recipient_name != EVERYONE:
364 recipient = _person_id(self.engine, recipient_name)
365 recipient_type = RecipientType.USER if recipient else RecipientType.EVERYONE
367 minutes: int | None = args.get("minutes")
368 snooze_for = dt.timedelta(minutes=minutes) if minutes else snoozer.snooze_period
369 snoozer.register_snooze(cmd, scope, name, recipient_type, recipient, snooze_for, reason="Assistant")
370 snoozes: dict[str, Any] = {"snoozes": self.engine.enquire_snoozes()}
371 return {"success": True, "result": snoozes}
373 def _check_name(self, scope: QualifiedTargetType, name: str | None) -> str | None:
374 registry = self.engine.context.delivery_registry
375 valid: list[str] | None = {
376 QualifiedTargetType.DELIVERY: list(registry.deliveries),
377 QualifiedTargetType.TRANSPORT: list(registry.transports),
378 QualifiedTargetType.PRIORITY: list(PRIORITY_VALUES),
379 }.get(scope)
380 if not name:
381 return f"A name is needed to snooze a {scope.lower()}"
382 if valid is not None and name not in valid:
383 return f"Unknown {scope.lower()} '{name}', choose from: {', '.join(valid)}"
384 if scope == QualifiedTargetType.CAMERA and not name.startswith("camera."):
385 return "A camera must be given as its entity_id, e.g. camera.front_door"
386 return None
388 def _requesting_person(self, llm_context: LLMContext) -> str | None:
389 user_id: str | None = llm_context.context.user_id if llm_context.context else None
390 return self.engine.context.people_registry.person_id_for_user_id(user_id)
393class RecentNotificationsTool(SupernotifyTool):
394 name = "supernotify__recent_notifications"
395 description = (
396 "List recent notifications and what happened to each one: which deliveries sent, were skipped or failed, "
397 "and why. Use this for questions like 'why didn't I get the doorbell alert?' or 'what notifications went "
398 "out today?'. 'occupancy' shows who was home at the time, which decides some deliveries."
399 )
401 def __init__(self, engine: SupernotifyEngine) -> None:
402 super().__init__(engine)
403 fields: dict[vol.Marker, Any] = {
404 vol.Optional("hours", default=24, description="How many hours back to look"): vol.All(
405 vol.Coerce(int), vol.Range(min=1, max=MAX_HOURS)
406 ),
407 vol.Optional("limit", default=10, description="The most notifications to return, newest first"): vol.All(
408 vol.Coerce(int), vol.Range(min=1, max=MAX_NOTIFICATIONS)
409 ),
410 }
411 if recipients := _recipient_names(engine):
412 fields[vol.Optional("recipient", description="Only notifications that reached this person")] = vol.In(recipients)
413 self.parameters = vol.Schema(fields)
415 @override
416 async def async_call(self, hass: HomeAssistant, tool_input: ToolInput, llm_context: LLMContext) -> JsonObjectType:
417 args = self.parameters(tool_input.tool_args)
418 since: dt.datetime = dt_util.now() - dt.timedelta(hours=args["hours"])
419 archive = self.engine.context.archive
420 result: dict[str, Any] = {}
421 if archive.archive_directory and archive.archive_directory.enabled:
422 archived = await archive.recent(since, MAX_NOTIFICATIONS)
423 else:
424 last = self.engine.last_notification
425 archived = [last.contents()] if last and last.created >= since else []
426 result["note"] = (
427 "The notification archive is off, so only the latest notification since Home Assistant started "
428 "is known. It can be turned on in the Supernotify options."
429 )
430 summaries = [summarize_notification(self.engine, contents) for contents in archived]
431 if recipient := args.get("recipient"):
432 summaries = [s for s in summaries if any(recipient in d.get("recipients", []) for d in s["deliveries"].values())]
433 result["notifications"] = summaries[: args["limit"]]
434 return {"success": True, "result": result}
437class DryRunTool(SupernotifyTool):
438 name = "supernotify__dry_run"
439 description = (
440 "Work out who a notification would reach right now, and by which deliveries, without sending it. "
441 "Use this for questions like 'if the alarm goes off, who gets told and how?'. 'occupancy' shows who is "
442 "home, which decides some deliveries. The duplicate check is not made."
443 )
445 def __init__(self, engine: SupernotifyEngine) -> None:
446 super().__init__(engine)
447 self.parameters = vol.Schema(_notification_fields(engine, message_required=False))
449 @override
450 async def async_call(self, hass: HomeAssistant, tool_input: ToolInput, llm_context: LLMContext) -> JsonObjectType:
451 args = self.parameters(tool_input.tool_args)
452 target, data, unknown = _notification_call(self.engine, args)
453 if unknown:
454 return {"success": False, "error": f"Unknown recipients: {', '.join(unknown)}"}
455 plan = await self.engine.async_dry_run(args.get("message", ""), title=args.get("title"), target=target, data=data)
456 plan["occupancy"] = {state: _names_for(self.engine, people) for state, people in plan["occupancy"].items()}
457 for delivery in plan["deliveries"].values():
458 if "recipients" in delivery:
459 delivery["recipients"] = _names_for(self.engine, delivery["recipients"])
460 return {"success": True, "result": plan}
463class SnoozesTool(SupernotifyTool):
464 name = "supernotify__snoozes"
465 description = "List the notification snoozes and silences in place, and who they apply to."
467 @override
468 async def async_call(self, hass: HomeAssistant, tool_input: ToolInput, llm_context: LLMContext) -> JsonObjectType:
469 snoozes: dict[str, Any] = {"snoozes": self.engine.enquire_snoozes()}
470 return {"success": True, "result": snoozes}
473def split_docs(text: str, index: str = "") -> list[DocsPage]:
474 """Split the combined documentation into pages, each starting at a top level heading with a
475 blank line either side - which a comment in a code example rarely has. Each page's link is its
476 own "Source:" line, published since v2.10.0, else taken from the index where the titles match"""
477 urls: dict[str, str] = {
478 title.casefold(): url.removesuffix("index.md")
479 for title, url in re.findall(r"^- \[([^\]]+)\]\(([^)]+)\)", index, flags=re.MULTILINE)
480 }
481 pages: list[DocsPage] = []
482 title: str | None = None
483 lines: list[str] = text.splitlines()
484 body: list[str] = []
486 def url_for(page_title: str) -> str | None:
487 # recipe pages are titled "Recipe - X", where the index often has a longer "X ..."
488 wanted = page_title.casefold().removeprefix("recipe - ")
489 return urls.get(wanted) or next((u for t, u in urls.items() if t.startswith(wanted)), None)
491 def finish() -> None:
492 if title and (page_text := "\n".join(body).strip()):
493 first_line, _, rest = page_text.partition("\n")
494 if first_line.startswith(DOCS_SOURCE_PREFIX):
495 pages.append(DocsPage(title, first_line.removeprefix(DOCS_SOURCE_PREFIX).strip(), rest.strip()))
496 else:
497 pages.append(DocsPage(title, url_for(title), page_text))
499 for i, line in enumerate(lines):
500 if line.startswith("# ") and (i == 0 or not lines[i - 1].strip()) and (i + 1 == len(lines) or not lines[i + 1].strip()):
501 finish()
502 title, body = line[2:].strip(), []
503 else:
504 body.append(line)
505 finish()
506 return pages
509def search_docs(pages: list[DocsPage], question: str) -> list[DocsPage]:
510 """The pages that best match the question - most of all in the title, then by how many of its
511 words they contain, then by how densely, so a long page doesn't win just by being long.
512 Words are matched by their stem, so 'snooze' finds 'Snoozing'."""
513 stems = {
514 w[: max(4, len(w) - 3)]
515 for w in re.findall(r"[a-z0-9_]+", question.casefold())
516 if len(w) > 2 and w not in DOCS_STOP_WORDS
517 }
518 scored: list[tuple[float, int, DocsPage]] = []
519 for position, page in enumerate(pages):
520 title, text = page.title.casefold(), page.text.casefold()
521 score = 0.0
522 for stem in stems:
523 if count := text.count(stem):
524 score += 2 + min(count * 1000 / len(text), 3)
525 if stem in title:
526 score += 6
527 if score:
528 scored.append((-score, position, page))
529 return [page for _score, _position, page in sorted(scored)[:DOCS_SECTIONS_RETURNED]]
532async def _docs(hass: HomeAssistant) -> list[DocsPage]:
533 """The documentation pages, fetched on first use and then at most once a day"""
534 cached: _DocsCache | None = hass.data.get(DOCS_CACHE)
535 if cached and dt_util.utcnow() - cached.fetched < DOCS_CACHE_TIME:
536 return cached.pages
537 session = async_get_clientsession(hass)
538 timeout = aiohttp.ClientTimeout(total=15)
539 async with session.get(DOCS_URL, timeout=timeout) as response:
540 response.raise_for_status()
541 text = await response.text()
542 index = ""
543 try:
544 async with session.get(DOCS_INDEX_URL, timeout=timeout) as response:
545 response.raise_for_status()
546 index = await response.text()
547 except (aiohttp.ClientError, TimeoutError) as e:
548 _LOGGER.debug("SUPERNOTIFY Documentation index not fetched, pages will have no links: %s", e)
549 pages = split_docs(text, index)
550 hass.data[DOCS_CACHE] = _DocsCache(dt_util.utcnow(), pages)
551 return pages
554class HelpTool(SupernotifyTool):
555 name = "supernotify__help"
556 description = (
557 "Look up the Supernotify documentation, for how-to questions such as 'how do I e-mail a camera snapshot?' "
558 "or 'what does inclusion fallback mean?'. Returns the best matching pages, with configuration examples "
559 "and recipes, to answer from. The documentation is for the latest release, in English."
560 )
562 def __init__(self, engine: SupernotifyEngine) -> None:
563 super().__init__(engine)
564 self.parameters = vol.Schema({
565 vol.Required("question", description="What to look up, in a few words or a whole question"): str
566 })
568 @override
569 async def async_call(self, hass: HomeAssistant, tool_input: ToolInput, llm_context: LLMContext) -> JsonObjectType:
570 args = self.parameters(tool_input.tool_args)
571 try:
572 pages = await _docs(hass)
573 except (aiohttp.ClientError, TimeoutError) as e:
574 _LOGGER.warning("SUPERNOTIFY Unable to fetch documentation from %s: %s", DOCS_URL, e)
575 return {"success": False, "error": f"The documentation site could not be reached ({e})"}
576 found: dict[str, Any] = {
577 "site": DOCS_SITE,
578 "pages": [
579 {"title": page.title, "url": page.url, "text": page.text[:DOCS_SECTION_MAX_CHARS]}
580 for page in search_docs(pages, args["question"])
581 ],
582 }
583 if not found["pages"]:
584 found["note"] = "Nothing matched, try other words"
585 return {"success": True, "result": found}