Add option to require an admin user for the MCP server endpoint (#180713)

Co-authored-by: Claude <noreply@anthropic.com>

homeassistant/components/mcp_server/__init__.py

@@ -5,7 +5,7 @@ from homeassistant.helpers import config_validation as cv
 from homeassistant.helpers.typing import ConfigType
 
 from . import http
-from .const import DOMAIN
+from .const import CONF_REQUIRE_ADMIN, DOMAIN
 from .session import SessionManager
 from .types import MCPServerConfigEntry
 
@@ -23,6 +23,21 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
     return True
 
 
+async def async_migrate_entry(hass: HomeAssistant, entry: MCPServerConfigEntry) -> bool:
+    """Migrate a config entry."""
+    if entry.version == 1 and entry.minor_version == 1:
+        # 1.1 -> 1.2: Endpoints served before this option existed stay open.
+        # A disabled config entry migrates only once enabled, so keep the
+        # choice the options flow may have saved in the meantime.
+        hass.config_entries.async_update_entry(
+            entry,
+            data={CONF_REQUIRE_ADMIN: False, **entry.data},
+            minor_version=2,
+        )
+
+    return True
+
+
 async def async_setup_entry(hass: HomeAssistant, entry: MCPServerConfigEntry) -> bool:
     """Set up Model Context Protocol Server from a config entry."""
 

homeassistant/components/mcp_server/config_flow.py

@@ -15,12 +15,13 @@ from homeassistant.const import CONF_LLM_HASS_API
 from homeassistant.core import HomeAssistant, callback
 from homeassistant.helpers import llm
 from homeassistant.helpers.selector import (
+    BooleanSelector,
     SelectOptionDict,
     SelectSelector,
     SelectSelectorConfig,
 )
 
-from .const import DOMAIN
+from .const import CONF_REQUIRE_ADMIN, DOMAIN
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -68,10 +69,22 @@ def _llm_api_schema(llm_apis: dict[str, str], default: list[str]) -> vol.Schema:
     )
 
 
+def _options_schema(
+    llm_apis: dict[str, str], default: list[str], require_admin: bool
+) -> vol.Schema:
+    """Return the schema for the options flow."""
+    return _llm_api_schema(llm_apis, default).extend(
+        {
+            vol.Required(CONF_REQUIRE_ADMIN, default=require_admin): BooleanSelector(),
+        }
+    )
+
+
 class ModelContextServerProtocolConfigFlow(ConfigFlow, domain=DOMAIN):
     """Handle a config flow for Model Context Protocol Server."""
 
     VERSION = 1
+    MINOR_VERSION = 2
 
     @staticmethod
     @callback
