refactor: replace Handler objects with on_* constructor kwargs

Replace the RequestHandler/NotificationHandler wrapper classes with direct
on_* keyword arguments on the Server constructor. Handlers are now raw
callables stored in internal dicts keyed by method string.

- Delete handler.py (Handler, RequestHandler, NotificationHandler)
- Server constructor takes on_list_tools, on_call_tool, etc. instead of
  handlers: Sequence[Handler]
- Add _add_request_handler / _add_notification_handler for post-construction
  registration (used by ExperimentalHandlers and MCPServer completion)
- Remove RequestT generic from Server — the request context type is
  transport-specific and not known at construction time
- Type ServerMessageMetadata.request_context as Any (transport-agnostic)
- Update ExperimentalHandlers to use RequestContext[ServerSession, Any, Any]
- Update MCPServer to pass on_* kwargs via _create_handler_kwargs()

Author: maxisbey

docs/migration.md

@@ -426,9 +426,9 @@ await client.read_resource("test://resource")
 await client.read_resource(str(my_any_url))
 ```
 
-### Lowlevel `Server`: decorator-based handlers replaced with `RequestHandler`/`NotificationHandler`
+### Lowlevel `Server`: decorator-based handlers replaced with constructor `on_*` params
 
-The lowlevel `Server` class no longer uses decorator methods for handler registration. Instead, handlers are `RequestHandler` and `NotificationHandler` objects passed to the constructor.
+The lowlevel `Server` class no longer uses decorator methods for handler registration. Instead, handlers are passed as `on_*` keyword arguments to the constructor.
 
 **Before (v1):**
 
@@ -449,7 +449,7 @@ async def handle_call_tool(name: str, arguments: dict):
 **After (v2):**
 
 ```python
