Make ClientRequestContext a concrete class

RequestContext[ClientSession] was the only instantiation of the generic
left in the tree (the server seat has ServerRequestContext), so the
public ClientRequestContext alias becomes the real dataclass and the
private mcp.shared._context module is deleted. request_id is now always
populated: the client only builds a context for inbound requests, and
ping is answered before any context exists.

Author: maxisbey

docs/migration.md

@@ -634,11 +634,11 @@ server = Server("my-server", on_call_tool=handle_call_tool)
 
 The `mcp.shared.context` module has been removed. `RequestContext` is now split into `ClientRequestContext` (in `mcp.client.context`) and `ServerRequestContext` (in `mcp.server.context`).
 
-The `RequestContext` class has been split to separate shared fields from server-specific fields. The shared `RequestContext` now only takes 1 type parameter (the session type) instead of 3.
+The split separates shared fields from server-specific fields. There is no shared `RequestContext` generic anymore — each concrete class fixes its session type.
 
 **`RequestContext` changes:**
 
-- Type parameters reduced from `RequestContext[SessionT, LifespanContextT, RequestT]` to `RequestContext[SessionT]`
+- The `RequestContext[SessionT, LifespanContextT, RequestT]` generic no longer exists; use `ClientRequestContext` or `ServerRequestContext[LifespanContextT, RequestT]`
 - Server-specific fields (`lifespan_context`, `request`, `close_sse_stream`, `close_standalone_sse_stream`) moved to new `ServerRequestContext` class in `mcp.server.context`
 
 **Before (v1):**
@@ -1188,6 +1188,7 @@ Behavior changes:
 - **A raising request callback** is answered with `code=0` and the exception text. v1 flattened every callback exception to `INVALID_PARAMS`. Callbacks that want a specific error response should return `ErrorData` (unchanged) or raise `MCPError`. One carve-out: a callback that raises pydantic's `ValidationError` is still answered with `INVALID_PARAMS` (`"Invalid request parameters"`, empty `data`) because the dispatcher cannot distinguish it from inbound-params validation — this conflation is pre-existing v1 behavior, and a revisit is pending.
 - **`send_request` before entering the context manager** raises `RuntimeError` immediately; v1 wrote to the transport and hung until the timeout. `send_notification` before entry still works.
 - **`send_notification` no longer takes `related_request_id`, and `send_request` no longer accepts `ServerMessageMetadata`.** The hint was never serialized by any client transport in v1 or v2 — it exists for the server's streamable-HTTP stream routing. Progress and response correlation via `progressToken` and the request id is unaffected.
+- **The private `mcp.shared._context.RequestContext` generic is deleted.** Client callbacks now receive the concrete `mcp.client.ClientRequestContext`, whose `request_id` is always populated (the client only builds a context for inbound requests). Annotations spelled `RequestContext[ClientSession]` become `ClientRequestContext`.
 
 `mcp.shared.session` is now a compatibility module: `ProgressFnT` is re-exported (its home is `mcp.shared.dispatcher`), and `RequestResponder` remains as a typing-only stub so `MessageHandlerFnT` annotations keep importing — it has been unreachable at runtime since the server-side swap. `RequestResponder.respond()` no longer exists.
 

src/mcp/client/context.py

@@ -1,16 +1,5 @@
 """Request context for MCP client handlers."""
 
-from mcp.client.session import ClientSession
-from mcp.shared._context import RequestContext
+from mcp.client.session import ClientRequestContext
 
-ClientRequestContext = RequestContext[ClientSession]
-"""Context for handling incoming requests in a client session.
-
-This context is passed to client-side callbacks (sampling, elicitation, list_roots) when the server sends requests
-to the client.
-
-Attributes:
-    request_id: The unique identifier for this request.
-    meta: Optional metadata associated with the request.
-    session: The client session handling this request.
-"""
+__all__ = ["ClientRequestContext"]

src/mcp/client/session.py

@@ -2,6 +2,7 @@
 
 import logging
 from collections.abc import Mapping
+from dataclasses import dataclass
 from types import TracebackType
 from typing import Any, Protocol, cast, get_args
 
@@ -14,15 +15,14 @@
 from mcp import types
 from mcp.client._transport import ReadStream, WriteStream
 from mcp.shared._compat import resync_tracer
-from mcp.shared._context import RequestContext
 from mcp.shared.dispatcher import CallOptions, DispatchContext, Dispatcher
 from mcp.shared.exceptions import MCPError
 from mcp.shared.jsonrpc_dispatcher import JSONRPCDispatcher
 from mcp.shared.message import ClientMessageMetadata, SessionMessage
 from mcp.shared.session import ProgressFnT, RequestResponder
 from mcp.shared.transport_context import TransportContext
 from mcp.shared.version import SUPPORTED_PROTOCOL_VERSIONS
