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

docs/deprecated.md

@@ -136,6 +136,7 @@ These are not spec changes, only SDK usage with a better replacement. They warn
 | Deprecated | What you do instead |
 |---|---|
 | `FuncMetadata.call_fn_with_arg_validation()` | `FuncMetadata.validate_arguments()` and then `FuncMetadata.call_fn()`. Only code that drives `FuncMetadata` directly (a custom `Tool` subclass, say) ever called it. |
+| `AuthSettings(resource_server_url=...)` without `validate_token_resource=` | Set it: `True` has the server refuse bearer tokens your verifier does not report as issued for `resource_server_url`, `False` says your verifier checks the token's audience itself (see **[Authorization](run/authorization.md#a-token-verifier)**). Unset behaves as `False`; 3.0 makes `True` the default whenever `resource_server_url` is set. |
 | `ClientCredentialsOAuthProvider(...)` or `PrivateKeyJWTOAuthProvider(...)` without `issuer=` | Pass `issuer=` naming the authorization server that issued the credentials (see **[Writing OAuth clients](client/oauth-clients.md#machine-to-machine)**). Without it the MCP server decides which authorization server receives them; 3.0 makes the keyword required. |
 
 ## Recap

docs/run/authorization.md

@@ -18,19 +18,23 @@ That's the whole triangle. Everything on this page is the middle bullet.
 
 The SDK has no opinion about what a valid token looks like. You tell it, by implementing **`TokenVerifier`**:
 
-```python title="server.py" hl_lines="12-14 19-24"
+```python title="server.py" hl_lines="14-16 21-27"
 --8<-- "docs_src/authorization/tutorial001.py"
 ```
 
 * `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; each entry records the resource it was issued for. A real one verifies a JWT signature or calls the authorization server's token-introspection endpoint, and reports who the token was issued for (its `aud`) in `AccessToken.resource`. 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 whose `AccessToken.resource` is not `resource_server_url`. Leaving it unset while `resource_server_url` is set warns (`MCPDeprecationWarning`) 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`.
 
 !!! tip
     `examples/servers/simple-auth/` in the SDK repository has an `IntrospectionTokenVerifier` that calls
@@ -86,7 +90,7 @@ This document is how a client that has never heard of your server finds its way
 
 Inside any handler, **`get_access_token()`** is the `AccessToken` your verifier returned for the current request:
 
-```python title="server.py" hl_lines="4 32-35"
+```python title="server.py" hl_lines="4 35-38"
 --8<-- "docs_src/authorization/tutorial002.py"
 ```
 

docs_src/authorization/tutorial001.py

@@ -4,8 +4,10 @@
 from mcp.server.auth.provider import AccessToken, TokenVerifier
 from mcp.server.auth.settings import AuthSettings
 
+RESOURCE = "http://127.0.0.1:8000/mcp"
+
 KNOWN_TOKENS = {
-    "alice-token": AccessToken(token="alice-token", client_id="alice", scopes=["notes:read"]),
+    "alice-token": AccessToken(token="alice-token", client_id="alice", scopes=["notes:read"], resource=RESOURCE),
 }
 
 
@@ -19,8 +21,9 @@ async def verify_token(self, token: str) -> AccessToken | None:
     token_verifier=StaticTokenVerifier(),
     auth=AuthSettings(
         issuer_url=AnyHttpUrl("https://auth.example.com"),
-        resource_server_url=AnyHttpUrl("http://127.0.0.1:8000/mcp"),
+        resource_server_url=AnyHttpUrl(RESOURCE),
         required_scopes=["notes:read"],
+        validate_token_resource=True,
     ),
 )
 

docs_src/authorization/tutorial002.py

@@ -5,8 +5,10 @@
 from mcp.server.auth.provider import AccessToken, TokenVerifier
 from mcp.server.auth.settings import AuthSettings
 
+RESOURCE = "http://127.0.0.1:8000/mcp"
+
 KNOWN_TOKENS = {
-    "alice-token": AccessToken(token="alice-token", client_id="alice", scopes=["notes:read"]),
+    "alice-token": AccessToken(token="alice-token", client_id="alice", scopes=["notes:read"], resource=RESOURCE),
 }
 
 
@@ -20,8 +22,9 @@ async def verify_token(self, token: str) -> AccessToken | None:
     token_verifier=StaticTokenVerifier(),
     auth=AuthSettings(
         issuer_url=AnyHttpUrl("https://auth.example.com"),
-        resource_server_url=AnyHttpUrl("http://127.0.0.1:8000/mcp"),
+        resource_server_url=AnyHttpUrl(RESOURCE),
         required_scopes=["notes:read"],
+        validate_token_resource=True,
     ),
 )
 

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()

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

@@ -24,8 +24,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,
     ),
 )
 

examples/stories/_shared/auth.py

@@ -169,4 +169,5 @@ def auth_settings(
         required_scopes=scopes,
         client_registration_options=ClientRegistrationOptions(enabled=True, valid_scopes=scopes, default_scopes=scopes),
         identity_assertion_enabled=identity_assertion_enabled,
+        validate_token_resource=True,
     )

examples/stories/bearer_auth/server.py

@@ -28,6 +28,7 @@ async def verify_token(self, token: str) -> AccessToken | None:
             client_id="demo-client",
             scopes=[REQUIRED_SCOPE],
             expires_at=int(time.time()) + 3600,
+            resource=RESOURCE_URL,
             subject="demo-user",
         )
 
@@ -40,6 +41,7 @@ def build_app() -> Starlette:
             issuer_url=AnyHttpUrl(ISSUER),
             resource_server_url=AnyHttpUrl(RESOURCE_URL),
             required_scopes=[REQUIRED_SCOPE],
+            validate_token_resource=True,
         ),
     )
 

examples/stories/bearer_auth/server_lowlevel.py

@@ -46,6 +46,7 @@ async def call_tool(ctx: ServerRequestContext[Any], params: types.CallToolReques
             issuer_url=AnyHttpUrl(ISSUER),
             resource_server_url=AnyHttpUrl(RESOURCE_URL),
             required_scopes=[REQUIRED_SCOPE],
+            validate_token_resource=True,
         ),
         token_verifier=StaticTokenVerifier(),
         transport_security=NO_DNS_REBIND,

examples/stories/oauth_client_credentials/server.py

@@ -66,7 +66,14 @@ async def token_endpoint(request: Request) -> JSONResponse:
         if creds != f"{DEMO_CLIENT_ID}:{DEMO_CLIENT_SECRET}":
             return JSONResponse({"error": "invalid_client"}, status_code=401)
         access = f"access_{secrets.token_hex(16)}"
-        issued[access] = AccessToken(token=access, client_id=DEMO_CLIENT_ID, scopes=[DEMO_SCOPE], expires_at=None)
+        resource = form.get("resource")  # RFC 8707: bind the token to the resource the client asked for
+        issued[access] = AccessToken(
+            token=access,
+            client_id=DEMO_CLIENT_ID,
+            scopes=[DEMO_SCOPE],
+            expires_at=None,
+            resource=resource if isinstance(resource, str) else None,
+        )
         body = OAuthToken(access_token=access, token_type="Bearer", expires_in=3600, scope=DEMO_SCOPE)
         return JSONResponse(body.model_dump(exclude_none=True), headers={"cache-control": "no-store"})
 

examples/stories/oauth_client_credentials/server_lowlevel.py

@@ -63,7 +63,14 @@ async def token_endpoint(request: Request) -> JSONResponse:
         if creds != f"{DEMO_CLIENT_ID}:{DEMO_CLIENT_SECRET}":
             return JSONResponse({"error": "invalid_client"}, status_code=401)
         access = f"access_{secrets.token_hex(16)}"
-        issued[access] = AccessToken(token=access, client_id=DEMO_CLIENT_ID, scopes=[DEMO_SCOPE], expires_at=None)
+        resource = form.get("resource")  # RFC 8707: bind the token to the resource the client asked for
+        issued[access] = AccessToken(
+            token=access,
+            client_id=DEMO_CLIENT_ID,
+            scopes=[DEMO_SCOPE],
+            expires_at=None,
+            resource=resource if isinstance(resource, str) else None,
+        )
         body = OAuthToken(access_token=access, token_type="Bearer", expires_in=3600, scope=DEMO_SCOPE)
         return JSONResponse(body.model_dump(exclude_none=True), headers={"cache-control": "no-store"})
 

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 %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:
     """Middleware that requires a valid Bearer token in the Authorization header.

src/mcp/server/auth/provider.py

@@ -46,6 +46,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
 
 
@@ -124,7 +125,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,9 @@
-from pydantic import AnyHttpUrl, BaseModel, ConfigDict, Field
+import warnings
+
+from pydantic import AnyHttpUrl, BaseModel, ConfigDict, Field, model_validator
+from typing_extensions import Self
+
+from mcp.shared.exceptions import MCPDeprecationWarning
 
 
 class ClientRegistrationOptions(BaseModel):
@@ -40,3 +45,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.",
+                MCPDeprecationWarning,
+                stacklevel=3,
+            )
+        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/docs_src/test_identity_assertion.py

@@ -151,6 +151,7 @@ async def test_the_whole_grant_is_one_token_request() -> None:
             issuer_url=AnyHttpUrl(tutorial002.ISSUER),
             resource_server_url=AnyHttpUrl(MCP_SERVER_URL),
             required_scopes=["notes:read"],
+            validate_token_resource=True,
         ),
     )
 

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 conformant 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."
             ),
         ),
     ),