Python: fail fast on MCP streamable HTTP 401/403

python/semantic_kernel/connectors/mcp.py

@@ -13,6 +13,7 @@
 from itertools import chain
 from typing import TYPE_CHECKING, Any
 
+import httpx
 from mcp import types
 from mcp.client.session import ClientSession
 from mcp.client.sse import sse_client
@@ -316,13 +317,15 @@ async def _inner_connect(self, ready_event: asyncio.Event) -> None:
                 )
             except Exception as ex:
                 await self._exit_stack.aclose()
+                ready_event.set()
                 raise KernelPluginInvalidConfigurationError(
                     "Failed to create a session. Please check your configuration."
                 ) from ex
             try:
                 await session.initialize()
             except Exception as ex:
                 await self._exit_stack.aclose()
+                ready_event.set()
                 raise KernelPluginInvalidConfigurationError(
                     "Failed to initialize session. Please check your configuration."
                 ) from ex
@@ -760,6 +763,29 @@ def get_mcp_client(self) -> _AsyncGeneratorContextManager[Any, None]:
             args.update(self._client_kwargs)
         return streamablehttp_client(**args)
 
+    async def __aenter__(self) -> Self:
+        """Fail fast on authentication/authorization errors before connecting."""
+        if not self.session:
+            timeout = self.timeout if self.timeout is not None else 30
+            client_kwargs: dict[str, Any] = {"timeout": timeout, "headers": self.headers}
+            if "auth" in self._client_kwargs:
+                client_kwargs["auth"] = self._client_kwargs["auth"]
+            try:
+                async with httpx.AsyncClient(**client_kwargs) as client:
+                    response = await client.get(self.url)
+                    if response.status_code in (httpx.codes.UNAUTHORIZED, httpx.codes.FORBIDDEN):
+                        raise KernelPluginInvalidConfigurationError(
+                            f"Failed to connect to the MCP server: received HTTP {response.status_code} (unauthorized/forbidden)."
+                        )
+            except KernelPluginInvalidConfigurationError:
+                raise
+            except Exception as ex:  # pragma: no cover - guarded for unexpected failures
+                raise KernelPluginInvalidConfigurationError(
+                    "Failed to connect to the MCP server. Please check your configuration."
+                ) from ex
+
+        return await super().__aenter__()
+
 
 class MCPWebsocketPlugin(MCPPluginBase):
     """MCP websocket server configuration."""

python/tests/unit/connectors/mcp/test_mcp.py

