Add AuthSettings.validate_token_resource to check a bearer token's resource

BearerAuthBackend takes an optional resource_server_url; when it is set, only
a token the verifier reports as issued for that URL (AccessToken.resource,
the RFC 8707 resource indicator, compared as a URL with a trailing slash
tolerated) is accepted, and anything else is answered 401 like an
unrecognized token. AuthSettings.validate_token_resource (default False)
turns this on for the Streamable HTTP and SSE apps; it requires
resource_server_url.

TokenVerifier.verify_token's docstring and docs/run/authorization.md say
where the token's audience goes and when to enable the option versus
checking the audience in the verifier. The simple-auth example enables it.
The interaction suite enables it for the bearer tests and records that it is
off by default on hosting:auth:aud-validation.

Author: maxisbey

docs/run/authorization.md

@@ -23,14 +23,15 @@ The SDK has no opinion about what a valid token looks like. You tell it, by impl
 ```
 
 * `TokenVerifier` is a protocol with one async method. `verify_token` gets the raw token from the `Authorization` header and returns an **`AccessToken`** if it's valid, `None` if it isn't. There is nothing else to implement.
-* This one looks the token up in a table. A real one verifies a JWT signature or calls the authorization server's token-introspection endpoint. That code is yours; the SDK only calls it.
+* This one looks the token up in a table that belongs to this server. A real one verifies a JWT signature or calls the authorization server's token-introspection endpoint, and puts the token's audience (`aud`) in `AccessToken.resource` so the SDK can check the token was issued for this server (see `validate_token_resource` below); if `aud` is a list, use the entry that equals your `resource_server_url`. That code is yours; the SDK only calls it.
 * `token_verifier=` and `auth=` always travel together. Pass one without the other and `MCPServer(...)` raises a `ValueError` before it ever serves a request.
 
 `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.
 * `required_scopes`: every token must carry all of them.
+* `validate_token_resource`: refuse any token your verifier does not report as issued for `resource_server_url` (its `AccessToken.resource`, compared as a URL, a trailing slash aside). Turn it on when your authorization server binds tokens to the `resource` a client asks for, which is what MCP clients send. If it issues its own audience identifiers instead (an Auth0 API identifier, an Entra application ID), leave it off and check `aud` against that identifier in your verifier: a token you cannot tie to this server should come back as `None`.
 
 !!! tip
     `examples/servers/simple-auth/` in the SDK repository has an `IntrospectionTokenVerifier` that calls

examples/servers/simple-auth/mcp_simple_auth/server.py

@@ -71,6 +71,7 @@ def create_resource_server(settings: ResourceServerSettings) -> MCPServer:
             issuer_url=settings.auth_server_url,
             required_scopes=[settings.mcp_scope],
             resource_server_url=settings.server_url,
+            validate_token_resource=True,  # tokens must be reported as issued for server_url
         ),
     )
     # Store settings for later use in run()

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

@@ -1,14 +1,17 @@
 import json
+import logging
 import time
 from typing import Any, TypedDict
 
-from pydantic import AnyHttpUrl
+from pydantic import AnyHttpUrl, ValidationError
 from starlette.authentication import AuthCredentials, AuthenticationBackend, SimpleUser
 from starlette.requests import HTTPConnection
 from starlette.types import Receive, Scope, Send
 
 from mcp.server.auth.provider import AccessToken, TokenVerifier, principal_components
 
+logger = logging.getLogger(__name__)
+
 
 class AuthenticatedUser(SimpleUser):
     """User with authentication info."""
@@ -39,10 +42,15 @@ def authorization_context(user: AuthenticatedUser) -> AuthorizationContext:
 
 
 class BearerAuthBackend(AuthenticationBackend):
-    """Authentication backend that validates Bearer tokens using a TokenVerifier."""
+    """Authentication backend that validates Bearer tokens using a TokenVerifier.
+
+    When `resource_server_url` is given, only a token whose `AccessToken.resource`
+    (its RFC 8707 resource indicator / audience) is that URL is accepted.
+    """
 
-    def __init__(self, token_verifier: TokenVerifier):
+    def __init__(self, token_verifier: TokenVerifier, *, resource_server_url: AnyHttpUrl | None = None):
         self.token_verifier = token_verifier
