fix(clientsessiongroup): only query negotiated capabilities

ClientSessionGroup._aggregate_components queried prompts, resources, and
tools unconditionally on every connect, ignoring the ServerCapabilities
returned by initialize(). A server that advertised only some of these
(e.g. tools) returned JSON-RPC "Method not found" for the rest, which was
swallowed into spurious WARNING logs.

The MCP lifecycle spec requires clients to only use capabilities that were
successfully negotiated. Gate each list_* call on the matching capability
from session.initialize_result.capabilities, falling back to the prior
unconditional behavior when initialize_result is absent so the existing
MCPError handler still covers servers that advertise a capability but fail
the method.

Fixes #2689

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: mr-brobot

src/mcp/client/session_group.py

@@ -344,36 +344,48 @@ async def _aggregate_components(self, server_info: types.Implementation, session
         tools_temp: dict[str, types.Tool] = {}
         tool_to_session_temp: dict[str, mcp.ClientSession] = {}
 
+        # Per the MCP lifecycle spec, clients must only use capabilities the
+        # server successfully negotiated, so gate each list_* call on the
+        # matching capability rather than issuing requests the server never
+        # advertised. initialize_result is populated once initialize() has
+        # completed; if it is missing, fall back to the prior unconditional
+        # behavior and let the MCPError handler cope with servers that
+        # advertise a capability but still fail the method.
+        capabilities = session.initialize_result.capabilities if session.initialize_result is not None else None
+
         # Query the server for its prompts and aggregate to list.
-        try:
-            prompts = (await session.list_prompts()).prompts
-            for prompt in prompts:
-                name = self._component_name(prompt.name, server_info)
-                prompts_temp[name] = prompt
-                component_names.prompts.add(name)
-        except MCPError as err:  # pragma: no cover
-            logging.warning(f"Could not fetch prompts: {err}")
+        if capabilities is None or capabilities.prompts is not None:
+            try:
+                prompts = (await session.list_prompts()).prompts
+                for prompt in prompts:
+                    name = self._component_name(prompt.name, server_info)
+                    prompts_temp[name] = prompt
+                    component_names.prompts.add(name)
+            except MCPError as err:  # pragma: no cover
+                logging.warning(f"Could not fetch prompts: {err}")
 
         # Query the server for its resources and aggregate to list.
-        try:
-            resources = (await session.list_resources()).resources
-            for resource in resources:
-                name = self._component_name(resource.name, server_info)
-                resources_temp[name] = resource
-                component_names.resources.add(name)
-        except MCPError as err:  # pragma: no cover
-            logging.warning(f"Could not fetch resources: {err}")
+        if capabilities is None or capabilities.resources is not None:
+            try:
+                resources = (await session.list_resources()).resources
+                for resource in resources:
+                    name = self._component_name(resource.name, server_info)
+                    resources_temp[name] = resource
+                    component_names.resources.add(name)
+            except MCPError as err:  # pragma: no cover
+                logging.warning(f"Could not fetch resources: {err}")
 
         # Query the server for its tools and aggregate to list.
-        try:
-            tools = (await session.list_tools()).tools
-            for tool in tools:
-                name = self._component_name(tool.name, server_info)
-                tools_temp[name] = tool
-                tool_to_session_temp[name] = session
-                component_names.tools.add(name)
-        except MCPError as err:  # pragma: no cover
-            logging.warning(f"Could not fetch tools: {err}")
+        if capabilities is None or capabilities.tools is not None:
+            try:
+                tools = (await session.list_tools()).tools
+                for tool in tools:
+                    name = self._component_name(tool.name, server_info)
+                    tools_temp[name] = tool
+                    tool_to_session_temp[name] = session
+                    component_names.tools.add(name)
+            except MCPError as err:  # pragma: no cover
+                logging.warning(f"Could not fetch tools: {err}")
 
         # Clean up exit stack for session if we couldn't retrieve anything
         # from the server.

tests/client/test_session_group.py

@@ -385,3 +385,46 @@ async def test_client_session_group_establish_session_parameterized(
             # 3. Assert returned values
             assert returned_server_info is mock_initialize_result.server_info
             assert returned_session is mock_entered_session
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize("advertised", ["tools", "prompts", "resources"])
+async def test_client_session_group_skips_unsupported_capabilities(advertised: str):
+    """Only the capability the server advertised is queried during aggregation."""
+    mock_session = mock.AsyncMock(spec=mcp.ClientSession)
+    mock_session.initialize_result = types.InitializeResult(
+        protocol_version=types.LATEST_PROTOCOL_VERSION,
+        capabilities=types.ServerCapabilities(
+            tools=types.ToolsCapability() if advertised == "tools" else None,
+            prompts=types.PromptsCapability() if advertised == "prompts" else None,
+            resources=types.ResourcesCapability() if advertised == "resources" else None,
+        ),
+        server_info=types.Implementation(name="srv", version="1"),
+    )
+    mock_tool = mock.Mock(spec=types.Tool)
+    mock_tool.name = "tool_a"
+    mock_resource = mock.Mock(spec=types.Resource)
+    mock_resource.name = "resource_b"
+    mock_prompt = mock.Mock(spec=types.Prompt)
+    mock_prompt.name = "prompt_c"
+    mock_session.list_tools.return_value = mock.AsyncMock(tools=[mock_tool])
+    mock_session.list_resources.return_value = mock.AsyncMock(resources=[mock_resource])
+    mock_session.list_prompts.return_value = mock.AsyncMock(prompts=[mock_prompt])
+
+    group = ClientSessionGroup()
+    await group.connect_with_session(types.Implementation(name="srv", version="1"), mock_session)
+
+    list_methods = {
+        "tools": mock_session.list_tools,
+        "prompts": mock_session.list_prompts,
+        "resources": mock_session.list_resources,
+    }
+    for capability, list_method in list_methods.items():
+        if capability == advertised:
+            list_method.assert_awaited_once()
+        else:
+            list_method.assert_not_awaited()
+
+    assert group.tools == ({"tool_a": mock_tool} if advertised == "tools" else {})
+    assert group.prompts == ({"prompt_c": mock_prompt} if advertised == "prompts" else {})
+    assert group.resources == ({"resource_b": mock_resource} if advertised == "resources" else {})