refactor: split RequestContext into handler context hierarchy

- HandlerContext (base): session, lifespan_context, experimental
- RequestHandlerContext: adds request_id, meta, request, SSE callbacks
- NotificationHandlerContext: empty subclass for notifications
- Rename Ctx aliases to RequestCtx / NotificationCtx
- Remove request_ctx contextvar and request_context property
- Tighten string fallback handler signatures to Callable[[Ctx, Any], ...]

Author: maxisbey

src/mcp/server/lowlevel/notification_handler.py

@@ -6,7 +6,7 @@
 from typing_extensions import TypeVar
 
 from mcp.server.session import ServerSession
-from mcp.shared.context import RequestContext
+from mcp.shared.context import NotificationHandlerContext
 from mcp.types import (
     CancelledNotificationParams,
     NotificationParams,
@@ -16,7 +16,7 @@
 LifespanResultT = TypeVar("LifespanResultT", default=Any)
 RequestT = TypeVar("RequestT", default=Any)
 
-Ctx = RequestContext[ServerSession, LifespanResultT, RequestT]
+NotificationCtx = NotificationHandlerContext[ServerSession, LifespanResultT]
 
 
 class NotificationHandler(Generic[LifespanResultT, RequestT]):
@@ -30,40 +30,40 @@ class NotificationHandler(Generic[LifespanResultT, RequestT]):
     def __init__(
         self,
         method: Literal["notifications/initialized"],
-        handler: Callable[[Ctx, NotificationParams | None], Awaitable[None]],
+        handler: Callable[[NotificationCtx, NotificationParams | None], Awaitable[None]],
     ) -> None: ...
 
     @overload
     def __init__(
         self,
         method: Literal["notifications/cancelled"],
-        handler: Callable[[Ctx, CancelledNotificationParams], Awaitable[None]],
+        handler: Callable[[NotificationCtx, CancelledNotificationParams], Awaitable[None]],
     ) -> None: ...
 
     @overload
     def __init__(
         self,
         method: Literal["notifications/progress"],
-        handler: Callable[[Ctx, ProgressNotificationParams], Awaitable[None]],
+        handler: Callable[[NotificationCtx, ProgressNotificationParams], Awaitable[None]],
     ) -> None: ...
 
     @overload
     def __init__(
         self,
         method: Literal["notifications/roots/list_changed"],
-        handler: Callable[[Ctx, NotificationParams | None], Awaitable[None]],
+        handler: Callable[[NotificationCtx, NotificationParams | None], Awaitable[None]],
     ) -> None: ...
 
     @overload
     def __init__(
         self,
         method: str,
-        handler: Callable[..., Awaitable[None]],
+        handler: Callable[[NotificationCtx, Any], Awaitable[None]],
     ) -> None: ...
 
-    def __init__(self, method: str, handler: Callable[..., Awaitable[None]]) -> None:
+    def __init__(self, method: str, handler: Callable[[NotificationCtx, Any], Awaitable[None]]) -> None:
         self.method = method
         self.endpoint = handler
 
-    async def handle(self, ctx: Ctx, params: Any) -> None:
+    async def handle(self, ctx: NotificationCtx, params: Any) -> None:
         await self.endpoint(ctx, params)

src/mcp/server/lowlevel/request_handler.py

@@ -6,7 +6,7 @@
 from typing_extensions import TypeVar
 
 from mcp.server.session import ServerSession
-from mcp.shared.context import RequestContext
+from mcp.shared.context import RequestHandlerContext
 from mcp.types import (
     CallToolRequestParams,
     CallToolResult,
@@ -31,7 +31,7 @@
 LifespanResultT = TypeVar("LifespanResultT", default=Any)
 RequestT = TypeVar("RequestT", default=Any)
 
-Ctx = RequestContext[ServerSession, LifespanResultT, RequestT]
+RequestCtx = RequestHandlerContext[ServerSession, LifespanResultT, RequestT]
 
 
 class RequestHandler(Generic[LifespanResultT, RequestT]):
@@ -45,96 +45,96 @@ class RequestHandler(Generic[LifespanResultT, RequestT]):
     def __init__(
         self,
         method: Literal["ping"],
-        handler: Callable[[Ctx, RequestParams | None], Awaitable[EmptyResult]],
+        handler: Callable[[RequestCtx, RequestParams | None], Awaitable[EmptyResult]],
     ) -> None: ...
 
     @overload
     def __init__(
         self,
         method: Literal["prompts/list"],
-        handler: Callable[[Ctx, PaginatedRequestParams | None], Awaitable[ListPromptsResult]],
+        handler: Callable[[RequestCtx, PaginatedRequestParams | None], Awaitable[ListPromptsResult]],
     ) -> None: ...
 
     @overload
     def __init__(
         self,
         method: Literal["prompts/get"],
-        handler: Callable[[Ctx, GetPromptRequestParams], Awaitable[GetPromptResult]],
+        handler: Callable[[RequestCtx, GetPromptRequestParams], Awaitable[GetPromptResult]],
     ) -> None: ...
 
     @overload
     def __init__(
         self,
         method: Literal["resources/list"],
-        handler: Callable[[Ctx, PaginatedRequestParams | None], Awaitable[ListResourcesResult]],
+        handler: Callable[[RequestCtx, PaginatedRequestParams | None], Awaitable[ListResourcesResult]],
     ) -> None: ...
 
     @overload
     def __init__(
         self,
         method: Literal["resources/templates/list"],
-        handler: Callable[[Ctx, PaginatedRequestParams | None], Awaitable[ListResourceTemplatesResult]],
+        handler: Callable[[RequestCtx, PaginatedRequestParams | None], Awaitable[ListResourceTemplatesResult]],
     ) -> None: ...
 
     @overload
     def __init__(
         self,
         method: Literal["resources/read"],
-        handler: Callable[[Ctx, ReadResourceRequestParams], Awaitable[ReadResourceResult]],
+        handler: Callable[[RequestCtx, ReadResourceRequestParams], Awaitable[ReadResourceResult]],
     ) -> None: ...
 
     @overload
     def __init__(
         self,
         method: Literal["resources/subscribe"],
-        handler: Callable[[Ctx, SubscribeRequestParams], Awaitable[EmptyResult]],
+        handler: Callable[[RequestCtx, SubscribeRequestParams], Awaitable[EmptyResult]],
     ) -> None: ...
 
     @overload
     def __init__(
         self,
         method: Literal["resources/unsubscribe"],
-        handler: Callable[[Ctx, UnsubscribeRequestParams], Awaitable[EmptyResult]],
+        handler: Callable[[RequestCtx, UnsubscribeRequestParams], Awaitable[EmptyResult]],
     ) -> None: ...
 
     @overload
     def __init__(
         self,
         method: Literal["logging/setLevel"],
-        handler: Callable[[Ctx, SetLevelRequestParams], Awaitable[EmptyResult]],
+        handler: Callable[[RequestCtx, SetLevelRequestParams], Awaitable[EmptyResult]],
     ) -> None: ...
 
     @overload
     def __init__(
         self,
         method: Literal["tools/list"],
-        handler: Callable[[Ctx, PaginatedRequestParams | None], Awaitable[ListToolsResult]],
+        handler: Callable[[RequestCtx, PaginatedRequestParams | None], Awaitable[ListToolsResult]],
     ) -> None: ...
 
     @overload
     def __init__(
         self,
         method: Literal["tools/call"],
-        handler: Callable[[Ctx, CallToolRequestParams], Awaitable[CallToolResult]],
+        handler: Callable[[RequestCtx, CallToolRequestParams], Awaitable[CallToolResult]],
     ) -> None: ...
 
     @overload
     def __init__(
         self,
         method: Literal["completion/complete"],
-        handler: Callable[[Ctx, CompleteRequestParams], Awaitable[CompleteResult]],
+        handler: Callable[[RequestCtx, CompleteRequestParams], Awaitable[CompleteResult]],
     ) -> None: ...
 
     @overload
     def __init__(
         self,
         method: str,
-        handler: Callable[..., Awaitable[Any]],
+        handler: Callable[[RequestCtx, Any], Awaitable[Any]],
     ) -> None: ...
 
-    def __init__(self, method: str, handler: Callable[..., Awaitable[Any]]) -> None:
+    def __init__(self, method: str, handler: Callable[[RequestCtx, Any], Awaitable[Any]]) -> None:
         self.method = method
         self.endpoint = handler
 
-    async def handle(self, ctx: Ctx, params: Any) -> Any:
+    async def handle(self, ctx: RequestCtx, params: Any) -> Any:
         return await self.endpoint(ctx, params)

src/mcp/server/lowlevel/server.py

@@ -68,7 +68,7 @@ async def main():
 from mcp.server.streamable_http import EventStore
 from mcp.server.streamable_http_manager import StreamableHTTPASGIApp, StreamableHTTPSessionManager
 from mcp.server.transport_security import TransportSecuritySettings
-from mcp.shared.context import RequestContext
+from mcp.shared.context import NotificationHandlerContext, RequestHandlerContext
 from mcp.shared.exceptions import MCPError
 from mcp.shared.message import ServerMessageMetadata, SessionMessage
 from mcp.shared.session import RequestResponder
@@ -374,17 +374,17 @@ async def _handle_request(
                 task_metadata = None
                 if hasattr(req, "params") and req.params is not None:
                     task_metadata = getattr(req.params, "task", None)
-                ctx = RequestContext(
-                    message.request_id,
-                    message.request_meta,
+                ctx = RequestHandlerContext(
                     session,
                     lifespan_context,
-                    Experimental(
+                    experimental=Experimental(
                         task_metadata=task_metadata,
                         _client_capabilities=client_capabilities,
                         _session=session,
                         _task_support=task_support,
                     ),
+                    request_id=message.request_id,
+                    meta=message.request_meta,
                     request=request_data,
                     close_sse_stream=close_sse_stream_cb,
                     close_standalone_sse_stream=close_standalone_sse_stream_cb,
@@ -418,11 +418,9 @@ async def _handle_notification(
             try:
                 client_capabilities = session.client_params.capabilities if session.client_params else None
                 task_support = self._experimental_handlers.task_support if self._experimental_handlers else None
-                ctx: RequestContext[ServerSession, Any, Any] = RequestContext(
-                    request_id="__notification__",
-                    meta=None,
-                    session=session,
-                    lifespan_context=lifespan_context,
+                ctx = NotificationHandlerContext(
+                    session,
+                    lifespan_context,
                     experimental=Experimental(
                         task_metadata=None,
                         _client_capabilities=client_capabilities,

src/mcp/shared/context.py

@@ -1,4 +1,4 @@
-"""Request context for MCP handlers."""
+"""Handler contexts for MCP handlers."""
 
 from dataclasses import dataclass, field
 from typing import Any, Generic
@@ -15,16 +15,29 @@
 
 
 @dataclass
-class RequestContext(Generic[SessionT, LifespanContextT, RequestT]):
-    request_id: RequestId
-    meta: RequestParamsMeta | None
+class HandlerContext(Generic[SessionT, LifespanContextT]):
+    """Base context shared by all handlers."""
+
     session: SessionT
     lifespan_context: LifespanContextT
     # NOTE: This is typed as Any to avoid circular imports. The actual type is
     # mcp.server.experimental.request_context.Experimental, but importing it here
     # triggers mcp.server.__init__ -> mcpserver -> tools -> back to this module.
     # The Server sets this to an Experimental instance at runtime.
-    experimental: Any = field(default=None)
-    request: RequestT | None = None
-    close_sse_stream: CloseSSEStreamCallback | None = None
-    close_standalone_sse_stream: CloseSSEStreamCallback | None = None
+    experimental: Any = field(default=None, kw_only=True)
+
+
+@dataclass
+class RequestHandlerContext(HandlerContext[SessionT, LifespanContextT], Generic[SessionT, LifespanContextT, RequestT]):
+    """Context for request handlers."""
+
+    request_id: RequestId = field(kw_only=True)
+    meta: RequestParamsMeta | None = field(kw_only=True)
+    request: RequestT | None = field(default=None, kw_only=True)
+    close_sse_stream: CloseSSEStreamCallback | None = field(default=None, kw_only=True)
+    close_standalone_sse_stream: CloseSSEStreamCallback | None = field(default=None, kw_only=True)
+
+
+@dataclass
+class NotificationHandlerContext(HandlerContext[SessionT, LifespanContextT]):
+    """Context for notification handlers."""