backport: phase-4 wave 1 (proof) — 25/25 pass across lowlevel/mcpserver/transports/auth

- transports/test_bridge (4, no edits), test_stdio (2 + _stdio_server rewrite)
- lowlevel/test_completion (5), test_logging (3), test_tools b1/3 (5)
- mcpserver/test_completion (1)
- auth/test_flow (5)
- _requirements.py: tools:call:unknown-name + protocol:error:internal-error divergences updated for v1

Author: maxisbey

tests/interaction/_requirements.py

@@ -349,8 +349,11 @@ def __post_init__(self) -> None:
         ),
         divergence=Divergence(
             note=(
-                "The low-level Server returns code 0 (not a defined JSON-RPC code) instead of -32603 and "
-                "leaks str(exc) as the error message."
+                "For tools/call the lowlevel @server.call_tool() decorator wraps the handler in a broad "
+                "try/except that converts every Exception to CallToolResult(isError=True, "
+                "content=[TextContent(text=str(exc))]), so the dispatcher's JSON-RPC error path is never "
+                "reached for tool calls and the test pins the isError=True result. For other request "
+                "handlers the dispatcher returns code 0 (not -32603) with str(exc) as the message."
             ),
         ),
     ),
@@ -559,6 +562,14 @@ def __post_init__(self) -> None:
     "tools:call:unknown-name": Requirement(
         source=f"{SPEC_BASE_URL}/server/tools#error-handling",
         behavior="tools/call for a name the server does not recognise returns a JSON-RPC error.",
+        divergence=Divergence(
+            note=(
+                "The lowlevel @server.call_tool() decorator catches every handler exception (including "
+                "McpError) and converts it to CallToolResult(isError=True, content=[TextContent(text=str(exc))]), "
+                "so a handler cannot produce a protocol-level JSON-RPC error for tools/call; the test pins "
+                "the isError=True result instead."
+            ),
+        ),
     ),
     "tools:capability:declared": Requirement(
         source=f"{SPEC_BASE_URL}/server/tools#capabilities",

tests/interaction/auth/test_flow.py

@@ -23,7 +23,7 @@
 from mcp.server.auth.middleware.auth_context import get_access_token
 from mcp.shared.auth import OAuthClientInformationFull
 from mcp.types import CallToolResult, ListToolsResult, TextContent, Tool
-from tests.interaction._connect import BASE_URL
+from tests.interaction._connect import BASE_URL, build_streamable_http_app
 from tests.interaction._requirements import requirement
 from tests.interaction.auth._harness import (
     REDIRECT_URI,
@@ -229,11 +229,11 @@ async def test_shimmed_app_serves_overrides_404s_and_otherwise_forwards_to_the_w
     own routing; provided here so the discovery tests can rely on the shim without each adding
     their own contract test.
     """
-    server = Server("bare")
+    server: Server[object] = Server("bare")
     provider = InMemoryAuthorizationServerProvider()
-    real_app = server.streamable_http_app(auth=auth_settings(), auth_server_provider=provider)
+    real_app, manager = build_streamable_http_app(server, auth=auth_settings(), auth_server_provider=provider)
     app = shimmed_app(real_app, not_found=frozenset({"/missing"}), serve={"/override": b'{"shimmed": true}'})
-    async with server.session_manager.run():
+    async with manager.run():
         async with httpx.AsyncClient(transport=StreamingASGITransport(app), base_url=BASE_URL) as http:
             served = await http.get("/override")
             assert served.status_code == 200

tests/interaction/lowlevel/test_completion.py

@@ -1,15 +1,17 @@
-"""Completion interactions against the low-level Server, driven through the public Client API."""
+"""Completion interactions against the low-level Server, driven through the public client API."""
 
 import pytest
 from inline_snapshot import snapshot
 
-from mcp import MCPError, types
-from mcp.server import Server, ServerRequestContext
+from mcp import McpError
+from mcp.server.lowlevel import Server
 from mcp.types import (
     INVALID_PARAMS,
     METHOD_NOT_FOUND,
     CompleteResult,
     Completion,
+    CompletionArgument,
+    CompletionContext,
     ErrorData,
     PromptReference,
     ResourceTemplateReference,
@@ -27,16 +29,20 @@ async def test_complete_prompt_argument(connect: Connect) -> None:
 
     The returned values are filtered by the argument's value, proving the value reached the handler.
     """
-
-    async def completion(ctx: ServerRequestContext, params: types.CompleteRequestParams) -> CompleteResult:
-        assert isinstance(params.ref, PromptReference)
-        assert params.ref.name == "code_review"
-        assert params.argument.name == "language"
+    server = Server("completer")
+
+    @server.completion()
+    async def completion(
+        ref: PromptReference | ResourceTemplateReference,
+        argument: CompletionArgument,
+        context: CompletionContext | None,
+    ) -> Completion | None:
+        assert isinstance(ref, PromptReference)
+        assert ref.name == "code_review"
+        assert argument.name == "language"
         candidates = ["python", "pytorch", "ruby"]
-        matches = [candidate for candidate in candidates if candidate.startswith(params.argument.value)]
-        return CompleteResult(completion=Completion(values=matches, total=len(matches), hasMore=False))
-
-    server = Server("completer", on_completion=completion)
+        matches = [candidate for candidate in candidates if candidate.startswith(argument.value)]
+        return Completion(values=matches, total=len(matches), hasMore=False)
 
     async with connect(server) as client:
         result = await client.complete(
@@ -51,14 +57,18 @@ async def completion(ctx: ServerRequestContext, params: types.CompleteRequestPar
 @requirement("completion:resource-template-arg")
 async def test_complete_resource_template_variable(connect: Connect) -> None:
     """Completing a URI template variable delivers the template URI and variable name to the handler."""
-
-    async def completion(ctx: ServerRequestContext, params: types.CompleteRequestParams) -> CompleteResult:
-        assert isinstance(params.ref, ResourceTemplateReference)
-        assert params.ref.uri == "github://repos/{owner}/{repo}"
-        assert params.argument.name == "owner"
-        return CompleteResult(completion=Completion(values=[f"{params.argument.value}contextprotocol"]))
-
-    server = Server("completer", on_completion=completion)
+    server = Server("completer")
+
+    @server.completion()
+    async def completion(
+        ref: PromptReference | ResourceTemplateReference,
+        argument: CompletionArgument,
+        context: CompletionContext | None,
+    ) -> Completion | None:
+        assert isinstance(ref, ResourceTemplateReference)
+        assert ref.uri == "github://repos/{owner}/{repo}"
+        assert argument.name == "owner"
+        return Completion(values=[f"{argument.value}contextprotocol"])
 
     async with connect(server) as client:
         result = await client.complete(
@@ -75,14 +85,18 @@ async def test_complete_receives_context_arguments(connect: Connect) -> None:
 
     The returned value is derived from the context, proving it arrived.
     """
-
-    async def completion(ctx: ServerRequestContext, params: types.CompleteRequestParams) -> CompleteResult:
-        assert params.argument.name == "repo"
-        assert params.context is not None
-        assert params.context.arguments is not None
-        return CompleteResult(completion=Completion(values=[f"{params.context.arguments['owner']}/python-sdk"]))
-
-    server = Server("completer", on_completion=completion)
+    server = Server("completer")
+
+    @server.completion()
+    async def completion(
+        ref: PromptReference | ResourceTemplateReference,
+        argument: CompletionArgument,
+        context: CompletionContext | None,
+    ) -> Completion | None:
+        assert argument.name == "repo"
+        assert context is not None
+        assert context.arguments is not None
+        return Completion(values=[f"{context.arguments['owner']}/python-sdk"])
 
     async with connect(server) as client:
         result = await client.complete(
@@ -102,15 +116,19 @@ async def test_completion_against_an_unknown_ref_is_rejected_with_invalid_params
     against); rejecting an unknown ref is the handler's job, and this test pins the spec-recommended
     way to do it.
     """
+    server = Server("completer")
 
-    async def completion(ctx: ServerRequestContext, params: types.CompleteRequestParams) -> CompleteResult:
-        assert isinstance(params.ref, PromptReference)
-        raise MCPError(code=INVALID_PARAMS, message=f"Unknown prompt: {params.ref.name!r}")
-
-    server = Server("completer", on_completion=completion)
+    @server.completion()
+    async def completion(
+        ref: PromptReference | ResourceTemplateReference,
+        argument: CompletionArgument,
+        context: CompletionContext | None,
+    ) -> Completion | None:
+        assert isinstance(ref, PromptReference)
+        raise McpError(ErrorData(code=INVALID_PARAMS, message=f"Unknown prompt: {ref.name!r}"))
 
     async with connect(server) as client:
-        with pytest.raises(MCPError) as exc_info:
+        with pytest.raises(McpError) as exc_info:
             await client.complete(PromptReference(type="ref/prompt", name="ghost"), argument={"name": "x", "value": ""})
 
     assert exc_info.value.error.code == INVALID_PARAMS
@@ -123,9 +141,11 @@ async def test_complete_without_handler_is_method_not_found(connect: Connect) ->
     server = Server("incomplete")
 
     async with connect(server) as client:
-        assert client.initialize_result.capabilities.completions is None
+        capabilities = client.get_server_capabilities()
+        assert capabilities is not None
+        assert capabilities.completions is None
 
-        with pytest.raises(MCPError) as exc_info:
+        with pytest.raises(McpError) as exc_info:
             await client.complete(
                 PromptReference(type="ref/prompt", name="anything"), argument={"name": "topic", "value": ""}
             )

tests/interaction/lowlevel/test_logging.py

@@ -9,11 +9,13 @@
 assert after the request completes on every transport leg -- no events, no waiting.
 """
 
+from typing import Any
+
 import pytest
 from inline_snapshot import snapshot
 
 from mcp import types
-from mcp.server import Server, ServerRequestContext
+from mcp.server import Server
 from mcp.types import CallToolResult, EmptyResult, LoggingMessageNotificationParams, TextContent
 from tests.interaction._connect import Connect
 from tests.interaction._requirements import requirement
@@ -35,12 +37,11 @@
 @requirement("logging:set-level")
 async def test_set_logging_level_reaches_handler(connect: Connect) -> None:
     """The level requested by the client is delivered to the server's handler verbatim."""
+    server = Server("logger")
 
-    async def set_logging_level(ctx: ServerRequestContext, params: types.SetLevelRequestParams) -> EmptyResult:
-        assert params.level == "warning"
-        return EmptyResult()
-
-    server = Server("logger", on_set_logging_level=set_logging_level)
+    @server.set_logging_level()
+    async def set_logging_level(level: types.LoggingLevel) -> None:
+        assert level == "warning"
 
     async with connect(server) as client:
         result = await client.set_logging_level("warning")
@@ -61,27 +62,29 @@ async def test_log_messages_reach_logging_callback_in_order(connect: Connect) ->
     async def collect(params: LoggingMessageNotificationParams) -> None:
         received.append(params)
 
-    async def list_tools(
-        ctx: ServerRequestContext, params: types.PaginatedRequestParams | None
-    ) -> types.ListToolsResult:
-        return types.ListToolsResult(tools=[types.Tool(name="chatty", inputSchema={"type": "object"})])
+    server = Server("logger")
+
+    @server.list_tools()
+    async def list_tools() -> list[types.Tool]:
+        return [types.Tool(name="chatty", inputSchema={"type": "object"})]
 
-    async def call_tool(ctx: ServerRequestContext, params: types.CallToolRequestParams) -> CallToolResult:
-        assert params.name == "chatty"
+    @server.call_tool()
+    async def call_tool(name: str, arguments: dict[str, Any]) -> list[types.ContentBlock]:
+        assert name == "chatty"
+        ctx = server.request_context
         await ctx.session.send_log_message(
             level="info", data="starting up", logger="app.lifecycle", related_request_id=ctx.request_id
         )
         await ctx.session.send_log_message(
             level="error", data={"code": 502, "retryable": True}, related_request_id=ctx.request_id
         )
-        return CallToolResult(content=[TextContent(type="text", text="done")])
+        return [TextContent(type="text", text="done")]
 
-    async def set_logging_level(ctx: ServerRequestContext, params: types.SetLevelRequestParams) -> EmptyResult:
+    @server.set_logging_level()
+    async def set_logging_level(level: types.LoggingLevel) -> None:
         """Registered so the logging capability is advertised; the client never sets a level."""
         raise NotImplementedError
 
-    server = Server("logger", on_list_tools=list_tools, on_call_tool=call_tool, on_set_logging_level=set_logging_level)
-
     async with connect(server, logging_callback=collect) as client:
         result = await client.call_tool("chatty", {})
 
@@ -102,25 +105,27 @@ async def test_log_messages_at_every_severity_level(connect: Connect) -> None:
     async def collect(params: LoggingMessageNotificationParams) -> None:
         received.append(params)
 
-    async def list_tools(
-        ctx: ServerRequestContext, params: types.PaginatedRequestParams | None
-    ) -> types.ListToolsResult:
-        return types.ListToolsResult(tools=[types.Tool(name="siren", inputSchema={"type": "object"})])
+    server = Server("logger")
+
+    @server.list_tools()
+    async def list_tools() -> list[types.Tool]:
+        return [types.Tool(name="siren", inputSchema={"type": "object"})]
 
-    async def call_tool(ctx: ServerRequestContext, params: types.CallToolRequestParams) -> CallToolResult:
-        assert params.name == "siren"
+    @server.call_tool()
+    async def call_tool(name: str, arguments: dict[str, Any]) -> list[types.ContentBlock]:
+        assert name == "siren"
+        ctx = server.request_context
         for level in ALL_LEVELS:
             await ctx.session.send_log_message(
                 level=level, data=f"a {level} message", related_request_id=ctx.request_id
             )
-        return CallToolResult(content=[TextContent(type="text", text="logged")])
+        return [TextContent(type="text", text="logged")]
 
-    async def set_logging_level(ctx: ServerRequestContext, params: types.SetLevelRequestParams) -> EmptyResult:
+    @server.set_logging_level()
+    async def set_logging_level(level: types.LoggingLevel) -> None:
         """Registered so the logging capability is advertised; the client never sets a level."""
         raise NotImplementedError
 
-    server = Server("logger", on_list_tools=list_tools, on_call_tool=call_tool, on_set_logging_level=set_logging_level)
-
     async with connect(server, logging_callback=collect) as client:
         await client.call_tool("siren", {})