refactor: split RequestContext between server and client

Author: Kludex

README.v2.md

@@ -229,7 +229,6 @@ from contextlib import asynccontextmanager
 from dataclasses import dataclass
 
 from mcp.server.mcpserver import Context, MCPServer
-from mcp.server.session import ServerSession
 
 
 # Mock database class for example
@@ -275,7 +274,7 @@ mcp = MCPServer("My App", lifespan=app_lifespan)
 
 # Access type-safe lifespan context in tools
 @mcp.tool()
-def query_db(ctx: Context[ServerSession, AppContext]) -> str:
+def query_db(ctx: Context[AppContext]) -> str:
     """Tool that uses initialized resources."""
     db = ctx.request_context.lifespan_context.db
     return db.query()
@@ -2135,7 +2134,7 @@ server_params = StdioServerParameters(
 
 # Optional: create a sampling callback
 async def handle_sampling_message(
-    context: RequestContext[ClientSession, None], params: types.CreateMessageRequestParams
+    context: RequestContext[ClientSession], params: types.CreateMessageRequestParams
 ) -> types.CreateMessageResult:
     print(f"Sampling request: {params.messages}")
     return types.CreateMessageResult(

examples/clients/simple-task-interactive-client/mcp_simple_task_interactive_client/main.py

@@ -7,7 +7,6 @@
 """
 
 import asyncio
-from typing import Any
 
 import click
 from mcp import ClientSession
@@ -24,7 +23,7 @@
 
 
 async def elicitation_callback(
-    context: RequestContext[ClientSession, Any],
+    context: RequestContext[ClientSession],
     params: ElicitRequestParams,
 ) -> ElicitResult:
     """Handle elicitation requests from the server."""
@@ -39,7 +38,7 @@ async def elicitation_callback(
 
 
 async def sampling_callback(
-    context: RequestContext[ClientSession, Any],
+    context: RequestContext[ClientSession],
     params: CreateMessageRequestParams,
 ) -> CreateMessageResult:
     """Handle sampling requests from the server."""

examples/snippets/clients/stdio_client.py

@@ -19,7 +19,7 @@
 
 # Optional: create a sampling callback
 async def handle_sampling_message(
-    context: RequestContext[ClientSession, None], params: types.CreateMessageRequestParams
+    context: RequestContext[ClientSession], params: types.CreateMessageRequestParams
 ) -> types.CreateMessageResult:
     print(f"Sampling request: {params.messages}")
     return types.CreateMessageResult(

examples/snippets/clients/url_elicitation_client.py

@@ -38,7 +38,7 @@
 
 
 async def handle_elicitation(
-    context: RequestContext[ClientSession, Any],
+    context: RequestContext[ClientSession],
     params: types.ElicitRequestParams,
 ) -> types.ElicitResult | types.ErrorData:
     """Handle elicitation requests from the server.

examples/snippets/servers/lifespan_example.py

@@ -5,7 +5,6 @@
 from dataclasses import dataclass
 
 from mcp.server.mcpserver import Context, MCPServer
-from mcp.server.session import ServerSession
 
 
 # Mock database class for example
@@ -51,7 +50,7 @@ async def app_lifespan(server: MCPServer) -> AsyncIterator[AppContext]:
 
 # Access type-safe lifespan context in tools
 @mcp.tool()
-def query_db(ctx: Context[ServerSession, AppContext]) -> str:
+def query_db(ctx: Context[AppContext]) -> str:
     """Tool that uses initialized resources."""
     db = ctx.request_context.lifespan_context.db
     return db.query()

src/mcp/client/experimental/task_handlers.py

@@ -11,8 +11,10 @@
 - Server polls client's task status via tasks/get, tasks/result, etc.
 """
 
+from __future__ import annotations
+
 from dataclasses import dataclass, field
-from typing import TYPE_CHECKING, Any, Protocol
+from typing import TYPE_CHECKING, Protocol
 
 from pydantic import TypeAdapter
 
@@ -32,7 +34,7 @@ class GetTaskHandlerFnT(Protocol):
 
     async def __call__(
         self,
-        context: RequestContext["ClientSession", Any],
+        context: RequestContext[ClientSession],
         params: types.GetTaskRequestParams,
     ) -> types.GetTaskResult | types.ErrorData: ...  # pragma: no branch
 
@@ -45,7 +47,7 @@ class GetTaskResultHandlerFnT(Protocol):
 
     async def __call__(
         self,
-        context: RequestContext["ClientSession", Any],
+        context: RequestContext[ClientSession],
         params: types.GetTaskPayloadRequestParams,
     ) -> types.GetTaskPayloadResult | types.ErrorData: ...  # pragma: no branch
 
@@ -58,7 +60,7 @@ class ListTasksHandlerFnT(Protocol):
 
     async def __call__(
         self,
-        context: RequestContext["ClientSession", Any],
+        context: RequestContext[ClientSession],
         params: types.PaginatedRequestParams | None,
     ) -> types.ListTasksResult | types.ErrorData: ...  # pragma: no branch
 
@@ -71,7 +73,7 @@ class CancelTaskHandlerFnT(Protocol):
 
     async def __call__(
         self,
-        context: RequestContext["ClientSession", Any],
+        context: RequestContext[ClientSession],
         params: types.CancelTaskRequestParams,
     ) -> types.CancelTaskResult | types.ErrorData: ...  # pragma: no branch
 
@@ -88,7 +90,7 @@ class TaskAugmentedSamplingFnT(Protocol):
 
     async def __call__(
         self,
-        context: RequestContext["ClientSession", Any],
+        context: RequestContext[ClientSession],
         params: types.CreateMessageRequestParams,
         task_metadata: types.TaskMetadata,
     ) -> types.CreateTaskResult | types.ErrorData: ...  # pragma: no branch
@@ -106,14 +108,14 @@ class TaskAugmentedElicitationFnT(Protocol):
 
     async def __call__(
         self,
-        context: RequestContext["ClientSession", Any],
+        context: RequestContext[ClientSession],
         params: types.ElicitRequestParams,
         task_metadata: types.TaskMetadata,
     ) -> types.CreateTaskResult | types.ErrorData: ...  # pragma: no branch
 
 
 async def default_get_task_handler(
-    context: RequestContext["ClientSession", Any],
+    context: RequestContext[ClientSession],
     params: types.GetTaskRequestParams,
 ) -> types.GetTaskResult | types.ErrorData:
     return types.ErrorData(
@@ -123,7 +125,7 @@ async def default_get_task_handler(
 
 
 async def default_get_task_result_handler(
-    context: RequestContext["ClientSession", Any],
+    context: RequestContext[ClientSession],
     params: types.GetTaskPayloadRequestParams,
 ) -> types.GetTaskPayloadResult | types.ErrorData:
     return types.ErrorData(
@@ -133,7 +135,7 @@ async def default_get_task_result_handler(
 
 
 async def default_list_tasks_handler(
-    context: RequestContext["ClientSession", Any],
+    context: RequestContext[ClientSession],
     params: types.PaginatedRequestParams | None,
 ) -> types.ListTasksResult | types.ErrorData:
     return types.ErrorData(
@@ -143,7 +145,7 @@ async def default_list_tasks_handler(
 
 
 async def default_cancel_task_handler(
-    context: RequestContext["ClientSession", Any],
+    context: RequestContext[ClientSession],
     params: types.CancelTaskRequestParams,
 ) -> types.CancelTaskResult | types.ErrorData:
     return types.ErrorData(
@@ -153,7 +155,7 @@ async def default_cancel_task_handler(
 
 
 async def default_task_augmented_sampling(
-    context: RequestContext["ClientSession", Any],
+    context: RequestContext[ClientSession],
     params: types.CreateMessageRequestParams,
     task_metadata: types.TaskMetadata,
 ) -> types.CreateTaskResult | types.ErrorData:
@@ -164,7 +166,7 @@ async def default_task_augmented_sampling(
 
 
 async def default_task_augmented_elicitation(
-    context: RequestContext["ClientSession", Any],
+    context: RequestContext[ClientSession],
     params: types.ElicitRequestParams,
     task_metadata: types.TaskMetadata,
 ) -> types.CreateTaskResult | types.ErrorData:
@@ -248,7 +250,7 @@ def handles_request(request: types.ServerRequest) -> bool:
 
     async def handle_request(
         self,
-        ctx: RequestContext["ClientSession", Any],
+        ctx: RequestContext[ClientSession],
         responder: RequestResponder[types.ServerRequest, types.ClientResult],
     ) -> None:
         """Handle a task-related request from the server.

src/mcp/client/session.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import logging
 from typing import Any, Protocol
 
@@ -22,30 +24,27 @@
 class SamplingFnT(Protocol):
     async def __call__(
         self,
-        context: RequestContext["ClientSession", Any],
+        context: RequestContext[ClientSession],
         params: types.CreateMessageRequestParams,
     ) -> types.CreateMessageResult | types.CreateMessageResultWithTools | types.ErrorData: ...  # pragma: no branch
 
 
 class ElicitationFnT(Protocol):
     async def __call__(
         self,
-        context: RequestContext["ClientSession", Any],
+        context: RequestContext[ClientSession],
         params: types.ElicitRequestParams,
     ) -> types.ElicitResult | types.ErrorData: ...  # pragma: no branch
 
 
 class ListRootsFnT(Protocol):
     async def __call__(
-        self, context: RequestContext["ClientSession", Any]
+        self, context: RequestContext[ClientSession]
     ) -> types.ListRootsResult | types.ErrorData: ...  # pragma: no branch
 
 
 class LoggingFnT(Protocol):
-    async def __call__(
-        self,
-        params: types.LoggingMessageNotificationParams,
-    ) -> None: ...  # pragma: no branch
+    async def __call__(self, params: types.LoggingMessageNotificationParams) -> None: ...  # pragma: no branch
 
 
 class MessageHandlerFnT(Protocol):
@@ -62,7 +61,7 @@ async def _default_message_handler(
 
 
 async def _default_sampling_callback(
-    context: RequestContext["ClientSession", Any],
+    context: RequestContext[ClientSession],
     params: types.CreateMessageRequestParams,
 ) -> types.CreateMessageResult | types.CreateMessageResultWithTools | types.ErrorData:
     return types.ErrorData(
@@ -72,7 +71,7 @@ async def _default_sampling_callback(
 
 
 async def _default_elicitation_callback(
-    context: RequestContext["ClientSession", Any],
+    context: RequestContext[ClientSession],
     params: types.ElicitRequestParams,
 ) -> types.ElicitResult | types.ErrorData:
     return types.ErrorData(  # pragma: no cover
@@ -82,7 +81,7 @@ async def _default_elicitation_callback(
 
 
 async def _default_list_roots_callback(
-    context: RequestContext["ClientSession", Any],
+    context: RequestContext[ClientSession],
 ) -> types.ListRootsResult | types.ErrorData:
     return types.ErrorData(
         code=types.INVALID_REQUEST,
@@ -153,10 +152,7 @@ async def initialize(self) -> types.InitializeResult:
             else None
         )
         elicitation = (
-            types.ElicitationCapability(
-                form=types.FormElicitationCapability(),
-                url=types.UrlElicitationCapability(),
-            )
+            types.ElicitationCapability(form=types.FormElicitationCapability(), url=types.UrlElicitationCapability())
             if self._elicitation_callback is not _default_elicitation_callback
             else None
         )
@@ -414,12 +410,7 @@ async def send_roots_list_changed(self) -> None:  # pragma: no cover
         await self.send_notification(types.RootsListChangedNotification())
 
     async def _received_request(self, responder: RequestResponder[types.ServerRequest, types.ClientResult]) -> None:
-        ctx = RequestContext[ClientSession, Any](
-            request_id=responder.request_id,
-            meta=responder.request_meta,
-            session=self,
-            lifespan_context=None,
-        )
+        ctx = RequestContext[ClientSession](request_id=responder.request_id, meta=responder.request_meta, session=self)
 
         # Delegate to experimental task handler if applicable
         if self._task_handlers.handles_request(responder.request):

src/mcp/server/context.py

@@ -0,0 +1,21 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any, Generic, TypeVar
+
+from mcp.server.experimental.request_context import Experimental
+from mcp.server.session import ServerSession
+from mcp.shared.context import RequestContext
+from mcp.shared.message import CloseSSEStreamCallback
+
+LifespanContextT = TypeVar("LifespanContextT")
+RequestT = TypeVar("RequestT", default=Any)
+
+
+@dataclass(kw_only=True)
+class ServerRequestContext(RequestContext[ServerSession], Generic[LifespanContextT, RequestT]):
+    lifespan_context: LifespanContextT
+    experimental: Experimental
+    request: RequestT | None = None
+    close_sse_stream: CloseSSEStreamCallback | None = None
+    close_standalone_sse_stream: CloseSSEStreamCallback | None = None

src/mcp/server/experimental/request_context.py

@@ -57,10 +57,7 @@ def client_supports_tasks(self) -> bool:
         return self._client_capabilities.tasks is not None
 
     def validate_task_mode(
-        self,
-        tool_task_mode: TaskExecutionMode | None,
-        *,
-        raise_error: bool = True,
+        self, tool_task_mode: TaskExecutionMode | None, *, raise_error: bool = True
     ) -> ErrorData | None:
         """Validate that the request is compatible with the tool's task execution mode.
 
@@ -95,12 +92,7 @@ def validate_task_mode(
 
         return error
 
-    def validate_for_tool(
-        self,
-        tool: Tool,
-        *,
-        raise_error: bool = True,
-    ) -> ErrorData | None:
+    def validate_for_tool(self, tool: Tool, *, raise_error: bool = True) -> ErrorData | None:
         """Validate that the request is compatible with the given tool.
 
         Convenience wrapper around validate_task_mode that extracts the mode from a Tool.