Emit RFC 6750 scope= in WWW-Authenticate and validate token audience

BearerAuthBackend / RequireAuthMiddleware now produce spec-conformant
challenges and reject tokens issued for a different resource server.

- A request with no credentials gets a bare `Bearer` challenge (with
  scope/resource_metadata only), not error="invalid_token" -- RFC 6750
  Section 3.1 says the error attribute SHOULD NOT appear when no
  authentication information was presented.
- A malformed/unknown token, an expired token, or a token whose audience
  does not match the configured resource_server_url is answered 401
  invalid_token with a specific error_description, carried via a new
  InvalidTokenUser marker so the middleware can distinguish it from
  no-credentials.
- All challenges (401 and the 403 insufficient_scope path) now advertise
  the required scopes in a `scope=` parameter, which the SDK client
  already reads to drive step-up.
- New check_token_audience() helper canonicalises default ports before
  comparing, and is wired through both the lowlevel and MCPServer
  Starlette stacks via the auth settings' resource_server_url.

Docs and migration guide updated; the corresponding interaction-suite
divergence entries are now closed.

docs/advanced/authorization.md

@@ -27,7 +27,7 @@ The SDK has no opinion about what a valid token looks like. You tell it, by impl
 `AuthSettings` is the public face of your resource server:
 
 * `issuer_url`: the authorization server that issues your tokens.
-* `resource_server_url`: the public URL of this MCP endpoint. It names *which* resource a token is for, and it's where the discovery document lives.
+* `resource_server_url`: the public URL of this MCP endpoint. It names *which* resource a token is for, and it's where the discovery document lives. When your verifier returns an `AccessToken.resource`, the SDK rejects the token unless it matches this URL — a token issued for a different resource never reaches a tool.
 * `required_scopes`: every token must carry all of them.
 
 !!! tip
@@ -61,14 +61,13 @@ You registered one tool. The second route is the SDK's.
 This document is how a client that has never heard of your server finds its way in: it reads `authorization_servers` and goes there for a token. You wrote none of it.
 
 !!! check
-    Call `/mcp` with no token (or with one your verifier returned `None` for) and the request is
-    stopped at the door:
+    Call `/mcp` with no token and the request is stopped at the door:
 
     ```text
     HTTP/1.1 401 Unauthorized
-    WWW-Authenticate: Bearer error="invalid_token", error_description="Authentication required", resource_metadata="http://127.0.0.1:8000/.well-known/oauth-protected-resource/mcp"
+    WWW-Authenticate: Bearer scope="notes:read", resource_metadata="http://127.0.0.1:8000/.well-known/oauth-protected-resource/mcp"
 
-    {"error": "invalid_token", "error_description": "Authentication required"}
+    {}
     ```
 
     Nothing was parsed and no tool ran. And that `resource_metadata` pointer in `WWW-Authenticate` is

docs/migration.md

@@ -1462,6 +1462,10 @@ issuer inconsistent with what clients compare against under RFC 8414 / RFC 9207.
 already-built `AnyHttpUrl` object still normalizes at construction; pass a string to get the
 preserved form.
 
+### Bearer tokens with a mismatched audience are rejected
+
+`BearerAuthBackend` now compares `AccessToken.resource` against `AuthSettings.resource_server_url` and answers a token whose RFC 8707 resource indicator does not name this server with `401 invalid_token`. The check is canonical-URI equality, so a token issued for `https://host/` is not accepted by a server at `https://host/mcp`. It is skipped when either side is `None` — populate `AccessToken.resource` only when your verifier surfaces the underlying audience claim. `BearerAuthBackend.__init__` gains a keyword-only `resource_server_url: AnyHttpUrl | None = None`, wired automatically from `AuthSettings`; pass it only if you construct the backend directly.
+
 ### Lowlevel `Server`: `subscribe` capability now correctly reported
 
 Previously, the lowlevel `Server` hardcoded `subscribe=False` in resource capabilities even when a `subscribe_resource()` handler was registered. The `subscribe` capability is now dynamically set to `True` when an `on_subscribe_resource` handler is provided. Clients that previously didn't see `subscribe: true` in capabilities will now see it when a handler is registered, which may change client behavior.

src/mcp/server/auth/middleware/bearer_auth.py

@@ -3,11 +3,12 @@ import time
 from typing import Any, TypedDict
 
 from pydantic import AnyHttpUrl
-from starlette.authentication import AuthCredentials, AuthenticationBackend, SimpleUser
+from starlette.authentication import AuthCredentials, AuthenticationBackend, BaseUser, SimpleUser
 from starlette.requests import HTTPConnection
 from starlette.types import Receive, Scope, Send
 
 from mcp.server.auth.provider import AccessToken, TokenVerifier
