fix: allow ping handler override, add task overloads, update stale docstrings

- Server.__init__: register default ping handler after user handlers so it
  can be overridden via constructor
- handler.py: add typed overloads for experimental task request methods
  (tasks/get, tasks/result, tasks/list, tasks/cancel) and task notification
  (notifications/tasks/status)
- Update stale docstrings in task_result_handler.py, request_context.py,
  and helpers.py to reflect new handler pattern
- Update docs/experimental/index.md example
- Add task handler migration section to docs/migration.md

Author: maxisbey

docs/experimental/index.md

@@ -27,10 +27,13 @@ Tasks are useful for:
 Experimental features are accessed via the `.experimental` property:
 
 ```python
-# Server-side
-@server.experimental.get_task()
-async def handle_get_task(request: GetTaskRequest) -> GetTaskResult:
-    ...
+# Server-side: register a custom task handler
+server = Server(
+    name="my-server",
+    handlers=[
+        RequestHandler("tasks/get", handler=handle_get_task),
+    ],
+)
 
 # Client-side
 result = await session.experimental.call_tool_as_task("tool_name", {"arg": "value"})

docs/migration.md

@@ -533,6 +533,41 @@ from mcp.shared.context import RequestContext
 # but None in notification handlers
 ```
 
+### 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.
+
+Default task handlers are still registered automatically via `server.experimental.enable_tasks()`.
+
+**Before (v1):**
+
+```python
+server = Server("my-server")
+server.experimental.enable_tasks(task_store)
+
+@server.experimental.get_task()
+async def custom_get_task(request: GetTaskRequest) -> GetTaskResult:
+    ...
+```
+
+**After (v2):**
+
+```python
+from mcp.server.lowlevel import Server, RequestHandler
+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.experimental.enable_tasks(task_store)
+```
+
 ## New Features
 
 ### `streamable_http_app()` available on lowlevel Server

src/mcp/server/experimental/request_context.py

@@ -168,10 +168,7 @@ async def run_task(
             RuntimeError: If task support is not enabled or task_metadata is missing
 
         Example:
-            @server.call_tool()
-            async def handle_tool(name: str, args: dict):
-                ctx = server.request_context
-
+            async def handle_tool(ctx: RequestContext, params: CallToolRequestParams) -> CallToolResult:
                 async def work(task: ServerTaskContext) -> CallToolResult:
                     result = await task.elicit(
                         message="Are you sure?",

src/mcp/server/experimental/task_result_handler.py

@@ -47,14 +47,14 @@ class TaskResultHandler:
         # Create handler with store and queue
         handler = TaskResultHandler(task_store, message_queue)
 
-        # Register it with the server
-        @server.experimental.get_task_result()
-        async def handle_task_result(req: GetTaskPayloadRequest) -> GetTaskPayloadResult:
-            ctx = server.request_context
-            return await handler.handle(req, ctx.session, ctx.request_id)
-
-        # Or use the convenience method
-        handler.register(server)
+        # Register as a handler with the lowlevel server
+        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),
+        ])
     """
 
     def __init__(

src/mcp/server/lowlevel/handler.py

@@ -11,14 +11,21 @@
     CallToolRequestParams,
     CallToolResult,
     CancelledNotificationParams,
+    CancelTaskRequestParams,
+    CancelTaskResult,
     CompleteRequestParams,
     CompleteResult,
     EmptyResult,
     GetPromptRequestParams,
     GetPromptResult,
+    GetTaskPayloadRequestParams,
+    GetTaskPayloadResult,
+    GetTaskRequestParams,
+    GetTaskResult,
     ListPromptsResult,
     ListResourcesResult,
     ListResourceTemplatesResult,
+    ListTasksResult,
     ListToolsResult,
     NotificationParams,
     PaginatedRequestParams,
@@ -28,6 +35,7 @@
     RequestParams,
     SetLevelRequestParams,
     SubscribeRequestParams,
+    TaskStatusNotificationParams,
     UnsubscribeRequestParams,
 )
 
@@ -137,6 +145,38 @@ def __init__(
         handler: Callable[[Ctx, CompleteRequestParams], Awaitable[CompleteResult]],
     ) -> None: ...
 
+    @overload
+    def __init__(
+        self,
+        method: Literal["tasks/get"],
+        handler: Callable[[Ctx, GetTaskRequestParams], Awaitable[GetTaskResult]],
+    ) -> None:
+        """Experimental: Tasks may evolve in future protocol versions."""
+
+    @overload
+    def __init__(
+        self,
+        method: Literal["tasks/result"],
+        handler: Callable[[Ctx, GetTaskPayloadRequestParams], Awaitable[GetTaskPayloadResult]],
+    ) -> None:
+        """Experimental: Tasks may evolve in future protocol versions."""
+
+    @overload
+    def __init__(
+        self,
+        method: Literal["tasks/list"],
+        handler: Callable[[Ctx, PaginatedRequestParams | None], Awaitable[ListTasksResult]],
+    ) -> None:
+        """Experimental: Tasks may evolve in future protocol versions."""
+
+    @overload
+    def __init__(
+        self,
+        method: Literal["tasks/cancel"],
+        handler: Callable[[Ctx, CancelTaskRequestParams], Awaitable[CancelTaskResult]],
+    ) -> None:
+        """Experimental: Tasks may evolve in future protocol versions."""
+
     @overload
     def __init__(
         self,
@@ -187,6 +227,14 @@ def __init__(
         handler: Callable[[Ctx, NotificationParams | None], Awaitable[None]],
     ) -> None: ...
 
+    @overload
+    def __init__(
+        self,
+        method: Literal["notifications/tasks/status"],
+        handler: Callable[[Ctx, TaskStatusNotificationParams], Awaitable[None]],
+    ) -> None:
+        """Experimental: Tasks may evolve in future protocol versions."""
+
     @overload
     def __init__(
         self,

src/mcp/server/lowlevel/server.py

@@ -136,9 +136,6 @@ def __init__(
         self._session_manager: StreamableHTTPSessionManager | None = None
         logger.debug("Initializing server %r", name)
 
-        # Register default ping handler
-        self._request_handlers["ping"] = RequestHandler("ping", handler=_ping_handler)
-
         # Process user-provided handlers with duplicate detection
         for handler in handlers:
             if isinstance(handler, RequestHandler):
@@ -152,6 +149,10 @@ def __init__(
             else:
                 raise TypeError(f"Unknown handler type: {type(handler)}")
 
+        # Register default ping handler (after user handlers, so users can override)
+        if "ping" not in self._request_handlers:
+            self._request_handlers["ping"] = RequestHandler("ping", handler=_ping_handler)
+
     def _add_handler(self, handler: Handler) -> None:
         """Add a handler, silently replacing any existing handler for the same method."""
         if isinstance(handler, RequestHandler):

src/mcp/shared/experimental/tasks/helpers.py

@@ -72,9 +72,8 @@ async def cancel_task(
             - Task is already in a terminal state (completed, failed, cancelled)
 
     Example:
-        @server.experimental.cancel_task()
-        async def handle_cancel(request: CancelTaskRequest) -> CancelTaskResult:
-            return await cancel_task(store, request.params.taskId)
+        async def handle_cancel(ctx, params: CancelTaskRequestParams) -> CancelTaskResult:
+            return await cancel_task(store, params.task_id)
     """
     task = await store.get_task(task_id)
     if task is None: