Add the Model Context Protocol integration

.strict-typing

@@ -311,6 +311,7 @@ homeassistant.components.manual.*
 homeassistant.components.mastodon.*
 homeassistant.components.matrix.*
 homeassistant.components.matter.*
+homeassistant.components.mcp.*
 homeassistant.components.mcp_server.*
 homeassistant.components.mealie.*
 homeassistant.components.media_extractor.*

homeassistant/components/mcp/__init__.py

@@ -0,0 +1,108 @@
+"""The Model Context Protocol integration."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+import httpx
+from mcp import types
+from voluptuous_openapi import convert_to_voluptuous
+
+from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import HomeAssistantError
+from homeassistant.helpers import llm
+from homeassistant.util.json import JsonObjectType
+
+from .const import DOMAIN
+from .coordinator import ModelContextProtocolCoordinator
+from .types import ModelContextProtocolConfigEntry
+
+__all__ = [
+    "DOMAIN",
+    "async_setup_entry",
+    "async_unload_entry",
+]
+
+API_PROMPT = "The following tools are available from a remote server named {name}."
+
+
+async def async_setup_entry(
+    hass: HomeAssistant, entry: ModelContextProtocolConfigEntry
+) -> bool:
+    """Set up Model Context Protocol from a config entry."""
+    coordinator = ModelContextProtocolCoordinator(hass, entry)
+    await coordinator.async_config_entry_first_refresh()
+
+    unsub = llm.async_register_api(
+        hass,
+        ModelContextProtocolAPI(
+            hass=hass,
+            id=f"{DOMAIN}-{entry.entry_id}",
+            name=entry.title,
+            coordinator=coordinator,
+        ),
+    )
+    entry.async_on_unload(unsub)
+
+    entry.runtime_data = coordinator
+    entry.async_on_unload(coordinator.close)
+
+    return True
+
+
+async def async_unload_entry(
+    hass: HomeAssistant, entry: ModelContextProtocolConfigEntry
+) -> bool:
+    """Unload a config entry."""
+    return True
+
+
+class ModelContextProtocolTool(llm.Tool):
+    """A Tool exposed over the Model Context Protocol."""
+
+    def __init__(
+        self,
+        tool: types.Tool,
+        coordinator: ModelContextProtocolCoordinator,
+    ) -> None:
+        """Initialize the tool."""
+        self.name = tool.name
+        self.description = tool.description
+        self.parameters = convert_to_voluptuous(tool.inputSchema)
+        self.coordinator = coordinator
+
+    async def async_call(
+        self,
+        hass: HomeAssistant,
+        tool_input: llm.ToolInput,
+        llm_context: llm.LLMContext,
+    ) -> JsonObjectType:
+        """Call the tool."""
+        session = self.coordinator.session
+        try:
+            result = await session.call_tool(tool_input.tool_name, tool_input.tool_args)
+        except httpx.HTTPStatusError as error:
+            raise HomeAssistantError(f"Error when calling tool: {error}") from error
+        return result.model_dump(exclude_unset=True, exclude_none=True)
+
+
+@dataclass(kw_only=True)
+class ModelContextProtocolAPI(llm.API):
+    """Define an object to hold the Model Context Protocol API."""
+
+    coordinator: ModelContextProtocolCoordinator
+
+    async def async_get_api_instance(
+        self, llm_context: llm.LLMContext
+    ) -> llm.APIInstance:
+        """Return the instance of the API."""
+        tools: list[llm.Tool] = [
+            ModelContextProtocolTool(tool, self.coordinator)
+            for tool in self.coordinator.data
+        ]
+        return llm.APIInstance(
+            self,
+            API_PROMPT.format(name=self.name),
+            llm_context,
+            tools=tools,
+        )

homeassistant/components/mcp/config_flow.py

@@ -0,0 +1,111 @@
+"""Config flow for the Model Context Protocol integration."""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+import httpx
+import voluptuous as vol
+
+from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
+from homeassistant.const import CONF_URL
+from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import HomeAssistantError
+from homeassistant.helpers import config_validation as cv
+
+from .const import DOMAIN
+from .coordinator import mcp_client
+
+_LOGGER = logging.getLogger(__name__)
+
+STEP_USER_DATA_SCHEMA = vol.Schema(
+    {
+        vol.Required(CONF_URL): str,
+    }
+)
+
+
+async def validate_input(hass: HomeAssistant, data: dict[str, Any]) -> dict[str, Any]:
+    """Validate the user input and connect to the MCP server."""
+    url = data[CONF_URL]
+    try:
+        cv.url(url)  # Cannot be added to schema directly
+    except vol.Invalid as error:
+        raise InvalidUrl from error
+    try:
+        async with mcp_client(url) as session:
+            response = await session.initialize()
+    except httpx.TimeoutException as error:
+        _LOGGER.info("Timeout connecting to MCP server: %s", error)
+        raise TimeoutConnectError from error
+    except httpx.HTTPStatusError as error:
+        _LOGGER.info("Cannot connect to MCP server: %s", error)
+        if error.response.status_code == 401:
+            raise InvalidAuth from error
+        raise CannotConnect from error
+    except httpx.HTTPError as error:
+        _LOGGER.info("Cannot connect to MCP server: %s", error)
+        raise CannotConnect from error
+
+    if not response.capabilities.tools:
+        raise MissingCapabilities(
+            f"MCP Server {url} does not support 'Tools' capability"
+        )
+
+    return {"title": response.serverInfo.name}
+
+
+class ModelContextProtocolConfigFlow(ConfigFlow, domain=DOMAIN):
+    """Handle a config flow for Model Context Protocol."""
+
+    VERSION = 1
+
+    async def async_step_user(
+        self, user_input: dict[str, Any] | None = None
+    ) -> ConfigFlowResult:
+        """Handle the initial step."""
+        errors: dict[str, str] = {}
+        if user_input is not None:
+            try:
+                info = await validate_input(self.hass, user_input)
+            except InvalidUrl:
+                errors[CONF_URL] = "invalid_url"
+            except TimeoutConnectError:
+                errors["base"] = "timeout_connect"
+            except CannotConnect:
+                errors["base"] = "cannot_connect"
+            except InvalidAuth:
+                return self.async_abort(reason="invalid_auth")
+            except MissingCapabilities:
+                return self.async_abort(reason="missing_capabilities")
+            except Exception:
+                _LOGGER.exception("Unexpected exception")
+                errors["base"] = "unknown"
+            else:
+                self._async_abort_entries_match({CONF_URL: user_input[CONF_URL]})
+                return self.async_create_entry(title=info["title"], data=user_input)
+
+        return self.async_show_form(
+            step_id="user", data_schema=STEP_USER_DATA_SCHEMA, errors=errors
+        )
+
+
+class InvalidUrl(HomeAssistantError):
+    """Error to indicate the URL format is invalid."""
+
+
+class CannotConnect(HomeAssistantError):
+    """Error to indicate we cannot connect."""
+
+
+class TimeoutConnectError(HomeAssistantError):
+    """Error to indicate we cannot connect."""
+
+
+class InvalidAuth(HomeAssistantError):
+    """Error to indicate there is invalid auth."""
+
+
+class MissingCapabilities(HomeAssistantError):
+    """Error to indicate that the MCP server is missing required capabilities."""

homeassistant/components/mcp/const.py

@@ -0,0 +1,3 @@
+"""Constants for the Model Context Protocol integration."""
+
+DOMAIN = "mcp"

homeassistant/components/mcp/coordinator.py

@@ -0,0 +1,79 @@
+"""Types for the Model Context Protocol integration."""
+
+from collections.abc import AsyncGenerator
+from contextlib import AbstractAsyncContextManager, asynccontextmanager
+import datetime
+import logging
+
+import httpx
+from mcp.client.session import ClientSession
+from mcp.client.sse import sse_client
+from mcp.types import Tool
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import CONF_URL
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
+
+from .const import DOMAIN
+
+_LOGGER = logging.getLogger(__name__)
+
+UPDATE_INTERVAL = datetime.timedelta(minutes=30)
+
+
+@asynccontextmanager
+async def mcp_client(url: str) -> AsyncGenerator[ClientSession]:
+    """Create a server-sent event MCP client.
+
+    This is an asynccontxt manager that wraps to other async context managers
+    so that the coordinator has a single object to manage.
+    """
+    try:
+        async with sse_client(url=url) as streams, ClientSession(*streams) as session:
+            await session.initialize()
+            yield session
+    except ExceptionGroup as err:
+        raise err.exceptions[0] from err
+
+
+class ModelContextProtocolCoordinator(DataUpdateCoordinator[list[Tool]]):
+    """Define an object to hold MCP data."""
+
+    config_entry: ConfigEntry
+    session: ClientSession
+
+    def __init__(self, hass: HomeAssistant, config_entry: ConfigEntry) -> None:
+        """Initialize ModelContextProtocolCoordinator."""
+        super().__init__(
+            hass,
+            logger=_LOGGER,
+            name=DOMAIN,
+            config_entry=config_entry,
+            update_interval=UPDATE_INTERVAL,
+        )
+        self.ctx_mgr: AbstractAsyncContextManager[ClientSession]
+
+    async def _async_setup(self) -> None:
+        """Set up the client connection."""
+        self.ctx_mgr = mcp_client(self.config_entry.data[CONF_URL])
+        try:
+            self.session = await self.ctx_mgr.__aenter__()  # pylint: disable=unnecessary-dunder-call
+        except httpx.HTTPError as err:
+            raise UpdateFailed(f"Error communicating with MCP server: {err}") from err
+
+    async def close(self) -> None:
+        """Close the client connection."""
+        await self.ctx_mgr.__aexit__(None, None, None)
+
+    async def _async_update_data(self) -> list[Tool]:
+        """Fetch data from API endpoint.
+
+        This is the place to pre-process the data to lookup tables
+        so entities can quickly look up their data.
+        """
+        try:
+            result = await self.session.list_tools()
+        except httpx.HTTPError as err:
+            raise UpdateFailed(f"Error communicating with API: {err}") from err
+        return result.tools

homeassistant/components/mcp/manifest.json

@@ -0,0 +1,14 @@
+{
+  "domain": "mcp",
+  "name": "Model Context Protocol",
+  "codeowners": ["@allenporter"],
+  "config_flow": true,
+  "dependencies": [],
+  "documentation": "https://www.home-assistant.io/integrations/mcp",
+  "homekit": {},
+  "iot_class": "local_polling",
+  "quality_scale": "silver",
+  "requirements": ["mcp==1.1.2"],
+  "ssdp": [],
+  "zeroconf": []
+}