fix: update tests to use new Server constructor kwargs pattern

Migrate test files from the old decorator-based handler registration
to the new on_* constructor kwargs pattern. Key changes:

- Replace @server.list_tools(), @server.call_tool(), etc. decorators
  with on_list_tools, on_call_tool, etc. kwargs on Server()
- Replace server.request_context access with ctx parameter
  (first argument to all handlers)
- Handlers now receive (ctx, params) and return full result types
  (e.g. ListToolsResult instead of list[Tool])
- Convert experimental task decorators to enable_tasks() kwargs
- Add LifespanContextT default to ServerRequestContext
- Widen on_call_tool return type to include CreateTaskResult
- Delete redundant tests/shared/test_memory.py
- Simplify tests to use Client where possible

Author: maxisbey

src/mcp/server/context.py

@@ -10,7 +10,7 @@
 from mcp.shared._context import RequestContext
 from mcp.shared.message import CloseSSEStreamCallback
 
-LifespanContextT = TypeVar("LifespanContextT")
+LifespanContextT = TypeVar("LifespanContextT", default=dict[str, Any])
 RequestT = TypeVar("RequestT", default=Any)
 
 

src/mcp/server/lowlevel/server.py

@@ -124,7 +124,7 @@ def __init__(
         | None = None,
         on_call_tool: Callable[
             [ServerRequestContext[LifespanResultT], types.CallToolRequestParams],
-            Awaitable[types.CallToolResult],
+            Awaitable[types.CallToolResult | types.CreateTaskResult],
         ]
         | None = None,
         on_list_resources: Callable[
@@ -643,5 +643,3 @@ def streamable_http_app(
             middleware=middleware,
             lifespan=lambda app: session_manager.run(),
         )
-
-

tests/client/test_client.py

@@ -11,7 +11,7 @@
 from mcp import types
 from mcp.client._memory import InMemoryTransport
 from mcp.client.client import Client
-from mcp.server import Server
+from mcp.server import Server, ServerRequestContext
 from mcp.server.mcpserver import MCPServer
 from mcp.types import (
     CallToolResult,
@@ -41,33 +41,36 @@
 @pytest.fixture
 def simple_server() -> Server:
     """Create a simple MCP server for testing."""
-    server = Server(name="test_server")
 
-    @server.list_resources()
-    async def handle_list_resources():
-        return [Resource(uri="memory://test", name="Test Resource", description="A test resource")]
+    async def handle_list_resources(
+        ctx: ServerRequestContext, params: types.PaginatedRequestParams | None
+    ) -> ListResourcesResult:
+        return ListResourcesResult(
+            resources=[Resource(uri="memory://test", name="Test Resource", description="A test resource")]
+        )
 
-    @server.subscribe_resource()
-    async def handle_subscribe_resource(uri: str):
-        pass
+    async def handle_subscribe_resource(ctx: ServerRequestContext, params: types.SubscribeRequestParams) -> EmptyResult:
+        return EmptyResult()
 
-    @server.unsubscribe_resource()
-    async def handle_unsubscribe_resource(uri: str):
-        pass
+    async def handle_unsubscribe_resource(
+        ctx: ServerRequestContext, params: types.UnsubscribeRequestParams
+    ) -> EmptyResult:
+        return EmptyResult()
 
-    @server.set_logging_level()
-    async def handle_set_logging_level(level: str):
-        pass
+    async def handle_set_logging_level(ctx: ServerRequestContext, params: types.SetLevelRequestParams) -> EmptyResult:
+        return EmptyResult()
 
-    @server.completion()
-    async def handle_completion(
-        ref: types.PromptReference | types.ResourceTemplateReference,
-        argument: types.CompletionArgument,
-        context: types.CompletionContext | None,
-    ) -> types.Completion | None:
-        return types.Completion(values=[])
+    async def handle_completion(ctx: ServerRequestContext, params: types.CompleteRequestParams) -> types.CompleteResult:
+        return types.CompleteResult(completion=types.Completion(values=[]))
 
-    return server
+    return Server(
+        name="test_server",
+        on_list_resources=handle_list_resources,
+        on_subscribe_resource=handle_subscribe_resource,
+        on_unsubscribe_resource=handle_unsubscribe_resource,
+        on_set_logging_level=handle_set_logging_level,
+        on_completion=handle_completion,
+    )
 
 
 @pytest.fixture
@@ -202,19 +205,14 @@ async def test_client_send_progress_notification():
     """Test sending progress notification."""
     received_from_client = None
     event = anyio.Event()
-    server = Server(name="test_server")
-
-    @server.progress_notification()
-    async def handle_progress_notification(
-        progress_token: str | int,
-        progress: float = 0.0,
-        total: float | None = None,
-        message: str | None = None,
-    ) -> None:
+
+    async def handle_progress(ctx: ServerRequestContext, params: types.ProgressNotificationParams) -> None:
         nonlocal received_from_client
-        received_from_client = {"progress_token": progress_token, "progress": progress}
+        received_from_client = {"progress_token": params.progress_token, "progress": params.progress}
         event.set()
 
+    server = Server(name="test_server", on_progress=handle_progress)
+
     async with Client(server) as client:
         await client.send_progress_notification(progress_token="token123", progress=50.0)
         await event.wait()

tests/client/test_http_unicode.py

@@ -8,7 +8,6 @@
 import socket
 from collections.abc import AsyncGenerator, Generator
 from contextlib import asynccontextmanager
-from typing import Any
 
 import pytest
 from starlette.applications import Starlette
@@ -17,7 +16,7 @@
 from mcp import types
 from mcp.client.session import ClientSession
 from mcp.client.streamable_http import streamable_http_client
-from mcp.server import Server
+from mcp.server import Server, ServerRequestContext
 from mcp.server.streamable_http_manager import StreamableHTTPSessionManager
 from mcp.types import TextContent, Tool
 from tests.test_helpers import wait_for_server
@@ -47,54 +46,56 @@ def run_unicode_server(port: int) -> None:  # pragma: no cover
     import uvicorn
 
     # Need to recreate the server setup in this process
-    server = Server(name="unicode_test_server")
-
-    @server.list_tools()
-    async def list_tools() -> list[Tool]:
-        """List tools with Unicode descriptions."""
-        return [
-            Tool(
-                name="echo_unicode",
-                description="🔤 Echo Unicode text - Hello 👋 World 🌍 - Testing 🧪 Unicode ✨",
-                input_schema={
-                    "type": "object",
-                    "properties": {
-                        "text": {"type": "string", "description": "Text to echo back"},
+    async def handle_list_tools(
+        ctx: ServerRequestContext, params: types.PaginatedRequestParams | None
+    ) -> types.ListToolsResult:
+        return types.ListToolsResult(
+            tools=[
+                Tool(
+                    name="echo_unicode",
+                    description="🔤 Echo Unicode text - Hello 👋 World 🌍 - Testing 🧪 Unicode ✨",
+                    input_schema={
+                        "type": "object",
+                        "properties": {
+                            "text": {"type": "string", "description": "Text to echo back"},
+                        },
+                        "required": ["text"],
                     },
-                    "required": ["text"],
-                },
-            ),
-        ]
-
-    @server.call_tool()
-    async def call_tool(name: str, arguments: dict[str, Any] | None) -> list[TextContent]:
-        """Handle tool calls with Unicode content."""
-        if name == "echo_unicode":
-            text = arguments.get("text", "") if arguments else ""
-            return [
-                TextContent(
-                    type="text",
-                    text=f"Echo: {text}",
-                )
+                ),
             ]
-        else:
-            raise ValueError(f"Unknown tool: {name}")
-
-    @server.list_prompts()
-    async def list_prompts() -> list[types.Prompt]:
-        """List prompts with Unicode names and descriptions."""
-        return [
-            types.Prompt(
-                name="unicode_prompt",
-                description="Unicode prompt - Слой хранилища, где располагаются",
-                arguments=[],
+        )
+
+    async def handle_call_tool(ctx: ServerRequestContext, params: types.CallToolRequestParams) -> types.CallToolResult:
+        if params.name == "echo_unicode":
+            text = params.arguments.get("text", "") if params.arguments else ""
+            return types.CallToolResult(
+                content=[
+                    TextContent(
+                        type="text",
+                        text=f"Echo: {text}",
+                    )
+                ]
             )
-        ]
+        else:
+            raise ValueError(f"Unknown tool: {params.name}")
+
+    async def handle_list_prompts(
+        ctx: ServerRequestContext, params: types.PaginatedRequestParams | None
+    ) -> types.ListPromptsResult:
+        return types.ListPromptsResult(
+            prompts=[
+                types.Prompt(
+                    name="unicode_prompt",
+                    description="Unicode prompt - Слой хранилища, где располагаются",
+                    arguments=[],
+                )
+            ]
+        )
 
-    @server.get_prompt()
-    async def get_prompt(name: str, arguments: dict[str, Any] | None) -> types.GetPromptResult:
-        """Get a prompt with Unicode content."""
-        if name == "unicode_prompt":
+    async def handle_get_prompt(
+        ctx: ServerRequestContext, params: types.GetPromptRequestParams
+    ) -> types.GetPromptResult:
+        if params.name == "unicode_prompt":
             return types.GetPromptResult(
                 messages=[
                     types.PromptMessage(
@@ -106,7 +107,15 @@ async def get_prompt(name: str, arguments: dict[str, Any] | None) -> types.GetPr
                     )
                 ]
             )
-        raise ValueError(f"Unknown prompt: {name}")
+        raise ValueError(f"Unknown prompt: {params.name}")
+
+    server = Server(
+        name="unicode_test_server",
+        on_list_tools=handle_list_tools,
+        on_call_tool=handle_call_tool,
+        on_list_prompts=handle_list_prompts,
+        on_get_prompt=handle_get_prompt,
+    )
 
     # Create the session manager
     session_manager = StreamableHTTPSessionManager(

tests/client/test_list_methods_cursor.py

@@ -3,9 +3,9 @@
 import pytest
 
 from mcp import Client, types
-from mcp.server import Server
+from mcp.server import Server, ServerRequestContext
 from mcp.server.mcpserver import MCPServer
-from mcp.types import ListToolsRequest, ListToolsResult
+from mcp.types import ListToolsResult
 
 from .conftest import StreamSpyCollection
 
@@ -105,14 +105,16 @@ async def test_list_tools_with_strict_server_validation(
 
 async def test_list_tools_with_lowlevel_server():
     """Test that list_tools works with a lowlevel Server using params."""
-    server = Server("test-lowlevel")
 
-    @server.list_tools()
-    async def handle_list_tools(request: ListToolsRequest) -> ListToolsResult:
+    async def handle_list_tools(
+        ctx: ServerRequestContext, params: types.PaginatedRequestParams | None
+    ) -> ListToolsResult:
         # Echo back what cursor we received in the tool description
-        cursor = request.params.cursor if request.params else None
+        cursor = params.cursor if params else None
         return ListToolsResult(tools=[types.Tool(name="test_tool", description=f"cursor={cursor}", input_schema={})])
 
+    server = Server("test-lowlevel", on_list_tools=handle_list_tools)
+
     async with Client(server) as client:
         result = await client.list_tools()
         assert result.tools[0].description == "cursor=None"

tests/client/transports/test_memory.py

@@ -2,31 +2,31 @@
 
 import pytest
 
-from mcp import Client
+from mcp import Client, types
 from mcp.client._memory import InMemoryTransport
-from mcp.server import Server
+from mcp.server import Server, ServerRequestContext
 from mcp.server.mcpserver import MCPServer
-from mcp.types import Resource
+from mcp.types import ListResourcesResult, Resource
 
 
 @pytest.fixture
 def simple_server() -> Server:
     """Create a simple MCP server for testing."""
-    server = Server(name="test_server")
-
-    # pragma: no cover - handler exists only to register a resource capability.
-    # Transport tests verify stream creation, not handler invocation.
-    @server.list_resources()
-    async def handle_list_resources():  # pragma: no cover
-        return [
-            Resource(
-                uri="memory://test",
-                name="Test Resource",
-                description="A test resource",
-            )
-        ]
 
-    return server
+    async def handle_list_resources(
+        ctx: ServerRequestContext, params: types.PaginatedRequestParams | None
+    ) -> ListResourcesResult:  # pragma: no cover
+        return ListResourcesResult(
+            resources=[
+                Resource(
+                    uri="memory://test",
+                    name="Test Resource",
+                    description="A test resource",
+                )
+            ]
+        )
+
+    return Server(name="test_server", on_list_resources=handle_list_resources)
 
 
 @pytest.fixture