Add the Model Context Protocol integration

.strict-typing

@@ -312,6 +312,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,69 @@
+"""The Model Context Protocol integration."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers import llm
+
+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
+
+
+@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."""
+        return llm.APIInstance(
+            self,
+            API_PROMPT.format(name=self.name),
+            llm_context,
+            tools=self.coordinator.data,
+        )

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,171 @@
+"""Types for the Model Context Protocol integration."""
+
+import asyncio
+from collections.abc import AsyncGenerator
+from contextlib import asynccontextmanager
+import datetime
+import logging
+
+import httpx
+from mcp.client.session import ClientSession
+from mcp.client.sse import sse_client
+import voluptuous as vol
+from voluptuous_openapi import convert_to_voluptuous
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import CONF_URL
+from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import HomeAssistantError
+from homeassistant.helpers import llm
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
+from homeassistant.util.json import JsonObjectType
+
+from .const import DOMAIN
+
+_LOGGER = logging.getLogger(__name__)
+
+UPDATE_INTERVAL = datetime.timedelta(minutes=30)
+TIMEOUT = 10
+
+
+@asynccontextmanager
+async def mcp_client(url: str) -> AsyncGenerator[ClientSession]:
+    """Create a server-sent event MCP client.
+
+    This is an asynccontext manager that exists to wrap 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 ModelContextProtocolTool(llm.Tool):
+    """A Tool exposed over the Model Context Protocol."""
+
+    def __init__(
+        self,
+        name: str,
+        description: str | None,
+        parameters: vol.Schema,
+        session: ClientSession,
+    ) -> None:
+        """Initialize the tool."""
+        self.name = name
+        self.description = description
+        self.parameters = parameters
+        self.session = session
+
+    async def async_call(
+        self,
+        hass: HomeAssistant,
+        tool_input: llm.ToolInput,
+        llm_context: llm.LLMContext,
+    ) -> JsonObjectType:
+        """Call the tool."""
+        try:
+            result = await self.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)
+
+
+class ModelContextProtocolCoordinator(DataUpdateCoordinator[list[llm.Tool]]):
+    """Define an object to hold MCP data."""
+
+    config_entry: ConfigEntry
+    _session: ClientSession | None = None
+    _setup_error: Exception | None = None
+
+    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._stop = asyncio.Event()
+
+    async def _async_setup(self) -> None:
+        """Set up the client connection."""
+        connected = asyncio.Event()
+        stop = asyncio.Event()
+        self.config_entry.async_create_background_task(
+            self.hass, self._connect(connected, stop), "mcp-client"
+        )
+        try:
+            async with asyncio.timeout(TIMEOUT):
+                await connected.wait()
+                self._stop = stop
+        finally:
+            if self._setup_error is not None:
+                raise self._setup_error
+
+    async def _connect(self, connected: asyncio.Event, stop: asyncio.Event) -> None:
+        """Create a server-sent event MCP client."""
+        url = self.config_entry.data[CONF_URL]
+        try:
+            async with (
+                sse_client(url=url) as streams,
+                ClientSession(*streams) as session,
+            ):
+                await session.initialize()
+                self._session = session
+                connected.set()
+                await stop.wait()
+        except httpx.HTTPStatusError as err:
+            self._setup_error = err
+            _LOGGER.debug("Error connecting to MCP server: %s", err)
+            raise UpdateFailed(f"Error connecting to MCP server: {err}") from err
+        except ExceptionGroup as err:
+            self._setup_error = err.exceptions[0]
+            _LOGGER.debug("Error connecting to MCP server: %s", err)
+            raise UpdateFailed(
+                "Error connecting to MCP server: {err.exceptions[0]}"
+            ) from err.exceptions[0]
+        finally:
+            self._session = None
+
+    async def close(self) -> None:
+        """Close the client connection."""
+        if self._stop is not None:
+            self._stop.set()
+
+    async def _async_update_data(self) -> list[llm.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.
+        """
+        if self._session is None:
+            raise UpdateFailed("No session available")
+        try:
+            result = await self._session.list_tools()
+        except httpx.HTTPError as err:
+            raise UpdateFailed(f"Error communicating with API: {err}") from err
+
+        _LOGGER.debug("Received tools: %s", result.tools)
+        tools: list[llm.Tool] = []
+        for tool in result.tools:
+            try:
+                parameters = convert_to_voluptuous(tool.inputSchema)
+            except Exception as err:
+                raise UpdateFailed(
+                    f"Error converting schema {err}: {tool.inputSchema}"
+                ) from err
+            tools.append(
+                ModelContextProtocolTool(
+                    tool.name,
+                    tool.description,
+                    parameters,
+                    self._session,
+                )
+            )
+        return tools

homeassistant/components/mcp/manifest.json

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