sketch: refactor lowlevel server to handler pattern #1968

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed

maxisbey wants to merge 17 commits into main from sketch/lowlevel-server-v2

+736 −692

docs/experimental/index.md

-Original file line number
+Diff line change
@@ Expand Up / @@ -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"})
@@ Expand Down @@

docs/migration.md

            
                      Original file line number
                      Diff line number
                      Diff line change
                  
    @@ -351,7 +351,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:**

    @@ -364,11 +363,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`

    @@ -426,6 +426,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 -->

    @@ -461,16 +623,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

-Original file line number
+Diff line change
@@ Expand Up / @@ -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?",
@@ Expand Down @@

src/mcp/server/experimental/task_result_handler.py

            
                      Original file line number
                      Diff line number
                      Diff line change
                  
    @@ -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

            
                      Original file line number
                      Diff line number
                      Diff line change
                  
    @@ -1,3 +1,4 @@
  
    from .handler import Handler, NotificationHandler, RequestHandler

    from .server import NotificationOptions, Server

    __all__ = ["Server", "NotificationOptions"]

    __all__ = ["Handler", "NotificationHandler", "NotificationOptions", "RequestHandler", "Server"]

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

sketch: refactor lowlevel server to handler pattern #1968

Diff view

Diff view

There are no files selected for viewing

Uh oh!

Uh oh!