sketch: refactor lowlevel server to handler pattern

docs/experimental/index.md

@@ -27,10 +27,9 @@ 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: enable task support (auto-registers default handlers)
+server = Server(name="my-server")
+server.experimental.enable_tasks()
 
 # Client-side
 result = await session.experimental.call_tool_as_task("tool_name", {"arg": "value"})

docs/migration.md

@@ -303,7 +303,6 @@ The nested `RequestParams.Meta` Pydantic model class has been replaced with a to
 - `RequestParams.Meta` (Pydantic model) → `RequestParamsMeta` (TypedDict)
 - Attribute access (`meta.progress_token`) → Dictionary access (`meta.get("progress_token")`)
 - `progress_token` field changed from `ProgressToken | None = None` to `NotRequired[ProgressToken]`
-`
 
 **In request context handlers:**
 
@@ -316,11 +315,12 @@ async def handle_tool(name: str, arguments: dict) -> list[TextContent]:
         await ctx.session.send_progress_notification(ctx.meta.progress_token, 0.5, 100)
 
 # After (v2)
-@server.call_tool()
-async def handle_tool(name: str, arguments: dict) -> list[TextContent]:
-    ctx = server.request_context
+async def handle_call_tool(
+    ctx: RequestContext, params: CallToolRequestParams
+) -> CallToolResult:
     if ctx.meta and "progress_token" in ctx.meta:
         await ctx.session.send_progress_notification(ctx.meta["progress_token"], 0.5, 100)
+    ...
 ```
 
 ### Resource URI type changed from `AnyUrl` to `str`
@@ -378,6 +378,168 @@ await client.read_resource("test://resource")
 await client.read_resource(str(my_any_url))
 ```
 
+### 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.
+
+**Before (v1):**
+
+```python
+from mcp.server.lowlevel.server import Server
+
+server = Server("my-server")
+
+@server.list_tools()
+async def handle_list_tools():
+    return [types.Tool(name="my_tool", description="A tool", inputSchema={})]
+
+@server.call_tool()
+async def handle_call_tool(name: str, arguments: dict):
+    return [types.TextContent(type="text", text=f"Called {name}")]
+```
+
+**After (v2):**
+
+```python
+from mcp.server.lowlevel import Server, RequestHandler
+from mcp.shared.context import RequestContext
+from mcp.types import (
+    CallToolRequestParams,
+    CallToolResult,
+    ListToolsResult,
+    PaginatedRequestParams,
+    TextContent,
+    Tool,
+)
+
+async def handle_list_tools(
+    ctx: RequestContext, params: PaginatedRequestParams | None
+) -> ListToolsResult:
+    return ListToolsResult(tools=[
+        Tool(name="my_tool", description="A tool", inputSchema={})
+    ])
+
+async def handle_call_tool(
+    ctx: RequestContext, params: CallToolRequestParams
+) -> CallToolResult:
+    return CallToolResult(
+        content=[TextContent(type="text", text=f"Called {params.name}")],
+        is_error=False,
+    )
+
+server = Server(
+    "my-server",
+    handlers=[
+        RequestHandler("tools/list", handler=handle_list_tools),
+        RequestHandler("tools/call", handler=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.shared.context import RequestContext
+from mcp.types import ProgressNotificationParams
+
+async def handle_progress(
+    ctx: RequestContext, params: ProgressNotificationParams
+) -> None:
+    print(f"Progress: {params.progress}/{params.total}")
+
+server = Server(
+    "my-server",
+    handlers=[
+        NotificationHandler("notifications/progress", handler=handle_progress),
+    ],
+)
+```
+
+### Lowlevel `Server`: `request_context` property removed
+
+The `server.request_context` property has been removed. Request context is now passed directly to handlers as the first argument (`ctx`). The `request_ctx` module-level contextvar still exists but should not be needed — use `ctx` directly instead.
+
+**Before (v1):**
+
+```python
+from mcp.server.lowlevel.server import request_ctx
+
+@server.call_tool()
+async def handle_call_tool(name: str, arguments: dict):
+    ctx = server.request_context  # or request_ctx.get()
+    await ctx.session.send_log_message(level="info", data="Processing...")
+    return [types.TextContent(type="text", text="Done")]
+```
+
+**After (v2):**
+
+```python
+from mcp.shared.context import RequestContext
+from mcp.types import CallToolRequestParams, CallToolResult, TextContent
+
+async def handle_call_tool(
+    ctx: RequestContext, params: CallToolRequestParams
+) -> CallToolResult:
+    await ctx.session.send_log_message(level="info", data="Processing...")
+    return CallToolResult(
+        content=[TextContent(type="text", text="Done")],
+        is_error=False,
+    )
+```
+
+### `RequestContext`: request-specific fields are now optional
+
+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
+
+# request_id, meta, etc. are available in request handlers
+# 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)
+```
+
 ## Deprecations
 
 <!-- Add deprecations below -->
@@ -413,16 +575,22 @@ 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.server import Server
-
-server = Server("my-server")
-
-# Register handlers...
-@server.list_tools()
-async def list_tools():
-    return [...]
+from mcp.server.lowlevel import Server, RequestHandler
+from mcp.shared.context import RequestContext
+from mcp.types import ListToolsResult, PaginatedRequestParams
+
+async def handle_list_tools(
+    ctx: RequestContext, params: PaginatedRequestParams | None
+) -> ListToolsResult:
+    return ListToolsResult(tools=[...])
+
+server = Server(
+    "my-server",
+    handlers=[
+        RequestHandler("tools/list", handler=handle_list_tools),
+    ],
+)
 
-# Create a Starlette app for streamable HTTP
 app = server.streamable_http_app(
     streamable_http_path="/mcp",
     json_response=False,

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/__init__.py

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