fix typehints for client transport session

Author: asheshvidyut

examples/clients/simple-auth-client/mcp_simple_auth_client/main.py

@@ -17,7 +17,7 @@
 from urllib.parse import parse_qs, urlparse
 
 from mcp.client.auth import OAuthClientProvider, TokenStorage
-from mcp.client.session import ClientSession
+from mcp.client.session import ClientSession, ClientTransportSession
 from mcp.client.sse import sse_client
 from mcp.client.streamable_http import streamablehttp_client
 from mcp.shared.auth import OAuthClientInformationFull, OAuthClientMetadata, OAuthToken
@@ -153,7 +153,7 @@ class SimpleAuthClient:
     def __init__(self, server_url: str, transport_type: str = "streamable-http"):
         self.server_url = server_url
         self.transport_type = transport_type
-        self.session: ClientSession | None = None
+        self.session: ClientTransportSession | None = None
 
     async def connect(self):
         """Connect to the MCP server."""

examples/clients/simple-chatbot/mcp_simple_chatbot/main.py

@@ -10,6 +10,7 @@
 from dotenv import load_dotenv
 from mcp import ClientSession, StdioServerParameters
 from mcp.client.stdio import stdio_client
+from mcp.client.transport_session import ClientTransportSession
 
 # Configure logging
 logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
@@ -67,7 +68,7 @@ def __init__(self, name: str, config: dict[str, Any]) -> None:
         self.name: str = name
         self.config: dict[str, Any] = config
         self.stdio_context: Any | None = None
-        self.session: ClientSession | None = None
+        self.session: ClientTransportSession | None = None
         self._cleanup_lock: asyncio.Lock = asyncio.Lock()
         self.exit_stack: AsyncExitStack = AsyncExitStack()
 

examples/snippets/clients/display_utilities.py

@@ -8,6 +8,7 @@
 
 from mcp import ClientSession, StdioServerParameters
 from mcp.client.stdio import stdio_client
+from mcp.client.transport_session import ClientTransportSession
 from mcp.shared.metadata_utils import get_display_name
 
 # Create server parameters for stdio connection
@@ -18,7 +19,7 @@
 )
 
 
-async def display_tools(session: ClientSession):
+async def display_tools(session: ClientTransportSession):
     """Display available tools with human-readable names"""
     tools_response = await session.list_tools()
 
@@ -30,7 +31,7 @@ async def display_tools(session: ClientSession):
             print(f"   {tool.description}")
 
 
-async def display_resources(session: ClientSession):
+async def display_resources(session: ClientTransportSession):
     """Display available resources with human-readable names"""
     resources_response = await session.list_resources()
 

examples/snippets/clients/stdio_client.py

@@ -9,6 +9,7 @@
 from pydantic import AnyUrl
 
 from mcp import ClientSession, StdioServerParameters, types
+from mcp.client.session import ClientTransportSession
 from mcp.client.stdio import stdio_client
 from mcp.shared.context import RequestContext
 
