Handle late MCP OAuth authentication failures (#175381)

homeassistant/components/mcp/auth.py

@@ -0,0 +1,36 @@
+"""Authentication helper classes for the Model Context Protocol integration."""
+
+from dataclasses import dataclass
+import re
+
+import httpx
+from yarl import URL
+
+# Headers and regex for WWW-Authenticate parsing for rfc9728
+WWW_AUTHENTICATE_HEADER = "WWW-Authenticate"
+RESOURCE_METADATA_REGEXP = r'resource_metadata="([^"]+)"'
+SCOPES_REGEXP = r'scope="([^"]+)"'
+
+
+@dataclass
+class AuthenticateHeader:
+    """Class to hold info from the WWW-Authenticate header for supporting rfc9728."""
+
+    resource_metadata_url: str
+    scopes: list[str] | None = None
+
+    @classmethod
+    def from_header(
+        cls, url: str, error_response: httpx.Response
+    ) -> AuthenticateHeader | None:
+        """Create AuthenticateHeader from WWW-Authenticate header."""
+        if not (header := error_response.headers.get(WWW_AUTHENTICATE_HEADER)) or not (
+            match := re.search(RESOURCE_METADATA_REGEXP, header)
+        ):
+            return None
+        resource_metadata_url = str(URL(url).join(URL(match.group(1))))
+        scope_match = re.search(SCOPES_REGEXP, header)
+        return cls(
+            resource_metadata_url=resource_metadata_url,
+            scopes=scope_match.group(1).split(" ") if scope_match else None,
+        )

homeassistant/components/mcp/config_flow.py

@@ -4,7 +4,6 @@ import asyncio
 from collections.abc import Iterable, Mapping
 from dataclasses import dataclass
 import logging
-import re
 from typing import Any, cast, override
 
 import httpx
@@ -24,6 +23,7 @@ from homeassistant.helpers.config_entry_oauth2_flow import (
 
 from . import async_get_config_entry_implementation
 from .application_credentials import authorization_server_context
+from .auth import AuthenticateHeader
 from .const import CONF_AUTHORIZATION_URL, CONF_SCOPE, CONF_TOKEN_URL, DOMAIN
 from .coordinator import TokenManager, mcp_client
 
@@ -35,35 +35,7 @@ STEP_USER_DATA_SCHEMA = vol.Schema(
     }
 )
 
-# Headers and regex for WWW-Authenticate parsing for rfc9728
-WWW_AUTHENTICATE_HEADER = "WWW-Authenticate"
-RESOURCE_METADATA_REGEXP = r'resource_metadata="([^"]+)"'
 OAUTH_PROTECTED_RESOURCE_ENDPOINT = "/.well-known/oauth-protected-resource"
-SCOPES_REGEXP = r'scope="([^"]+)"'
-
-
-@dataclass
-class AuthenticateHeader:
-    """Class to hold info from the WWW-Authenticate header for supporting rfc9728."""
-
-    resource_metadata_url: str
-    scopes: list[str] | None = None
-
-    @classmethod
-    def from_header(
-        cls, url: str, error_response: httpx.Response
-    ) -> AuthenticateHeader | None:
-        """Create AuthenticateHeader from WWW-Authenticate header."""
-        if not (header := error_response.headers.get(WWW_AUTHENTICATE_HEADER)) or not (
-            match := re.search(RESOURCE_METADATA_REGEXP, header)
-        ):
-            return None
-        resource_metadata_url = str(URL(url).join(URL(match.group(1))))
-        scope_match = re.search(SCOPES_REGEXP, header)
-        return cls(
-            resource_metadata_url=resource_metadata_url,
-            scopes=scope_match.group(1).split(" ") if scope_match else None,
-        )
 
 
 @dataclass
@@ -369,6 +341,8 @@ class ModelContextProtocolConfigFlow(AbstractOAuth2FlowHandler, domain=DOMAIN):
         self, entry_data: Mapping[str, Any]
     ) -> ConfigFlowResult:
         """Perform reauth upon an API authentication error."""
