776 lines
31 KiB
Python
776 lines
31 KiB
Python
"""Expose the in-process server's toolset as a Home Assistant LLM API (#1745).
|
|
|
|
While the in-process server entry is up, the ha-mcp toolset is registered as
|
|
one or two LLM APIs (``homeassistant.helpers.llm``). Any Home Assistant
|
|
conversation agent — OpenAI, Google, Ollama, Anthropic, or any other — can
|
|
then select it in its "Control Home Assistant" option, and the user chats
|
|
with the toolset through the surfaces Home Assistant already has: the Assist
|
|
chat UI, the companion apps, and voice satellites. No separate chat frontend
|
|
is needed.
|
|
|
|
Two exposure modes (the ``llm_api_exposure`` entry option picks which are
|
|
registered; default is tool-search only):
|
|
|
|
* **tool search** — the agent gets a tiny catalog: the server's pinned tools
|
|
mirrored directly, plus two meta-tools synthesized here: ``ha_search_tools``
|
|
(find tools by task) and ``ha_call_tool`` (execute a discovered tool). This
|
|
keeps per-turn context small — the shape context-limited models need.
|
|
* **full** — every exposed tool is mirrored directly into the agent's tool
|
|
list, one schema each.
|
|
|
|
**Per-tool exposure is decided by the server, not here.** The server stamps
|
|
every ``tools/list`` entry with ``_meta.ha_mcp = {llm_api_exposed, pinned}``
|
|
(see ``src/ha_mcp/llm_exposure.py``): user toggles from the settings UI, with
|
|
deny-by-default for beta/developer/restart-reload-backup tools. Both modes
|
|
filter on the stamp, and the tool-search ``ha_call_tool`` forwarder re-checks
|
|
it at call time — a hidden tool is invisible (absent from lists and search
|
|
results) and a hallucinated call to one gets a plain unknown-tool error, the
|
|
same answer a nonexistent tool gets, so nothing leaks. Globally-disabled
|
|
tools never appear in ``tools/list`` at all and the server rejects calling
|
|
them by name, and every forwarded call traverses the server's policy /
|
|
read-only middleware exactly like any MCP client's call.
|
|
|
|
The server runs on its own worker thread behind a loopback HTTP listener, and
|
|
``ha_mcp`` must never be imported in the HA main process (see
|
|
:mod:`embedded_server`), so this module talks real MCP to the server over
|
|
loopback streamable HTTP. The ``mcp`` client SDK arrives with the
|
|
runtime-installed ha-mcp package (a fastmcp dependency), so every SDK import
|
|
here is lazy and the first one runs on the executor.
|
|
|
|
The tool list is fetched fresh on every ``async_get_api_instance`` call (once
|
|
per conversation turn): exposure toggles and runtime-registered custom tools
|
|
apply on the agent's next message, and two loopback round-trips per turn are
|
|
noise next to the LLM call itself. Tool calls likewise open a short-lived
|
|
stateless session each — the in-process server serves ``stateless_http=True``.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import importlib
|
|
import logging
|
|
from collections.abc import AsyncIterator, Callable, Iterable
|
|
from contextlib import AsyncExitStack, asynccontextmanager
|
|
from dataclasses import dataclass
|
|
from functools import cache
|
|
from typing import TYPE_CHECKING, Any, cast
|
|
|
|
import voluptuous as vol
|
|
from homeassistant.core import HomeAssistant
|
|
from homeassistant.exceptions import HomeAssistantError
|
|
from homeassistant.helpers import llm
|
|
|
|
from .const import (
|
|
DATA_LLM_API_UNSUB,
|
|
DEFAULT_LLM_API_EXPOSURE,
|
|
DOMAIN,
|
|
EXPOSURE_BOTH,
|
|
EXPOSURE_FULL,
|
|
EXPOSURE_TOOL_SEARCH,
|
|
OPT_LLM_API_EXPOSURE,
|
|
)
|
|
|
|
if TYPE_CHECKING:
|
|
import httpx
|
|
from homeassistant.config_entries import ConfigEntry
|
|
from homeassistant.util.json import JsonObjectType
|
|
from mcp import types as mcp_types
|
|
from mcp.client.session import ClientSession
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
# Listing tools is two loopback round-trips (initialize + tools/list); a slow
|
|
# answer means the server thread is wedged, not that the network is slow.
|
|
_LIST_TOOLS_TIMEOUT_SECONDS = 10.0
|
|
|
|
# Tool calls run real work — WebSocket-verified device control, dashboard
|
|
# screenshots, config writes that poll for completion — well beyond the 10s a
|
|
# remote-server integration would allow. The conversation agent shows a spinner
|
|
# for the duration, so err generous rather than kill a legitimate slow tool.
|
|
_CALL_TOOL_TIMEOUT_SECONDS = 300.0
|
|
|
|
# The server-side stamp this module filters on (mirrors
|
|
# src/ha_mcp/llm_exposure.py — keep the names in sync).
|
|
_META_NAMESPACE = "ha_mcp"
|
|
_META_EXPOSED_KEY = "llm_api_exposed"
|
|
_META_PINNED_KEY = "pinned"
|
|
|
|
# Fallback exposure policy for servers that predate the stamp: hide the
|
|
# operational-hazard names and the known beta/developer tools. Imperfect by
|
|
# construction (a newer beta tool on an old server can't be known here) but
|
|
# strictly safer than exposing everything, and logged once per instance
|
|
# build. The real policy lives server-side.
|
|
_FALLBACK_DENY_PREFIXES = ("ha_dev_",)
|
|
_FALLBACK_DENY_TOOLS = frozenset(
|
|
{
|
|
"ha_restart",
|
|
"ha_reload_core",
|
|
"ha_manage_backup",
|
|
# Beta-tagged tools as of the stamp's introduction (server-side the
|
|
# gate is tag-based and future-proof; this list is only the legacy
|
|
# fallback).
|
|
"ha_config_set_yaml",
|
|
"ha_manage_custom_tool",
|
|
"ha_get_dashboard_screenshot",
|
|
"ha_install_mcp_tools",
|
|
"ha_list_files",
|
|
"ha_read_file",
|
|
"ha_write_file",
|
|
"ha_delete_file",
|
|
}
|
|
)
|
|
|
|
# Names of the meta-tools synthesized for the tool-search mode. ha_search_tools
|
|
# deliberately matches the server's own tool-search terminology; if the server
|
|
# itself runs ENABLE_TOOL_SEARCH its identically-named tool is excluded from
|
|
# mirroring/search results to avoid duplicates.
|
|
_SEARCH_TOOL_NAME = "ha_search_tools"
|
|
_CALL_TOOL_NAME = "ha_call_tool"
|
|
_SEARCH_RESULT_LIMIT = 8
|
|
|
|
|
|
@cache
|
|
def _schema_converter() -> Callable[[Any], Any]:
|
|
"""Resolve the Core-provided schema converter once, off the event loop."""
|
|
try:
|
|
legacy = importlib.import_module("voluptuous_openapi")
|
|
except ModuleNotFoundError as err:
|
|
if err.name != "voluptuous_openapi":
|
|
raise
|
|
probatio = importlib.import_module("probatio")
|
|
return cast(Callable[[Any], Any], probatio.from_openapi)
|
|
return cast(Callable[[Any], Any], legacy.convert_to_voluptuous)
|
|
|
|
|
|
def convert_to_voluptuous(schema: Any) -> vol.Schema:
|
|
"""Convert an OpenAPI schema on stable and Probatio-based HA Core."""
|
|
return cast(vol.Schema, _schema_converter()(schema))
|
|
|
|
|
|
# Used when the server's initialize result carries no instructions (it always
|
|
# should — ha-mcp ships server-level instructions — but never render an empty
|
|
# prompt if a build does not).
|
|
_FALLBACK_API_PROMPT = (
|
|
"The following tools are provided by the HA-MCP server running inside "
|
|
"Home Assistant. They give full control over this Home Assistant "
|
|
"instance: entities, automations, scripts, dashboards, helpers, and "
|
|
"configuration."
|
|
)
|
|
|
|
_TOOL_SEARCH_PROMPT = (
|
|
"\n\n## Tool Discovery\n"
|
|
"This assistant uses search-based tool discovery: most tools are NOT "
|
|
"listed directly.\n"
|
|
f"1. Call {_SEARCH_TOOL_NAME}(query=...) to find tools for the task; "
|
|
"results include each tool's name, description, and input schema.\n"
|
|
f"2. Execute a discovered tool with {_CALL_TOOL_NAME}(name=..., "
|
|
"arguments={...}) — discovered tools are NOT directly callable here.\n"
|
|
"3. The few tools listed directly can be called as usual.\n"
|
|
"Search once per task, not per call — tool names stay valid all "
|
|
"conversation."
|
|
)
|
|
|
|
|
|
def _transport_error_leaves() -> tuple[type[BaseException], ...]:
|
|
"""Return the non-group exception classes a loopback exchange can raise.
|
|
|
|
OSError covers a refused/dropped loopback connect; TimeoutError comes
|
|
from our asyncio.timeout budget. httpx errors and protocol-level McpError
|
|
can also escape a session call UNWRAPPED (HA core's mcp integration
|
|
catches both the same way), but neither class is importable at module
|
|
level — both arrive with the runtime-installed server package — hence a
|
|
function instead of a module constant.
|
|
"""
|
|
errors: tuple[type[BaseException], ...] = (TimeoutError, OSError)
|
|
try:
|
|
import httpx
|
|
from mcp import McpError
|
|
except ImportError: # pragma: no cover - SDK-less builds never open a session
|
|
return errors
|
|
return (*errors, httpx.HTTPError, McpError)
|
|
|
|
|
|
def _transport_errors() -> tuple[type[BaseException], ...]:
|
|
"""Return the ``except`` target for one loopback MCP exchange.
|
|
|
|
Evaluated at exception time (an ``except`` expression is), so the lazy
|
|
imports in :func:`_transport_error_leaves` have already succeeded by
|
|
then. Includes ExceptionGroup because the SDK's anyio task groups wrap
|
|
in-session failures — but a caught group must still pass
|
|
:func:`_is_transport_failure` before being mapped to a friendly error,
|
|
or a genuine bug that happened inside the task group would be relabeled
|
|
as a transport failure (review finding).
|
|
"""
|
|
return (*_transport_error_leaves(), ExceptionGroup)
|
|
|
|
|
|
def _is_transport_failure(err: BaseException) -> bool:
|
|
"""Return True when ``err`` is purely a transport failure.
|
|
|
|
A group counts only when EVERY leaf (nested groups included) is a
|
|
transport error: a group carrying any non-transport member is a genuine
|
|
bug that must propagate with its loud traceback instead of being
|
|
remapped to a "could not reach the server" message.
|
|
"""
|
|
if isinstance(err, ExceptionGroup):
|
|
return all(_is_transport_failure(exc) for exc in err.exceptions)
|
|
return isinstance(err, _transport_error_leaves())
|
|
|
|
|
|
def _import_mcp_sdk() -> None:
|
|
"""Import lazy LLM dependencies (blocking; run on the executor).
|
|
|
|
Raises ImportError when the MCP SDK or Core's schema converter is not
|
|
importable — the caller decides whether that skips registration or
|
|
surfaces as a conversation error.
|
|
"""
|
|
importlib.import_module("mcp.client.session")
|
|
importlib.import_module("mcp.client.streamable_http")
|
|
_schema_converter()
|
|
|
|
|
|
async def async_probe_mcp_sdk(hass: HomeAssistant) -> bool:
|
|
"""Return True when lazy LLM dependencies import (first import off-loop)."""
|
|
try:
|
|
await hass.async_add_executor_job(_import_mcp_sdk)
|
|
except ImportError as err:
|
|
_LOGGER.warning(
|
|
"A required LLM dependency is not importable (%s); the "
|
|
"conversation-agent LLM API will not be available",
|
|
err,
|
|
)
|
|
return False
|
|
return True
|
|
|
|
|
|
def _loopback_httpx_client_factory(
|
|
headers: dict[str, str] | None = None,
|
|
timeout: httpx.Timeout | None = None,
|
|
auth: httpx.Auth | None = None,
|
|
) -> httpx.AsyncClient:
|
|
"""``httpx_client_factory`` for the pre-rename SDK's ``streamablehttp_client``.
|
|
|
|
That deprecated entry point takes no ``http_client`` — it always builds
|
|
its own via this factory — but the factory itself IS overridable, so the
|
|
same ``verify=False`` / ``trust_env=False`` posture as the canonical path
|
|
in :func:`_mcp_session` still applies: this fallback is loopback-only
|
|
too, so a real SSL context is pure waste and this call must never be
|
|
diverted through an environment proxy (which would also leak the URL's
|
|
embedded ``secret_path`` to it).
|
|
|
|
``timeout`` mirrors ``create_mcp_http_client`` (the factory this
|
|
substitutes for): a ``None`` here means "no timeout was supplied", not
|
|
"disable timeouts" — the real ``streamablehttp_client`` caller always
|
|
passes an explicit ``httpx.Timeout``, but a bare ``None`` reaching
|
|
``httpx.AsyncClient`` directly disables every timeout outright (review
|
|
finding), which is the wrong zero-argument default for a fallback aimed
|
|
at unknown old environments.
|
|
|
|
The substitute value is inlined rather than imported from
|
|
``mcp.shared._httpx_utils``: this factory only ever runs on SDKs old
|
|
enough to lack ``streamable_http_client`` (the canonical name this
|
|
module tries first), and ``MCP_DEFAULT_TIMEOUT`` /
|
|
``MCP_DEFAULT_SSE_READ_TIMEOUT`` don't exist before mcp 1.24 either — an
|
|
import would raise on every SDK version this fallback actually serves
|
|
(round-2 review finding: the first attempt at this fix broke the exact
|
|
path it was meant to harden). ``create_mcp_http_client``'s own ``None``
|
|
default on those older SDKs is this same flat ``httpx.Timeout(30.0)``,
|
|
with no separate read timeout (the ``read=300`` shape is itself a 1.24+
|
|
addition) — matched here rather than guessed at.
|
|
"""
|
|
import httpx
|
|
|
|
if timeout is None:
|
|
timeout = httpx.Timeout(30.0)
|
|
|
|
return httpx.AsyncClient(
|
|
headers=headers, timeout=timeout, auth=auth, verify=False, trust_env=False
|
|
)
|
|
|
|
|
|
@asynccontextmanager
|
|
async def _mcp_session(
|
|
url: str,
|
|
) -> AsyncIterator[tuple[ClientSession, mcp_types.InitializeResult]]:
|
|
"""Open an initialized MCP session against the loopback server.
|
|
|
|
Imports resolve from ``sys.modules`` — :func:`async_probe_mcp_sdk` did the
|
|
real (blocking) import on the executor before the API was registered.
|
|
|
|
Builds a throwaway httpx client scoped to this one session rather than
|
|
reusing Home Assistant's shared one (``helpers.httpx_client.
|
|
get_async_client`` — the prior approach): the SDK applies NO timeout of
|
|
its own when a caller-provided client is passed, so whatever timeout
|
|
THAT client happens to carry becomes the real wire-level ceiling. HA's
|
|
shared client is built with no explicit ``timeout=``, so it silently
|
|
carries httpx's own hardcoded 5-second default — capping every tool call
|
|
at 5 seconds of read-idle no matter how generous
|
|
``_CALL_TOOL_TIMEOUT_SECONDS`` / ``_LIST_TOOLS_TIMEOUT_SECONDS`` looked
|
|
(live-found investigating a ~60s Assist-pipeline hang: a real tool doing
|
|
real work never got anywhere near its own asyncio budget).
|
|
|
|
``verify=False`` is not a security relaxation: ``url`` is always
|
|
``http://127.0.0.1:<port>...`` (see ``async_register_llm_api``) — a
|
|
plain-HTTP loopback call that never negotiates TLS — so building a real
|
|
SSL context would be pure waste. It also keeps this loop-safe the same
|
|
way the shared client did: an SSL context built with ``verify=True``
|
|
loads the system CA bundle SYNCHRONOUSLY (live-found — HA's
|
|
blocking-call monitor flagged this exact line when the SDK built its own
|
|
default client), and skipping verification skips that load entirely.
|
|
|
|
``trust_env=False`` for the same "this is loopback, not the network"
|
|
reason: httpx defaults to reading ``HTTP_PROXY``/``NO_PROXY`` from the
|
|
environment, and ``127.0.0.1`` is not exempt unless ``NO_PROXY``
|
|
explicitly lists it. Under an ``HTTP_PROXY`` that doesn't, this call
|
|
would leave the loopback listener entirely and go out through the
|
|
configured proxy instead — which also hands the proxy ``url``'s
|
|
embedded ``secret_path`` (the private endpoint credential). Disabling
|
|
env trust removes both failure modes; nothing here should ever consult
|
|
a proxy.
|
|
|
|
The client is entered on the exit stack so it closes with the rest of
|
|
the session.
|
|
"""
|
|
from mcp.client.session import ClientSession
|
|
|
|
async with AsyncExitStack() as stack:
|
|
try:
|
|
from mcp.client.streamable_http import streamable_http_client
|
|
except ImportError:
|
|
# Pre-rename SDK (an older ha-mcp resolved by a pip-spec override
|
|
# pins an older fastmcp/mcp): same call shape, deprecated name,
|
|
# and no http_client kwarg — but it does accept a factory for the
|
|
# client it builds internally, so _loopback_httpx_client_factory
|
|
# keeps this fallback on the same verify=False/trust_env=False
|
|
# posture as the canonical path below.
|
|
from mcp.client.streamable_http import streamablehttp_client
|
|
|
|
transport = streamablehttp_client(
|
|
url=url, httpx_client_factory=_loopback_httpx_client_factory
|
|
)
|
|
else:
|
|
import httpx
|
|
|
|
http_client = await stack.enter_async_context(
|
|
httpx.AsyncClient(
|
|
verify=False,
|
|
trust_env=False,
|
|
timeout=httpx.Timeout(_CALL_TOOL_TIMEOUT_SECONDS),
|
|
)
|
|
)
|
|
transport = streamable_http_client(url=url, http_client=http_client)
|
|
|
|
read_stream, write_stream, _ = await stack.enter_async_context(transport)
|
|
session = await stack.enter_async_context(
|
|
ClientSession(read_stream, write_stream)
|
|
)
|
|
init_result = await session.initialize()
|
|
yield session, init_result
|
|
|
|
|
|
def _tool_meta_namespace(tool: Any) -> dict[str, Any] | None:
|
|
"""Return the tool's ``_meta.ha_mcp`` namespace, or None when absent."""
|
|
meta = getattr(tool, "meta", None)
|
|
if not isinstance(meta, dict):
|
|
return None
|
|
namespace = meta.get(_META_NAMESPACE)
|
|
return namespace if isinstance(namespace, dict) else None
|
|
|
|
|
|
def _fallback_exposed(name: str) -> bool:
|
|
"""Legacy exposure policy for servers that predate the meta stamp."""
|
|
if name.startswith(_FALLBACK_DENY_PREFIXES):
|
|
return False
|
|
return name not in _FALLBACK_DENY_TOOLS
|
|
|
|
|
|
def _partition_tools(tools: Iterable[Any]) -> tuple[list[Any], set[str], bool]:
|
|
"""Split a raw tools/list into (exposed tools, pinned names, stamped).
|
|
|
|
``stamped`` is False when NO tool carried the server's exposure stamp —
|
|
an older server package — in which case the conservative component-side
|
|
fallback policy was applied instead.
|
|
"""
|
|
stamped = False
|
|
exposed: list[Any] = []
|
|
pinned: set[str] = set()
|
|
for tool in tools:
|
|
namespace = _tool_meta_namespace(tool)
|
|
if namespace is not None and _META_EXPOSED_KEY in namespace:
|
|
stamped = True
|
|
if namespace.get(_META_PINNED_KEY):
|
|
pinned.add(tool.name)
|
|
if namespace.get(_META_EXPOSED_KEY):
|
|
exposed.append(tool)
|
|
elif _fallback_exposed(tool.name):
|
|
exposed.append(tool)
|
|
if not stamped:
|
|
# The fallback path already filtered; recompute pinned as empty (an
|
|
# unstamped server gives no pinned signal — the tool-search mode then
|
|
# simply mirrors nothing directly).
|
|
pinned = set()
|
|
return exposed, pinned, stamped
|
|
|
|
|
|
class HaMcpTool(llm.Tool):
|
|
"""One ha-mcp tool, called over loopback MCP."""
|
|
|
|
def __init__(
|
|
self,
|
|
name: str,
|
|
description: str | None,
|
|
parameters: vol.Schema,
|
|
server_url: str,
|
|
) -> None:
|
|
"""Store the converted schema and the loopback endpoint."""
|
|
self.name = name
|
|
self.description = description
|
|
self.parameters = parameters
|
|
self._server_url = server_url
|
|
|
|
async def async_call(
|
|
self,
|
|
hass: HomeAssistant,
|
|
tool_input: llm.ToolInput,
|
|
llm_context: llm.LLMContext,
|
|
) -> JsonObjectType:
|
|
"""Call the tool on the in-process server and return its result."""
|
|
return await _forward_tool_call(
|
|
hass, self._server_url, self.name, tool_input.tool_args
|
|
)
|
|
|
|
|
|
async def _forward_tool_call(
|
|
hass: HomeAssistant, server_url: str, name: str, arguments: dict[str, Any]
|
|
) -> JsonObjectType:
|
|
"""Forward one tool call over loopback and dump the result for the agent."""
|
|
try:
|
|
async with (
|
|
asyncio.timeout(_CALL_TOOL_TIMEOUT_SECONDS),
|
|
_mcp_session(server_url) as (session, _init),
|
|
):
|
|
result = await session.call_tool(name, arguments)
|
|
except _transport_errors() as err:
|
|
if not _is_transport_failure(err):
|
|
raise
|
|
raise HomeAssistantError(
|
|
f"Error calling the HA-MCP tool {name}: {err}"
|
|
) from err
|
|
# Full CallToolResult (content blocks, structuredContent, isError) —
|
|
# the same shape HA core's mcp integration hands to agents; ha-mcp
|
|
# signals tool failure via isError + structured error JSON, which the
|
|
# agent reads and reacts to like any tool output.
|
|
return result.model_dump(exclude_unset=True, exclude_none=True)
|
|
|
|
|
|
def _search_score(query_words: list[str], name: str, description: str) -> int:
|
|
"""Score a tool against the query (simple word overlap + substring)."""
|
|
haystack = f"{name} {description}".lower()
|
|
name_lower = name.lower()
|
|
score = 0
|
|
for word in query_words:
|
|
if word in name_lower:
|
|
score += 3
|
|
elif word in haystack:
|
|
score += 1
|
|
return score
|
|
|
|
|
|
class HaMcpSearchTool(llm.Tool):
|
|
"""Meta-tool: find ha-mcp tools relevant to a task (tool-search mode).
|
|
|
|
Searches only the EXPOSED catalog snapshot taken at instance build, so a
|
|
hidden tool can never appear in results.
|
|
"""
|
|
|
|
name = _SEARCH_TOOL_NAME
|
|
description = (
|
|
"Search the Home Assistant MCP toolset for tools relevant to a task. "
|
|
"Returns each match's name, description, and input schema. Execute "
|
|
f"matches with {_CALL_TOOL_NAME}."
|
|
)
|
|
parameters = vol.Schema({vol.Required("query"): str})
|
|
|
|
def __init__(self, catalog: list[dict[str, Any]]) -> None:
|
|
"""Hold the exposed-catalog snapshot (name/description/schema dicts)."""
|
|
self._catalog = catalog
|
|
|
|
async def async_call(
|
|
self,
|
|
hass: HomeAssistant,
|
|
tool_input: llm.ToolInput,
|
|
llm_context: llm.LLMContext,
|
|
) -> JsonObjectType:
|
|
"""Return the top-scoring exposed tools for the query."""
|
|
query_words = [
|
|
w for w in str(tool_input.tool_args.get("query", "")).lower().split() if w
|
|
]
|
|
scored = sorted(
|
|
(
|
|
(_search_score(query_words, t["name"], t["description"]), t)
|
|
for t in self._catalog
|
|
),
|
|
key=lambda pair: pair[0],
|
|
reverse=True,
|
|
)
|
|
results = [t for score, t in scored[:_SEARCH_RESULT_LIMIT] if score > 0]
|
|
if not results:
|
|
return {
|
|
"results": [],
|
|
"message": (
|
|
"No matching tools. Try different task words (e.g. "
|
|
"'automation', 'light', 'history', 'dashboard')."
|
|
),
|
|
}
|
|
return {"results": results}
|
|
|
|
|
|
class HaMcpCallTool(llm.Tool):
|
|
"""Meta-tool: execute a tool discovered via search (tool-search mode).
|
|
|
|
The exposure re-check at call time is the enforcement half of the
|
|
tool-search mode: hiding a tool from search results alone would not stop
|
|
a model that guesses a name. A non-exposed name gets the same
|
|
unknown-tool answer a nonexistent name gets — existence never leaks.
|
|
"""
|
|
|
|
name = _CALL_TOOL_NAME
|
|
description = (
|
|
"Execute a Home Assistant MCP tool by name with a dictionary of "
|
|
f"arguments. Discover tools and their schemas with {_SEARCH_TOOL_NAME} "
|
|
"first."
|
|
)
|
|
parameters = vol.Schema(
|
|
{
|
|
vol.Required("name"): str,
|
|
vol.Optional("arguments", default=dict): dict,
|
|
}
|
|
)
|
|
|
|
def __init__(self, server_url: str, exposed_names: set[str]) -> None:
|
|
"""Hold the loopback endpoint and the exposed-name allowlist."""
|
|
self._server_url = server_url
|
|
self._exposed_names = exposed_names
|
|
|
|
async def async_call(
|
|
self,
|
|
hass: HomeAssistant,
|
|
tool_input: llm.ToolInput,
|
|
llm_context: llm.LLMContext,
|
|
) -> JsonObjectType:
|
|
"""Forward the call when the target is exposed; unknown-tool otherwise."""
|
|
name = str(tool_input.tool_args.get("name", ""))
|
|
arguments = tool_input.tool_args.get("arguments") or {}
|
|
if name not in self._exposed_names:
|
|
return {
|
|
"error": f"Unknown tool '{name}'.",
|
|
"suggestion": (f"Use {_SEARCH_TOOL_NAME} to discover available tools."),
|
|
}
|
|
return await _forward_tool_call(hass, self._server_url, name, arguments)
|
|
|
|
|
|
@dataclass(kw_only=True)
|
|
class HaMcpLlmApi(llm.API):
|
|
"""The in-process ha-mcp server's toolset as a Home Assistant LLM API."""
|
|
|
|
server_url: str
|
|
# Valid instance modes are only tool_search and full — EXPOSURE_BOTH is
|
|
# an option value that _apis_for_mode expands into two instances and must
|
|
# never reach here. The default is the compact/safe shape, matching the
|
|
# option default (review finding: defaulting to full made an omitted
|
|
# mode maximally exposed).
|
|
mode: str = EXPOSURE_TOOL_SEARCH
|
|
|
|
async def async_get_api_instance(
|
|
self, llm_context: llm.LLMContext
|
|
) -> llm.APIInstance:
|
|
"""Fetch the current tool list and return an API instance.
|
|
|
|
Fetched fresh each conversation turn (see the module docstring); the
|
|
server's own initialize ``instructions`` become the API prompt, so the
|
|
agent gets the same guidance every MCP client gets.
|
|
"""
|
|
try:
|
|
async with (
|
|
asyncio.timeout(_LIST_TOOLS_TIMEOUT_SECONDS),
|
|
_mcp_session(self.server_url) as (
|
|
session,
|
|
init_result,
|
|
),
|
|
):
|
|
list_result = await session.list_tools()
|
|
except _transport_errors() as err:
|
|
if not _is_transport_failure(err):
|
|
raise
|
|
raise HomeAssistantError(
|
|
f"Could not reach the in-process HA-MCP server: {err}"
|
|
) from err
|
|
|
|
exposed, pinned, stamped = _partition_tools(list_result.tools)
|
|
# Never mirror or search a server-side tool that shares a synthesized
|
|
# meta-tool's name (the server's own tool-search mode registers an
|
|
# ha_search_tools) — one name, one behavior.
|
|
exposed = [
|
|
t for t in exposed if t.name not in (_SEARCH_TOOL_NAME, _CALL_TOOL_NAME)
|
|
]
|
|
if not stamped:
|
|
_LOGGER.warning(
|
|
"The running server does not stamp LLM-API exposure metadata "
|
|
"(older ha-mcp package); applying the component's built-in "
|
|
"conservative deny-list instead. Update the server package "
|
|
"for per-tool control from the settings UI."
|
|
)
|
|
|
|
prompt = init_result.instructions or _FALLBACK_API_PROMPT
|
|
# full is the explicit opt-in; anything else — including an unknown
|
|
# value — falls through to the compact/safe tool-search shape.
|
|
if self.mode == EXPOSURE_FULL:
|
|
tools = self._build_full_tools(exposed)
|
|
else:
|
|
tools = self._build_tool_search_tools(exposed, pinned)
|
|
prompt += _TOOL_SEARCH_PROMPT
|
|
|
|
return llm.APIInstance(self, prompt, llm_context, tools)
|
|
|
|
def _convert_parameters(self, tool: Any) -> vol.Schema | None:
|
|
"""Convert one tool's JSON schema, or None (logged) when it fails."""
|
|
try:
|
|
return convert_to_voluptuous(tool.inputSchema)
|
|
except Exception:
|
|
# One unconvertible schema must not take down the whole
|
|
# toolset for the conversation — skip that tool, loudly.
|
|
_LOGGER.warning(
|
|
"Skipping tool %s: could not convert its input schema",
|
|
tool.name,
|
|
exc_info=True,
|
|
)
|
|
return None
|
|
|
|
def _build_full_tools(self, exposed: list[Any]) -> list[llm.Tool]:
|
|
"""Mirror every exposed tool directly (full-catalog mode)."""
|
|
tools: list[llm.Tool] = []
|
|
for tool in exposed:
|
|
parameters = self._convert_parameters(tool)
|
|
if parameters is None:
|
|
continue
|
|
tools.append(
|
|
HaMcpTool(tool.name, tool.description, parameters, self.server_url)
|
|
)
|
|
return tools
|
|
|
|
def _build_tool_search_tools(
|
|
self, exposed: list[Any], pinned: set[str]
|
|
) -> list[llm.Tool]:
|
|
"""Build the compact catalog: mirrored pinned tools + meta-tools."""
|
|
tools: list[llm.Tool] = []
|
|
exposed_names: set[str] = set()
|
|
catalog: list[dict[str, Any]] = []
|
|
for tool in exposed:
|
|
exposed_names.add(tool.name)
|
|
catalog.append(
|
|
{
|
|
"name": tool.name,
|
|
"description": tool.description or "",
|
|
"input_schema": tool.inputSchema,
|
|
}
|
|
)
|
|
if tool.name in pinned:
|
|
parameters = self._convert_parameters(tool)
|
|
if parameters is not None:
|
|
tools.append(
|
|
HaMcpTool(
|
|
tool.name, tool.description, parameters, self.server_url
|
|
)
|
|
)
|
|
tools.append(HaMcpSearchTool(catalog))
|
|
tools.append(HaMcpCallTool(self.server_url, exposed_names))
|
|
return tools
|
|
|
|
|
|
def _apis_for_mode(
|
|
hass: HomeAssistant, entry: ConfigEntry, server_url: str, exposure: str
|
|
) -> list[HaMcpLlmApi]:
|
|
"""Build the API registration set for the configured exposure mode."""
|
|
full = HaMcpLlmApi(
|
|
hass=hass,
|
|
id=f"{DOMAIN}-{entry.entry_id}",
|
|
name=entry.title,
|
|
server_url=server_url,
|
|
mode=EXPOSURE_FULL,
|
|
)
|
|
search = HaMcpLlmApi(
|
|
hass=hass,
|
|
id=f"{DOMAIN}-{entry.entry_id}-toolsearch",
|
|
name=f"{entry.title} (tool search)",
|
|
server_url=server_url,
|
|
mode=EXPOSURE_TOOL_SEARCH,
|
|
)
|
|
if exposure == EXPOSURE_FULL:
|
|
return [full]
|
|
if exposure == EXPOSURE_BOTH:
|
|
return [full, search]
|
|
# Default and explicit tool_search both land here; an unknown stored
|
|
# value degrades to the default rather than failing bring-up.
|
|
return [search]
|
|
|
|
|
|
async def async_register_llm_api(
|
|
hass: HomeAssistant,
|
|
entry: ConfigEntry,
|
|
*,
|
|
port: int,
|
|
secret_path: str,
|
|
) -> None:
|
|
"""Register the toolset as LLM API(s) per the exposure option (advisory).
|
|
|
|
Called from the bring-up success path. Never raises — and that has to be
|
|
literal, not aspirational: any exception escaping here lands in the
|
|
bring-up's outer ``except Exception``, which tears the already-running
|
|
server down and files a "start" repair issue for what is a cosmetic
|
|
failure (review finding). Hence the broad containment: whatever goes
|
|
wrong is logged and the feature is simply absent until the next (re)load.
|
|
Cancellation (a BaseException) still propagates.
|
|
"""
|
|
try:
|
|
if not await async_probe_mcp_sdk(hass):
|
|
return
|
|
|
|
# Re-registration guard: a bring-up after a teardown that could not
|
|
# run (or a duplicate bring-up) must replace the stale registration,
|
|
# not fail on the duplicate id.
|
|
async_unregister_llm_api(hass)
|
|
|
|
exposure = str(
|
|
entry.options.get(OPT_LLM_API_EXPOSURE, DEFAULT_LLM_API_EXPOSURE)
|
|
)
|
|
server_url = f"http://127.0.0.1:{port}{secret_path}"
|
|
unsubs = [
|
|
llm.async_register_api(hass, api)
|
|
for api in _apis_for_mode(hass, entry, server_url, exposure)
|
|
]
|
|
hass.data.setdefault(DOMAIN, {})[DATA_LLM_API_UNSUB] = unsubs
|
|
except Exception:
|
|
_LOGGER.warning(
|
|
"Could not register the HA-MCP LLM API; conversation agents will "
|
|
"not see the toolset until the entry is reloaded",
|
|
exc_info=True,
|
|
)
|
|
return
|
|
# The embedded e2e (test_llm_api_registered_inside_ha) asserts on this
|
|
# message to prove the registration ran inside a real HA — keep the
|
|
# "Registered the HA-MCP toolset as LLM API" prefix stable.
|
|
_LOGGER.info(
|
|
"Registered the HA-MCP toolset as LLM API (%s mode) — select it in a "
|
|
"conversation agent's settings to chat with it (text or voice)",
|
|
exposure,
|
|
)
|
|
|
|
|
|
def async_unregister_llm_api(hass: HomeAssistant) -> None:
|
|
"""Unregister the LLM API(s) if registered (idempotent, teardown-safe)."""
|
|
unsubs = hass.data.get(DOMAIN, {}).pop(DATA_LLM_API_UNSUB, None)
|
|
if not unsubs:
|
|
return
|
|
for unsub in unsubs:
|
|
unsub()
|