Extend the interaction suite to the 2026-07-28 spec

tests/interaction/README.md

@@ -61,6 +61,12 @@ stateless configurations), and over the legacy SSE transport the same way. A tes
 `async with connect(server, ...) as client:` and asserts the same output on every leg, because the
 transport is not supposed to change observable behaviour. Requirements that need a server-to-client
 back-channel or persisted session state are carved out of the stateless arm via `arm_exclusions`.
+
+The 2026 cells run the client's response cache in its default-on configuration. Servers stamp
+`ttlMs: 0` by default, so nothing is served from cache unless a test opts in server-side by
+authoring a positive `ttl_ms` — a test that does so and then repeats a call must expect the
+repeat to be served from cache instead of reaching the handler.
+
 Tests that are tied to one transport do not use the fixture: the wire-recording tests
 (their seam is the in-memory stream pair), the bare-`ClientSession` lifecycle tests, the
 real-clock timeout tests (the timeout machinery is transport-independent and must not race
@@ -163,10 +169,11 @@ What admits or excludes a cell:
   closes, grep for the reason string to find every cell to re-admit.
 - **`known_failures`** keep a cell in the grid but mark it as a strict xfail — the test runs and
   must fail; an unexpected pass fails the suite.
-- **`TRANSPORT_SPEC_VERSIONS`** era-locks a transport to a subset of spec versions (currently only
-  `sse` is locked to `2025-11-25`). A `(transport, version)` cell is dropped if the version is not
-  in the transport's entry; transports absent from the map serve every spec version. This is the
-  mechanism for cutting an entire transport off from a new revision (or admitting it).
+- **`TRANSPORT_SPEC_VERSIONS`** era-locks a transport to a subset of spec versions (currently
+  `sse` and `streamable-http-stateless` are locked to `2025-11-25`). A `(transport, version)`
+  cell is dropped if the version is not in the transport's entry; transports absent from the
+  map serve every spec version. This is the mechanism for cutting an entire transport off from
+  a new revision (or admitting it).
 - **`transports`** is descriptive metadata for the non-`connect` transport-specific suites under
   `transports/` and does **not** drive cell generation. Only `arm_exclusions`, `added_in`,
   `removed_in`, and `TRANSPORT_SPEC_VERSIONS` filter the grid.

tests/interaction/_connect.py

@@ -388,7 +388,9 @@ def httpx_client_factory(
     transport = sse_client(f"{BASE_URL}/sse", httpx_client_factory=httpx_client_factory)
     async with Client(
         transport,
-        # SSE is a legacy-only transport; the modern path has no SSE story.
+        # A policy lock, not a capability one: the dual-era server loop behind build_sse_app
+        # would negotiate 2026 if probed, but SSE is the deprecated legacy transport and its
+        # clients run the handshake era by design.
         mode="legacy",
         read_timeout_seconds=read_timeout_seconds,
         sampling_callback=sampling_callback,

tests/interaction/auth/_harness.py

@@ -16,9 +16,10 @@
 from typing import Any
 from urllib.parse import parse_qs, parse_qsl, urlsplit
 
+import anyio
 import httpx
 from pydantic import AnyHttpUrl, AnyUrl, BaseModel
-from starlette.types import ASGIApp, Receive, Scope, Send
+from starlette.types import ASGIApp, Message, Receive, Scope, Send
 
 from mcp.client.auth import OAuthClientProvider
 from mcp.client.client import Client
@@ -132,25 +133,38 @@ class HeadlessOAuth:
     `redirect_handler` GETs the authorize URL on the bound client (with `auth=None` so the
     request does not re-enter the locked auth flow), parses `code` and `state` from the 302
     `Location`, and stashes them; `callback_handler` returns the stashed pair. Tests inspect
-    `authorize_url` to assert what the SDK put on the authorize request.
+    `authorize_url` to assert what the SDK put on the authorize request, and `iss`/`error` to
+    assert what the redirect carried.
 
     `state_override`: when set, `callback_handler` returns this value as the state instead of
     the one parsed from the redirect, so tests can drive the state-mismatch path.
 
     `iss_override`: when set, `callback_handler` returns this value as the RFC 9207 issuer
     instead of the one parsed from the redirect, so tests can drive the iss-mismatch path.
+
+    `code_override`: when set, returned as the code instead of the parsed one (token-endpoint rejection path).
+    `omit_iss`: when set, no iss is returned, overriding everything (`iss_override` cannot express absence).
     """
 
-    def __init__(self, *, state_override: str | None = None, iss_override: str | None = None) -> None:
+    def __init__(
+        self,
+        *,
+        state_override: str | None = None,
+        iss_override: str | None = None,
+        code_override: str | None = None,
+        omit_iss: bool = False,
+    ) -> None:
         self.authorize_url: str | None = None
         self.authorize_urls: list[str] = []
         self.error: str | None = None
+        self.iss: str | None = None
         self._state_override = state_override
         self._iss_override = iss_override
+        self._code_override = code_override
+        self._omit_iss = omit_iss
         self._http: httpx.AsyncClient | None = None
         self._code: str = ""
         self._state: str | None = None
-        self._iss: str | None = None
 
     def bind(self, http_client: httpx.AsyncClient) -> None:
         self._http = http_client
@@ -166,14 +180,15 @@ async def redirect_handler(self, authorization_url: str) -> None:
         params = parse_qs(urlsplit(response.headers["location"]).query)
         self._code = params.get("code", [""])[0]
         self._state = params.get("state", [None])[0]
-        self._iss = params.get("iss", [None])[0]
+        self.iss = params.get("iss", [None])[0]
         self.error = params.get("error", [None])[0]
 
     async def callback_handler(self) -> AuthorizationCodeResult:
+        iss = self._iss_override if self._iss_override is not None else self.iss
         return AuthorizationCodeResult(
-            code=self._code,
+            code=self._code_override if self._code_override is not None else self._code,
             state=self._state_override if self._state_override is not None else self._state,
-            iss=self._iss_override if self._iss_override is not None else self._iss,
+            iss=None if self._omit_iss else iss,
         )
 
 
@@ -308,7 +323,7 @@ def first_challenge_shim(www_authenticate: str, *, path: str = "/mcp") -> Callab
     return lambda app: _FirstChallenge(app, path, www_authenticate)
 
 
-def step_up_shim(www_authenticate: str, *, on_nth_authenticated_post: int = 2) -> AppShim:
+def step_up_shim(www_authenticate: str, *, on_nth_authenticated_post: int = 2, persist: bool = False) -> AppShim:
     """Build an `app_shim` that 403s the Nth authenticated POST to `/mcp` with the given challenge.
 
     Subsequent requests pass through. Used to drive the client's `insufficient_scope` step-up
@@ -320,6 +335,8 @@ def step_up_shim(www_authenticate: str, *, on_nth_authenticated_post: int = 2) -
     first authenticated POST is the auth flow's retry of the original initialize request (yielded
     after the 401 branch, where the generator ends without inspecting the response), so a 403
     there would not reach the step-up handler.
+
+    `persist`: when set, 403s every authenticated POST from the Nth onward, re-challenging the step-up retry.
     """
     seen = 0
     fired = False
@@ -328,7 +345,7 @@ def factory(app: ASGIApp) -> ASGIApp:
         async def wrapped(scope: Scope, receive: Receive, send: Send) -> None:
             nonlocal seen, fired
             if (
-                not fired
+                (persist or not fired)
                 and scope["type"] == "http"
                 and scope["path"] == "/mcp"
                 and scope["method"] == "POST"
@@ -355,6 +372,55 @@ async def wrapped(scope: Scope, receive: Receive, send: Send) -> None:
     return factory
 
 
+def get_stream_step_up_shim(www_authenticate: str) -> tuple[list[int], anyio.Event, AppShim]:
+    """Build an `app_shim` that 403s the first authenticated GET to `/mcp` with the given challenge.
+
+    Returns:
+        The statuses of every authenticated GET response (live-updated), an event set when one
+        of those responses starts with status 200 (the reopened stream), and the shim factory.
+    """
+    statuses: list[int] = []
+    reopened = anyio.Event()
+    fired = False
+
+    def factory(app: ASGIApp) -> ASGIApp:
+        async def wrapped(scope: Scope, receive: Receive, send: Send) -> None:
+            nonlocal fired
+            if not (
+                scope["type"] == "http"
+                and scope["path"] == "/mcp"
+                and scope["method"] == "GET"
+                and b"authorization" in dict(scope["headers"])
+            ):
+                await app(scope, receive, send)
+                return
+
+            async def recording_send(message: Message) -> None:
+                if message["type"] == "http.response.start":
+                    statuses.append(message["status"])
+                    if message["status"] == 200:
+                        reopened.set()
+                await send(message)
+
+            if not fired:
+                fired = True
+                await recording_send(
+                    {
+                        "type": "http.response.start",
+                        "status": 403,
+                        "headers": [(b"www-authenticate", www_authenticate.encode())],
+                    }
+                )
+                await recording_send({"type": "http.response.body", "body": b""})
+                return
+            # The reopened SSE stream stays open until the test's exit cancels it; nothing may follow this await.
+            await app(scope, receive, recording_send)
+
+        return wrapped
+
+    return statuses, reopened, factory
+
+
 def m2m_token_shim(provider: InMemoryAuthorizationServerProvider, *, scopes: list[str]) -> AppShim:
     """Build an `app_shim` that handles `grant_type=client_credentials` at `/token`.
 

tests/interaction/auth/_provider.py

@@ -49,6 +49,8 @@ class InMemoryAuthorizationServerProvider(
         `fail_next_refresh`: the next refresh-token exchange raises `invalid_grant` once.
         `reject_all_tokens`: `load_access_token` returns None for every token, so the bearer
             middleware 401s every authenticated request.
+        `rotate_refresh_tokens`: when False, the refresh response carries no `refresh_token` and
+            the presented one stays valid (an RFC 6749 §6 non-rotating server).
     """
 
     def __init__(
@@ -59,6 +61,7 @@ def __init__(
         issue_expired_first: bool = False,
         fail_next_refresh: bool = False,
         reject_all_tokens: bool = False,
+        rotate_refresh_tokens: bool = True,
         issuer: str | None = None,
     ) -> None:
         self._default_scopes = list(default_scopes) if default_scopes is not None else ["mcp"]
@@ -71,6 +74,7 @@ def __init__(
         self._issue_expired_first = issue_expired_first
         self._fail_next_refresh = fail_next_refresh
         self._reject_all_tokens = reject_all_tokens
+        self._rotate_refresh_tokens = rotate_refresh_tokens
         self._tokens_issued = 0
         self.clients: dict[str, OAuthClientInformationFull] = {}
         self.codes: dict[str, AuthorizationCode] = {}
@@ -178,11 +182,19 @@ async def load_refresh_token(self, client: OAuthClientInformationFull, refresh_t
     async def exchange_refresh_token(
         self, client: OAuthClientInformationFull, refresh_token: RefreshToken, scopes: list[str]
     ) -> OAuthToken:
-        """Mint a new access token and rotate the refresh token, consuming the old one."""
+        """Mint a new access token, and rotate the refresh token unless rotation is disabled."""
         assert client.client_id is not None
         if self._fail_next_refresh:
             self._fail_next_refresh = False
             raise TokenError(error="invalid_grant", error_description="refresh denied by harness")
+        if not self._rotate_refresh_tokens:
+            access = self.mint_access_token(client_id=client.client_id, scopes=scopes)
+            return OAuthToken(
+                access_token=access,
+                token_type="Bearer",
+                expires_in=self._next_expires_in(),
+                scope=" ".join(scopes),
+            )
         access = self.mint_access_token(client_id=client.client_id, scopes=scopes)
         new_refresh = f"refresh_{secrets.token_hex(16)}"
         self.refresh_tokens[new_refresh] = RefreshToken(token=new_refresh, client_id=client.client_id, scopes=scopes)

tests/interaction/auth/test_as_handlers.py

@@ -16,10 +16,11 @@
 import httpx
 import pytest
 from inline_snapshot import snapshot
+from pydantic import AnyUrl
 
 from mcp.server import Server
 from mcp.server.auth.provider import ProviderTokenVerifier
-from mcp.shared.auth import OAuthClientInformationFull
+from mcp.shared.auth import OAuthClientInformationFull, OAuthClientMetadata
 from tests.interaction._connect import mounted_app
 from tests.interaction._requirements import requirement
 from tests.interaction.auth._harness import REDIRECT_URI, auth_settings, oauth_client_metadata
@@ -298,3 +299,25 @@ async def test_a_non_loopback_http_redirect_uri_is_accepted_at_registration(
     info = OAuthClientInformationFull.model_validate_json(response.content)
     assert [str(u) for u in (info.redirect_uris or [])] == ["http://evil.example/callback"]
     assert info.client_id in provider.clients
+
+
+@requirement("hosting:auth:as:register-echo-application-type")
+async def test_register_echoes_native_for_a_client_that_registered_application_type_web(
+    as_app: tuple[httpx.AsyncClient, InMemoryAuthorizationServerProvider],
+) -> None:
+    """A client registering `application_type: "web"` is told `"native"` in the registration echo.
+
+    When the passthrough fix lands: re-pin the echo to `"web"` and delete the Divergence.
+    """
+    http, _ = as_app
+    metadata = OAuthClientMetadata(
+        client_name="interaction-suite", redirect_uris=[AnyUrl(REDIRECT_URI)], application_type="web"
+    )
+
+    response = await http.post("/register", content=metadata.model_dump_json())
+
+    assert response.status_code == 201
+    body = response.json()
+    assert body["application_type"] == "native"
+    # The omission is specific to application_type, not a generally lossy echo.
+    assert body["client_name"] == "interaction-suite"