feat: add on_* handler kwargs to enable_tasks for custom task handlers

Allow users to override default task handlers by passing on_get_task,
on_task_result, on_list_tasks, and on_cancel_task to enable_tasks().
User-provided handlers are registered first; defaults fill in the rest.

Author: maxisbey

docs/migration.md

@@ -594,7 +594,7 @@ from mcp.server import ServerRequestContext
 
 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()`.
+Default task handlers are still registered automatically via `server.experimental.enable_tasks()`. Custom handlers can be passed as `on_*` kwargs to override specific defaults.
 
 **Before (v1):**
 
@@ -610,13 +610,16 @@ async def custom_get_task(request: GetTaskRequest) -> GetTaskResult:
 **After (v2):**
 
 ```python
-from mcp.server.lowlevel import Server
+from mcp.server import Server, ServerRequestContext
 from mcp.types import GetTaskRequestParams, GetTaskResult
 
+
+async def custom_get_task(ctx: ServerRequestContext, params: GetTaskRequestParams) -> GetTaskResult:
+    ...
+
+
 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.
+server.experimental.enable_tasks(on_get_task=custom_get_task)
 ```
 
 ## Deprecations

src/mcp/server/experimental/task_result_handler.py

@@ -44,17 +44,14 @@ class TaskResultHandler:
     5. Returns the final result
 
     Usage:
-        handler = TaskResultHandler(task_store, message_queue)
-
         async def handle_task_result(
             ctx: ServerRequestContext, params: GetTaskPayloadRequestParams
         ) -> GetTaskPayloadResult:
-            assert ctx.request_id is not None
-            return await handler.handle(
-                GetTaskPayloadRequest(params=params), ctx.session, ctx.request_id
-            )
+            ...
 
-        server._add_request_handler("tasks/result", handle_task_result)
+        server.experimental.enable_tasks(
+            on_task_result=handle_task_result,
+        )
     """
 
     def __init__(

src/mcp/server/lowlevel/experimental.py

@@ -78,15 +78,35 @@ def enable_tasks(
         self,
         store: TaskStore | None = None,
         queue: TaskMessageQueue | None = None,
+        *,
+        on_get_task: Callable[[ServerRequestContext[Any, Any], GetTaskRequestParams], Awaitable[GetTaskResult]]
+        | None = None,
+        on_task_result: Callable[
+            [ServerRequestContext[Any, Any], GetTaskPayloadRequestParams], Awaitable[GetTaskPayloadResult]
+        ]
+        | None = None,
+        on_list_tasks: Callable[
+            [ServerRequestContext[Any, Any], PaginatedRequestParams | None], Awaitable[ListTasksResult]
+        ]
+        | None = None,
+        on_cancel_task: Callable[
+            [ServerRequestContext[Any, Any], CancelTaskRequestParams], Awaitable[CancelTaskResult]
+        ]
+        | None = None,
     ) -> TaskSupport:
         """Enable experimental task support.
 
-        This sets up the task infrastructure and auto-registers default handlers
-        for tasks/get, tasks/result, tasks/list, and tasks/cancel.
+        This sets up the task infrastructure and registers handlers for
+        tasks/get, tasks/result, tasks/list, and tasks/cancel. Custom handlers
+        can be provided via the on_* kwargs; any not provided will use defaults.
 
         Args:
             store: Custom TaskStore implementation (defaults to InMemoryTaskStore)
             queue: Custom TaskMessageQueue implementation (defaults to InMemoryTaskMessageQueue)
+            on_get_task: Custom handler for tasks/get
+            on_task_result: Custom handler for tasks/result
+            on_list_tasks: Custom handler for tasks/list
+            on_cancel_task: Custom handler for tasks/cancel
 
         Returns:
             The TaskSupport configuration object
@@ -110,6 +130,17 @@ def enable_tasks(
 
         self._task_support = TaskSupport(store=store, queue=queue)
 
+        # Register user-provided handlers
+        if on_get_task is not None:
+            self._add_request_handler("tasks/get", on_get_task)
+        if on_task_result is not None:
+            self._add_request_handler("tasks/result", on_task_result)
+        if on_list_tasks is not None:
+            self._add_request_handler("tasks/list", on_list_tasks)
+        if on_cancel_task is not None:
+            self._add_request_handler("tasks/cancel", on_cancel_task)
+
+        # Fill in defaults for any not provided
         if not self._has_handler("tasks/get"):
 
             async def _default_get_task(