Narrow message_handler's parameter to notifications and exceptions (#3168)

src/mcp/client/__init__.py

@@ -20,7 +20,7 @@ from mcp.client.extension import (
     UnexpectedClaimedResult,
     advertise,
 )
-from mcp.client.session import ClientSession
+from mcp.client.session import ClientSession, IncomingMessage
 
 __all__ = [
     "CacheConfig",
@@ -32,6 +32,7 @@ __all__ = [
     "ClientExtension",
     "ClientRequestContext",
     "ClientSession",
+    "IncomingMessage",
     "InMemoryResponseCacheStore",
     "InputRequiredRoundsExceededError",
     "NotificationBinding",

src/mcp/client/__main__.py

@@ -9,11 +9,10 @@ import anyio
 import mcp_types as types
 
 from mcp.client._transport import ReadStream, WriteStream
-from mcp.client.session import ClientSession
+from mcp.client.session import ClientSession, IncomingMessage
 from mcp.client.sse import sse_client
 from mcp.client.stdio import StdioServerParameters, stdio_client
 from mcp.shared.message import SessionMessage
-from mcp.shared.session import RequestResponder
 
 if not sys.warnoptions:
     warnings.simplefilter("ignore")
@@ -22,9 +21,7 @@ logging.basicConfig(level=logging.INFO)
 logger = logging.getLogger("client")
 
 
-async def message_handler(
-    message: RequestResponder[types.ServerRequest, types.ClientResult] | types.ServerNotification | Exception,
-) -> None:
+async def message_handler(message: IncomingMessage) -> None:
     if isinstance(message, Exception):
         logger.error("Error: %s", message)
         return

src/mcp/client/client.py

@@ -52,6 +52,7 @@ from mcp.client.session import (
     ClientRequestContext,
     ClientSession,
     ElicitationFnT,
+    IncomingMessage,
     ListRootsFnT,
     LoggingFnT,
     MessageHandlerFnT,
@@ -68,7 +69,6 @@ from mcp.shared.dispatcher import Dispatcher, ProgressFnT
 from mcp.shared.exceptions import MCPDeprecationWarning, MCPError
 from mcp.shared.extension import validate_extension_identifier
 from mcp.shared.jsonrpc_dispatcher import JSONRPCDispatcher
-from mcp.shared.session import RequestResponder
 from mcp.shared.subscriptions import event_to_notification
 
 logger = logging.getLogger(__name__)
@@ -155,9 +155,7 @@ def _strip_userinfo(url: str) -> str:
 def _evicting_message_handler(cache: ClientResponseCache, user_handler: MessageHandlerFnT | None) -> MessageHandlerFnT:
     """Wrap the session message handler with cache eviction on server notifications."""
 
-    async def handler(
-        message: RequestResponder[types.ServerRequest, types.ClientResult] | types.ServerNotification | Exception,
-    ) -> None:
+    async def handler(message: IncomingMessage) -> None:
         if isinstance(message, types.ServerNotification):
             try:
                 await cache.evict_for_notification(message)

src/mcp/client/session.py

@@ -6,7 +6,7 @@ from dataclasses import dataclass
 from functools import reduce
 from operator import or_
 from types import TracebackType
-from typing import Annotated, Any, Final, Literal, Protocol, cast, overload
+from typing import Annotated, Any, Final, Literal, Protocol, TypeAlias, cast, overload
 
 import anyio
 import anyio.abc
@@ -53,7 +53,6 @@ from mcp.shared.inbound import (
 )
 from mcp.shared.jsonrpc_dispatcher import JSONRPCDispatcher, cancelled_request_id_from_params
 from mcp.shared.message import ClientMessageMetadata, SessionMessage
-from mcp.shared.session import RequestResponder
 from mcp.shared.subscriptions import SUBSCRIPTION_ID_META_KEY, event_from_wire
 from mcp.shared.transport_context import TransportContext
 
@@ -172,16 +171,20 @@ class LoggingFnT(Protocol):
     async def __call__(self, params: types.LoggingMessageNotificationParams) -> None: ...  # pragma: no branch
 
 
+IncomingMessage: TypeAlias = types.ServerNotification | Exception
+"""What `message_handler` receives: the server notifications the session surfaces, plus transport-level exceptions.
+
+`notifications/cancelled` is applied by the dispatcher and never surfaced, and a
+`notifications/subscriptions/acknowledged` for a live `listen()` stream is consumed by that
+stream, so neither reaches the handler.
+"""
+
+
 class MessageHandlerFnT(Protocol):
-    async def __call__(
-        self,
-        message: RequestResponder[types.ServerRequest, types.ClientResult] | types.ServerNotification | Exception,
-    ) -> None: ...  # pragma: no branch
+    async def __call__(self, message: IncomingMessage) -> None: ...  # pragma: no branch
 
 
-async def _default_message_handler(
-    message: RequestResponder[types.ServerRequest, types.ClientResult] | types.ServerNotification | Exception,
-) -> None:
+async def _default_message_handler(message: IncomingMessage) -> None:
     await anyio.lowlevel.checkpoint()
 
 
@@ -331,8 +334,10 @@ class ClientSession:
     `dispatcher=`), enter as an async context manager, then call
     `initialize()`. The dispatcher owns the receive loop and request
     correlation; this class owns the typed MCP layer and the constructor
-    callbacks. Transport `Exception` items reach `message_handler` only when
-    the session builds its own dispatcher from a stream pair.
+    callbacks. Transport `Exception` items reach `message_handler` on any
+    stream-backed dispatcher (`JSONRPCDispatcher`), whether built here from a
+    stream pair or supplied without a stream-exception hook of its own; an
+    in-process `DirectDispatcher` carries none.
 
     Extension `result_claims` fold into tools/call parsing at `adopt()`;
     `notification_bindings` observe vendor notifications via bounded FIFOs.

src/mcp/client/session_group.py

@@ -25,8 +25,8 @@ from mcp.client.sse import sse_client
 from mcp.client.stdio import StdioServerParameters
 from mcp.client.streamable_http import streamable_http_client
 from mcp.shared._httpx_utils import create_mcp_http_client
+from mcp.shared.dispatcher import ProgressFnT
 from mcp.shared.exceptions import MCPError
-from mcp.shared.session import ProgressFnT
 
 
 class SseServerParameters(BaseModel):

src/mcp/shared/session.py

@@ -1,22 +0,0 @@
-"""Compatibility names that outlived the removed v1 session layer (`BaseSession`)."""
-
-from typing import Generic, TypeVar
-
-from mcp_types import RequestParamsMeta
-
-from mcp.shared.dispatcher import ProgressFnT as ProgressFnT
-from mcp.shared.message import MessageMetadata
-
-RequestId = str | int
-
-ReceiveRequestT = TypeVar("ReceiveRequestT")
-SendResultT = TypeVar("SendResultT")
-
-
-class RequestResponder(Generic[ReceiveRequestT, SendResultT]):
-    """Typing stub for the v1 responder; the SDK never instantiates it."""
-
-    request_id: RequestId
-    request_meta: RequestParamsMeta | None
-    request: ReceiveRequestT
-    message_metadata: MessageMetadata