@@ -95,7 +108,7 @@ class ModelContextServerProtocolConfigFlow(ConfigFlow, domain=DOMAIN):
             else:
                 return self.async_create_entry(
                     title=_llm_api_title(llm_apis, user_input[CONF_LLM_HASS_API]),
-                    data=user_input,
+                    data={**user_input, CONF_REQUIRE_ADMIN: True},
                 )
 
         return self.async_show_form(
@@ -137,7 +150,12 @@ class ModelContextServerProtocolOptionsFlow(OptionsFlow):
 
         return self.async_show_form(
             step_id="init",
-            data_schema=_llm_api_schema(llm_apis, current),
+            data_schema=_options_schema(
+                llm_apis,
+                current,
+                # A disabled config entry has not migrated yet
+                self.config_entry.data.get(CONF_REQUIRE_ADMIN, False),
+            ),
             description_placeholders={"more_info_url": MORE_INFO_URL},
             errors=errors,
         )

homeassistant/components/mcp_server/const.py

@@ -1,6 +1,7 @@
 """Constants for the Model Context Protocol Server integration."""
 
 DOMAIN = "mcp_server"
+CONF_REQUIRE_ADMIN = "require_admin"
 TITLE = "Model Context Protocol Server"
 # The Stateless API is no longer registered explicitly, but this
 # name may still exist in the users config entry.

homeassistant/components/mcp_server/http.py

@@ -8,8 +8,8 @@ The Streamable HTTP protocol uses these HTTP endpoints:
 - /api/mcp: The Streamable HTTP endpoint currently implements the
   stateless protocol for simplicity. This receives client requests and
   sends them to the MCP server, then waits for a response to send back to
-  the client. This serves the configured LLM APIs and does not require
-  admin access.
+  the client. This serves the configured LLM APIs and requires admin access
+  when the config entry is configured to require it.
 - /api/mcp/<API ID>: The same Streamable HTTP endpoint, but exposing a
   specific LLM API selected by its ID. These endpoints require admin access,
   except for the Assist API.
@@ -50,7 +50,7 @@ from homeassistant.core import Context, HomeAssistant, callback
 from homeassistant.exceptions import Unauthorized
 from homeassistant.helpers import llm
 
-from .const import DOMAIN
+from .const import CONF_REQUIRE_ADMIN, DOMAIN
 from .server import create_server
 from .session import Session
 from .types import MCPServerConfigEntry
@@ -93,6 +93,12 @@ def async_get_config_entry(hass: HomeAssistant) -> MCPServerConfigEntry:
     return config_entries[0]
 
 
+def _validate_admin(request: web.Request, entry: MCPServerConfigEntry) -> None:
+    """Verify the user may use the endpoints serving the configured LLM APIs."""
+    if entry.data[CONF_REQUIRE_ADMIN] and not request["hass_user"].is_admin:
+        raise Unauthorized
+
+
 @dataclass
 class Streams:
     """Pairs of streams for MCP server communication."""
@@ -173,6 +179,7 @@ class ModelContextProtocolSSEView(HomeAssistantView):
         """
         hass = request.app[KEY_HASS]
         entry = async_get_config_entry(hass)
+        _validate_admin(request, entry)
         session_manager = entry.runtime_data
 
         server, options = await create_mcp_server(
@@ -225,6 +232,7 @@ class ModelContextProtocolMessagesView(HomeAssistantView):
         """
         hass = request.app[KEY_HASS]
         config_entry = async_get_config_entry(hass)
+        _validate_admin(request, config_entry)
 
         session_manager = config_entry.runtime_data
         if (session := session_manager.get(session_id)) is None:
@@ -297,7 +305,8 @@ async def _async_handle_streamable_message(
 class ModelContextProtocolStreamableView(HomeAssistantView):
     """Model Context Protocol Streamable HTTP endpoint.
 
-    This serves the configured LLM APIs and does not require admin access.
+    This serves the configured LLM APIs and requires admin access when the
+    config entry is configured to require it.
     """
 
     name = f"{DOMAIN}:streamable"
@@ -307,6 +316,7 @@ class ModelContextProtocolStreamableView(HomeAssistantView):
         """Process JSON-RPC messages for the configured LLM APIs."""
         hass = request.app[KEY_HASS]
         entry = async_get_config_entry(hass)
+        _validate_admin(request, entry)
         return await _async_handle_streamable_message(
             request, self.context(request), entry.data[CONF_LLM_HASS_API]
         )

homeassistant/components/mcp_server/strings.json

@@ -25,10 +25,12 @@
     "step": {
       "init": {
         "data": {
-          "llm_hass_api": "[%key:common::config_flow::data::llm_hass_api%]"
+          "llm_hass_api": "[%key:common::config_flow::data::llm_hass_api%]",
+          "require_admin": "Require an administrator account"
         },
         "data_description": {
-          "llm_hass_api": "[%key:component::mcp_server::config::step::user::data_description::llm_hass_api%]"
+          "llm_hass_api": "[%key:component::mcp_server::config::step::user::data_description::llm_hass_api%]",
+          "require_admin": "Only allow administrator accounts to use the Model Context Protocol endpoint."
         },
         "description": "[%key:component::mcp_server::config::step::user::description%]"
       }

tests/components/mcp_server/conftest.py

@@ -5,7 +5,7 @@ from unittest.mock import AsyncMock, patch
 
 import pytest
 
-from homeassistant.components.mcp_server.const import DOMAIN
+from homeassistant.components.mcp_server.const import CONF_REQUIRE_ADMIN, DOMAIN
 from homeassistant.const import CONF_LLM_HASS_API
 from homeassistant.core import HomeAssistant
 from homeassistant.helpers import llm
@@ -52,16 +52,24 @@ def llm_hass_api_fixture() -> list[str]:
     return [llm.LLM_API_ASSIST]
 
 
+@pytest.fixture(name="require_admin")
+def require_admin_fixture() -> bool:
+    """Fixture for the config entry require admin option."""
+    return False
+
+
 @pytest.fixture(name="config_entry")
 def mock_config_entry(
-    hass: HomeAssistant, llm_hass_api: str | list[str]
+    hass: HomeAssistant, llm_hass_api: str | list[str], require_admin: bool
 ) -> MockConfigEntry:
     """Fixture to load the integration."""
     config_entry = MockConfigEntry(
         domain=DOMAIN,
         data={
             CONF_LLM_HASS_API: llm_hass_api,
+            CONF_REQUIRE_ADMIN: require_admin,
         },
+        minor_version=2,
     )
     config_entry.add_to_hass(hass)
     return config_entry

tests/components/mcp_server/test_config_flow.py

@@ -6,7 +6,8 @@ from unittest.mock import AsyncMock
 import pytest
 
 from homeassistant import config_entries
-from homeassistant.components.mcp_server.const import DOMAIN
+from homeassistant.components.mcp_server.const import CONF_REQUIRE_ADMIN, DOMAIN
+from homeassistant.config_entries import ConfigEntryDisabler
 from homeassistant.const import CONF_LLM_HASS_API
 from homeassistant.core import HomeAssistant
 from homeassistant.data_entry_flow import FlowResultType
@@ -43,7 +44,11 @@ async def test_form(
     assert result["type"] is FlowResultType.CREATE_ENTRY
     assert result["title"] == "Assist"
     assert len(mock_setup_entry.mock_calls) == 1
-    assert result["data"] == {CONF_LLM_HASS_API: ["assist"]}
+    assert result["minor_version"] == 2
+    assert result["data"] == {
+        CONF_LLM_HASS_API: ["assist"],
+        CONF_REQUIRE_ADMIN: True,
+    }
 
 
 @pytest.mark.parametrize(
@@ -84,17 +89,24 @@ async def test_options_flow(hass: HomeAssistant, config_entry: MockConfigEntry)
     assert result["type"] is FlowResultType.FORM
     assert result["step_id"] == "init"
     assert not result["errors"]
-    assert result["data_schema"]({}) == {CONF_LLM_HASS_API: [llm.LLM_API_ASSIST]}
+    assert result["data_schema"]({}) == {
+        CONF_LLM_HASS_API: [llm.LLM_API_ASSIST],
+        CONF_REQUIRE_ADMIN: False,
+    }
 
     result = await hass.config_entries.options.async_configure(
         result["flow_id"],
-        {CONF_LLM_HASS_API: [llm.LLM_API_ASSIST, TEST_LLM_API_ID]},
+        {
+            CONF_LLM_HASS_API: [llm.LLM_API_ASSIST, TEST_LLM_API_ID],
+            CONF_REQUIRE_ADMIN: True,
+        },
     )
     await hass.async_block_till_done()
 
     assert result["type"] is FlowResultType.CREATE_ENTRY
     assert config_entry.data == {
-        CONF_LLM_HASS_API: [llm.LLM_API_ASSIST, TEST_LLM_API_ID]
+        CONF_LLM_HASS_API: [llm.LLM_API_ASSIST, TEST_LLM_API_ID],
+        CONF_REQUIRE_ADMIN: True,
     }
     assert config_entry.title == "Assist, Test"
 
@@ -109,16 +121,22 @@ async def test_options_flow_legacy_single_api(
     result = await hass.config_entries.options.async_init(config_entry.entry_id)
 
     assert result["type"] is FlowResultType.FORM
-    assert result["data_schema"]({}) == {CONF_LLM_HASS_API: [llm.LLM_API_ASSIST]}
+    assert result["data_schema"]({}) == {
+        CONF_LLM_HASS_API: [llm.LLM_API_ASSIST],
+        CONF_REQUIRE_ADMIN: False,
+    }
 
     result = await hass.config_entries.options.async_configure(
         result["flow_id"],
-        {CONF_LLM_HASS_API: [llm.LLM_API_ASSIST]},
+        {CONF_LLM_HASS_API: [llm.LLM_API_ASSIST], CONF_REQUIRE_ADMIN: False},
     )
     await hass.async_block_till_done()
 
     assert result["type"] is FlowResultType.CREATE_ENTRY
-    assert config_entry.data == {CONF_LLM_HASS_API: [llm.LLM_API_ASSIST]}
+    assert config_entry.data == {
+        CONF_LLM_HASS_API: [llm.LLM_API_ASSIST],
+        CONF_REQUIRE_ADMIN: False,
+    }
 
 
 async def test_options_flow_keeps_custom_title(
@@ -132,12 +150,15 @@ async def test_options_flow_keeps_custom_title(
     result = await hass.config_entries.options.async_init(config_entry.entry_id)
     result = await hass.config_entries.options.async_configure(
         result["flow_id"],
-        {CONF_LLM_HASS_API: [TEST_LLM_API_ID]},
+        {CONF_LLM_HASS_API: [TEST_LLM_API_ID], CONF_REQUIRE_ADMIN: False},
     )
     await hass.async_block_till_done()
 
     assert result["type"] is FlowResultType.CREATE_ENTRY
-    assert config_entry.data == {CONF_LLM_HASS_API: [TEST_LLM_API_ID]}
+    assert config_entry.data == {
+        CONF_LLM_HASS_API: [TEST_LLM_API_ID],
+        CONF_REQUIRE_ADMIN: False,
+    }
     assert config_entry.title == "My MCP server"
 
 
@@ -152,7 +173,7 @@ async def test_options_flow_errors(
 
     result = await hass.config_entries.options.async_configure(
         result["flow_id"],
-        {CONF_LLM_HASS_API: []},
+        {CONF_LLM_HASS_API: [], CONF_REQUIRE_ADMIN: False},
     )
 
     assert result["type"] is FlowResultType.FORM
@@ -160,9 +181,39 @@ async def test_options_flow_errors(
 
     result = await hass.config_entries.options.async_configure(
         result["flow_id"],
-        {CONF_LLM_HASS_API: [llm.LLM_API_ASSIST]},
+        {CONF_LLM_HASS_API: [llm.LLM_API_ASSIST], CONF_REQUIRE_ADMIN: False},
     )
     await hass.async_block_till_done()
 
     assert result["type"] is FlowResultType.CREATE_ENTRY
-    assert config_entry.data == {CONF_LLM_HASS_API: [llm.LLM_API_ASSIST]}
+    assert config_entry.data == {
+        CONF_LLM_HASS_API: [llm.LLM_API_ASSIST],
+        CONF_REQUIRE_ADMIN: False,
+    }
+
+
+async def test_options_flow_unmigrated_entry(hass: HomeAssistant) -> None:
+    """Test the options flow on a disabled config entry that has not migrated."""
+    config_entry = MockConfigEntry(
+        domain=DOMAIN,
+        data={CONF_LLM_HASS_API: [llm.LLM_API_ASSIST]},
+        minor_version=1,
+        disabled_by=ConfigEntryDisabler.USER,
+    )
+    config_entry.add_to_hass(hass)
+
+    result = await hass.config_entries.options.async_init(config_entry.entry_id)
+    assert result["type"] is FlowResultType.FORM
+    assert result["data_schema"]({}) == {
+        CONF_LLM_HASS_API: [llm.LLM_API_ASSIST],
+        CONF_REQUIRE_ADMIN: False,
+    }
+
+    result = await hass.config_entries.options.async_configure(
+        result["flow_id"],
+        {CONF_LLM_HASS_API: [llm.LLM_API_ASSIST], CONF_REQUIRE_ADMIN: True},
+    )
+    await hass.async_block_till_done()
+
+    assert result["type"] is FlowResultType.CREATE_ENTRY
+    assert config_entry.data[CONF_REQUIRE_ADMIN] is True

tests/components/mcp_server/test_http.py

@@ -735,3 +735,62 @@ async def test_streamable_api_id_unknown(
     )
     assert response.status == HTTPStatus.NOT_FOUND
     assert "Unknown LLM API" in await response.text()
+
+
+@pytest.mark.parametrize(
+    ("require_admin", "expected_status"),
+    [
+        pytest.param(False, HTTPStatus.OK, id="not_required"),
+        pytest.param(True, HTTPStatus.UNAUTHORIZED, id="required"),
+    ],
+)
+async def test_require_admin_option(
+    hass: HomeAssistant,
+    setup_integration: None,
+    hass_client: ClientSessionGenerator,
+    hass_read_only_access_token: str,
+    expected_status: HTTPStatus,
+) -> None:
+    """Test the require admin option applied to a non-admin user."""
+    client = await hass_client(hass_read_only_access_token)
+
+    response = await client.post(
+        STREAMABLE_API,
+        json=INITIALIZE_MESSAGE,
+        headers={"accept": CONTENT_TYPE_JSON},
+    )
+    assert response.status == expected_status
+
+
+@pytest.mark.parametrize("require_admin", [True])
+async def test_require_admin_blocks_sse_endpoints(
+    hass: HomeAssistant,
+    setup_integration: None,
+    hass_client: ClientSessionGenerator,
+    hass_read_only_access_token: str,
+) -> None:
+    """Test the require admin option applied to the SSE endpoints."""
+    client = await hass_client(hass_read_only_access_token)
+
+    response = await client.get(SSE_API)
+    assert response.status == HTTPStatus.UNAUTHORIZED
+
+    response = await client.post(MESSAGES_API.format(session_id="session-id"))
+    assert response.status == HTTPStatus.UNAUTHORIZED
+
+
+@pytest.mark.parametrize("require_admin", [True])
+async def test_require_admin_allows_admin(
+    hass: HomeAssistant,
+    setup_integration: None,
+    hass_client: ClientSessionGenerator,
+) -> None:
+    """Test an admin user may use the endpoint that requires an admin."""
+    client = await hass_client()
+
+    response = await client.post(
+        STREAMABLE_API,
+        json=INITIALIZE_MESSAGE,
+        headers={"accept": CONTENT_TYPE_JSON},
+    )
+    assert response.status == HTTPStatus.OK

tests/components/mcp_server/test_init.py

@@ -1,7 +1,10 @@
 """Test the Model Context Protocol Server init module."""
 
+from homeassistant.components.mcp_server.const import CONF_REQUIRE_ADMIN, DOMAIN
 from homeassistant.config_entries import ConfigEntryState
+from homeassistant.const import CONF_LLM_HASS_API
 from homeassistant.core import HomeAssistant
+from homeassistant.helpers import llm
 
 from tests.common import MockConfigEntry
 
@@ -13,3 +16,38 @@ async def test_init(hass: HomeAssistant, config_entry: MockConfigEntry) -> None:
 
     await hass.config_entries.async_unload(config_entry.entry_id)
     assert config_entry.state is ConfigEntryState.NOT_LOADED
+
+
+async def test_migrate_entry_require_admin(hass: HomeAssistant) -> None:
+    """Test an entry created before the require admin option keeps the endpoints open."""
+    config_entry = MockConfigEntry(
+        domain=DOMAIN,
+        data={CONF_LLM_HASS_API: [llm.LLM_API_ASSIST]},
+        minor_version=1,
+    )
+    config_entry.add_to_hass(hass)
+
+    await hass.config_entries.async_setup(config_entry.entry_id)
+    assert config_entry.state is ConfigEntryState.LOADED
+
+    assert config_entry.minor_version == 2
+    assert config_entry.data == {
+        CONF_LLM_HASS_API: [llm.LLM_API_ASSIST],
+        CONF_REQUIRE_ADMIN: False,
+    }
+
+
+async def test_migrate_entry_keeps_require_admin(hass: HomeAssistant) -> None:
+    """Test the migration keeps an option the options flow saved before it ran."""
+    config_entry = MockConfigEntry(
+        domain=DOMAIN,
+        data={CONF_LLM_HASS_API: [llm.LLM_API_ASSIST], CONF_REQUIRE_ADMIN: True},
+        minor_version=1,
+    )
+    config_entry.add_to_hass(hass)
+
+    await hass.config_entries.async_setup(config_entry.entry_id)
+    assert config_entry.state is ConfigEntryState.LOADED
+
+    assert config_entry.minor_version == 2
+    assert config_entry.data[CONF_REQUIRE_ADMIN] is True