@@ -22,7 +23,7 @@
 
 # Optional: create a sampling callback
 async def handle_sampling_message(
-    context: RequestContext[ClientSession, None], params: types.CreateMessageRequestParams
+    context: RequestContext[ClientTransportSession, None], params: types.CreateMessageRequestParams
 ) -> types.CreateMessageResult:
     print(f"Sampling request: {params.messages}")
     return types.CreateMessageResult(

src/mcp/client/session.py

@@ -23,22 +23,22 @@
 class SamplingFnT(Protocol):
     async def __call__(
         self,
-        context: RequestContext["ClientSession", Any],
+        context: RequestContext["ClientTransportSession", Any],
         params: types.CreateMessageRequestParams,
     ) -> types.CreateMessageResult | types.ErrorData: ...
 
 
 class ElicitationFnT(Protocol):
     async def __call__(
         self,
-        context: RequestContext["ClientSession", Any],
+        context: RequestContext["ClientTransportSession", Any],
         params: types.ElicitRequestParams,
     ) -> types.ElicitResult | types.ErrorData: ...
 
 
 class ListRootsFnT(Protocol):
     async def __call__(
-        self, context: RequestContext["ClientSession", Any]
+        self, context: RequestContext["ClientTransportSession", Any]
     ) -> types.ListRootsResult | types.ErrorData: ...
 
 
@@ -63,7 +63,7 @@ async def _default_message_handler(
 
 
 async def _default_sampling_callback(
-    context: RequestContext["ClientSession", Any],
+    context: RequestContext["ClientTransportSession", Any],
     params: types.CreateMessageRequestParams,
 ) -> types.CreateMessageResult | types.ErrorData:
     return types.ErrorData(
@@ -73,7 +73,7 @@ async def _default_sampling_callback(
 
 
 async def _default_elicitation_callback(
-    context: RequestContext["ClientSession", Any],
+    context: RequestContext["ClientTransportSession", Any],
     params: types.ElicitRequestParams,
 ) -> types.ElicitResult | types.ErrorData:
     return types.ErrorData(
@@ -83,7 +83,7 @@ async def _default_elicitation_callback(
 
 
 async def _default_list_roots_callback(
-    context: RequestContext["ClientSession", Any],
+    context: RequestContext["ClientTransportSession", Any],
 ) -> types.ListRootsResult | types.ErrorData:
     return types.ErrorData(
         code=types.INVALID_REQUEST,
@@ -498,7 +498,7 @@ async def send_roots_list_changed(self) -> None:
         await self.send_notification(types.ClientNotification(types.RootsListChangedNotification()))
 
     async def _received_request(self, responder: RequestResponder[types.ServerRequest, types.ClientResult]) -> None:
-        ctx = RequestContext[ClientSession, Any](
+        ctx = RequestContext[ClientTransportSession, Any](
             request_id=responder.request_id,
             meta=responder.request_meta,
             session=self,

src/mcp/client/session_group.py

@@ -96,10 +96,10 @@ class _ComponentNames(BaseModel):
     _tools: dict[str, types.Tool]
 
     # Client-server connection management.
-    _sessions: dict[mcp.ClientSession, _ComponentNames]
-    _tool_to_session: dict[str, mcp.ClientSession]
+    _sessions: dict[mcp.ClientTransportSession, _ComponentNames]
+    _tool_to_session: dict[str, mcp.ClientTransportSession]
     _exit_stack: contextlib.AsyncExitStack
-    _session_exit_stacks: dict[mcp.ClientSession, contextlib.AsyncExitStack]
+    _session_exit_stacks: dict[mcp.ClientTransportSession, contextlib.AsyncExitStack]
 
     # Optional fn consuming (component_name, serverInfo) for custom names.
     # This is provide a means to mitigate naming conflicts across servers.
@@ -153,7 +153,7 @@ async def __aexit__(
                 tg.start_soon(exit_stack.aclose)
 
     @property
-    def sessions(self) -> list[mcp.ClientSession]:
+    def sessions(self) -> list[mcp.ClientTransportSession]:
         """Returns the list of sessions being managed."""
         return list(self._sessions.keys())
 
@@ -178,7 +178,7 @@ async def call_tool(self, name: str, args: dict[str, Any]) -> types.CallToolResu
         session_tool_name = self.tools[name].name
         return await session.call_tool(session_tool_name, args)
 
-    async def disconnect_from_server(self, session: mcp.ClientSession) -> None:
+    async def disconnect_from_server(self, session: mcp.ClientTransportSession) -> None:
         """Disconnects from a single MCP server."""
 
         session_known_for_components = session in self._sessions
@@ -216,23 +216,23 @@ async def disconnect_from_server(self, session: mcp.ClientSession) -> None:
             await session_stack_to_close.aclose()
 
     async def connect_with_session(
-        self, server_info: types.Implementation, session: mcp.ClientSession
-    ) -> mcp.ClientSession:
+        self, server_info: types.Implementation, session: mcp.ClientTransportSession
+    ) -> mcp.ClientTransportSession:
         """Connects to a single MCP server."""
         await self._aggregate_components(server_info, session)
         return session
 
     async def connect_to_server(
         self,
         server_params: ServerParameters,
-    ) -> mcp.ClientSession:
+    ) -> mcp.ClientTransportSession:
         """Connects to a single MCP server."""
         server_info, session = await self._establish_session(server_params)
         return await self.connect_with_session(server_info, session)
 
     async def _establish_session(
         self, server_params: ServerParameters
-    ) -> tuple[types.Implementation, mcp.ClientSession]:
+    ) -> tuple[types.Implementation, mcp.ClientTransportSession]:
         """Establish a client session to an MCP server."""
 
         session_stack = contextlib.AsyncExitStack()
@@ -276,7 +276,9 @@ async def _establish_session(
             await session_stack.aclose()
             raise
 
-    async def _aggregate_components(self, server_info: types.Implementation, session: mcp.ClientSession) -> None:
+    async def _aggregate_components(
+        self, server_info: types.Implementation, session: mcp.ClientTransportSession
+    ) -> None:
         """Aggregates prompts, resources, and tools from a given session."""
 
         # Create a reverse index so we can find all prompts, resources, and
@@ -289,7 +291,7 @@ async def _aggregate_components(self, server_info: types.Implementation, session
         prompts_temp: dict[str, types.Prompt] = {}
         resources_temp: dict[str, types.Resource] = {}
         tools_temp: dict[str, types.Tool] = {}
-        tool_to_session_temp: dict[str, mcp.ClientSession] = {}
+        tool_to_session_temp: dict[str, mcp.ClientTransportSession] = {}
 
         # Query the server for its prompts and aggregate to list.
         try:

src/mcp/client/transport_session.py

@@ -1,8 +1,9 @@
 from abc import ABC, abstractmethod
 from datetime import timedelta
-from typing import Any
+from typing import Any, overload
 
 from pydantic import AnyUrl
+from typing_extensions import deprecated
 
 from mcp import types
 from mcp.shared.session import ProgressFnT
@@ -109,12 +110,29 @@ async def complete(
         """Send a completion/complete request."""
         raise NotImplementedError
 
+    @overload
+    @deprecated("Use list_tools(params=PaginatedRequestParams(...)) instead")
+    async def list_tools(self, cursor: str | None) -> types.ListToolsResult: ...
+
+    @overload
+    async def list_tools(self, *, params: types.PaginatedRequestParams | None) -> types.ListToolsResult: ...
+
+    @overload
+    async def list_tools(self) -> types.ListToolsResult: ...
+
     @abstractmethod
     async def list_tools(
         self,
         cursor: str | None = None,
+        *,
+        params: types.PaginatedRequestParams | None = None,
     ) -> types.ListToolsResult:
-        """Send a tools/list request."""
+        """Send a tools/list request.
+
+        Args:
+            cursor: Simple cursor string for pagination (deprecated, use params instead)
+            params: Full pagination parameters including cursor and any future fields
+        """
         raise NotImplementedError
 
     @abstractmethod

src/mcp/shared/context.py

@@ -3,10 +3,11 @@
 
 from typing_extensions import TypeVar
 
+from mcp.client.transport_session import ClientTransportSession
 from mcp.shared.session import BaseSession
 from mcp.types import RequestId, RequestParams
 
-SessionT = TypeVar("SessionT", bound=BaseSession[Any, Any, Any, Any, Any])
+SessionT = TypeVar("SessionT", bound=BaseSession[Any, Any, Any, Any, Any] | ClientTransportSession)
 LifespanContextT = TypeVar("LifespanContextT")
 RequestT = TypeVar("RequestT", default=Any)
 

tests/client/test_list_roots_callback.py

@@ -1,7 +1,7 @@
 import pytest
 from pydantic import FileUrl
 
-from mcp.client.session import ClientSession
+from mcp.client.session import ClientTransportSession
 from mcp.server.fastmcp.server import Context
 from mcp.server.session import ServerSession
 from mcp.shared.context import RequestContext
@@ -31,7 +31,7 @@ async def test_list_roots_callback():
     )
 
     async def list_roots_callback(
-        context: RequestContext[ClientSession, None],
+        context: RequestContext[ClientTransportSession, None],
     ) -> ListRootsResult:
         return callback_return
 

tests/client/test_sampling_callback.py

@@ -1,6 +1,6 @@
 import pytest
 
-from mcp.client.session import ClientSession
+from mcp.client.session import ClientTransportSession
 from mcp.shared.context import RequestContext
 from mcp.shared.memory import (
     create_connected_server_and_client_session as create_session,
@@ -27,7 +27,7 @@ async def test_sampling_callback():
     )
 
     async def sampling_callback(
-        context: RequestContext[ClientSession, None],
+        context: RequestContext[ClientTransportSession, None],
         params: CreateMessageRequestParams,
     ) -> CreateMessageResult:
         return callback_return