[v1.x] Add AuthSettings.validate_token_resource to check a bearer token's resource

v1.x backport. 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 turns
this on for FastMCP's SSE and Streamable HTTP apps; leaving it unset while
resource_server_url is set emits a DeprecationWarning and behaves as False,
and 3.0 makes True the default there. RefreshToken gains an optional
resource so a provider can carry the binding through the refresh grant.

TokenVerifier.verify_token's docstring and docs/authorization.md say where
the token's audience goes and when to enable the option versus checking the
audience in the verifier. The oauth_server snippet and the simple-auth
example set it.

Author: maxisbey

docs/authorization.md

@@ -40,8 +40,9 @@ mcp = FastMCP(
     # Auth settings for RFC 9728 Protected Resource Metadata
     auth=AuthSettings(
         issuer_url=AnyHttpUrl("https://auth.example.com"),  # Authorization Server URL
-        resource_server_url=AnyHttpUrl("http://localhost:3001"),  # This server's URL
+        resource_server_url=AnyHttpUrl("http://127.0.0.1:8000/mcp"),  # This server's URL (mcp.run() default)
         required_scopes=["user"],
+        validate_token_resource=True,
     ),
 )
 
@@ -74,6 +75,12 @@ For a complete example with separate Authorization Server and Resource Server im
 
 See [TokenVerifier](https://github.com/modelcontextprotocol/python-sdk/blob/v1.x/src/mcp/server/auth/provider.py) for more details on implementing token validation.
 
+A verifier should report who the token was issued for (its `aud`) in `AccessToken.resource`. `AuthSettings(validate_token_resource=True)` then refuses any token whose `resource` is not `resource_server_url`. Leaving it unset while `resource_server_url` is set warns (`DeprecationWarning`) and behaves as `False`; 3.0 makes `True` the default for resource servers.
+
+- Turn it on when your authorization server binds tokens to the `resource` the client requested, which MCP clients always send. Keep `resource_server_url` the exact URL clients connect to.
+- Leave it off when your authorization server uses its own audience identifiers (an Auth0 API identifier, an Entra application ID) and check `aud` in your verifier instead, returning `None` for a token that isn't for this server.
+- If `aud` is a list, put the entry that equals `resource_server_url` in `resource`.
+
 ## Client-Side Authentication
 
 The SDK includes [authorization support](https://modelcontextprotocol.io/specification/2025-11-25/basic/authorization) for connecting to protected MCP servers:

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

@@ -75,6 +75,7 @@ def create_resource_server(settings: ResourceServerSettings) -> FastMCP:
             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
         ),
     )
 

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

@@ -69,12 +69,19 @@ async def verify_token(self, token: str) -> AccessToken | None:
                     logger.warning(f"Token resource validation failed. Expected: {self.resource_url}")
                     return None
 
+                # `aud` may be a string or a list; report the entry naming this server when there is
+                # one, otherwise what the token was issued for, so the server can compare it.
+                aud: str | list[str] | None = data.get("aud")
+                audiences = aud if isinstance(aud, list) else [aud] if aud else []
+                own = self.resource_url.rstrip("/")
+                resource = next((a for a in audiences if a.rstrip("/") == own), audiences[0] if audiences else None)
+
                 return AccessToken(
                     token=token,
                     client_id=data.get("client_id", "unknown"),
                     scopes=data.get("scope", "").split() if data.get("scope") else [],
                     expires_at=data.get("exp"),
-                    resource=data.get("aud"),  # Include resource in token
+                    resource=resource,
                     subject=data.get("sub"),  # RFC 7662 subject (resource owner)
                     claims=data,
                 )

examples/snippets/servers/oauth_server.py

@@ -26,8 +26,9 @@ async def verify_token(self, token: str) -> AccessToken | None:
     # Auth settings for RFC 9728 Protected Resource Metadata
     auth=AuthSettings(
         issuer_url=AnyHttpUrl("https://auth.example.com"),  # Authorization Server URL
-        resource_server_url=AnyHttpUrl("http://localhost:3001"),  # This server's URL
+        resource_server_url=AnyHttpUrl("http://127.0.0.1:8000/mcp"),  # This server's URL (mcp.run() default)
         required_scopes=["user"],
+        validate_token_resource=True,
     ),
 )
 

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
 
+logger = logging.getLogger(__name__)
+
 
 class AuthenticatedUser(SimpleUser):
     """User with authentication info."""
@@ -46,10 +49,14 @@ def authorization_context(user: AuthenticatedUser) -> AuthorizationContext:
 class BearerAuthBackend(AuthenticationBackend):
     """
     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(
@@ -70,8 +77,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 %r 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:
     """

src/mcp/server/auth/provider.py

@@ -33,6 +33,7 @@ class RefreshToken(BaseModel):
     client_id: str
     scopes: list[str]
     expires_at: int | None = None
+    resource: str | None = None  # RFC 8707 resource indicator; propagate to refreshed AccessTokens
     subject: str | None = None  # resource owner; propagate to refreshed AccessTokens
 
 
@@ -97,7 +98,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: FastMCP doesn't render any of these types in the user response, so it's

src/mcp/server/auth/settings.py

@@ -1,4 +1,7 @@
-from pydantic import AnyHttpUrl, BaseModel, Field
+import warnings
+
+from pydantic import AnyHttpUrl, BaseModel, Field, model_validator
+from typing_extensions import Self
 
 
 class ClientRegistrationOptions(BaseModel):