+from mcp.shared.auth_utils import check_token_audience
 
 
 class AuthenticatedUser(SimpleUser):
@@ -19,6 +20,27 @@ class AuthenticatedUser(SimpleUser):
         self.scopes = auth_info.scopes
 
 
+class InvalidTokenUser(BaseUser):
+    """Marker for a request that presented a Bearer token the verifier rejected,
+    that has expired, or whose audience does not match this resource server.
+    Carries the human-readable reason for the WWW-Authenticate error_description."""
+
+    def __init__(self, reason: str) -> None:
+        self.reason = reason
+
+    @property
+    def is_authenticated(self) -> bool:
+        return False
+
+    @property
+    def display_name(self) -> str:
+        return ""
+
+    @property
+    def identity(self) -> str:
+        return ""
+
+
 class AuthorizationContext(TypedDict):
     client_id: str
     issuer: str | None
@@ -46,27 +68,30 @@ def authorization_context(user: AuthenticatedUser) -> AuthorizationContext:
 class BearerAuthBackend(AuthenticationBackend):
     """Authentication backend that validates Bearer tokens using a TokenVerifier."""
 
-    def __init__(self, token_verifier: TokenVerifier):
+    def __init__(self, token_verifier: TokenVerifier, *, resource_server_url: AnyHttpUrl | None = None) -> None:
         self.token_verifier = token_verifier
+        self.resource_server_url = resource_server_url
 
-    async def authenticate(self, conn: HTTPConnection):
+    async def authenticate(self, conn: HTTPConnection) -> tuple[AuthCredentials, BaseUser] | None:
         auth_header = next(
             (conn.headers.get(key) for key in conn.headers if key.lower() == "authorization"),
             None,
         )
         if not auth_header or not auth_header.lower().startswith("bearer "):
-            return None
+            return None  # no credentials presented → bare challenge per RFC 6750 §3
 
-        token = auth_header[7:]  # Remove "Bearer " prefix
-
-        # Validate the token with the verifier
+        token = auth_header[7:]
         auth_info = await self.token_verifier.verify_token(token)
-
-        if not auth_info:
-            return None
-
-        if auth_info.expires_at and auth_info.expires_at < int(time.time()):
-            return None
+        if auth_info is None:
+            return AuthCredentials(), InvalidTokenUser("The access token is malformed or unknown")
+        if auth_info.expires_at is not None and auth_info.expires_at < int(time.time()):
+            return AuthCredentials(), InvalidTokenUser("The access token has expired")
+        if (
+            self.resource_server_url is not None
+            and auth_info.resource is not None
+            and not check_token_audience(auth_info.resource, self.resource_server_url)
+        ):
+            return AuthCredentials(), InvalidTokenUser("The access token was issued for a different resource")
 
         return AuthCredentials(auth_info.scopes), AuthenticatedUser(auth_info)
 
@@ -97,35 +122,47 @@ class RequireAuthMiddleware:
 
     async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
         auth_user = scope.get("user")
+        if isinstance(auth_user, InvalidTokenUser):
+            await self._send_auth_error(send, status_code=401, error="invalid_token", description=auth_user.reason)
+            return
         if not isinstance(auth_user, AuthenticatedUser):
-            await self._send_auth_error(
-                send, status_code=401, error="invalid_token", description="Authentication required"
-            )
+            await self._send_auth_error(send, status_code=401)
             return
 
-        auth_credentials = scope.get("auth")
-
+        auth_credentials = scope["auth"]
         for required_scope in self.required_scopes:
-            # auth_credentials should always be provided; this is just paranoia
-            if auth_credentials is None or required_scope not in auth_credentials.scopes:
+            if required_scope not in auth_credentials.scopes:
                 await self._send_auth_error(
-                    send, status_code=403, error="insufficient_scope", description=f"Required scope: {required_scope}"
+                    send,
+                    status_code=403,
+                    error="insufficient_scope",
+                    description="The access token lacks a required scope",
                 )
                 return
 
         await self.app(scope, receive, send)
 
-    async def _send_auth_error(self, send: Send, status_code: int, error: str, description: str) -> None:
-        """Send an authentication error response with WWW-Authenticate header."""
-        # Build WWW-Authenticate header value
-        www_auth_parts = [f'error="{error}"', f'error_description="{description}"']
+    async def _send_auth_error(
+        self, send: Send, *, status_code: int, error: str | None = None, description: str | None = None
+    ) -> None:
+        """Send a Bearer challenge. RFC 6750 §3: error/error_description only when a token
+        was presented; scope advertises what is required; resource_metadata for discovery."""
+        parts: list[str] = []
+        if error is not None:
+            parts.append(f'error="{error}"')
+        if description is not None:
+            parts.append(f'error_description="{description}"')
+        if self.required_scopes:
+            parts.append(f'scope="{" ".join(self.required_scopes)}"')
         if self.resource_metadata_url:
-            www_auth_parts.append(f'resource_metadata="{self.resource_metadata_url}"')
+            parts.append(f'resource_metadata="{self.resource_metadata_url}"')
+        www_authenticate = f"Bearer {', '.join(parts)}" if parts else "Bearer"
 
-        www_authenticate = f"Bearer {', '.join(www_auth_parts)}"
-
-        # Send response
-        body = {"error": error, "error_description": description}
+        body: dict[str, str] = {}
+        if error is not None:
+            body["error"] = error
+        if description is not None:
+            body["error_description"] = description
         body_bytes = json.dumps(body).encode()
 
         await send(

src/mcp/server/lowlevel/server.py

@@ -710,7 +710,7 @@ class Server(Generic[LifespanResultT]):
                 middleware = [
                     Middleware(
                         AuthenticationMiddleware,
-                        backend=BearerAuthBackend(token_verifier),
+                        backend=BearerAuthBackend(token_verifier, resource_server_url=auth.resource_server_url),
                     ),
                     Middleware(AuthContextMiddleware),
                 ]

src/mcp/server/mcpserver/server.py

@@ -1000,7 +1000,9 @@ class MCPServer(Generic[LifespanResultT]):
                     # extract auth info from request (but do not require it)
                     Middleware(
                         AuthenticationMiddleware,
-                        backend=BearerAuthBackend(self._token_verifier),
+                        backend=BearerAuthBackend(
+                            self._token_verifier, resource_server_url=self.settings.auth.resource_server_url
+                        ),
                     ),
                     # Add the auth context middleware to store
                     # authenticated user in a contextvar

src/mcp/shared/auth_utils.py

@@ -5,12 +5,15 @@ from urllib.parse import urlparse, urlsplit, urlunsplit
 
 from pydantic import AnyUrl, HttpUrl
 
+_DEFAULT_PORTS = {"http": 80, "https": 443}
+
 
 def resource_url_from_server_url(url: str | HttpUrl | AnyUrl) -> str:
     """Convert server URL to canonical resource URL per RFC 8707.
 
     RFC 8707 section 2 states that resource URIs "MUST NOT include a fragment component".
-    Returns absolute URI with lowercase scheme/host for canonical form.
+    Returns absolute URI with lowercase scheme/host and the scheme's default port
+    elided (RFC 3986 §6.2.3) for canonical form.
 
     Args:
         url: Server URL to convert
@@ -23,9 +26,13 @@ def resource_url_from_server_url(url: str | HttpUrl | AnyUrl) -> str:
 
     # Parse the URL and remove fragment, create canonical form
     parsed = urlsplit(url_str)
-    canonical = urlunsplit(parsed._replace(scheme=parsed.scheme.lower(), netloc=parsed.netloc.lower(), fragment=""))
-
-    return canonical
+    scheme = parsed.scheme.lower()
+    netloc = parsed.netloc.lower()
+    # RFC 3986 §6.2.3: an explicit default port is equivalent to omitting it.
+    if parsed.port is not None and _DEFAULT_PORTS.get(scheme) == parsed.port:
+        userinfo, sep, hostport = netloc.rpartition("@")
+        netloc = f"{userinfo}{sep}{hostport.rsplit(':', 1)[0]}"
+    return urlunsplit(parsed._replace(scheme=scheme, netloc=netloc, fragment=""))
 
 
 def check_resource_allowed(requested_resource: str, configured_resource: str) -> bool:
@@ -65,6 +72,19 @@ def check_resource_allowed(requested_resource: str, configured_resource: str) ->
     return requested_path.startswith(configured_path)
 
 
+def check_token_audience(token_resource: str, server_resource: str | HttpUrl | AnyUrl) -> bool:
+    """Return True iff a token's RFC 8707 resource indicator identifies this server.
+
+    Server-side audience validation is canonical-URI equality (authorization.mdx
+    Token Audience Binding): a token for a parent or sibling path on the same
+    origin is NOT for this server. Contrast check_resource_allowed, which is the
+    client-side hierarchical question and intentionally more permissive.
+    """
+    return resource_url_from_server_url(token_resource).rstrip("/") == resource_url_from_server_url(
+        server_resource
+    ).rstrip("/")
+
+
 def calculate_token_expiry(expires_in: int | str | None) -> float | None:
     """Calculate token expiry timestamp from expires_in seconds.
 

tests/interaction/_requirements.py

@@ -2622,12 +2622,6 @@ REQUIREMENTS: dict[str, Requirement] = {
         behavior="The resource server validates that the token audience matches its resource identifier.",
         transports=("streamable-http",),
         note="Auth is enforced at the HTTP layer.",
-        divergence=Divergence(
-            note=(
-                "BearerAuthBackend never inspects AccessToken.resource; a token issued for a different "
-                "resource is accepted. Spec MUST."
-            ),
-        ),
     ),
     "hosting:auth:authinfo-propagates": Requirement(
         source="sdk",
@@ -2640,18 +2634,12 @@ REQUIREMENTS: dict[str, Requirement] = {
         behavior="An expired token returns 401 invalid_token.",
         transports=("streamable-http",),
         note="Auth is enforced at the HTTP layer; 401 is an HTTP status code.",
-        divergence=Divergence(
-            note="The challenge carries no `scope` parameter; see the note on hosting:auth:missing-401.",
-        ),
     ),
     "hosting:auth:invalid-401": Requirement(
         source=f"{SPEC_BASE_URL}/basic/authorization#token-handling",
         behavior="A malformed bearer token or token-verification failure returns 401 with WWW-Authenticate.",
         transports=("streamable-http",),
         note="Auth is enforced at the HTTP layer; 401 is an HTTP status code.",
-        divergence=Divergence(
-            note="The challenge carries no `scope` parameter; see the note on hosting:auth:missing-401.",
-        ),
     ),
     "hosting:auth:metadata-endpoints": Requirement(
         source=f"{SPEC_BASE_URL}/basic/authorization#authorization-server-location",
@@ -2671,15 +2659,6 @@ REQUIREMENTS: dict[str, Requirement] = {
         ),
         transports=("streamable-http",),
         note="Auth is enforced at the HTTP layer; 401 is an HTTP status code.",
-        divergence=Divergence(
-            note=(
-                "The SDK never emits a `scope` parameter in any WWW-Authenticate challenge — neither the "
-                "discovery-time 401 (#protected-resource-metadata-discovery-requirements SHOULD) nor the "
-                "runtime 403 (#runtime-insufficient-scope-errors SHOULD); and for the no-credentials case "
-                'it emits error="invalid_token", which RFC 6750 Section 3.1 says SHOULD NOT appear when no '
-                "authentication information was presented."
-            ),
-        ),
     ),
     "hosting:auth:prm:authorization-servers-field": Requirement(
         source=f"{SPEC_BASE_URL}/basic/authorization#authorization-server-location",
@@ -2706,13 +2685,6 @@ REQUIREMENTS: dict[str, Requirement] = {
         ),
         transports=("streamable-http",),
         note="Auth is enforced at the HTTP layer; 403 is an HTTP status code.",
-        divergence=Divergence(
-            note=(
-                'The SDK emits error="insufficient_scope" and error_description but never the `scope` '
-                "parameter the spec SHOULD include; the SDK client reads `scope` from this header to drive "
-                "step-up (utils.py extract_scope_from_www_auth) — a resource-server/client asymmetry."
-            ),
-        ),
     ),
     "hosting:auth:as:authorize-requires-pkce": Requirement(
         source=f"{SPEC_BASE_URL}/basic/authorization#authorization-code-protection",

tests/interaction/auth/_harness.py

@@ -277,9 +277,9 @@ def shim(
 class _FirstChallenge:
     """ASGI shim that answers the first request to a path with 401 + a given WWW-Authenticate.
 
-    Subsequent requests pass through to the wrapped app. Used to make the initial 401 carry
-    parameters (such as `scope=`) that the SDK's own bearer middleware cannot be configured
-    to emit, so client behaviour driven by those parameters is reachable end to end. Reserve
+    Subsequent requests pass through to the wrapped app. Used to make the initial 401 carry a
+    `scope=` value that differs from the gate's `required_scopes` (which is all the real
+    middleware can emit), so client scope-selection priority is reachable end to end. Reserve
     this pattern for behaviour the real server cannot be made to produce.
     """
 
@@ -312,9 +312,10 @@ def step_up_shim(www_authenticate: str, *, on_nth_authenticated_post: int = 2) -
     """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
-    handling: the SDK's bearer middleware never emits `scope=` in its 403 challenge (see the
-    divergence on `hosting:auth:scope-403`), so the test supplies the 403 itself. Reserve this
-    pattern for behaviour the real server cannot be made to produce.
+    handling: the real middleware's 403 carries `scope=` from the gate's static
+    `required_scopes`, but step-up tests need a wider scope than the gate would emit so the
+    client's scope-union logic has something to add. Reserve this pattern for behaviour the
+    real server cannot be made to produce.
 
     The default `on_nth_authenticated_post=2` targets the `notifications/initialized` POST: the
     first authenticated POST is the auth flow's retry of the original initialize request (yielded

tests/interaction/auth/test_authorize_token.py

@@ -328,12 +328,12 @@ async def test_the_registered_auth_method_is_used_regardless_of_as_metadata_adve
 async def test_scope_is_selected_from_the_www_authenticate_challenge_over_prm_metadata() -> None:
     """When the 401 challenge carries `scope=`, that value is requested instead of the PRM scopes.
 
-    The SDK's bearer middleware never emits `scope=` in WWW-Authenticate (see the divergence
-    on `hosting:auth:scope-403`), so the test supplies the first 401 itself via
-    `first_challenge_shim` and disables token verification so the post-auth retry succeeds
-    regardless of the granted scope. PRM advertises `["from-prm"]` (it mirrors
-    `required_scopes`); the challenge says `from-header`; the authorize URL must carry
-    `from-header`.
+    The SDK's bearer middleware emits `scope=` with the configured `required_scopes`, which
+    here would be `from-prm` — the same value PRM advertises — so the test supplies the first
+    401 itself via `first_challenge_shim` to put a *different* value in the challenge, and
+    disables token verification so the post-auth retry succeeds regardless of the granted
+    scope. PRM advertises `["from-prm"]`; the challenge says `from-header`; the authorize URL
+    must carry `from-header`.
     """
     recorded, on_request = record_requests()
     provider = InMemoryAuthorizationServerProvider(default_scopes=["from-header"])

tests/interaction/auth/test_bearer.py

@@ -3,8 +3,8 @@
 These tests mount only the resource-server side of the auth wiring (a `StaticTokenVerifier`
 seeded with hand-built tokens, no authorization-server provider) and speak raw HTTP, since
 every assertion is about HTTP semantics the SDK `Client` cannot observe: the 401/403 status,
-the `WWW-Authenticate` header structure, and that a wrong-audience token reaches the MCP
-endpoint behind the gate. The flow side of the same 401 is `test_flow.py`'s flagship test.
+the `WWW-Authenticate` header structure, and that a token with no audience claim reaches the
+MCP endpoint behind the gate. The flow side of the same 401 is `test_flow.py`'s flagship test.
 """
 
 import time
@@ -40,6 +40,14 @@ TOKENS = {
         expires_at=_FUTURE,
         resource="https://other.example/mcp",
     ),
+    "tok-parent-aud": AccessToken(
+        token="tok-parent-aud",
+        client_id="c",
+        scopes=[REQUIRED_SCOPE],
+        expires_at=_FUTURE,
+        resource="http://127.0.0.1:8000/",
+    ),
+    "tok-no-aud": AccessToken(token="tok-no-aud", client_id="c", scopes=[REQUIRED_SCOPE], expires_at=_FUTURE),
 }
 
 
@@ -80,41 +88,37 @@ async def test_a_request_with_no_authorization_header_is_challenged_with_resourc
 ) -> None:
     """No `Authorization` header → 401 with a `WWW-Authenticate` carrying `resource_metadata`.
 
-    The snapshot pins current behaviour: the SDK collapses the no-header, unknown-token, and
-    expired-token cases into one challenge (`error="invalid_token"`, no `scope` parameter). The
-    spec says the discovery-time challenge SHOULD include `scope` and RFC 6750 says the
-    no-credentials case SHOULD NOT carry an error code; both gaps are recorded as the divergence
-    on this requirement. Asserting the dict equals an exact key set also pins that no parameter
-    appears twice.
+    RFC 6750 §3: a no-credentials challenge carries no error code. The snapshot pins the
+    full header (parameter order included); asserting the dict equals an exact key set also
+    pins that no parameter appears twice.
     """
     response = await post_mcp(protected)
 
     assert response.status_code == 401
     assert response.headers["www-authenticate"] == snapshot(
-        'Bearer error="invalid_token", error_description="Authentication required", '
-        'resource_metadata="http://127.0.0.1:8000/.well-known/oauth-protected-resource/mcp"'
+        'Bearer scope="mcp:read", resource_metadata="http://127.0.0.1:8000/.well-known/oauth-protected-resource/mcp"'
     )
     assert parse_www_authenticate(response.headers["www-authenticate"]) == {
-        "error": "invalid_token",
-        "error_description": "Authentication required",
+        "scope": REQUIRED_SCOPE,
         "resource_metadata": RESOURCE_METADATA_URL,
     }
-    assert response.json() == snapshot({"error": "invalid_token", "error_description": "Authentication required"})
+    assert response.json() == snapshot({})
 
 
 @requirement("hosting:auth:invalid-401")
 async def test_an_unrecognized_bearer_token_is_answered_401_invalid_token(protected: httpx.AsyncClient) -> None:
     """A token the verifier does not recognize is answered 401 `invalid_token`.
 
-    The challenge is identical to the no-header case (the backend returns `None` for both); the
-    missing `scope` parameter is the recorded divergence on this requirement.
+    The challenge is distinct from the no-header case: a bearer token was presented, so RFC
+    6750 §3.1's `error` and `error_description` apply.
     """
     response = await post_mcp(protected, bearer="tok-unknown")
 
     assert response.status_code == 401
     assert parse_www_authenticate(response.headers["www-authenticate"]) == {
         "error": "invalid_token",
-        "error_description": "Authentication required",
+        "error_description": "The access token is malformed or unknown",
+        "scope": REQUIRED_SCOPE,
         "resource_metadata": RESOURCE_METADATA_URL,
     }
 
@@ -124,48 +128,89 @@ async def test_an_expired_token_is_answered_401(protected: httpx.AsyncClient) ->
     """A token whose `expires_at` is in the past is answered 401 `invalid_token`.
 
     The expiry check is the bearer backend's, against the wall clock; the test seeds a concrete
-    past timestamp so no time mocking is involved. The missing `scope` parameter is the recorded
-    divergence on this requirement.
+    past timestamp so no time mocking is involved.
     """
     response = await post_mcp(protected, bearer="tok-expired")
 
     assert response.status_code == 401
-    assert parse_www_authenticate(response.headers["www-authenticate"])["error"] == "invalid_token"
+    assert parse_www_authenticate(response.headers["www-authenticate"]) == {
+        "error": "invalid_token",
+        "error_description": "The access token has expired",
+        "scope": REQUIRED_SCOPE,
+        "resource_metadata": RESOURCE_METADATA_URL,
+    }
 
 
 @requirement("hosting:auth:scope-403")
-async def test_a_token_missing_a_required_scope_is_answered_403_insufficient_scope_without_a_scope_param(
+async def test_a_token_missing_a_required_scope_is_answered_403_with_the_required_scope_in_the_challenge(
     protected: httpx.AsyncClient,
 ) -> None:
-    """A token lacking the required scope is answered 403 `insufficient_scope`, with no `scope` parameter.
+    """A token lacking the required scope is answered 403 `insufficient_scope` with `scope=` naming what's needed.
 
-    The spec's runtime-insufficient-scope guidance says the challenge SHOULD include `scope`
-    naming the required scope; the SDK never emits it, recorded as the divergence on this
-    requirement. The SDK client reads `scope` from this header to drive step-up, so the gap is
-    a resource-server/client asymmetry.
+    The SDK client reads `scope` from this header to drive step-up, so the parameter is the
+    contract between resource server and client.
     """
     response = await post_mcp(protected, bearer="tok-noscope")
 
     assert response.status_code == 403
-    parsed = parse_www_authenticate(response.headers["www-authenticate"])
-    assert parsed == {
+    assert parse_www_authenticate(response.headers["www-authenticate"]) == {
         "error": "insufficient_scope",
-        "error_description": f"Required scope: {REQUIRED_SCOPE}",
+        "error_description": "The access token lacks a required scope",
+        "scope": REQUIRED_SCOPE,
         "resource_metadata": RESOURCE_METADATA_URL,
     }
-    assert "scope" not in parsed
 
 
 @requirement("hosting:auth:aud-validation")
-async def test_a_token_with_a_mismatched_audience_is_accepted(protected: httpx.AsyncClient) -> None:
-    """A token whose `resource` does not match the server's resource identifier is accepted.
+async def test_a_token_with_a_mismatched_audience_is_answered_401_invalid_token(protected: httpx.AsyncClient) -> None:
+    """A token whose `resource` does not match the server's resource identifier is answered 401.
 
-    The spec mandates the resource server validate the token's audience; the bearer backend
-    never inspects `AccessToken.resource`, so the request passes the gate and the MCP endpoint
-    serves it. This pins current behaviour with the divergence recorded on the requirement.
+    Spec-mandated: the resource server MUST validate the token's audience and reject tokens
+    not issued specifically for it.
     """
     response = await post_mcp(protected, bearer="tok-wrong-aud")
 
+    assert response.status_code == 401
+    assert parse_www_authenticate(response.headers["www-authenticate"]) == {
+        "error": "invalid_token",
+        "error_description": "The access token was issued for a different resource",
+        "scope": REQUIRED_SCOPE,
+        "resource_metadata": RESOURCE_METADATA_URL,
+    }
+
+
+@requirement("hosting:auth:aud-validation")
+async def test_a_token_for_a_parent_path_on_the_same_origin_is_answered_401_invalid_token(
+    protected: httpx.AsyncClient,
+) -> None:
+    """A token whose audience is the same origin but a parent path is answered 401.
+
+    This is the discriminating case for canonical-URI equality: under hierarchical prefix
+    semantics a token for `http://host/` would be accepted by a server at `http://host/mcp`;
+    under audience binding it must be rejected. The cross-origin case above cannot catch a
+    regression to prefix semantics.
+    """
+    response = await post_mcp(protected, bearer="tok-parent-aud")
+
+    assert response.status_code == 401
+    assert parse_www_authenticate(response.headers["www-authenticate"]) == {
+        "error": "invalid_token",
+        "error_description": "The access token was issued for a different resource",
+        "scope": REQUIRED_SCOPE,
+        "resource_metadata": RESOURCE_METADATA_URL,
+    }
+
+
+@requirement("hosting:auth:aud-validation")
+async def test_a_token_without_a_resource_claim_passes_the_audience_check(protected: httpx.AsyncClient) -> None:
+    """A token whose `AccessToken.resource` is unset passes the audience check.
+
+    SDK-defined pass-through: the SDK cannot distinguish a verifier that performed its own
+    audience check and chose not to surface the claim from a token that genuinely carries
+    none, so `resource is None` is accepted. This pins that policy.
+    """
+    response = await post_mcp(protected, bearer="tok-no-aud")
+
     assert response.status_code == 200
     assert response.headers["content-type"].startswith("text/event-stream")
     # The body is finite SSE: a result event followed by stream close. Pull the JSON-RPC response
@@ -186,4 +231,6 @@ async def test_an_access_token_in_the_query_string_is_not_accepted(protected: ht
     response = await post_mcp(protected, query={"access_token": "tok-valid"})
 
     assert response.status_code == 401
-    assert parse_www_authenticate(response.headers["www-authenticate"])["error"] == "invalid_token"
+    parsed = parse_www_authenticate(response.headers["www-authenticate"])
+    assert "error" not in parsed
+    assert parsed["scope"] == REQUIRED_SCOPE

tests/server/auth/middleware/test_bearer_auth.py

@@ -6,10 +6,15 @@ from typing import Any, cast
 import pytest
 from starlette.authentication import AuthCredentials
 from starlette.datastructures import Headers
-from starlette.requests import Request
+from starlette.requests import Request, empty_receive
 from starlette.types import Message, Receive, Scope, Send
 
-from mcp.server.auth.middleware.bearer_auth import AuthenticatedUser, BearerAuthBackend, RequireAuthMiddleware
+from mcp.server.auth.middleware.bearer_auth import (
+    AuthenticatedUser,
+    BearerAuthBackend,
+    InvalidTokenUser,
+    RequireAuthMiddleware,
+)
 from mcp.server.auth.provider import AccessToken, OAuthAuthorizationServerProvider, ProviderTokenVerifier
 
 
@@ -126,7 +131,9 @@ class TestBearerAuthBackend:
         assert result is None
 
     async def test_invalid_token(self, mock_oauth_provider: OAuthAuthorizationServerProvider[Any, Any, Any]):
-        """Test authentication with invalid token."""
+        """A Bearer token the verifier rejects yields an InvalidTokenUser carrying the
+        reason, so RequireAuthMiddleware can send error="invalid_token" rather than a
+        bare challenge (RFC 6750 §3.1 distinguishes no-credentials from bad-credentials)."""
         backend = BearerAuthBackend(token_verifier=ProviderTokenVerifier(mock_oauth_provider))
         request = Request(
             {
@@ -135,14 +142,24 @@ class TestBearerAuthBackend:
             }
         )
         result = await backend.authenticate(request)
-        assert result is None
+        assert result is not None
+        credentials, user = result
+        assert isinstance(credentials, AuthCredentials)
+        assert credentials.scopes == []
+        assert isinstance(user, InvalidTokenUser)
+        assert user.reason == "The access token is malformed or unknown"
+        # BaseUser interface obligations — Starlette may render these
+        assert user.is_authenticated is False
+        assert user.display_name == ""
+        assert user.identity == ""
 
     async def test_expired_token(
         self,
         mock_oauth_provider: OAuthAuthorizationServerProvider[Any, Any, Any],
         expired_access_token: AccessToken,
     ):
-        """Test authentication with expired token."""
+        """An expired token yields an InvalidTokenUser whose reason names expiry, so the
+        WWW-Authenticate error_description tells the client why (RFC 6750 §3.1)."""
         backend = BearerAuthBackend(token_verifier=ProviderTokenVerifier(mock_oauth_provider))
         add_token_to_provider(mock_oauth_provider, "expired_token", expired_access_token)
         request = Request(
@@ -152,7 +169,10 @@ class TestBearerAuthBackend:
             }
         )
         result = await backend.authenticate(request)
-        assert result is None
+        assert result is not None
+        _, user = result
+        assert isinstance(user, InvalidTokenUser)
+        assert user.reason == "The access token has expired"
 
     async def test_valid_token(
         self,
@@ -344,17 +364,18 @@ class TestRequireAuthMiddleware:
         assert any(h[0] == b"www-authenticate" for h in sent_messages[0]["headers"])
         assert not app.called
 
-    async def test_no_auth_credentials(self, valid_access_token: AccessToken):
-        """Test middleware with no auth credentials in scope."""
+    async def test_invalid_token_user_gets_401_with_error_description(self):
+        """When the backend marked the request with InvalidTokenUser, the middleware
+        sends 401 with error="invalid_token" and the carried reason as error_description
+        (RFC 6750 §3.1) — distinct from the bare challenge sent when no token was presented."""
         app = MockApp()
         middleware = RequireAuthMiddleware(app, required_scopes=["read"])
+        scope: Scope = {
+            "type": "http",
+            "user": InvalidTokenUser("The access token has expired"),
+            "auth": AuthCredentials(),
+        }
 
-        # Create a user with read/write scopes
-        user = AuthenticatedUser(valid_access_token)
-
-        scope: Scope = {"type": "http", "user": user}  # No auth credentials
-
-        # Create dummy async functions for receive and send
         async def receive() -> Message:  # pragma: no cover
             return {"type": "http.request"}
 
@@ -365,11 +386,12 @@ class TestRequireAuthMiddleware:
 
         await middleware(scope, receive, send)
 
-        # Check that a 403 response was sent
         assert len(sent_messages) == 2
         assert sent_messages[0]["type"] == "http.response.start"
-        assert sent_messages[0]["status"] == 403
-        assert any(h[0] == b"www-authenticate" for h in sent_messages[0]["headers"])
+        assert sent_messages[0]["status"] == 401
+        www_authenticate = dict(sent_messages[0]["headers"])[b"www-authenticate"]
+        assert b'error="invalid_token"' in www_authenticate
+        assert b'error_description="The access token has expired"' in www_authenticate
         assert not app.called
 
     async def test_has_required_scopes(self, valid_access_token: AccessToken):
@@ -446,3 +468,24 @@ class TestRequireAuthMiddleware:
         assert app.scope == scope
         assert app.receive == receive
         assert app.send == send
+
+
+@pytest.mark.anyio
+async def test_unauthenticated_request_with_no_required_scopes_gets_bare_bearer_challenge():
+    """RFC 6750 §3: when no credentials were presented and the server has nothing to
+    advertise (no required scopes, no resource_metadata), the WWW-Authenticate header
+    is the bare scheme name with no parameters."""
+    app = MockApp()
+    middleware = RequireAuthMiddleware(app, required_scopes=[])
+    scope: Scope = {"type": "http"}
+
+    sent_messages: list[Message] = []
+
+    async def send(message: Message) -> None:
+        sent_messages.append(message)
+
+    await middleware(scope, empty_receive, send)
+
+    assert sent_messages[0]["status"] == 401
+    assert dict(sent_messages[0]["headers"])[b"www-authenticate"] == b"Bearer"
+    assert not app.called

tests/shared/test_auth_utils.py

@@ -2,7 +2,7 @@
 
 from pydantic import HttpUrl
 
-from mcp.shared.auth_utils import check_resource_allowed, resource_url_from_server_url
+from mcp.shared.auth_utils import check_resource_allowed, check_token_audience, resource_url_from_server_url
 
 # Tests for resource_url_from_server_url function
 
@@ -34,6 +34,23 @@ def test_resource_url_from_server_url_preserves_port():
     assert resource_url_from_server_url("http://example.com:8080/") == "http://example.com:8080/"
 
 
+def test_resource_url_from_server_url_strips_default_port():
+    """An explicit default port is equivalent to omitting it (RFC 3986 §6.2.3)."""
+    assert resource_url_from_server_url("https://example.com:443/mcp") == "https://example.com/mcp"
+    assert resource_url_from_server_url("http://example.com:80/mcp") == "http://example.com/mcp"
+    # Only the scheme's own default is stripped — :80 on https is significant.
+    assert resource_url_from_server_url("https://example.com:80/mcp") == "https://example.com:80/mcp"
+    # IPv6 brackets survive the rewrite.
+    assert resource_url_from_server_url("https://[::1]:443/mcp") == "https://[::1]/mcp"
+
+
+def test_check_token_audience_ignores_default_port():
+    """A token issued for `https://h:443/mcp` is for the server at `https://h/mcp`."""
+    assert check_token_audience("https://h:443/mcp", "https://h/mcp") is True
+    assert check_token_audience("https://h/mcp", "https://h:443/mcp") is True
+    assert check_token_audience("https://h:8443/mcp", "https://h/mcp") is False
+
+
 def test_resource_url_from_server_url_lowercase_scheme_and_host():
     """Scheme and host should be lowercase for canonical form."""
     assert resource_url_from_server_url("HTTPS://EXAMPLE.COM/path") == "https://example.com/path"