backport: phase-4 waves 3+4 + S2-S9 (Workflow, 43 agents) — ~174 pass / 3 deferred

Parallel lane (14 N-risk files) + sequential lane (8 Y-risk files), all batches ≤5 tests.
Deferred: roots:list-changed (no v1 lowlevel decorator), resources:templates:pagination
(nullary-only), client-auth:authorize:offline-access-consent (v1 lacks SEP-2207).
Several expected gaps did not materialize (InMemoryTransport, REQUEST_TIMEOUT, FastMCP ctor
kwargs, verify_tokens=False) — all solvable with helpers or snapshot regen.

Author: maxisbey

tests/interaction/_requirements.py

@@ -794,6 +794,12 @@ def __post_init__(self) -> None:
             "A server with resource handlers advertises the resources capability, including the subscribe "
             "sub-flag when a subscribe handler is registered."
         ),
+        divergence=Divergence(
+            note=(
+                "The low-level Server hard-codes subscribe=False in get_capabilities() regardless of "
+                "whether a subscribe_resource handler is registered."
+            ),
+        ),
     ),
     "resources:list-changed": Requirement(
         source=f"{SPEC_BASE_URL}/server/resources#list-changed-notification",
@@ -857,6 +863,11 @@ def __post_init__(self) -> None:
     "resources:templates:pagination": Requirement(
         source=f"{SPEC_BASE_URL}/server/utilities/pagination#operations-supporting-pagination",
         behavior="resources/templates/list supports cursor pagination.",
+        deferred=(
+            "Not expressible via the v1 public API: Server.list_resource_templates() only accepts a nullary "
+            "() -> list[ResourceTemplate] handler with no dual-signature dispatch, so the inbound cursor is "
+            "unreadable and the handler cannot return a nextCursor."
+        ),
     ),
     "resources:unsubscribe": Requirement(
         source=f"{SPEC_BASE_URL}/server/resources#subscriptions",
@@ -1439,6 +1450,10 @@ def __post_init__(self) -> None:
     "roots:list-changed": Requirement(
         source=f"{SPEC_BASE_URL}/client/roots#root-list-changes",
         behavior="A roots/list_changed notification sent by the client is delivered to the server's handler.",
+        deferred=(
+            "Not expressible via the v1 public API: the low-level Server exposes no decorator for "
+            "notifications/roots/list_changed, so a server handler cannot be registered to observe delivery."
+        ),
     ),
     "roots:list-changed:client-emits": Requirement(
         source=f"{SPEC_BASE_URL}/client/roots#root-list-changes",
@@ -2483,6 +2498,10 @@ def __post_init__(self) -> None:
             "and prompt=consent is added to the authorize request."
         ),
         transports=("streamable-http",),
+        deferred=(
+            "Not expressible via the v1 public API: v1's OAuthClientProvider has no SEP-2207 "
+            "offline_access auto-append or prompt=consent logic."
+        ),
     ),
     "client-auth:bearer-header:every-request": Requirement(
         source=f"{SPEC_BASE_URL}/basic/authorization#token-requirements",

tests/interaction/auth/test_as_handlers.py

@@ -234,7 +234,10 @@ async def test_registration_with_invalid_metadata_is_rejected_with_400(
     no_auth_code = await http.post("/register", json=body | {"grant_types": ["refresh_token"]})
     assert no_auth_code.status_code == 400
     assert no_auth_code.json() == snapshot(
-        {"error": "invalid_client_metadata", "error_description": "grant_types must include 'authorization_code'"}
+        {
+            "error": "invalid_client_metadata",
+            "error_description": "grant_types must be authorization_code and refresh_token",
+        }
     )
 
     bad_scope = await http.post("/register", json=body | {"scope": "forbidden"})

tests/interaction/auth/test_authorize_token.py

@@ -1,8 +1,9 @@
 """Authorization-request, token-request, and PKCE wire-level invariants of the SDK's OAuth client.
 
-Every test connects a real `Client` end to end via `connect_with_oauth`; the assertions are on
-the parsed authorize URL and the recorded `/token` form body, because those wire shapes are what
-the spec mandates and `Client` cannot observe them. The recording uses `record_requests`, which
+Every test connects a real `ClientSession` end to end via `connect_with_oauth`; the assertions
+are on the parsed authorize URL and the recorded `/token` form body, because those wire shapes
+are what the spec mandates and the session cannot observe them. The recording uses
+`record_requests`, which
 snapshots each request at send time so the auth flow's in-place header mutation on retry never
 affects what was captured for the first attempt.
 
@@ -24,11 +25,10 @@
 from inline_snapshot import snapshot
 from pydantic import AnyHttpUrl, AnyUrl
 
-from mcp import types
 from mcp.client.auth import OAuthFlowError
-from mcp.server import Server, ServerRequestContext
+from mcp.server import Server
 from mcp.shared.auth import OAuthClientInformationFull, OAuthMetadata
-from mcp.types import ListToolsResult, Tool
+from mcp.types import Tool
 from tests.interaction._connect import BASE_URL
 from tests.interaction._requirements import requirement
 from tests.interaction.auth._harness import (
@@ -50,8 +50,15 @@
 ASM_PATH = "/.well-known/oauth-authorization-server"
 
 
-async def list_tools(ctx: ServerRequestContext, params: types.PaginatedRequestParams | None) -> ListToolsResult:
-    return ListToolsResult(tools=[Tool(name="echo", inputSchema={"type": "object"})])
+def make_guarded_server() -> Server:
+    """Build a lowlevel `Server` exposing one `echo` tool, to mount behind the OAuth-gated MCP endpoint."""
+    server = Server("guarded")
+
+    @server.list_tools()
+    async def _list_tools() -> list[Tool]:
+        return [Tool(name="echo", inputSchema={"type": "object"})]
+
+    return server
 
 
 def authorize_params(authorize_url: str) -> dict[str, str]:
@@ -91,13 +98,12 @@ def token_request(self) -> RecordedRequest:
 async def recorded_oauth_flow() -> AsyncIterator[RecordedFlow]:
     """Run one full OAuth connect with default configuration and yield its recorded wire traffic.
 
-    `valid_scopes` includes `offline_access` so the AS metadata advertises it and the SDK's
-    SEP-2207 auto-append (and the resulting `prompt=consent`) is exercised; `required_scopes`
+    `valid_scopes` includes `offline_access` so the AS metadata advertises it; `required_scopes`
     stays at `["mcp"]` so the issued token still passes the bearer middleware.
     """
     recorded, on_request = record_requests()
     provider = InMemoryAuthorizationServerProvider()
-    server = Server("guarded", on_list_tools=list_tools)
+    server = make_guarded_server()
     settings = auth_settings(required_scopes=["mcp"], valid_scopes=["mcp", "offline_access"])
 
     with anyio.fail_after(5):
@@ -113,7 +119,6 @@ async def recorded_oauth_flow() -> AsyncIterator[RecordedFlow]:
 
 @requirement("client-auth:pkce:s256")
 @requirement("client-auth:resource-parameter")
-@requirement("client-auth:authorize:offline-access-consent")
 async def test_the_authorize_url_carries_s256_pkce_and_the_resource_indicator(
     recorded_oauth_flow: RecordedFlow,
 ) -> None:
@@ -130,7 +135,6 @@ async def test_the_authorize_url_carries_s256_pkce_and_the_resource_indicator(
             "client_id",
             "code_challenge",
             "code_challenge_method",
-            "prompt",
             "redirect_uri",
             "resource",
             "response_type",
@@ -145,9 +149,7 @@ async def test_the_authorize_url_carries_s256_pkce_and_the_resource_indicator(
     # the stable prefix so the test does not lock in a trailing-slash decision.
     assert params["resource"].startswith(BASE_URL)
     assert params["state"] != ""
-
-    assert params["scope"].split(" ") == snapshot(["mcp", "offline_access"])
-    assert params["prompt"] == "consent"
+    assert params["scope"].split(" ") == snapshot(["mcp"])
 
 
 @requirement("client-auth:pkce:s256")
@@ -175,15 +177,15 @@ async def test_a_mismatched_state_on_the_callback_aborts_the_flow() -> None:
     random tokens).
     """
     provider = InMemoryAuthorizationServerProvider()
-    server = Server("guarded", on_list_tools=list_tools)
+    server = make_guarded_server()
     headless = HeadlessOAuth(state_override="wrong-state")
 
     with anyio.fail_after(5):
         with pytest.RaisesGroup(
             pytest.RaisesExc(OAuthFlowError, match="^State parameter mismatch:"), flatten_subgroups=True
         ):
-            # Entering the connect raises during the OAuth handshake (inside `Client.__aenter__`),
-            # so an `async with` body would be unreachable; entering explicitly avoids dead code.
+            # Entering the connect raises during the OAuth handshake (before `ClientSession.initialize`
+            # returns), so an `async with` body would be unreachable; entering explicitly avoids dead code.
             await connect_with_oauth(server, provider=provider, headless=headless).__aenter__()
 
 
@@ -238,7 +240,7 @@ async def test_a_client_with_a_secret_authenticates_the_token_request_with_http_
     """
     recorded, on_request = record_requests()
     provider = InMemoryAuthorizationServerProvider()
-    server = Server("guarded", on_list_tools=list_tools)
+    server = make_guarded_server()
 
     client_info = OAuthClientInformationFull(
         client_id="cid",
@@ -276,7 +278,7 @@ async def test_the_registered_auth_method_is_used_regardless_of_as_metadata_adve
     """
     recorded, on_request = record_requests()
     provider = InMemoryAuthorizationServerProvider()
-    server = Server("guarded", on_list_tools=list_tools)
+    server = make_guarded_server()
 
     override = OAuthMetadata(
         issuer=AnyHttpUrl(f"{BASE_URL}/"),
@@ -319,7 +321,7 @@ async def test_scope_is_selected_from_the_www_authenticate_challenge_over_prm_me
     """
     recorded, on_request = record_requests()
     provider = InMemoryAuthorizationServerProvider(default_scopes=["from-header"])
-    server = Server("guarded", on_list_tools=list_tools)
+    server = make_guarded_server()
     settings = auth_settings(required_scopes=["from-prm"], valid_scopes=["from-header", "from-prm"])
     challenge = f'Bearer scope="from-header", resource_metadata="{BASE_URL}{PRM_PATH}"'
 
@@ -360,7 +362,7 @@ async def test_pkce_is_still_sent_when_as_metadata_omits_code_challenge_methods_
     serve = {ASM_PATH: override.model_dump_json(exclude_none=True).encode()}
 
     provider = InMemoryAuthorizationServerProvider()
-    server = Server("guarded", on_list_tools=list_tools)
+    server = make_guarded_server()
 
     with anyio.fail_after(5):
         async with connect_with_oauth(
@@ -386,7 +388,7 @@ async def test_an_authorize_error_on_the_callback_aborts_the_flow_before_the_tok
     """
     recorded, on_request = record_requests()
     provider = InMemoryAuthorizationServerProvider(deny_authorize=True)
-    server = Server("guarded", on_list_tools=list_tools)
+    server = make_guarded_server()
     headless = HeadlessOAuth()
 
     with anyio.fail_after(5):

tests/interaction/auth/test_discovery.py

@@ -1,13 +1,13 @@
 """Protected-resource and authorization-server metadata discovery, end to end.
 
-Every client-side test connects a real `Client` via `connect_with_oauth` and asserts on the
-recorded request paths the discovery probes produced; the discovery URL ordering is a wire
-detail `Client` cannot observe directly but the recording can. Tests that need a metadata
+Every client-side test connects a real `ClientSession` via `connect_with_oauth` and asserts on
+the recorded request paths the discovery probes produced; the discovery URL ordering is a wire
+detail the session cannot observe directly but the recording can. Tests that need a metadata
 endpoint to 404 or return alternate content wrap the SDK's app in `shimmed_app` while leaving
 the real authorize and token endpoints behind it, so the rest of the flow runs unaltered.
 
-The two server-side tests (#5, #6) drive raw httpx against `mounted_app` because their
-assertions are the metadata response bodies and headers, which `Client` does not surface.
+The two server-side tests drive raw httpx against `mounted_app` because their assertions are
+the metadata response bodies and headers, which `ClientSession` does not surface.
 """
 
 import json
@@ -17,11 +17,10 @@
 from inline_snapshot import snapshot
 from pydantic import AnyHttpUrl
 
-from mcp import types
 from mcp.client.auth import OAuthFlowError, OAuthRegistrationError
-from mcp.server import Server, ServerRequestContext
+from mcp.server import Server
 from mcp.shared.auth import OAuthMetadata, ProtectedResourceMetadata
-from mcp.types import ListToolsResult, Tool
+from mcp.types import Tool
 from tests.interaction._connect import BASE_URL, mounted_app
 from tests.interaction._requirements import requirement
 from tests.interaction.auth._harness import (
@@ -42,8 +41,15 @@
 OIDC_ROOT = "/.well-known/openid-configuration"
 
 
-async def list_tools(ctx: ServerRequestContext, params: types.PaginatedRequestParams | None) -> ListToolsResult:
-    return ListToolsResult(tools=[Tool(name="probe", inputSchema={"type": "object"})])
+def guarded_server() -> Server:
+    """Build a lowlevel `Server` exposing a single `probe` tool via `list_tools`."""
+    server = Server("guarded")
+
+    @server.list_tools()
+    async def _list_tools() -> list[Tool]:
+        return [Tool(name="probe", inputSchema={"type": "object"})]
+
+    return server
 
 
 def discovery_gets(recorded: list[RecordedRequest]) -> list[str]:
@@ -74,7 +80,7 @@ async def test_prm_discovery_uses_the_resource_metadata_url_from_www_authenticat
     """
     recorded, on_request = record_requests()
     provider = InMemoryAuthorizationServerProvider()
-    server = Server("guarded", on_list_tools=list_tools)
+    server = guarded_server()
 
     with anyio.fail_after(5):
         async with connect_with_oauth(server, provider=provider, on_request=on_request) as (client, _):
@@ -97,7 +103,7 @@ async def test_prm_discovery_falls_back_from_path_well_known_to_root_on_404() ->
     """
     recorded, on_request = record_requests()
     provider = InMemoryAuthorizationServerProvider()
-    server = Server("guarded", on_list_tools=list_tools)
+    server = guarded_server()
 
     prm = ProtectedResourceMetadata(
         resource=AnyHttpUrl(f"{BASE_URL}/mcp"), authorization_servers=[AnyHttpUrl(BASE_URL)]
@@ -132,7 +138,7 @@ async def test_when_every_prm_probe_fails_the_client_discovers_as_metadata_at_th
     """
     recorded, on_request = record_requests()
     provider = InMemoryAuthorizationServerProvider()
-    server = Server("guarded", on_list_tools=list_tools)
+    server = guarded_server()
     app_shim = shim(not_found=frozenset({PRM_PATH_SUFFIXED, PRM_ROOT}))
 
     with anyio.fail_after(5):
@@ -160,7 +166,7 @@ async def test_a_400_from_the_registration_endpoint_surfaces_as_a_registration_e
     """
     recorded, on_request = record_requests()
     provider = InMemoryAuthorizationServerProvider()
-    server = Server("guarded", on_list_tools=list_tools)
+    server = guarded_server()
     error_body = json.dumps({"error": "invalid_client_metadata", "error_description": "no"}).encode()
     app_shim = shim(serve={"/register": (400, error_body)})
 
@@ -185,7 +191,7 @@ async def test_prm_with_a_mismatched_resource_aborts_the_flow_before_authorize()
     """
     recorded, on_request = record_requests()
     provider = InMemoryAuthorizationServerProvider()
-    server = Server("guarded", on_list_tools=list_tools)
+    server = guarded_server()
 
     prm = ProtectedResourceMetadata(
         resource=AnyHttpUrl(f"{BASE_URL}/other"), authorization_servers=[AnyHttpUrl(BASE_URL)]
@@ -237,7 +243,7 @@ async def test_as_metadata_discovery_falls_back_through_the_spec_endpoint_order(
     """
     recorded, on_request = record_requests()
     provider = InMemoryAuthorizationServerProvider()
-    server = Server("guarded", on_list_tools=list_tools)
+    server = guarded_server()
 
     prm = ProtectedResourceMetadata(
         resource=AnyHttpUrl(f"{BASE_URL}/mcp"), authorization_servers=[AnyHttpUrl(authorization_server)]
@@ -320,7 +326,7 @@ async def test_as_metadata_with_a_mismatched_issuer_is_accepted_and_the_flow_pro
     unknown-field tolerance.
     """
     provider = InMemoryAuthorizationServerProvider()
-    server = Server("guarded", on_list_tools=list_tools)
+    server = guarded_server()
 
     metadata = real_asm()
     metadata.issuer = AnyHttpUrl(f"{BASE_URL}/wrong-issuer")