@@ -28,3 +31,26 @@ 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 | None = Field(
+        default=None,
+        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; set it to False when your token "
+        "verifier checks the token's audience itself. With `resource_server_url` set, leaving it unset warns "
+        "and behaves as False; 3.0 makes True the default there.",
+    )
+
+    @model_validator(mode="after")
+    def _check_validate_token_resource(self) -> Self:
+        if self.validate_token_resource and self.resource_server_url is None:
+            raise ValueError("validate_token_resource requires resource_server_url")
+        if self.validate_token_resource is None and self.resource_server_url is not None:
+            warnings.warn(
+                "`AuthSettings.validate_token_resource` is not set, so bearer tokens are not checked "
+                "against `resource_server_url`; it will default to True in 3.0 when `resource_server_url` is "
+                "set. Set it to True to have the server refuse tokens issued for another resource, or to "
+                "False if your TokenVerifier validates the token's audience itself.",
+                DeprecationWarning,
+                stacklevel=3,
+            )
+        return self

src/mcp/server/fastmcp/server.py

@@ -881,7 +881,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
@@ -999,7 +1004,12 @@ def streamable_http_app(self) -> Starlette:
                 middleware = [
                     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,
+                        ),
                     ),
                     Middleware(AuthContextMiddleware),
                 ]

tests/server/auth/middleware/test_bearer_auth.py

@@ -6,6 +6,7 @@
 from typing import Any, cast
 
 import pytest
+from pydantic import AnyHttpUrl
 from starlette.authentication import AuthCredentials
 from starlette.datastructures import Headers
 from starlette.requests import Request
@@ -265,6 +266,56 @@ async def test_mixed_case_authorization_header(
         assert user.access_token == valid_access_token
 
 
+class SingleTokenVerifier:
+    """A `TokenVerifier` that knows exactly one token."""
+
+    def __init__(self, access_token: AccessToken) -> None:
+        self.access_token = access_token
+
+    async def verify_token(self, token: str) -> AccessToken | None:
+        return self.access_token if token == self.access_token.token else None
+
+
+RS = "https://api.example.com/mcp"
+
+
+@pytest.mark.anyio
+@pytest.mark.parametrize(
+    ("resource_server_url", "token_resource", "accepted"),
+    [
+        (None, "https://other.example.com/mcp", True),  # nothing configured to compare against
+        (None, None, True),
+        (RS, None, False),  # the verifier did not report what the token was issued for
+        (RS, RS, True),
+        (RS, RS + "/", True),
+        (RS, "https://API.EXAMPLE.COM:443/mcp", True),  # same URL, different spelling
+        (RS, "https://api.example.com", False),
+        (RS, RS + "/child", False),
+        (RS, "https://api.example.com/other", False),
+        (RS, "https://other.example.com/mcp", False),
+        (RS, "api.example.com", False),  # not a URL
+    ],
+)
+async def test_backend_accepts_only_tokens_issued_for_its_resource(
+    resource_server_url: str | None, token_resource: str | None, accepted: bool
+):
+    """With `resource_server_url` set, only a token whose `resource` (RFC 8707) is that URL is
+    accepted and anything else is treated like an unrecognized token (spec-mandated audience
+    check); without it the verifier's answer stands (SDK-defined, the default wiring)."""
+    token = AccessToken(token="t", client_id="c", scopes=["read"], resource=token_resource)
+    backend = BearerAuthBackend(
+        SingleTokenVerifier(token),
+        resource_server_url=AnyHttpUrl(resource_server_url) if resource_server_url else None,
+    )
+
+    result = await backend.authenticate(Request({"type": "http", "headers": [(b"authorization", b"Bearer t")]}))
+
+    if accepted:
+        assert result is not None and result[1].access_token == token
+    else:
+        assert result is None
+
+
 @pytest.mark.anyio
 class TestRequireAuthMiddleware:
     """Tests for the RequireAuthMiddleware class."""

tests/server/auth/test_settings.py

@@ -0,0 +1,33 @@
+import warnings
+
+import pytest
+from pydantic import AnyHttpUrl, ValidationError
+
+from mcp.server.auth.settings import AuthSettings
+
+ISSUER = AnyHttpUrl("https://auth.example.com")
+RESOURCE = AnyHttpUrl("https://mcp.example.com/mcp")
+
+
+def test_validate_token_resource_requires_a_resource_server_url():
+    """SDK-defined: asking the bearer gate to compare tokens against `resource_server_url` without
+    configuring one is refused at construction time rather than silently comparing nothing."""
+    AuthSettings(issuer_url=ISSUER, resource_server_url=RESOURCE, validate_token_resource=True)
+    with pytest.raises(ValidationError, match="validate_token_resource requires resource_server_url"):
+        AuthSettings(issuer_url=ISSUER, resource_server_url=None, validate_token_resource=True)
+
+
+def test_leaving_validate_token_resource_unset_warns_when_a_resource_server_url_is_configured():
+    """Unset behaves as False but says so: a resource server that has not chosen gets a
+    `DeprecationWarning` pointing at its own `AuthSettings(...)` call (3.0 flips the default)."""
+    with pytest.warns(DeprecationWarning, match="validate_token_resource") as record:
+        settings = AuthSettings(issuer_url=ISSUER, resource_server_url=RESOURCE)
+    assert settings.validate_token_resource is None
+    assert record[0].filename == __file__
+
+
+@pytest.mark.parametrize("kwargs", [{"validate_token_resource": False}, {"resource_server_url": None}])
+def test_an_explicit_choice_or_no_resource_server_url_does_not_warn(kwargs: dict[str, object]):
+    with warnings.catch_warnings():
+        warnings.simplefilter("error")
+        AuthSettings.model_validate({"issuer_url": ISSUER, "resource_server_url": RESOURCE, **kwargs})