@@ -241,6 +241,119 @@ async def test_with_kwargs_streamablehttp(mock_session, mock_client, list_tool_c
         assert len(loaded_plugin.functions["func2"].parameters) == 0
 
 
+class _DummyHttpxClient:
+    def __init__(self, status_code: int):
+        self.status_code = status_code
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, exc_type, exc_val, exc_tb):
+        return False
+
+    async def get(self, url: str):  # pragma: no cover - trivial
+        import httpx
+
+        return httpx.Response(self.status_code, request=httpx.Request("GET", url))
+
+
+@pytest.mark.parametrize("status_code", [401, 403])
+@patch("semantic_kernel.connectors.mcp.ClientSession")
+@patch("semantic_kernel.connectors.mcp.streamablehttp_client")
+async def test_streamable_http_raises_on_401_403(mock_streamable_client, mock_session, monkeypatch, status_code):
+    monkeypatch.setattr("semantic_kernel.connectors.mcp.httpx.AsyncClient", lambda **_: _DummyHttpxClient(status_code))
+
+    with pytest.raises(KernelPluginInvalidConfigurationError):
+        async with MCPStreamableHttpPlugin(
+            name="TestMCPPlugin",
+            description="Test MCP Plugin",
+            url="http://localhost:8080/mcp",
+        ):
+            pass
+
+
+@patch("semantic_kernel.connectors.mcp.ClientSession")
+@patch("semantic_kernel.connectors.mcp.streamablehttp_client")
+async def test_streamable_http_allows_success(mock_streamable_client, mock_session, monkeypatch):
+    monkeypatch.setattr("semantic_kernel.connectors.mcp.httpx.AsyncClient", lambda **_: _DummyHttpxClient(200))
+
+    mock_read = MagicMock()
+    mock_write = MagicMock()
+    mock_callback = MagicMock()
+
+    mock_generator = MagicMock()
+    mock_generator.__aenter__.return_value = (mock_read, mock_write, mock_callback)
+    mock_generator.__aexit__.return_value = (mock_read, mock_write, mock_callback)
+    mock_streamable_client.return_value = mock_generator
+    mock_session.return_value.__aenter__.return_value.list_tools.return_value = []
+    mock_session.return_value.initialize = AsyncMock()
+
+    async with MCPStreamableHttpPlugin(
+        name="TestMCPPlugin",
+        description="Test MCP Plugin",
+        url="http://localhost:8080/mcp",
+    ) as plugin:
+        assert plugin is not None
+
+
+@patch("semantic_kernel.connectors.mcp.ClientSession")
+@patch("semantic_kernel.connectors.mcp.streamablehttp_client")
+async def test_streamable_http_skips_preflight_with_existing_session(mock_streamable_client, mock_session):
+    """Preflight is skipped when a session is already provided."""
+    existing_session = MagicMock()
+    existing_session._request_id = 1
+    existing_session.list_tools = AsyncMock(return_value=MagicMock(tools=[]))
+
+    plugin = MCPStreamableHttpPlugin(
+        name="TestMCPPlugin",
+        description="Test MCP Plugin",
+        url="http://localhost:8080/mcp",
+        session=existing_session,
+    )
+    await plugin.connect()
+    assert plugin.session is existing_session
+    await plugin.close()
+
+
+@patch("semantic_kernel.connectors.mcp.streamablehttp_client")
+async def test_streamable_http_session_create_failure_does_not_hang(mock_streamable_client):
+    """ready_event.set() prevents connect() from hanging when session creation fails."""
+    mock_streamable_client.side_effect = RuntimeError("connection refused")
+
+    with pytest.raises(KernelPluginInvalidConfigurationError, match="create a session"):
+        async with MCPStreamableHttpPlugin(
+            name="TestMCPPlugin",
+            description="Test MCP Plugin",
+            url="http://localhost:8080/mcp",
+        ):
+            pass
+
+
+@patch("semantic_kernel.connectors.mcp.ClientSession")
+@patch("semantic_kernel.connectors.mcp.streamablehttp_client")
+async def test_streamable_http_session_init_failure_does_not_hang(mock_streamable_client, mock_session, monkeypatch):
+    """ready_event.set() prevents connect() from hanging when session.initialize() fails."""
+    monkeypatch.setattr("semantic_kernel.connectors.mcp.httpx.AsyncClient", lambda **_: _DummyHttpxClient(200))
+
+    mock_read = MagicMock()
+    mock_write = MagicMock()
+    mock_callback = MagicMock()
+
+    mock_generator = MagicMock()
+    mock_generator.__aenter__.return_value = (mock_read, mock_write, mock_callback)
+    mock_generator.__aexit__.return_value = (mock_read, mock_write, mock_callback)
+    mock_streamable_client.return_value = mock_generator
+    mock_session.return_value.initialize = AsyncMock(side_effect=RuntimeError("init failed"))
+
+    with pytest.raises(KernelPluginInvalidConfigurationError, match="initialize session"):
+        async with MCPStreamableHttpPlugin(
+            name="TestMCPPlugin",
+            description="Test MCP Plugin",
+            url="http://localhost:8080/mcp",
+        ):
+            pass
+
+
 async def test_kernel_as_mcp_server(kernel: "Kernel", decorated_native_function, custom_plugin_class):
     kernel.add_plugin(custom_plugin_class, "test")
     kernel.add_functions("test", [decorated_native_function])