-from mcp.types._types import RequestParamsMeta
+from mcp.types import RequestId, RequestParamsMeta
 
 DEFAULT_CLIENT_INFO = types.Implementation(name="mcp", version="0.1.0")
 
@@ -31,25 +31,34 @@
 ReceiveResultT = TypeVar("ReceiveResultT", bound=BaseModel)
 
 
+@dataclass(kw_only=True)
+class ClientRequestContext:
+    """Context for a server-initiated request, passed to the sampling/elicitation/list-roots callbacks."""
+
+    session: ClientSession
+    request_id: RequestId
+    meta: RequestParamsMeta | None = None
+
+
 class SamplingFnT(Protocol):
     async def __call__(
         self,
-        context: RequestContext[ClientSession],
+        context: ClientRequestContext,
         params: types.CreateMessageRequestParams,
     ) -> types.CreateMessageResult | types.CreateMessageResultWithTools | types.ErrorData: ...  # pragma: no branch
 
 
 class ElicitationFnT(Protocol):
     async def __call__(
         self,
-        context: RequestContext[ClientSession],
+        context: ClientRequestContext,
         params: types.ElicitRequestParams,
     ) -> types.ElicitResult | types.ErrorData: ...  # pragma: no branch
 
 
 class ListRootsFnT(Protocol):
     async def __call__(
-        self, context: RequestContext[ClientSession]
+        self, context: ClientRequestContext
     ) -> types.ListRootsResult | types.ErrorData: ...  # pragma: no branch
 
 