+        if entry_data and "auth_header" in entry_data:
+            self.auth_header = entry_data["auth_header"]
         return await self.async_step_reauth_confirm()
 
     async def async_step_reauth_confirm(
@@ -379,6 +353,13 @@ class ModelContextProtocolConfigFlow(AbstractOAuth2FlowHandler, domain=DOMAIN):
             return self.async_show_form(step_id="reauth_confirm")
         config_entry = self._get_reauth_entry()
         self.data = {**config_entry.data}
+        if "auth_implementation" not in self.data:
+            # For entries configured without authentication (no-auth), any authentication
+            # failure (from a tool call or coordinator update) requires upgrading to OAuth.
+            # We bypass validate_input connection handshake (which might succeed if the server
+            # doesn't restrict the connection handshake itself) and proceed directly to OAuth discovery.
+            return await self.async_step_auth_discovery()
+
         self.flow_impl = await async_get_config_entry_implementation(  # type: ignore[assignment]
             self.hass, config_entry
         )

homeassistant/components/mcp/coordinator.py

@@ -18,12 +18,17 @@ 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 ConfigEntryAuthFailed, HomeAssistantError
+from homeassistant.exceptions import (
+    ConfigEntryAuthFailed,
+    HomeAssistantError,
+    OAuth2TokenRequestReauthError,
+)
 from homeassistant.helpers import llm
 from homeassistant.helpers.httpx_client import create_async_httpx_client
 from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
 from homeassistant.util.json import JsonObjectType
 
+from .auth import AuthenticateHeader
 from .const import DOMAIN
 
 _LOGGER = logging.getLogger(__name__)
@@ -98,6 +103,7 @@ class ModelContextProtocolTool(llm.Tool):
         description: str | None,
         parameters: vol.Schema,
         server_url: str,
+        config_entry: ConfigEntry,
         token_manager: TokenManager | None = None,
     ) -> None:
         """Initialize the tool."""
@@ -105,6 +111,7 @@ class ModelContextProtocolTool(llm.Tool):
         self.description = description
         self.parameters = parameters
         self.server_url = server_url
+        self.config_entry = config_entry
         self.token_manager = token_manager
 
     @override
@@ -126,9 +133,32 @@ class ModelContextProtocolTool(llm.Tool):
         except TimeoutError as error:
             _LOGGER.debug("Timeout when calling tool: %s", error)
             raise HomeAssistantError(f"Timeout when calling tool: {error}") from error
+        except OAuth2TokenRequestReauthError as error:
+            _LOGGER.debug("OAuth token request failed when calling tool: %s", error)
+            self.config_entry.async_start_reauth(hass)
+            raise ConfigEntryAuthFailed(
+                "OAuth token request failed when calling tool"
+            ) from error
         except httpx.HTTPStatusError as error:
             _LOGGER.debug("Error when calling tool: %s", error)
+            if error.response.status_code == 401:
+                auth_header = AuthenticateHeader.from_header(
+                    self.server_url, error.response
+                )
+                self.config_entry.async_start_reauth(
+                    hass, data={"auth_header": auth_header}
+                )
+                raise ConfigEntryAuthFailed(
+                    "The MCP server requires authentication"
+                ) from error
             raise HomeAssistantError(f"Error when calling tool: {error}") from error
+        except httpx.HTTPError as error:
+            _LOGGER.debug(
+                "Error communicating with MCP server when calling tool: %s", error
+            )
+            raise HomeAssistantError(
+                f"Error communicating with MCP server when calling tool: {error}"
+            ) from error
         return result.model_dump(exclude_unset=True, exclude_none=True)
 
 
@@ -169,9 +199,18 @@ class ModelContextProtocolCoordinator(DataUpdateCoordinator[list[llm.Tool]]):
         except TimeoutError as error:
             _LOGGER.debug("Timeout when listing tools: %s", error)
             raise UpdateFailed(f"Timeout when listing tools: {error}") from error
+        except OAuth2TokenRequestReauthError as error:
+            _LOGGER.debug("OAuth token request failed: %s", error)
+            raise ConfigEntryAuthFailed("OAuth token request failed") from error
         except httpx.HTTPStatusError as error:
             _LOGGER.debug("Error communicating with API: %s", error)
-            if error.response.status_code == 401 and self.token_manager is not None:
+            if error.response.status_code == 401:
+                auth_header = AuthenticateHeader.from_header(
+                    self.config_entry.data[CONF_URL], error.response
+                )
+                self.config_entry.async_start_reauth(
+                    self.hass, data={"auth_header": auth_header}
+                )
                 raise ConfigEntryAuthFailed(
                     "The MCP server requires authentication"
                 ) from error