+        self.resource_server_url = resource_server_url
 
     async def authenticate(self, conn: HTTPConnection):
         auth_header = next(
@@ -63,8 +71,22 @@ async def authenticate(self, conn: HTTPConnection):
         if auth_info.expires_at and auth_info.expires_at < int(time.time()):
             return None
 
+        if self.resource_server_url and not self._issued_for_this_resource(auth_info.resource):
+            logger.warning(
+                "Bearer token resource %s is not resource_server_url %s", auth_info.resource, self.resource_server_url
+            )
+            return None
+
         return AuthCredentials(auth_info.scopes), AuthenticatedUser(auth_info)
 
+    def _issued_for_this_resource(self, resource: str | None) -> bool:
+        """Compare as URLs (so case and default-port spelling do not matter), a trailing slash aside."""
+        try:
+            token_resource = str(AnyHttpUrl(resource or ""))
+        except ValidationError:
+            return False
+        return token_resource.removesuffix("/") == str(self.resource_server_url).removesuffix("/")
+
 
 class RequireAuthMiddleware:
     """Middleware that requires a valid Bearer token in the Authorization header.

src/mcp/server/auth/provider.py

@@ -124,7 +124,15 @@ class TokenVerifier(Protocol):
     """Protocol for verifying bearer tokens."""
 
     async def verify_token(self, token: str) -> AccessToken | None:
-        """Verify a bearer token and return access info if valid."""
+        """Verify a bearer token and return access info if valid.
+
+        Set `AccessToken.resource` to the resource the token was issued for (its RFC 8707
+        resource indicator / `aud`; for a list, the entry equal to the server's
+        `AuthSettings.resource_server_url`). With `AuthSettings.validate_token_resource` the
+        bearer middleware then refuses any token whose resource is not `resource_server_url`;
+        without it, confirming the token was issued for this server (for example by passing the
+        expected audience to your JWT library) is up to the verifier.
+        """
 
 
 # NOTE: MCPServer doesn't render any of these types in the user response, so it's

src/mcp/server/auth/settings.py

@@ -1,4 +1,5 @@
-from pydantic import AnyHttpUrl, BaseModel, ConfigDict, Field
+from pydantic import AnyHttpUrl, BaseModel, ConfigDict, Field, model_validator
+from typing_extensions import Self
 
 
 class ClientRegistrationOptions(BaseModel):
@@ -40,3 +41,15 @@ class AuthSettings(BaseModel):
         description="The URL of the MCP server to be used as the resource identifier "
         "and base route to look up OAuth Protected Resource Metadata.",
     )
+    validate_token_resource: bool = Field(
+        default=False,
+        description="Only accept tokens the token verifier reports as issued for `resource_server_url` "
+        "(`AccessToken.resource`, the RFC 8707 resource indicator). Enable it when your authorization "
+        "server binds tokens to the `resource` the client requested.",
+    )
+
+    @model_validator(mode="after")
+    def _validate_token_resource_needs_a_resource(self) -> Self:
+        if self.validate_token_resource and self.resource_server_url is None:
+            raise ValueError("validate_token_resource requires resource_server_url")
+        return self

src/mcp/server/lowlevel/server.py

@@ -776,7 +776,10 @@ def streamable_http_app(
                 middleware = [
                     Middleware(
                         AuthenticationMiddleware,
-                        backend=BearerAuthBackend(token_verifier),
+                        backend=BearerAuthBackend(
+                            token_verifier,
+                            resource_server_url=auth.resource_server_url if auth.validate_token_resource else None,
+                        ),
                     ),
                     Middleware(AuthContextMiddleware),
                 ]

src/mcp/server/mcpserver/server.py

@@ -1187,7 +1187,12 @@ async def handle_sse(scope: Scope, receive: Receive, send: Send):  # pragma: no
                     # 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
+                            if self.settings.auth.validate_token_resource
+                            else None,
+                        ),
                     ),
                     # Add the auth context middleware to store
                     # authenticated user in a contextvar

tests/interaction/_requirements.py

@@ -2848,11 +2848,11 @@ def __post_init__(self) -> None:
         source=f"{SPEC_BASE_URL}/basic/authorization#access-token-usage",
         behavior="The resource server validates that the token audience matches its resource identifier.",
         transports=("streamable-http",),
-        note="Auth is enforced at the HTTP layer.",
+        note="Auth is enforced at the HTTP layer; the tests enable AuthSettings.validate_token_resource.",
         divergence=Divergence(
             note=(
-                "BearerAuthBackend never inspects AccessToken.resource; a token issued for a different "
-                "resource is accepted. Spec MUST."
+                "Off by default: without AuthSettings.validate_token_resource the bearer gate does not compare "
+                "AccessToken.resource with resource_server_url and the check is the token verifier's. Spec MUST."
             ),
         ),
     ),

tests/interaction/auth/_harness.py

@@ -182,6 +182,7 @@ def auth_settings(
     required_scopes: Sequence[str] = ("mcp",),
     valid_scopes: Sequence[str] | None = None,
     identity_assertion_enabled: bool = False,
+    validate_token_resource: bool = False,
 ) -> AuthSettings:
     """Build `AuthSettings` for the co-hosted authorization + resource server.
 
@@ -194,14 +195,16 @@ def auth_settings(
 
     `identity_assertion_enabled` advertises and accepts the SEP-990 ID-JAG grant (RFC 7523
     jwt-bearer); the provider must implement `exchange_identity_assertion` for the endpoint to
-    issue tokens.
+    issue tokens. `validate_token_resource` makes the bearer gate refuse tokens whose verifier
+    does not report them as issued for the resource URL.
     """
     required = list(required_scopes)
     valid = list(valid_scopes) if valid_scopes is not None else required
     return AuthSettings(
         issuer_url=AnyHttpUrl(BASE_URL),
         resource_server_url=AnyHttpUrl(f"{BASE_URL}/mcp"),
         required_scopes=required,
+        validate_token_resource=validate_token_resource,
         client_registration_options=ClientRegistrationOptions(
             enabled=True, valid_scopes=valid, default_scopes=required
         ),

tests/interaction/auth/test_bearer.py

@@ -2,9 +2,9 @@
 
 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.
+every assertion is about HTTP semantics the SDK `Client` cannot observe: the 401/403 status
+and the `WWW-Authenticate` header structure. The flow side of the same 401 is `test_flow.py`'s
+flagship test.
 """
 
 import time
@@ -24,30 +24,33 @@
 pytestmark = pytest.mark.anyio
 
 REQUIRED_SCOPE = "mcp:read"
+RESOURCE = "http://127.0.0.1:8000/mcp"
 RESOURCE_METADATA_URL = "http://127.0.0.1:8000/.well-known/oauth-protected-resource/mcp"
 
 _FUTURE = int(time.time()) + 3600
 _PAST = int(time.time()) - 3600
 
+
+def tok(name: str, *, scopes: list[str], expires_at: int, resource: str | None = RESOURCE) -> AccessToken:
+    return AccessToken(token=name, client_id="c", scopes=scopes, expires_at=expires_at, resource=resource)
+
+
 TOKENS = {
-    "tok-valid": AccessToken(token="tok-valid", client_id="c", scopes=[REQUIRED_SCOPE], expires_at=_FUTURE),
-    "tok-expired": AccessToken(token="tok-expired", client_id="c", scopes=[REQUIRED_SCOPE], expires_at=_PAST),
-    "tok-noscope": AccessToken(token="tok-noscope", client_id="c", scopes=["other:thing"], expires_at=_FUTURE),
-    "tok-wrong-aud": AccessToken(
-        token="tok-wrong-aud",
-        client_id="c",
-        scopes=[REQUIRED_SCOPE],
-        expires_at=_FUTURE,
-        resource="https://other.example/mcp",
+    "tok-valid": tok("tok-valid", scopes=[REQUIRED_SCOPE], expires_at=_FUTURE),
+    "tok-expired": tok("tok-expired", scopes=[REQUIRED_SCOPE], expires_at=_PAST),
+    "tok-noscope": tok("tok-noscope", scopes=["other:thing"], expires_at=_FUTURE),
+    "tok-wrong-aud": tok(
+        "tok-wrong-aud", scopes=[REQUIRED_SCOPE], expires_at=_FUTURE, resource="https://other.example/mcp"
     ),
+    "tok-no-aud": tok("tok-no-aud", scopes=[REQUIRED_SCOPE], expires_at=_FUTURE, resource=None),
 }
 
 
 @pytest.fixture
 async def protected() -> AsyncIterator[httpx2.AsyncClient]:
     """A bearer-gated streamable-HTTP app (resource server only) on the in-process bridge."""
     server = Server("rs")
-    settings = auth_settings(required_scopes=[REQUIRED_SCOPE])
+    settings = auth_settings(required_scopes=[REQUIRED_SCOPE], validate_token_resource=True)
     async with mounted_app(server, auth=settings, token_verifier=StaticTokenVerifier(TOKENS)) as (http, _):
         yield http
 
@@ -157,19 +160,26 @@ async def test_a_token_missing_a_required_scope_is_answered_403_insufficient_sco
 
 
 @requirement("hosting:auth:aud-validation")
-async def test_a_token_with_a_mismatched_audience_is_accepted(protected: httpx2.AsyncClient) -> None:
-    """A token whose `resource` does not match the server's resource identifier is accepted.
-
-    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.
+@pytest.mark.parametrize("bearer", ["tok-wrong-aud", "tok-no-aud"])
+async def test_a_token_not_issued_for_this_resource_is_answered_401(protected: httpx2.AsyncClient, bearer: str) -> None:
+    """Spec-mandated audience check, which the SDK performs when `AuthSettings.validate_token_resource`
+    is set (off by default, the recorded divergence): a token whose verifier-reported `resource`
+    (RFC 8707) is another URL, or absent, is answered 401 `invalid_token` like an unrecognized token.
     """
-    response = await post_mcp(protected, bearer="tok-wrong-aud")
+    response = await post_mcp(protected, bearer=bearer)
+
+    assert response.status_code == 401
+    assert parse_www_authenticate(response.headers["www-authenticate"])["error"] == "invalid_token"
+
+
+@requirement("hosting:auth:aud-validation")
+async def test_a_token_issued_for_this_resource_is_served(protected: httpx2.AsyncClient) -> None:
+    """The other half: a token the verifier reports as issued for `resource_server_url` passes the
+    gate and the MCP endpoint answers the initialize request."""
+    response = await post_mcp(protected, bearer="tok-valid")
 
     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
-    # out of the buffered text to prove the MCP endpoint actually answered the initialize request.
+    # Finite SSE body: pull out the JSON-RPC result to prove the endpoint actually answered.
     [data] = [line.removeprefix("data: ") for line in response.text.splitlines() if line.startswith("data: ")]
     assert "protocolVersion" in JSONRPCResponse.model_validate_json(data).result