Fix session group refresh 3713

docs/client/session-groups.md

@@ -66,6 +66,8 @@ Run it again. `print(sorted(group.tools))` now shows both:
 
 If you already hold a connected `ClientSession` (`Client.session` is one), hand it to `await group.connect_with_session(server_info, session)` instead of opening a new transport. It aggregates the same way. The group never closes a session it didn't open. `server_info` names the server for component prefixes; on a 2026-era connection `client.server_info` can be `None` (identity is optional), so pass your own `Implementation(name=..., version=...)` in that case.
 
+Calling `connect_with_session` with an already-connected session refreshes that session's aggregated components to match its current catalog. Any components the server no longer exposes are removed from the group, and newly added ones are registered, without triggering duplicate-name collisions against the session's own previous registrations. If querying a component type fails with `MCPError` during refresh, existing registrations for that component type remain intact. If any component collides with a *different* session in the group, an `MCPError` is raised and the group's existing state is preserved.
+
 ## The classic handshake
 
 `ClientSessionGroup` is built on `ClientSession`, not on `Client`. Each `connect_to_server` runs the classic `initialize` handshake. It never sends the `server/discover` probe described in **[Protocol versions](../protocol-versions.md)**. Every MCP server understands that handshake, so this costs you compatibility with nothing; it only means a group takes the older, slower path to a server that could do better.
@@ -77,6 +79,6 @@ If you already hold a connected `ClientSession` (`Client.session` is one), hand
 * `group.call_tool(name, arguments)` routes to the owning server for you.
 * Names must be unique across the whole group; two servers with a `search` tool cannot coexist on their own.
 * `component_name_hook=` rewrites every registered name. The dict key changes, the wire name does not.
-* `connect_with_session` adds a session you already hold; `disconnect_from_server` removes one.
+* `connect_with_session` adds a session you already hold (or refreshes an existing one); `disconnect_from_server` removes one.
 
 The handshake a group speaks (and the faster one a `Client` prefers) is the subject of **[Protocol versions](../protocol-versions.md)**.

src/mcp/client/session_group.py

@@ -175,7 +175,7 @@ async def __aexit__(
     @property
     def sessions(self) -> list[mcp.ClientSession]:
         """Returns the list of sessions being managed."""
-        return list(self._sessions.keys())  # pragma: no cover
+        return list(self._sessions.keys())
 
     @property
     def prompts(self) -> dict[str, types.Prompt]:
@@ -283,8 +283,8 @@ async def disconnect_from_server(self, session: mcp.ClientSession) -> None:
 
         # Clean up the session's resources via its dedicated exit stack
         if session_known_for_stack:
-            session_stack_to_close = self._session_exit_stacks.pop(session)  # pragma: no cover
-            await session_stack_to_close.aclose()  # pragma: no cover
+            session_stack_to_close = self._session_exit_stacks.pop(session)
+            await session_stack_to_close.aclose()
 
     async def connect_with_session(
         self, server_info: types.Implementation, session: mcp.ClientSession
@@ -378,9 +378,14 @@ async def _aggregate_components(self, server_info: types.Implementation, session
         # tools belonging to this session. Used for removing components from
         # the session group via self.disconnect_from_server.
         component_names = self._ComponentNames()
-
-        # Temporary components dicts. We do not want to modify the aggregate
-        # lists in case of an intermediate failure.
+        own_components = self._sessions.get(session)
+
+        # Temporary component dictionaries to stage changes. We query and validate
+        # all components before mutating group state so that any failure leaves
+        # existing registrations intact. If querying a component type fails with
+        # MCPError during a refresh, existing registrations for that component type
+        # are preserved rather than evicted (a failed fetch is not evidence that the
+        # server removed those components).
         prompts_temp: dict[str, types.Prompt] = {}
         resources_temp: dict[str, types.Resource] = {}
         tools_temp: dict[str, types.Tool] = {}
@@ -393,8 +398,12 @@ async def _aggregate_components(self, server_info: types.Implementation, session
                 name = self._component_name(prompt.name, server_info)
                 prompts_temp[name] = prompt
                 component_names.prompts.add(name)
-        except MCPError as err:  # pragma: no cover
+        except MCPError as err:
             logging.warning(f"Could not fetch prompts: {err}")
+            if own_components:
+                for name in own_components.prompts:
+                    prompts_temp[name] = self._prompts[name]
+                    component_names.prompts.add(name)
 
         # Query the server for its resources and aggregate to list.
         try:
@@ -403,8 +412,12 @@ async def _aggregate_components(self, server_info: types.Implementation, session
                 name = self._component_name(resource.name, server_info)
                 resources_temp[name] = resource
                 component_names.resources.add(name)
-        except MCPError as err:  # pragma: no cover
+        except MCPError as err:
             logging.warning(f"Could not fetch resources: {err}")
+            if own_components:
+                for name in own_components.resources:
+                    resources_temp[name] = self._resources[name]
+                    component_names.resources.add(name)
 
         # Query the server for its tools and aggregate to list.
         try:
@@ -414,31 +427,46 @@ async def _aggregate_components(self, server_info: types.Implementation, session
                 tools_temp[name] = tool
                 tool_to_session_temp[name] = session
                 component_names.tools.add(name)
-        except MCPError as err:  # pragma: no cover
+        except MCPError as err:
             logging.warning(f"Could not fetch tools: {err}")
-
-        # Clean up exit stack for session if we couldn't retrieve anything
-        # from the server.
-        if not any((prompts_temp, resources_temp, tools_temp)):
-            del self._session_exit_stacks[session]  # pragma: no cover
-
-        # Check for duplicates.
-        matching_prompts = prompts_temp.keys() & self._prompts.keys()
+            if own_components:
+                for name in own_components.tools:
+                    tools_temp[name] = self._tools[name]
+                    tool_to_session_temp[name] = session
+                    component_names.tools.add(name)
+
+        # Check for duplicates, ignoring components already registered by this session.
+        other_prompts = self._prompts.keys() - own_components.prompts if own_components else self._prompts.keys()
+        matching_prompts = prompts_temp.keys() & other_prompts
         if matching_prompts:
             raise MCPError(  # pragma: no cover
                 code=types.INVALID_PARAMS,
                 message=f"{matching_prompts} already exist in group prompts.",
             )
-        matching_resources = resources_temp.keys() & self._resources.keys()
+        other_resources = (
+            self._resources.keys() - own_components.resources if own_components else self._resources.keys()
+        )
+        matching_resources = resources_temp.keys() & other_resources
         if matching_resources:
             raise MCPError(  # pragma: no cover
                 code=types.INVALID_PARAMS,
                 message=f"{matching_resources} already exist in group resources.",
             )
-        matching_tools = tools_temp.keys() & self._tools.keys()
+        other_tools = self._tools.keys() - own_components.tools if own_components else self._tools.keys()
+        matching_tools = tools_temp.keys() & other_tools
         if matching_tools:
             raise MCPError(code=types.INVALID_PARAMS, message=f"{matching_tools} already exist in group tools.")
 
+        # Clean up any previously registered components for this session before aggregating new ones.
+        if own_components:
+            for name in own_components.prompts:
+                self._prompts.pop(name, None)
+            for name in own_components.resources:
+                self._resources.pop(name, None)
+            for name in own_components.tools:
+                self._tools.pop(name, None)
+                self._tool_to_session.pop(name, None)
+
         # Aggregate components.
         self._sessions[session] = component_names
         self._prompts.update(prompts_temp)