@@ -71,7 +80,7 @@ async def _default_message_handler(
 
 
 async def _default_sampling_callback(
-    context: RequestContext[ClientSession],
+    context: ClientRequestContext,
     params: types.CreateMessageRequestParams,
 ) -> types.CreateMessageResult | types.CreateMessageResultWithTools | types.ErrorData:
     return types.ErrorData(
@@ -81,7 +90,7 @@ async def _default_sampling_callback(
 
 
 async def _default_elicitation_callback(
-    context: RequestContext[ClientSession],
+    context: ClientRequestContext,
     params: types.ElicitRequestParams,
 ) -> types.ElicitResult | types.ErrorData:
     return types.ErrorData(
@@ -91,7 +100,7 @@ async def _default_elicitation_callback(
 
 
 async def _default_list_roots_callback(
-    context: RequestContext[ClientSession],
+    context: ClientRequestContext,
 ) -> types.ListRootsResult | types.ErrorData:
     return types.ErrorData(
         code=types.INVALID_REQUEST,
@@ -496,19 +505,22 @@ async def _on_request(
             payload["params"] = dict(params)
         request = types.server_request_adapter.validate_python(payload, by_name=False)
 
-        ctx = RequestContext[ClientSession](
-            request_id=dctx.request_id, meta=request.params.meta if request.params else None, session=self
-        )
         response: types.ClientResult | types.ErrorData
-        match request:
-            case types.CreateMessageRequest(params=sampling_params):
-                response = await self._sampling_callback(ctx, sampling_params)
-            case types.ElicitRequest(params=elicit_params):
-                response = await self._elicitation_callback(ctx, elicit_params)
-            case types.ListRootsRequest():
-                response = await self._list_roots_callback(ctx)
-            case types.PingRequest():  # pragma: no branch
-                response = types.EmptyResult()
+        if isinstance(request, types.PingRequest):
+            # Answered without a context: direct dispatch carries no request id.
+            response = types.EmptyResult()
+        else:
+            assert dctx.request_id is not None  # the callback-driving dispatchers always assign ids
+            ctx = ClientRequestContext(
+                session=self, request_id=dctx.request_id, meta=request.params.meta if request.params else None
+            )
+            match request:
+                case types.CreateMessageRequest(params=sampling_params):
+                    response = await self._sampling_callback(ctx, sampling_params)
+                case types.ElicitRequest(params=elicit_params):
+                    response = await self._elicitation_callback(ctx, elicit_params)
+                case types.ListRootsRequest():  # pragma: no branch
+                    response = await self._list_roots_callback(ctx)
         client_response = ClientResponse.validate_python(response)
         if isinstance(client_response, types.ErrorData):
             raise MCPError.from_error_data(client_response)

src/mcp/shared/_context.py

@@ -2,9 +2,8 @@
 from pydantic import FileUrl
 
 from mcp import Client
-from mcp.client.session import ClientSession
+from mcp.client import ClientRequestContext
 from mcp.server.mcpserver import Context, MCPServer
-from mcp.shared._context import RequestContext
 from mcp.types import ListRootsResult, Root, TextContent
 
 
@@ -20,7 +19,7 @@ async def test_list_roots_callback():
     )
 
     async def list_roots_callback(
-        context: RequestContext[ClientSession],
+        context: ClientRequestContext,
     ) -> ListRootsResult:
         return callback_return
 

tests/client/test_list_roots_callback.py

@@ -1,9 +1,8 @@
 import pytest
 
 from mcp import Client
-from mcp.client.session import ClientSession
+from mcp.client import ClientRequestContext
 from mcp.server.mcpserver import Context, MCPServer
-from mcp.shared._context import RequestContext
 from mcp.types import (
     CreateMessageRequestParams,
     CreateMessageResult,
@@ -26,7 +25,7 @@ async def test_sampling_callback():
     )
 
     async def sampling_callback(
-        context: RequestContext[ClientSession],
+        context: ClientRequestContext,
         params: CreateMessageRequestParams,
     ) -> CreateMessageResult:
         return callback_return
@@ -71,7 +70,7 @@ async def test_create_message_backwards_compat_single_content():
     )
 
     async def sampling_callback(
-        context: RequestContext[ClientSession],
+        context: ClientRequestContext,
         params: CreateMessageRequestParams,
     ) -> CreateMessageResult:
         return callback_return

tests/client/test_sampling_callback.py

@@ -10,8 +10,8 @@
 import pytest
 
 from mcp import types
+from mcp.client import ClientRequestContext
 from mcp.client.session import DEFAULT_CLIENT_INFO, ClientSession
-from mcp.shared._context import RequestContext
 from mcp.shared.direct_dispatcher import create_direct_dispatcher_pair
 from mcp.shared.dispatcher import CallOptions, DispatchContext, OnNotify, OnRequest
 from mcp.shared.message import SessionMessage
@@ -425,7 +425,7 @@ async def test_client_capabilities_with_custom_callbacks():
     received_capabilities = None
 
     async def custom_sampling_callback(  # pragma: no cover
-        context: RequestContext[ClientSession],
+        context: ClientRequestContext,
         params: types.CreateMessageRequestParams,
     ) -> types.CreateMessageResult | types.ErrorData:
         return types.CreateMessageResult(
@@ -435,7 +435,7 @@ async def custom_sampling_callback(  # pragma: no cover
         )
 
     async def custom_list_roots_callback(  # pragma: no cover
-        context: RequestContext[ClientSession],
+        context: ClientRequestContext,
     ) -> types.ListRootsResult | types.ErrorData:
         return types.ListRootsResult(roots=[])
 
@@ -509,7 +509,7 @@ async def test_client_capabilities_with_sampling_tools():
     received_capabilities = None
 
     async def custom_sampling_callback(  # pragma: no cover
-        context: RequestContext[ClientSession],
+        context: ClientRequestContext,
         params: types.CreateMessageRequestParams,
     ) -> types.CreateMessageResult | types.ErrorData:
         return types.CreateMessageResult(

tests/client/test_session.py

@@ -6,9 +6,9 @@
 from pydantic import BaseModel, Field
 
 from mcp import Client, types
-from mcp.client.session import ClientSession, ElicitationFnT
+from mcp.client import ClientRequestContext
+from mcp.client.session import ElicitationFnT
 from mcp.server.mcpserver import Context, MCPServer
-from mcp.shared._context import RequestContext
 from mcp.types import ElicitRequestParams, ElicitResult, TextContent
 
 
@@ -64,7 +64,7 @@ async def test_elicitation_accept_returns_the_users_answer_to_the_tool():
     create_ask_user_tool(mcp)
 
     # Create a custom handler for elicitation requests
-    async def elicitation_callback(context: RequestContext[ClientSession], params: ElicitRequestParams):
+    async def elicitation_callback(context: ClientRequestContext, params: ElicitRequestParams):
         if params.message == "Tool wants to ask: What is your name?":
             return ElicitResult(action="accept", content={"answer": "Test User"})
         else:  # pragma: no cover
@@ -81,7 +81,7 @@ async def test_elicitation_decline_reaches_the_tool_without_content():
     mcp = MCPServer(name="ElicitationDeclineServer")
     create_ask_user_tool(mcp)
 
-    async def elicitation_callback(context: RequestContext[ClientSession], params: ElicitRequestParams):
+    async def elicitation_callback(context: ClientRequestContext, params: ElicitRequestParams):
         return ElicitResult(action="decline")
 
     await call_tool_and_assert(
@@ -119,9 +119,7 @@ class InvalidNestedSchema(BaseModel):
     create_validation_tool("nested_model", InvalidNestedSchema)
 
     # Dummy callback (won't be called due to validation failure)
-    async def elicitation_callback(
-        context: RequestContext[ClientSession], params: ElicitRequestParams
-    ):  # pragma: no cover
+    async def elicitation_callback(context: ClientRequestContext, params: ElicitRequestParams):  # pragma: no cover
         return ElicitResult(action="accept", content={})
 
     async with Client(mcp, elicitation_callback=elicitation_callback) as client:
@@ -176,7 +174,7 @@ async def optional_tool(ctx: Context) -> str:
 
     for content, expected in test_cases:
 
-        async def callback(context: RequestContext[ClientSession], params: ElicitRequestParams):
+        async def callback(context: ClientRequestContext, params: ElicitRequestParams):
             return ElicitResult(action="accept", content=content)
 
         await call_tool_and_assert(mcp, callback, "optional_tool", {}, expected)
@@ -194,9 +192,7 @@ async def invalid_optional_tool(ctx: Context) -> str:
         except TypeError as e:
             return f"Validation failed: {str(e)}"
 
-    async def elicitation_callback(
-        context: RequestContext[ClientSession], params: ElicitRequestParams
-    ):  # pragma: no cover
+    async def elicitation_callback(context: ClientRequestContext, params: ElicitRequestParams):  # pragma: no cover
         return ElicitResult(action="accept", content={})
 
     await call_tool_and_assert(
@@ -219,7 +215,7 @@ async def valid_multiselect_tool(ctx: Context) -> str:
             return f"Name: {result.data.name}, Tags: {', '.join(result.data.tags)}"
         return f"User {result.action}"  # pragma: no cover
 
-    async def multiselect_callback(context: RequestContext[ClientSession], params: ElicitRequestParams):
+    async def multiselect_callback(context: ClientRequestContext, params: ElicitRequestParams):
         if "Please provide tags" in params.message:
             return ElicitResult(action="accept", content={"name": "Test", "tags": ["tag1", "tag2"]})
         return ElicitResult(action="decline")  # pragma: no cover
@@ -239,7 +235,7 @@ async def optional_multiselect_tool(ctx: Context) -> str:
             return f"Name: {result.data.name}, Tags: {tags_str}"
         return f"User {result.action}"  # pragma: no cover
 
-    async def optional_multiselect_callback(context: RequestContext[ClientSession], params: ElicitRequestParams):
+    async def optional_multiselect_callback(context: ClientRequestContext, params: ElicitRequestParams):
         if "Please provide optional tags" in params.message:
             return ElicitResult(action="accept", content={"name": "Test", "tags": ["tag1", "tag2"]})
         return ElicitResult(action="decline")  # pragma: no cover
@@ -273,7 +269,7 @@ async def defaults_tool(ctx: Context) -> str:
             return f"User {result.action}"
 
     # First verify that defaults are present in the JSON schema sent to clients
-    async def callback_schema_verify(context: RequestContext[ClientSession], params: ElicitRequestParams):
+    async def callback_schema_verify(context: ClientRequestContext, params: ElicitRequestParams):
         # Verify the schema includes defaults
         assert isinstance(params, types.ElicitRequestFormParams), "Expected form mode elicitation"
         schema = params.requested_schema
@@ -295,7 +291,7 @@ async def callback_schema_verify(context: RequestContext[ClientSession], params:
     )
 
     # Test overriding defaults
-    async def callback_override(context: RequestContext[ClientSession], params: ElicitRequestParams):
+    async def callback_override(context: ClientRequestContext, params: ElicitRequestParams):
         return ElicitResult(
             action="accept", content={"email": "john@example.com", "name": "John", "age": 25, "subscribe": False}
         )
@@ -371,7 +367,7 @@ async def select_color_legacy(ctx: Context) -> str:
             return f"User: {result.data.user_name}, Color: {result.data.color}"
         return f"User {result.action}"  # pragma: no cover
 
-    async def enum_callback(context: RequestContext[ClientSession], params: ElicitRequestParams):
+    async def enum_callback(context: ClientRequestContext, params: ElicitRequestParams):
         if "colors" in params.message and "legacy" not in params.message:
             return ElicitResult(action="accept", content={"user_name": "Bob", "favorite_colors": ["red", "green"]})
         elif "color" in params.message: