refactor: address PR review feedback on transport abstractions

- Remove __init__ and WireMessageT from AbstractBaseSession, making it a pure
  abstract interface without state management. BaseSession now owns all state.

- Convert BaseClientSession from ABC to @runtime_checkable Protocol with structural
  subtyping. This eliminates the diamond inheritance in ClientSession and removes
  the need for inheritance to satisfy the interface.

- Add missing methods to BaseClientSession protocol: complete(), set_logging_level(),
  and explicitly declare send_request(), send_notification(), send_progress_notification().

- Remove commented-out import in context.py.

- Add comprehensive tests for BaseClientSession protocol satisfaction and E2E usage.

- Document all breaking changes in docs/migration.md including:
  * ClientRequestContext type change to BaseClientSession
  * Generic callback protocols (SamplingFnT[ClientSession], etc.)
  * SessionT renamed to SessionT_co
  * AbstractBaseSession simplification
  * BaseClientSession Protocol conversion

Fixes structural issues identified in PR review:
- AbstractBaseSession no longer manages state it doesn't own
- Eliminates double initialization of _response_streams and _task_group
- Removes Any leakage for WireMessageT
- BaseClientSession now earns its place as a pure interface

Author: sreenithi

docs/migration.md

@@ -471,6 +471,138 @@ await client.read_resource("test://resource")
 await client.read_resource(str(my_any_url))
 ```
 
+### Transport Abstractions Refactored
+
+The session hierarchy has been refactored to support pluggable transport implementations. This introduces several breaking changes:
+
+#### `ClientRequestContext` type changed
+
+`ClientRequestContext` is now `RequestContext[BaseClientSession]` instead of `RequestContext[ClientSession]`. This means callbacks receive the more general `BaseClientSession` type, which may not have all methods available on `ClientSession`.
+
+**Before:**
+
+```python
+from mcp.client.context import ClientRequestContext
+from mcp.client.session import ClientSession
+
+async def my_callback(context: ClientRequestContext) -> None:
+    # Could access ClientSession-specific methods
+    caps = context.session.get_server_capabilities()
+```
+
+**After:**
+
+```python
+from mcp.client.context import ClientRequestContext
+from mcp.client.session import ClientSession
+
+async def my_callback(context: ClientRequestContext) -> None:
+    # context.session is BaseClientSession - narrow the type if needed
+    if isinstance(context.session, ClientSession):
+        caps = context.session.get_server_capabilities()
+```
+
+#### Callback protocols are now generic
+
+`sampling_callback`, `elicitation_callback`, and `list_roots_callback` protocols now require explicit type parameters.
+
+**Before:**
+
+```python
+from mcp.client.session import SamplingFnT
+
+async def my_sampling(context, params) -> CreateMessageResult:
+    ...
+
+# Type inferred as SamplingFnT
+session = ClientSession(..., sampling_callback=my_sampling)
+```
+
+**After:**
+
+```python
+from mcp.client.session import SamplingFnT, ClientSession
+
+async def my_sampling(
+    context: RequestContext[ClientSession],
+    params: CreateMessageRequestParams
+) -> CreateMessageResult:
+    ...
+
+# Explicit type annotation recommended
+my_sampling_typed: SamplingFnT[ClientSession] = my_sampling
+session = ClientSession(..., sampling_callback=my_sampling_typed)
+```
+
+#### `SessionT` renamed to `SessionT_co`
+
+In `mcp.shared._context` and `mcp.shared.progress`, the `SessionT` TypeVar has been renamed to `SessionT_co` to follow naming conventions for covariant type variables.
+
+**Before:**
+
+```python
+from mcp.shared._context import SessionT
+```
+
+**After:**
+
+```python
+from mcp.shared._context import SessionT_co
+```
+
+#### `AbstractBaseSession` simplified
+
+`AbstractBaseSession` is now a pure abstract interface with no `__init__` method and no `WireMessageT` type parameter. If you were subclassing it directly, you now need to manage all state in your subclass.
+
+**Before:**
+
+```python
+from mcp.shared.session import AbstractBaseSession
+
+class MySession(AbstractBaseSession[MyMessage, ...]):
+    def __init__(self):
+        super().__init__()  # Would set up _response_streams, _task_group
+```
+
+**After:**
+
+```python
+from mcp.shared.session import AbstractBaseSession
+
+class MySession(AbstractBaseSession[...]):
+    def __init__(self):
+        # Manage your own state - no super().__init__() to call
+        self._my_state = {}
+```
+
+#### `BaseClientSession` is now a Protocol
+
+`BaseClientSession` is now a `typing.Protocol` (structural subtyping) instead of an abstract base class. It no longer inherits from `AbstractBaseSession` and requires no inheritance to satisfy.
+
+**Before:**
+
+```python
+from mcp.client.base_client_session import BaseClientSession
+
+class MyClientSession(BaseClientSession):
+    async def initialize(self) -> InitializeResult:
+        ...
+```
+
+**After:**
+
+```python
+from mcp.client.base_client_session import BaseClientSession
+
+class MyClientSession:
+    # Just implement the methods - no inheritance needed
+    async def initialize(self) -> InitializeResult:
+        ...
+
+# Verify protocol satisfaction at runtime
+assert isinstance(MyClientSession(), BaseClientSession)
+```
+
 ## Deprecations
 
 <!-- Add deprecations below -->

src/mcp/client/base_client_session.py

@@ -1,75 +1,68 @@
-from abc import abstractmethod
 from typing import Any, TypeVar
 
+from typing_extensions import Protocol, runtime_checkable
+
 from mcp import types
-from mcp.shared.session import CommonBaseSession, ProgressFnT
+from mcp.shared.session import ProgressFnT
 from mcp.types._types import RequestParamsMeta
 
-ClientSessionT_contra = TypeVar("ClientSessionT_contra", bound="BaseClientSession", contravariant=True)
+ClientSessionT_contra = TypeVar("ClientSessionT_contra", contravariant=True)
 
 
-class BaseClientSession(
-    CommonBaseSession[
-        Any,
-        types.ClientRequest,
-        types.ClientNotification,
-        types.ClientResult,
-        types.ServerRequest,
-        types.ServerNotification,
-    ]
-):
-    """Base class for client transport sessions.
+@runtime_checkable
+class BaseClientSession(Protocol):
+    """Protocol defining the interface for MCP client sessions.
 
-    The class provides all the methods that a client session should implement,
-    irrespective of the transport used.
+    This protocol specifies all methods that a client session must implement,
+    irrespective of the transport used. Implementations satisfy this protocol
+    through structural subtyping — no inheritance required.
     """
 
-    @abstractmethod
-    async def initialize(self) -> types.InitializeResult:
-        """Initialize the client session."""
-        raise NotImplementedError
+    # Methods from AbstractBaseSession (must be explicitly declared in Protocol)
+    async def send_request(
+        self,
+        request: types.ClientRequest,
+        result_type: type,
+        request_read_timeout_seconds: float | None = None,
+        metadata: Any = None,
+        progress_callback: ProgressFnT | None = None,
+    ) -> Any: ...
+
+    async def send_notification(
+        self,
+        notification: types.ClientNotification,
+        related_request_id: Any = None,
+    ) -> None: ...
+
+    async def send_progress_notification(
+        self,
+        progress_token: types.ProgressToken,
+        progress: float,
+        total: float | None = None,
+        message: str | None = None,
+        *,
+        meta: RequestParamsMeta | None = None,
+    ) -> None: ...
 
-    @abstractmethod
-    async def send_ping(self, *, meta: RequestParamsMeta | None = None) -> types.EmptyResult:
-        """Send a ping request."""
-        raise NotImplementedError
+    # Client-specific methods
+    async def initialize(self) -> types.InitializeResult: ...
 
-    @abstractmethod
-    async def list_resources(self, *, params: types.PaginatedRequestParams | None = None) -> types.ListResourcesResult:
-        """Send a resources/list request.
+    async def send_ping(self, *, meta: RequestParamsMeta | None = None) -> types.EmptyResult: ...
 
-        Args:
-            params: Full pagination parameters including cursor and any future fields
-        """
-        raise NotImplementedError
+    async def list_resources(
+        self, *, params: types.PaginatedRequestParams | None = None
+    ) -> types.ListResourcesResult: ...
 
-    @abstractmethod
     async def list_resource_templates(
         self, *, params: types.PaginatedRequestParams | None = None
-    ) -> types.ListResourceTemplatesResult:
-        """Send a resources/templates/list request.
-
-        Args:
-            params: Full pagination parameters including cursor and any future fields
-        """
-        raise NotImplementedError
-
-    @abstractmethod
-    async def read_resource(self, uri: str, *, meta: RequestParamsMeta | None = None) -> types.ReadResourceResult:
-        """Send a resources/read request."""
-        raise NotImplementedError
-
-    @abstractmethod
-    async def subscribe_resource(self, uri: str, *, meta: RequestParamsMeta | None = None) -> types.EmptyResult:
-        """Send a resources/subscribe request."""
-        raise NotImplementedError
-
-    @abstractmethod
-    async def unsubscribe_resource(self, uri: str, *, meta: RequestParamsMeta | None = None) -> types.EmptyResult:
-        """Send a resources/unsubscribe request."""
-        raise NotImplementedError
-
-    @abstractmethod
+    ) -> types.ListResourceTemplatesResult: ...
+
+    async def read_resource(self, uri: str, *, meta: RequestParamsMeta | None = None) -> types.ReadResourceResult: ...
+
+    async def subscribe_resource(self, uri: str, *, meta: RequestParamsMeta | None = None) -> types.EmptyResult: ...
+
+    async def unsubscribe_resource(self, uri: str, *, meta: RequestParamsMeta | None = None) -> types.EmptyResult: ...
+
     async def call_tool(
         self,
         name: str,
@@ -78,40 +71,33 @@ async def call_tool(
         progress_callback: ProgressFnT | None = None,
         *,
         meta: RequestParamsMeta | None = None,
-    ) -> types.CallToolResult:
-        """Send a tools/call request with optional progress callback support."""
-        raise NotImplementedError
-
-    @abstractmethod
-    async def list_prompts(self, *, params: types.PaginatedRequestParams | None = None) -> types.ListPromptsResult:
-        """Send a prompts/list request.
+    ) -> types.CallToolResult: ...
 
-        Args:
-            params: Full pagination parameters including cursor and any future fields
-        """
-        raise NotImplementedError
+    async def list_prompts(self, *, params: types.PaginatedRequestParams | None = None) -> types.ListPromptsResult: ...
 
-    @abstractmethod
     async def get_prompt(
         self,
         name: str,
         arguments: dict[str, str] | None = None,
         *,
         meta: RequestParamsMeta | None = None,
-    ) -> types.GetPromptResult:
-        """Send a prompts/get request."""
-        raise NotImplementedError
-
-    @abstractmethod
-    async def list_tools(self, *, params: types.PaginatedRequestParams | None = None) -> types.ListToolsResult:
-        """Send a tools/list request.
-
-        Args:
-            params: Full pagination parameters including cursor and any future fields
-        """
-        raise NotImplementedError
-
-    @abstractmethod
-    async def send_roots_list_changed(self) -> None:  # pragma: no cover
-        """Send a roots/list_changed notification."""
-        raise NotImplementedError
+    ) -> types.GetPromptResult: ...
+
+    async def list_tools(self, *, params: types.PaginatedRequestParams | None = None) -> types.ListToolsResult: ...
+
+    # Missing methods added per review
+    async def complete(
+        self,
+        ref: types.ResourceTemplateReference | types.PromptReference,
+        argument: dict[str, str],
+        context_arguments: dict[str, str] | None = None,
+    ) -> types.CompleteResult: ...
+
+    async def set_logging_level(
+        self,
+        level: types.LoggingLevel,
+        *,
+        meta: RequestParamsMeta | None = None,
+    ) -> types.EmptyResult: ...
+
+    async def send_roots_list_changed(self) -> None: ...

src/mcp/client/context.py

@@ -1,8 +1,6 @@
 """Request context for MCP client handlers."""
 
 from mcp.client import BaseClientSession
-
-# from mcp.client.session import ClientSession
 from mcp.shared._context import RequestContext
 
 ClientRequestContext = RequestContext[BaseClientSession]

src/mcp/client/session.py

@@ -107,7 +107,6 @@ class ClientSession(
         types.ServerRequest,
         types.ServerNotification,
     ],
-    BaseClientSession,
 ):
     def __init__(
         self,

src/mcp/shared/_context.py

@@ -1,14 +1,13 @@
 """Request context for MCP handlers."""
 
 from dataclasses import dataclass
-from typing import Any, Generic
+from typing import Generic
 
 from typing_extensions import TypeVar
 
-from mcp.shared.session import CommonBaseSession
 from mcp.types import RequestId, RequestParamsMeta
 
-SessionT_co = TypeVar("SessionT_co", bound=CommonBaseSession[Any, Any, Any, Any, Any, Any], covariant=True)
+SessionT_co = TypeVar("SessionT_co", covariant=True)
 
 
 @dataclass(kw_only=True)

src/mcp/shared/message.py

@@ -6,7 +6,6 @@
 
 from collections.abc import Awaitable, Callable
 from dataclasses import dataclass
-from typing import TypeVar
 
 from mcp.types import JSONRPCMessage, RequestId
 
@@ -41,8 +40,6 @@ class ServerMessageMetadata:
 
 MessageMetadata = ClientMessageMetadata | ServerMessageMetadata | None
 
-WireMessageT = TypeVar("WireMessageT")
-
 
 @dataclass
 class SessionMessage: