refactor: collapse context hierarchy into single RequestContext class

Merge HandlerContext, RequestHandlerContext, and NotificationHandlerContext
into a single RequestContext class. Request-specific fields (request_id,
meta, etc.) are now optional with None defaults, so the same class works
for both request and notification handlers.

Also:
- Make Server.add_handler/has_handler private (_add_handler/_has_handler)
- MCPServer._setup_handlers() -> _create_handlers() returning a list
  passed to Server(handlers=...) constructor
- Update migration docs to reflect single context class

Author: maxisbey

docs/migration.md

@@ -456,7 +456,7 @@ params = CallToolRequestParams(
 
 ### Lowlevel `Server`: decorator-based handlers replaced with `RequestHandler`/`NotificationHandler`
 
-The lowlevel `Server` class no longer uses decorator methods for handler registration. Instead, handlers are `RequestHandler` and `NotificationHandler` objects passed to the constructor or added via `add_handler()`.
+The lowlevel `Server` class no longer uses decorator methods for handler registration. Instead, handlers are `RequestHandler` and `NotificationHandler` objects passed to the constructor.
 
 **Before (v1):**
 
@@ -478,7 +478,7 @@ async def handle_call_tool(name: str, arguments: dict):
 
 ```python
 from mcp.server.lowlevel import Server, RequestHandler
-from mcp.shared.context import RequestHandlerContext
+from mcp.shared.context import RequestContext
 from mcp.types import (
     CallToolRequestParams,
     CallToolResult,
@@ -489,14 +489,14 @@ from mcp.types import (
 )
 
 async def handle_list_tools(
-    ctx: RequestHandlerContext, params: PaginatedRequestParams | None
+    ctx: RequestContext, params: PaginatedRequestParams | None
 ) -> ListToolsResult:
     return ListToolsResult(tools=[
         Tool(name="my_tool", description="A tool", inputSchema={})
     ])
 
 async def handle_call_tool(
-    ctx: RequestHandlerContext, params: CallToolRequestParams
+    ctx: RequestContext, params: CallToolRequestParams
 ) -> CallToolResult:
     return CallToolResult(
         content=[TextContent(type="text", text=f"Called {params.name}")],
@@ -514,21 +514,19 @@ server = Server(
 
 **Key differences:**
 
-- Handlers receive `(ctx, params)` instead of the full request object or unpacked arguments. `ctx` is a `RequestHandlerContext` (for requests) or `NotificationHandlerContext` (for notifications) with `session`, `lifespan_context`, and `experimental` fields. `params` is the typed request params object.
+- Handlers receive `(ctx, params)` instead of the full request object or unpacked arguments. `ctx` is a `RequestContext` with `session`, `lifespan_context`, and `experimental` fields (plus `request_id`, `meta`, etc. for request handlers). `params` is the typed request params object.
 - Handlers return the full result type (e.g. `ListToolsResult`) rather than unwrapped values (e.g. `list[Tool]`).
 - Registration uses method strings (`"tools/call"`) instead of request types (`CallToolRequest`).
-- Handlers can be added after construction with `server.add_handler()` (silently replaces existing handlers for the same method).
-- `server.has_handler(method)` checks if a handler is registered for a given method string.
 
 **Notification handlers:**
 
 ```python
 from mcp.server.lowlevel import NotificationHandler
-from mcp.shared.context import NotificationHandlerContext
+from mcp.shared.context import RequestContext
 from mcp.types import ProgressNotificationParams
 
 async def handle_progress(
-    ctx: NotificationHandlerContext, params: ProgressNotificationParams
+    ctx: RequestContext, params: ProgressNotificationParams
 ) -> None:
     print(f"Progress: {params.progress}/{params.total}")
 
@@ -559,11 +557,11 @@ async def handle_call_tool(name: str, arguments: dict):
 **After (v2):**
 
 ```python
-from mcp.shared.context import RequestHandlerContext
+from mcp.shared.context import RequestContext
 from mcp.types import CallToolRequestParams, CallToolResult, TextContent
 
 async def handle_call_tool(
-    ctx: RequestHandlerContext, params: CallToolRequestParams
+    ctx: RequestContext, params: CallToolRequestParams
 ) -> CallToolResult:
     await ctx.session.send_log_message(level="info", data="Processing...")
     return CallToolResult(
@@ -572,24 +570,15 @@ async def handle_call_tool(
     )
 ```
 
-### `RequestContext` split into `HandlerContext`, `RequestHandlerContext`, `NotificationHandlerContext`
+### `RequestContext`: request-specific fields are now optional
 
-The `RequestContext` class in `mcp.shared.context` has been replaced with a three-class hierarchy:
-
-- `HandlerContext` — base class with `session`, `lifespan_context`, `experimental`
-- `RequestHandlerContext(HandlerContext)` — adds `request_id`, `meta`, `request`, `close_sse_stream`, `close_standalone_sse_stream`
-- `NotificationHandlerContext(HandlerContext)` — empty subclass for notifications
-
-**Before (v1):**
+The `RequestContext` class now uses optional fields for request-specific data (`request_id`, `meta`, etc.) so it can be used for both request and notification handlers. In notification handlers, these fields are `None`.
 
 ```python
 from mcp.shared.context import RequestContext
-```
-
-**After (v2):**
 
-```python
-from mcp.shared.context import HandlerContext, RequestHandlerContext, NotificationHandlerContext
+# request_id, meta, etc. are available in request handlers
+# but None in notification handlers
 ```
 
 ## New Features
@@ -600,11 +589,11 @@ The `streamable_http_app()` method is now available directly on the lowlevel `Se
 
 ```python
 from mcp.server.lowlevel import Server, RequestHandler
-from mcp.shared.context import RequestHandlerContext
+from mcp.shared.context import RequestContext
 from mcp.types import ListToolsResult, PaginatedRequestParams
 
 async def handle_list_tools(
-    ctx: RequestHandlerContext, params: PaginatedRequestParams | None
+    ctx: RequestContext, params: PaginatedRequestParams | None
 ) -> ListToolsResult:
     return ListToolsResult(tools=[...])
 

src/mcp/server/lowlevel/experimental.py

@@ -12,7 +12,7 @@
 from mcp.server.experimental.task_support import TaskSupport
 from mcp.server.lowlevel.notification_handler import NotificationHandler
 from mcp.server.lowlevel.request_handler import RequestHandler
-from mcp.shared.context import RequestHandlerContext
+from mcp.shared.context import RequestContext
 from mcp.shared.exceptions import MCPError
 from mcp.shared.experimental.tasks.helpers import cancel_task
 from mcp.shared.experimental.tasks.in_memory_task_store import InMemoryTaskStore
@@ -125,7 +125,7 @@ def _register_default_task_handlers(self) -> None:
         if not self._has_handler("tasks/get"):
 
             async def _default_get_task(
-                ctx: RequestHandlerContext[Any, Any, Any], params: GetTaskRequestParams
+                ctx: RequestContext[Any, Any, Any], params: GetTaskRequestParams
             ) -> GetTaskResult:
                 task = await support.store.get_task(params.task_id)
                 if task is None:
@@ -145,8 +145,9 @@ async def _default_get_task(
         if not self._has_handler("tasks/result"):
 
             async def _default_get_task_result(
-                ctx: RequestHandlerContext[Any, Any, Any], params: GetTaskPayloadRequestParams
+                ctx: RequestContext[Any, Any, Any], params: GetTaskPayloadRequestParams
             ) -> GetTaskPayloadResult:
+                assert ctx.request_id is not None
                 req = GetTaskPayloadRequest(params=params)
                 result = await support.handler.handle(req, ctx.session, ctx.request_id)
                 return result
@@ -156,7 +157,7 @@ async def _default_get_task_result(
         if not self._has_handler("tasks/list"):
 
             async def _default_list_tasks(
-                ctx: RequestHandlerContext[Any, Any, Any], params: PaginatedRequestParams | None
+                ctx: RequestContext[Any, Any, Any], params: PaginatedRequestParams | None
             ) -> ListTasksResult:
                 cursor = params.cursor if params else None
                 tasks, next_cursor = await support.store.list_tasks(cursor)
@@ -167,7 +168,7 @@ async def _default_list_tasks(
         if not self._has_handler("tasks/cancel"):
 
             async def _default_cancel_task(
-                ctx: RequestHandlerContext[Any, Any, Any], params: CancelTaskRequestParams
+                ctx: RequestContext[Any, Any, Any], params: CancelTaskRequestParams
             ) -> CancelTaskResult:
                 result = await cancel_task(support.store, params.task_id)
                 return result

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 NotificationHandlerContext
+from mcp.shared.context import RequestContext
 from mcp.types import (
     CancelledNotificationParams,
     NotificationParams,
@@ -16,7 +16,7 @@
 LifespanResultT = TypeVar("LifespanResultT", default=Any)
 RequestT = TypeVar("RequestT", default=Any)
 
-NotificationCtx = NotificationHandlerContext[ServerSession, LifespanResultT]
+NotificationCtx = RequestContext[ServerSession, LifespanResultT, Any]
 
 
 class NotificationHandler(Generic[LifespanResultT, RequestT]):

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 RequestHandlerContext
+from mcp.shared.context import RequestContext
 from mcp.types import (
     CallToolRequestParams,
     CallToolResult,
@@ -31,7 +31,7 @@
 LifespanResultT = TypeVar("LifespanResultT", default=Any)
 RequestT = TypeVar("RequestT", default=Any)
 
-RequestCtx = RequestHandlerContext[ServerSession, LifespanResultT, RequestT]
+RequestCtx = RequestContext[ServerSession, LifespanResultT, RequestT]
 
 
 class RequestHandler(Generic[LifespanResultT, RequestT]):

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 NotificationHandlerContext, RequestHandlerContext
+from mcp.shared.context import RequestContext
 from mcp.shared.exceptions import MCPError
 from mcp.shared.message import ServerMessageMetadata, SessionMessage
 from mcp.shared.session import RequestResponder
@@ -155,7 +155,7 @@ def __init__(
                 else:
                     raise TypeError(f"Unknown handler type: {type(handler)}")
 
-    def add_handler(self, handler: RequestHandler[Any, Any] | NotificationHandler[Any, Any]) -> None:
+    def _add_handler(self, handler: RequestHandler[Any, Any] | NotificationHandler[Any, Any]) -> None:
         """Add a handler, silently replacing any existing handler for the same method."""
         if isinstance(handler, RequestHandler):
             self._request_handlers[handler.method] = handler
@@ -164,7 +164,7 @@ def add_handler(self, handler: RequestHandler[Any, Any] | NotificationHandler[An
         else:
             raise TypeError(f"Unknown handler type: {type(handler)}")
 
-    def has_handler(self, method: str) -> bool:
+    def _has_handler(self, method: str) -> bool:
         """Check if a handler is registered for the given method."""
         return method in self._request_handlers or method in self._notification_handlers
 
@@ -253,8 +253,8 @@ def experimental(self) -> ExperimentalHandlers:
         # We create this inline so we only add these capabilities _if_ they're actually used
         if self._experimental_handlers is None:
             self._experimental_handlers = ExperimentalHandlers(
-                add_handler=self.add_handler,
-                has_handler=self.has_handler,
+                add_handler=self._add_handler,
+                has_handler=self._has_handler,
             )
         return self._experimental_handlers
 
@@ -377,7 +377,7 @@ 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 = RequestHandlerContext(
+                ctx = RequestContext(
                     session,
                     lifespan_context,
                     experimental=Experimental(
@@ -421,7 +421,7 @@ 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 = NotificationHandlerContext(
+                ctx = RequestContext(
                     session,
                     lifespan_context,
                     experimental=Experimental(

src/mcp/server/mcpserver/server.py

@@ -31,6 +31,7 @@
 from mcp.server.elicitation import ElicitationResult, ElicitSchemaModelT, UrlElicitationResult, elicit_with_validation
 from mcp.server.elicitation import elicit_url as _elicit_url
 from mcp.server.lowlevel.helper_types import ReadResourceContents
+from mcp.server.lowlevel.notification_handler import NotificationHandler
 from mcp.server.lowlevel.request_handler import RequestHandler
 from mcp.server.lowlevel.server import LifespanResultT, Server
 from mcp.server.lowlevel.server import lifespan as default_lifespan
@@ -46,7 +47,7 @@
 from mcp.server.streamable_http import EventStore
 from mcp.server.streamable_http_manager import StreamableHTTPSessionManager
 from mcp.server.transport_security import TransportSecuritySettings
-from mcp.shared.context import LifespanContextT, RequestHandlerContext, RequestT
+from mcp.shared.context import LifespanContextT, RequestContext, RequestT
 from mcp.shared.exceptions import MCPError
 from mcp.types import (
     Annotations,
@@ -76,7 +77,7 @@
 
 _CallableT = TypeVar("_CallableT", bound=Callable[..., Any])
 
-_mcp_server_ctx: contextvars.ContextVar[RequestHandlerContext[ServerSession, Any, Any]] = contextvars.ContextVar(
+_mcp_server_ctx: contextvars.ContextVar[RequestContext[ServerSession, Any, Any]] = contextvars.ContextVar(
     "_mcp_server_ctx"
 )
 
@@ -159,6 +160,9 @@ def __init__(
             auth=auth,
         )
 
+        self._tool_manager = ToolManager(tools=tools, warn_on_duplicate_tools=self.settings.warn_on_duplicate_tools)
+        self._resource_manager = ResourceManager(warn_on_duplicate_resources=self.settings.warn_on_duplicate_resources)
+        self._prompt_manager = PromptManager(warn_on_duplicate_prompts=self.settings.warn_on_duplicate_prompts)
         self._lowlevel_server = Server(
             name=name or "mcp-server",
             title=title,
@@ -167,13 +171,11 @@ def __init__(
             website_url=website_url,
             icons=icons,
             version=version,
+            handlers=self._create_handlers(),
             # TODO(Marcelo): It seems there's a type mismatch between the lifespan type from an MCPServer and Server.
             # We need to create a Lifespan type that is a generic on the server type, like Starlette does.
             lifespan=(lifespan_wrapper(self, self.settings.lifespan) if self.settings.lifespan else default_lifespan),  # type: ignore
         )
-        self._tool_manager = ToolManager(tools=tools, warn_on_duplicate_tools=self.settings.warn_on_duplicate_tools)
-        self._resource_manager = ResourceManager(warn_on_duplicate_resources=self.settings.warn_on_duplicate_resources)
-        self._prompt_manager = PromptManager(warn_on_duplicate_prompts=self.settings.warn_on_duplicate_prompts)
         # Validate auth configuration
         if self.settings.auth is not None:
             if auth_server_provider and token_verifier:  # pragma: no cover
@@ -191,9 +193,6 @@ def __init__(
             self._token_verifier = ProviderTokenVerifier(auth_server_provider)
         self._custom_starlette_routes: list[Route] = []
 
-        # Set up MCP protocol handlers
-        self._setup_handlers()
-
         # Configure logging
         configure_logging(self.settings.log_level)
 
@@ -290,8 +289,8 @@ def run(
             case "streamable-http":  # pragma: no cover
                 anyio.run(lambda: self.run_streamable_http_async(**kwargs))
 
-    def _setup_handlers(self) -> None:
-        """Set up core MCP protocol handlers."""
+    def _create_handlers(self) -> list[RequestHandler[Any, Any] | NotificationHandler[Any, Any]]:
+        """Create core MCP protocol handlers."""
 
         async def handle_list_tools(ctx: Any, params: Any) -> ListToolsResult:
             token = _mcp_server_ctx.set(ctx)
@@ -382,15 +381,15 @@ async def handle_get_prompt(ctx: Any, params: Any) -> GetPromptResult:
             finally:
                 _mcp_server_ctx.reset(token)
 
-        self._lowlevel_server.add_handler(RequestHandler("tools/list", handler=handle_list_tools))
-        self._lowlevel_server.add_handler(RequestHandler("tools/call", handler=handle_call_tool))
-        self._lowlevel_server.add_handler(RequestHandler("resources/list", handler=handle_list_resources))
-        self._lowlevel_server.add_handler(RequestHandler("resources/read", handler=handle_read_resource))
-        self._lowlevel_server.add_handler(
-            RequestHandler("resources/templates/list", handler=handle_list_resource_templates)
-        )
-        self._lowlevel_server.add_handler(RequestHandler("prompts/list", handler=handle_list_prompts))
-        self._lowlevel_server.add_handler(RequestHandler("prompts/get", handler=handle_get_prompt))
+        return [
+            RequestHandler("tools/list", handler=handle_list_tools),
+            RequestHandler("tools/call", handler=handle_call_tool),
+            RequestHandler("resources/list", handler=handle_list_resources),
+            RequestHandler("resources/read", handler=handle_read_resource),
+            RequestHandler("resources/templates/list", handler=handle_list_resource_templates),
+            RequestHandler("prompts/list", handler=handle_list_prompts),
+            RequestHandler("prompts/get", handler=handle_get_prompt),
+        ]
 
     async def list_tools(self) -> list[MCPTool]:
         """List all available tools."""
@@ -614,7 +613,11 @@ async def handler(ctx: Any, params: Any) -> CompleteResult:
                 finally:
                     _mcp_server_ctx.reset(token)
 
-            self._lowlevel_server.add_handler(RequestHandler("completion/complete", handler=handler))
+            # TODO(maxisbey): remove private access — completion needs post-construction
+            #   handler registration, find a better pattern for this
+            self._lowlevel_server._add_handler(  # pyright: ignore[reportPrivateUsage]
+                RequestHandler("completion/complete", handler=handler)
+            )
             return func
 
         return decorator
@@ -1136,13 +1139,13 @@ def my_tool(x: int, ctx: Context) -> str:
     The context is optional - tools that don't need it can omit the parameter.
     """
 
-    _request_context: RequestHandlerContext[ServerSessionT, LifespanContextT, RequestT] | None
+    _request_context: RequestContext[ServerSessionT, LifespanContextT, RequestT] | None
     _mcp_server: MCPServer | None
 
     def __init__(
         self,
         *,
-        request_context: (RequestHandlerContext[ServerSessionT, LifespanContextT, RequestT] | None) = None,
+        request_context: (RequestContext[ServerSessionT, LifespanContextT, RequestT] | None) = None,
         mcp_server: MCPServer | None = None,
         **kwargs: Any,
     ):
@@ -1160,7 +1163,7 @@ def mcp_server(self) -> MCPServer:
     @property
     def request_context(
         self,
-    ) -> RequestHandlerContext[ServerSessionT, LifespanContextT, RequestT]:
+    ) -> RequestContext[ServerSessionT, LifespanContextT, RequestT]:
         """Access to the underlying request context."""
         if self._request_context is None:  # pragma: no cover
             raise ValueError("Context is not available outside of a request")