@@ -195,6 +234,7 @@ class ModelContextProtocolCoordinator(DataUpdateCoordinator[list[llm.Tool]]):
                     tool.description,
                     parameters,
                     self.config_entry.data[CONF_URL],
+                    self.config_entry,
                     self.token_manager,
                 )
             )

tests/components/mcp/test_config_flow.py

@@ -9,6 +9,7 @@ import pytest
 import respx
 
 from homeassistant import config_entries
+from homeassistant.components.mcp.auth import AuthenticateHeader
 from homeassistant.components.mcp.const import (
     CONF_AUTHORIZATION_URL,
     CONF_SCOPE,
@@ -892,3 +893,135 @@ async def test_reauth_flow(
     assert token == OAUTH_TOKEN_PAYLOAD
 
     assert len(mock_setup_entry.mock_calls) == 1
+
+
+@pytest.mark.usefixtures("current_request_with_host")
+@respx.mock
+async def test_reauth_flow_upgrade_to_oauth(
+    hass: HomeAssistant,
+    mock_setup_entry: AsyncMock,
+    mock_mcp_client: Mock,
+    credential: None,
+    aioclient_mock: AiohttpClientMocker,
+    hass_client_no_auth: ClientSessionGenerator,
+) -> None:
+    """Test reauth flow upgrading a no-auth entry to OAuth."""
+    config_entry = MockConfigEntry(
+        domain=DOMAIN,
+        data={CONF_URL: MCP_SERVER_URL},
+        title=TEST_API_NAME,
+    )
+    config_entry.add_to_hass(hass)
+
+    auth_header = AuthenticateHeader(
+        resource_metadata_url="https://example.com/custom-discovery",
+        scopes=SCOPES_SUPPORTED,
+    )
+
+    # Start reauth flow passing auth_header
+    config_entry.async_start_reauth(hass, data={"auth_header": auth_header})
+    await hass.async_block_till_done()
+
+    flows = hass.config_entries.flow.async_progress()
+    assert len(flows) == 1
+    result = flows[0]
+    assert result["step_id"] == "reauth_confirm"
+
+    # Mock discovery URLs (bypassing connection validation)
+    respx.get("https://example.com/custom-discovery").mock(
+        return_value=OAUTH_PROTECTED_RESOURCE_METADATA_RESPONSE
+    )
+    respx.get(OAUTH_AUTHORIZATION_SERVER_DISCOVERY_ENDPOINT).mock(
+        return_value=OAUTH_SERVER_METADATA_RESPONSE
+    )
+
+    # Click Submit on reauth_confirm
+    result = await hass.config_entries.flow.async_configure(result["flow_id"], {})
+
+    # Flow should proceed to credentials choice
+    assert result["type"] is FlowResultType.MENU
+    assert result["step_id"] == "credentials_choice"
+
+    result = await hass.config_entries.flow.async_configure(
+        result["flow_id"],
+        {
+            "next_step_id": "pick_implementation",
+        },
+    )
+    assert result["type"] is FlowResultType.EXTERNAL_STEP
+    result = await perform_oauth_flow(
+        hass,
+        aioclient_mock,
+        hass_client_no_auth,
+        result,
+        authorize_url=OAUTH_AUTHORIZE_URL,
+        token_url=OAUTH_TOKEN_URL,
+        scopes=SCOPES_SUPPORTED,
+    )
+
+    # Verify we can connect to the server now with the token
+    response = Mock()
+    response.serverInfo.name = TEST_API_NAME
+    # Return success for validation in async_oauth_create_entry
+    mock_mcp_client.return_value.initialize.return_value = response
+
+    result = await hass.config_entries.flow.async_configure(result["flow_id"])
+    assert result["type"] is FlowResultType.ABORT
+    assert result["reason"] == "reauth_successful"
+
+    assert config_entry.unique_id is None
+    assert config_entry.title == TEST_API_NAME
+    data = {**config_entry.data}
+    token = data.pop(CONF_TOKEN)
+    assert data == {
+        "auth_implementation": AUTH_DOMAIN,
+        CONF_URL: MCP_SERVER_URL,
+        CONF_AUTHORIZATION_URL: OAUTH_AUTHORIZE_URL,
+        CONF_TOKEN_URL: OAUTH_TOKEN_URL,
+        CONF_SCOPE: SCOPES_SUPPORTED,
+    }
+    assert token
+    token.pop("expires_at")
+    assert token == OAUTH_TOKEN_PAYLOAD
+
+    assert len(mock_setup_entry.mock_calls) == 1
+
+
+@pytest.mark.usefixtures("current_request_with_host")
+@respx.mock
+async def test_reauth_flow_upgrade_to_oauth_no_auth_header(
+    hass: HomeAssistant,
+    mock_setup_entry: AsyncMock,
+    mock_mcp_client: Mock,
+    credential: None,
+    aioclient_mock: AiohttpClientMocker,
+    hass_client_no_auth: ClientSessionGenerator,
+) -> None:
+    """Test reauth flow upgrading a no-auth entry to OAuth when no auth header is passed (fallback)."""
+    config_entry = MockConfigEntry(
+        domain=DOMAIN,
+        data={CONF_URL: MCP_SERVER_URL},
+        title=TEST_API_NAME,
+    )
+    config_entry.add_to_hass(hass)
+
+    # Start reauth flow without passing auth_header
+    config_entry.async_start_reauth(hass)
+    await hass.async_block_till_done()
+
+    flows = hass.config_entries.flow.async_progress()
+    assert len(flows) == 1
+    result = flows[0]
+    assert result["step_id"] == "reauth_confirm"
+
+    # Mock discovery on the default server URL (since there is no auth_header)
+    respx.get(OAUTH_DISCOVERY_ENDPOINT).mock(
+        return_value=OAUTH_SERVER_METADATA_RESPONSE
+    )
+
+    # Click Submit on reauth_confirm
+    result = await hass.config_entries.flow.async_configure(result["flow_id"], {})
+
+    # Flow should proceed directly to credentials choice menu (without validate_input)
+    assert result["type"] is FlowResultType.MENU
+    assert result["step_id"] == "credentials_choice"

tests/components/mcp/test_init.py

@@ -9,9 +9,15 @@ from mcp.types import CallToolResult, ErrorData, ListToolsResult, TextContent, T
 import pytest
 import voluptuous as vol
 
+from homeassistant.components.mcp.const import DOMAIN
 from homeassistant.config_entries import ConfigEntryState
 from homeassistant.core import Context, HomeAssistant
-from homeassistant.exceptions import HomeAssistantError
+from homeassistant.exceptions import (
+    ConfigEntryAuthFailed,
+    HomeAssistantError,
+    OAuth2TokenRequestError,
+    OAuth2TokenRequestReauthError,
+)
 from homeassistant.helpers import llm
 from homeassistant.helpers.config_entry_oauth2_flow import (
     ImplementationUnavailableError,
@@ -84,7 +90,6 @@ async def test_init(
     [
         (httpx.TimeoutException("Some timeout")),
         (httpx.HTTPStatusError("", request=None, response=httpx.Response(500))),
-        (httpx.HTTPStatusError("", request=None, response=httpx.Response(401))),
         (httpx.HTTPError("Some HTTP error")),
     ],
 )
@@ -104,6 +109,55 @@ async def test_mcp_server_failure(
     assert config_entry.state is ConfigEntryState.SETUP_RETRY
 
 
+async def test_mcp_server_setup_auth_failure(
+    hass: HomeAssistant,
+    config_entry: MockConfigEntry,
+    mock_mcp_client: Mock,
+) -> None:
+    """Test setup auth failure triggers reauth."""
+    mock_mcp_client.side_effect = httpx.HTTPStatusError(
+        "Authentication required", request=None, response=httpx.Response(401)
+    )
+
+    await hass.config_entries.async_setup(config_entry.entry_id)
+    assert config_entry.state is ConfigEntryState.SETUP_ERROR
+
+    flows = hass.config_entries.flow.async_progress()
+    assert len(flows) == 1
+    assert flows[0]["step_id"] == "reauth_confirm"
+
+
+async def test_mcp_server_setup_auth_failure_with_www_authenticate_header(
+    hass: HomeAssistant,
+    config_entry: MockConfigEntry,
+    mock_mcp_client: Mock,
+) -> None:
+    """Test setup auth failure with WWW-Authenticate header parses header and triggers reauth."""
+    headers = {
+        "WWW-Authenticate": 'mcp resource_metadata="https://example.com/custom-discovery", scope="read write"'
+    }
+    mock_mcp_client.side_effect = httpx.HTTPStatusError(
+        "Authentication required",
+        request=None,
+        response=httpx.Response(401, headers=headers),
+    )
+
+    await hass.config_entries.async_setup(config_entry.entry_id)
+    assert config_entry.state is ConfigEntryState.SETUP_ERROR
+
+    flows = hass.config_entries.flow.async_progress()
+    assert len(flows) == 1
+    assert flows[0]["step_id"] == "reauth_confirm"
+
+    # Get the flow handler instance and verify it has the correct auth_header
+    flow_handler = hass.config_entries.flow._progress[flows[0]["flow_id"]]
+    assert flow_handler.auth_header is not None
+    assert (
+        flow_handler.auth_header.resource_metadata_url
+        == "https://example.com/custom-discovery"
+    )
+
+
 async def test_mcp_server_http_transport_failure(
     hass: HomeAssistant,
     config_entry: MockConfigEntry,
@@ -361,3 +415,293 @@ async def test_oauth_implementation_not_available(
         await hass.async_block_till_done()
 
     assert config_entry_with_auth.state is ConfigEntryState.SETUP_RETRY
+
+
+async def test_tool_call_no_auth_auth_failure(
+    hass: HomeAssistant,
+    config_entry: MockConfigEntry,
+    mock_mcp_client: Mock,
+) -> None:
+    """Test tool call auth failure when no auth was initially required."""
+    mock_mcp_client.return_value.list_tools.return_value = ListToolsResult(
+        tools=[SEARCH_MEMORY_TOOL]
+    )
+
+    await hass.config_entries.async_setup(config_entry.entry_id)
+    assert config_entry.state is ConfigEntryState.LOADED
+
+    apis = llm.async_get_apis(hass)
+    api = next(iter([api for api in apis if api.name == TEST_API_NAME]))
+    api_instance = await api.async_get_api_instance(create_llm_context())
+    tool = api_instance.tools[0]
+
+    # Mock tool call encountering a 401 response
+    mock_mcp_client.return_value.call_tool.side_effect = httpx.HTTPStatusError(
+        "Authentication required", request=None, response=httpx.Response(401)
+    )
+
+    with pytest.raises(ConfigEntryAuthFailed):
+        await tool.async_call(
+            hass,
+            llm.ToolInput(
+                tool_name="search_memory", tool_args={"query": "User's birth month"}
+            ),
+            create_llm_context(),
+        )
+
+    flows = hass.config_entries.flow.async_progress()
+    assert len(flows) == 1
+    assert flows[0]["step_id"] == "reauth_confirm"
+
+
+async def test_tool_call_no_auth_auth_failure_with_www_authenticate_header(
+    hass: HomeAssistant,
+    config_entry: MockConfigEntry,
+    mock_mcp_client: Mock,
+) -> None:
+    """Test tool call 401 with WWW-Authenticate header triggers reauth and passes header."""
+    mock_mcp_client.return_value.list_tools.return_value = ListToolsResult(
+        tools=[SEARCH_MEMORY_TOOL]
+    )
+
+    await hass.config_entries.async_setup(config_entry.entry_id)
+    assert config_entry.state is ConfigEntryState.LOADED
+
+    apis = llm.async_get_apis(hass)
+    api = next(iter([api for api in apis if api.name == TEST_API_NAME]))
+    api_instance = await api.async_get_api_instance(create_llm_context())
+    tool = api_instance.tools[0]
+
+    # Mock tool call encountering a 401 response with WWW-Authenticate header
+    headers = {
+        "WWW-Authenticate": 'mcp resource_metadata="https://example.com/custom-discovery", scope="read write"'
+    }
+    mock_mcp_client.return_value.call_tool.side_effect = httpx.HTTPStatusError(
+        "Authentication required",
+        request=None,
+        response=httpx.Response(401, headers=headers),
+    )
+
+    with pytest.raises(ConfigEntryAuthFailed):
+        await tool.async_call(
+            hass,
+            llm.ToolInput(
+                tool_name="search_memory", tool_args={"query": "User's birth month"}
+            ),
+            create_llm_context(),
+        )
+
+    flows = hass.config_entries.flow.async_progress()
+    assert len(flows) == 1
+    assert flows[0]["step_id"] == "reauth_confirm"
+
+    # Get the flow handler instance and verify it has the correct auth_header
+    flow_handler = hass.config_entries.flow._progress[flows[0]["flow_id"]]
+    assert flow_handler.auth_header is not None
+    assert (
+        flow_handler.auth_header.resource_metadata_url
+        == "https://example.com/custom-discovery"
+    )
+
+
+async def test_tool_call_expired_oauth_failure(
+    hass: HomeAssistant,
+    credential: None,
+    config_entry_with_auth: MockConfigEntry,
+    mock_mcp_client: Mock,
+) -> None:
+    """Test tool call token refresh failure when OAuth is configured."""
+    mock_mcp_client.return_value.list_tools.return_value = ListToolsResult(
+        tools=[SEARCH_MEMORY_TOOL]
+    )
+
+    await hass.config_entries.async_setup(config_entry_with_auth.entry_id)
+    assert config_entry_with_auth.state is ConfigEntryState.LOADED
+
+    apis = llm.async_get_apis(hass)
+    api = next(iter([api for api in apis if api.name == TEST_API_NAME]))
+    api_instance = await api.async_get_api_instance(create_llm_context())
+    tool = api_instance.tools[0]
+
+    # Mock token validation failure during tool call
+    with (
+        patch(
+            "homeassistant.helpers.config_entry_oauth2_flow.OAuth2Session.async_ensure_token_valid",
+            side_effect=OAuth2TokenRequestReauthError(
+                request_info=Mock(), history=(), domain=DOMAIN
+            ),
+        ),
+        pytest.raises(ConfigEntryAuthFailed),
+    ):
+        await tool.async_call(
+            hass,
+            llm.ToolInput(
+                tool_name="search_memory", tool_args={"query": "User's birth month"}
+            ),
+            create_llm_context(),
+        )
+
+    flows = hass.config_entries.flow.async_progress()
+    assert len(flows) == 1
+    assert flows[0]["step_id"] == "reauth_confirm"
+
+
+async def test_mcp_server_setup_oauth_failure(
+    hass: HomeAssistant,
+    credential: None,
+    config_entry_with_auth: MockConfigEntry,
+) -> None:
+    """Test setup OAuth failure triggers reauth."""
+    # Mock token validation failure (e.g. refresh token expired)
+    with patch(
+        "homeassistant.helpers.config_entry_oauth2_flow.OAuth2Session.async_ensure_token_valid",
+        side_effect=OAuth2TokenRequestReauthError(
+            request_info=Mock(), history=(), domain=DOMAIN
+        ),
+    ):
+        await hass.config_entries.async_setup(config_entry_with_auth.entry_id)
+        assert config_entry_with_auth.state is ConfigEntryState.SETUP_ERROR
+
+    flows = hass.config_entries.flow.async_progress()
+    assert len(flows) == 1
+    assert flows[0]["step_id"] == "reauth_confirm"
+
+
+async def test_list_tools_timeout(
+    hass: HomeAssistant, config_entry: MockConfigEntry, mock_mcp_client: Mock
+) -> None:
+    """Test setup fails with SETUP_RETRY if list tools times out."""
+    mock_mcp_client.return_value.list_tools.side_effect = TimeoutError(
+        "Listing tools timed out"
+    )
+
+    await hass.config_entries.async_setup(config_entry.entry_id)
+    assert config_entry.state is ConfigEntryState.SETUP_RETRY
+
+
+async def test_tool_call_timeout(
+    hass: HomeAssistant,
+    config_entry: MockConfigEntry,
+    mock_mcp_client: Mock,
+) -> None:
+    """Test tool call timing out raises HomeAssistantError."""
+    mock_mcp_client.return_value.list_tools.return_value = ListToolsResult(
+        tools=[SEARCH_MEMORY_TOOL]
+    )
+
+    await hass.config_entries.async_setup(config_entry.entry_id)
+    assert config_entry.state is ConfigEntryState.LOADED
+
+    apis = llm.async_get_apis(hass)
+    api = next(iter([api for api in apis if api.name == TEST_API_NAME]))
+    api_instance = await api.async_get_api_instance(create_llm_context())
+    tool = api_instance.tools[0]
+
+    # Mock tool call timeout
+    mock_mcp_client.return_value.call_tool.side_effect = TimeoutError("Call timed out")
+
+    with pytest.raises(HomeAssistantError, match="Timeout when calling tool"):
+        await tool.async_call(
+            hass,
+            llm.ToolInput(
+                tool_name="search_memory", tool_args={"query": "User's birth month"}
+            ),
+            create_llm_context(),
+        )
+
+
+async def test_tool_call_transient_oauth_failure(
+    hass: HomeAssistant,
+    credential: None,
+    config_entry_with_auth: MockConfigEntry,
+    mock_mcp_client: Mock,
+) -> None:
+    """Test tool call transient token refresh failure does not trigger reauth."""
+    mock_mcp_client.return_value.list_tools.return_value = ListToolsResult(
+        tools=[SEARCH_MEMORY_TOOL]
+    )
+
+    await hass.config_entries.async_setup(config_entry_with_auth.entry_id)
+    assert config_entry_with_auth.state is ConfigEntryState.LOADED
+
+    apis = llm.async_get_apis(hass)
+    api = next(iter([api for api in apis if api.name == TEST_API_NAME]))
+    api_instance = await api.async_get_api_instance(create_llm_context())
+    tool = api_instance.tools[0]
+
+    # Mock transient token validation failure (e.g. 503 Service Unavailable)
+    with (
+        patch(
+            "homeassistant.helpers.config_entry_oauth2_flow.OAuth2Session.async_ensure_token_valid",
+            side_effect=OAuth2TokenRequestError(
+                request_info=Mock(), history=(), domain=DOMAIN
+            ),
+        ),
+        pytest.raises(HomeAssistantError),
+    ):
+        await tool.async_call(
+            hass,
+            llm.ToolInput(
+                tool_name="search_memory", tool_args={"query": "User's birth month"}
+            ),
+            create_llm_context(),
+        )
+
+    # Verify no reauth flow is initiated
+    flows = hass.config_entries.flow.async_progress()
+    assert len(flows) == 0
+
+
+async def test_mcp_server_setup_transient_oauth_failure(
+    hass: HomeAssistant,
+    credential: None,
+    config_entry_with_auth: MockConfigEntry,
+) -> None:
+    """Test setup transient OAuth failure does not trigger reauth."""
+    with patch(
+        "homeassistant.helpers.config_entry_oauth2_flow.OAuth2Session.async_ensure_token_valid",
+        side_effect=OAuth2TokenRequestError(
+            request_info=Mock(), history=(), domain=DOMAIN
+        ),
+    ):
+        await hass.config_entries.async_setup(config_entry_with_auth.entry_id)
+        assert config_entry_with_auth.state is ConfigEntryState.SETUP_RETRY
+
+    flows = hass.config_entries.flow.async_progress()
+    assert len(flows) == 0
+
+
+async def test_tool_call_http_error(
+    hass: HomeAssistant,
+    config_entry: MockConfigEntry,
+    mock_mcp_client: Mock,
+) -> None:
+    """Test tool call HTTP error raises HomeAssistantError."""
+    mock_mcp_client.return_value.list_tools.return_value = ListToolsResult(
+        tools=[SEARCH_MEMORY_TOOL]
+    )
+
+    await hass.config_entries.async_setup(config_entry.entry_id)
+    assert config_entry.state is ConfigEntryState.LOADED
+
+    apis = llm.async_get_apis(hass)
+    api = next(iter([api for api in apis if api.name == TEST_API_NAME]))
+    api_instance = await api.async_get_api_instance(create_llm_context())
+    tool = api_instance.tools[0]
+
+    # Mock tool call raising HTTPError
+    mock_mcp_client.return_value.call_tool.side_effect = httpx.HTTPError(
+        "Connection timed out or failed"
+    )
+
+    with pytest.raises(
+        HomeAssistantError,
+        match="Error communicating with MCP server when calling tool",
+    ):
+        await tool.async_call(
+            hass,
+            llm.ToolInput(
+                tool_name="search_memory", tool_args={"query": "User's birth month"}
+            ),
+            create_llm_context(),
+        )