-from mcp.server.lowlevel import Server, RequestHandler
+from mcp.server.lowlevel import Server
 from mcp.shared.context import RequestContext
 from mcp.types import (
     CallToolRequestParams,
@@ -477,23 +477,20 @@ async def handle_call_tool(
 
 server = Server(
     "my-server",
-    handlers=[
-        RequestHandler("tools/list", handler=handle_list_tools),
-        RequestHandler("tools/call", handler=handle_call_tool),
-    ],
+    on_list_tools=handle_list_tools,
+    on_call_tool=handle_call_tool,
 )
 ```
 
 **Key differences:**
 
 - 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`).
 
 **Notification handlers:**
 
 ```python
-from mcp.server.lowlevel import NotificationHandler
+from mcp.server.lowlevel import Server
 from mcp.shared.context import RequestContext
 from mcp.types import ProgressNotificationParams
 
@@ -504,9 +501,7 @@ async def handle_progress(
 
 server = Server(
     "my-server",
-    handlers=[
-        NotificationHandler("notifications/progress", handler=handle_progress),
-    ],
+    on_progress=handle_progress,
 )
 ```
 
@@ -555,7 +550,7 @@ from mcp.shared.context import RequestContext
 
 ### Experimental: task handler decorators removed
 
-The experimental decorator methods on `ExperimentalHandlers` (`@server.experimental.list_tasks()`, `@server.experimental.get_task()`, etc.) have been removed. Custom task handlers are now registered as `RequestHandler` objects passed to the `Server` constructor, consistent with the new handler pattern.
+The experimental decorator methods on `ExperimentalHandlers` (`@server.experimental.list_tasks()`, `@server.experimental.get_task()`, etc.) have been removed.
 
 Default task handlers are still registered automatically via `server.experimental.enable_tasks()`.
 
@@ -573,19 +568,13 @@ async def custom_get_task(request: GetTaskRequest) -> GetTaskResult:
 **After (v2):**
 
 ```python
-from mcp.server.lowlevel import Server, RequestHandler
+from mcp.server.lowlevel import Server
 from mcp.types import GetTaskRequestParams, GetTaskResult
 
-async def custom_get_task(ctx, params: GetTaskRequestParams) -> GetTaskResult:
-    ...
-
-server = Server(
-    "my-server",
-    handlers=[
-        RequestHandler("tasks/get", handler=custom_get_task),
-    ],
-)
+server = Server("my-server")
 server.experimental.enable_tasks(task_store)
+# Default handlers are registered automatically.
+# Custom task handlers are not yet supported via the constructor.
 ```
 
 ## Deprecations
@@ -623,7 +612,7 @@ params = CallToolRequestParams(
 The `streamable_http_app()` method is now available directly on the lowlevel `Server` class, not just `MCPServer`. This allows using the streamable HTTP transport without the MCPServer wrapper.
 
 ```python
-from mcp.server.lowlevel import Server, RequestHandler
+from mcp.server.lowlevel import Server
 from mcp.shared.context import RequestContext
 from mcp.types import ListToolsResult, PaginatedRequestParams
 
@@ -634,9 +623,7 @@ async def handle_list_tools(
 
 server = Server(
     "my-server",
-    handlers=[
-        RequestHandler("tools/list", handler=handle_list_tools),
-    ],
+    on_list_tools=handle_list_tools,
 )
 
 app = server.streamable_http_app(

src/mcp/server/experimental/task_result_handler.py

@@ -52,9 +52,7 @@ async def handle_task_result(ctx, params):
             return await handler.handle(
                 GetTaskPayloadRequest(params=params), ctx.session, ctx.request_id
             )
-        server = Server(handlers=[
-            RequestHandler("tasks/result", handler=handle_task_result),
-        ])
+        server = Server(on_call_tool=..., on_list_tools=...)
     """
 
     def __init__(

src/mcp/server/lowlevel/__init__.py

@@ -1,4 +1,3 @@
-from .handler import Handler, NotificationHandler, RequestHandler
 from .server import NotificationOptions, Server
 
-__all__ = ["Handler", "NotificationHandler", "NotificationOptions", "RequestHandler", "Server"]
+__all__ = ["NotificationOptions", "Server"]

src/mcp/server/lowlevel/experimental.py

@@ -6,11 +6,11 @@
 from __future__ import annotations
 
 import logging
-from collections.abc import Callable
+from collections.abc import Awaitable, Callable
 from typing import Any
 
 from mcp.server.experimental.task_support import TaskSupport
-from mcp.server.lowlevel.handler import Handler, RequestHandler
+from mcp.server.session import ServerSession
 from mcp.shared.context import RequestContext
 from mcp.shared.exceptions import MCPError
 from mcp.shared.experimental.tasks.helpers import cancel_task
@@ -48,10 +48,12 @@ class ExperimentalHandlers:
 
     def __init__(
         self,
-        add_handler: Callable[[Handler], None],
+        add_request_handler: Callable[
+            [str, Callable[[RequestContext[ServerSession, Any, Any], Any], Awaitable[Any]]], None
+        ],
         has_handler: Callable[[str], bool],
     ) -> None:
-        self._add_handler = add_handler
+        self._add_request_handler = add_request_handler
         self._has_handler = has_handler
         self._task_support: TaskSupport | None = None
 
@@ -124,7 +126,7 @@ def _register_default_task_handlers(self) -> None:
         if not self._has_handler("tasks/get"):
 
             async def _default_get_task(
-                ctx: RequestContext[Any, Any, Any], params: GetTaskRequestParams
+                ctx: RequestContext[ServerSession, Any, Any], params: GetTaskRequestParams
             ) -> GetTaskResult:
                 task = await support.store.get_task(params.task_id)
                 if task is None:
@@ -139,37 +141,37 @@ async def _default_get_task(
                     poll_interval=task.poll_interval,
                 )
 
-            self._add_handler(RequestHandler("tasks/get", handler=_default_get_task))
+            self._add_request_handler("tasks/get", _default_get_task)
 
         if not self._has_handler("tasks/result"):
 
             async def _default_get_task_result(
-                ctx: RequestContext[Any, Any, Any], params: GetTaskPayloadRequestParams
+                ctx: RequestContext[ServerSession, 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
 
-            self._add_handler(RequestHandler("tasks/result", handler=_default_get_task_result))
+            self._add_request_handler("tasks/result", _default_get_task_result)
 
         if not self._has_handler("tasks/list"):
 
             async def _default_list_tasks(
-                ctx: RequestContext[Any, Any, Any], params: PaginatedRequestParams | None
+                ctx: RequestContext[ServerSession, Any, Any], params: PaginatedRequestParams | None
             ) -> ListTasksResult:
                 cursor = params.cursor if params else None
                 tasks, next_cursor = await support.store.list_tasks(cursor)
                 return ListTasksResult(tasks=tasks, next_cursor=next_cursor)
 
-            self._add_handler(RequestHandler("tasks/list", handler=_default_list_tasks))
+            self._add_request_handler("tasks/list", _default_list_tasks)
 
         if not self._has_handler("tasks/cancel"):
 
             async def _default_cancel_task(
-                ctx: RequestContext[Any, Any, Any], params: CancelTaskRequestParams
+                ctx: RequestContext[ServerSession, Any, Any], params: CancelTaskRequestParams
             ) -> CancelTaskResult:
                 result = await cancel_task(support.store, params.task_id)
                 return result
 
-            self._add_handler(RequestHandler("tasks/cancel", handler=_default_cancel_task))
+            self._add_request_handler("tasks/cancel", _default_cancel_task)