Add subject and claims to AccessToken

Adds optional `subject` and `claims` fields to `AccessToken` so token
verifiers can surface the resource owner (`sub`) and any additional
claims to request handlers. `subject` is also added to
`AuthorizationCode` and `RefreshToken` so the value can be carried
through code-for-token exchange and refresh.

The simple-auth example threads the subject from login through the
introspection response and back into the verifier; the integration
test exercises the full code → token → refresh → load chain.

v1.x backport of #2686. Closes #1038.

Reported-by: Thomas Steinacher <@thomasst>
Reported-by: Yukuan Jia <@yukuanj>
Reported-by: Shivam Aggarwal <@shivama205>

Author: maxisbey

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

@@ -123,6 +123,8 @@ async def introspect_handler(request: Request) -> Response:
                 "iat": int(time.time()),
                 "token_type": "Bearer",
                 "aud": access_token.resource,  # RFC 8707 audience claim
+                "sub": access_token.subject,  # RFC 7662 subject
+                "iss": str(server_settings.server_url),
             }
         )
 

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

@@ -186,6 +186,7 @@ async def handle_simple_callback(self, username: str, password: str, state: str)
             scopes=[self.settings.mcp_scope],
             code_challenge=code_challenge,
             resource=resource,  # RFC 8707
+            subject=username,
         )
         self.auth_codes[new_code] = auth_code
 
@@ -224,6 +225,7 @@ async def exchange_authorization_code(
             scopes=authorization_code.scopes,
             expires_at=int(time.time()) + 3600,
             resource=authorization_code.resource,  # RFC 8707
+            subject=authorization_code.subject,
         )
 
         # Store user data mapping for this token

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

@@ -75,6 +75,8 @@ async def verify_token(self, token: str) -> AccessToken | None:
                     scopes=data.get("scope", "").split() if data.get("scope") else [],
                     expires_at=data.get("exp"),
                     resource=data.get("aud"),  # Include resource in token
+                    subject=data.get("sub"),  # RFC 7662 subject (resource owner)
+                    claims=data,
                 )
             except Exception as e:
                 logger.warning(f"Token introspection failed: {e}")

src/mcp/server/auth/provider.py

@@ -1,5 +1,5 @@
 from dataclasses import dataclass
-from typing import Generic, Literal, Protocol, TypeVar
+from typing import Any, Generic, Literal, Protocol, TypeVar
 from urllib.parse import parse_qs, urlencode, urlparse, urlunparse
 
 from pydantic import AnyUrl, BaseModel
@@ -25,13 +25,15 @@ class AuthorizationCode(BaseModel):
     redirect_uri: AnyUrl
     redirect_uri_provided_explicitly: bool
     resource: str | None = None  # RFC 8707 resource indicator
+    subject: str | None = None  # resource owner; propagate to the issued AccessToken
 
 
 class RefreshToken(BaseModel):
     token: str
     client_id: str
     scopes: list[str]
     expires_at: int | None = None
+    subject: str | None = None  # resource owner; propagate to refreshed AccessTokens
 
 
 class AccessToken(BaseModel):
@@ -40,6 +42,8 @@ class AccessToken(BaseModel):
     scopes: list[str]
     expires_at: int | None = None
     resource: str | None = None  # RFC 8707 resource indicator
+    subject: str | None = None  # RFC 7662/9068 `sub`: resource owner; unique only per issuer
+    claims: dict[str, Any] | None = None  # additional claims (e.g. `iss`, `act`)
 
 
 RegistrationErrorCode = Literal[

src/mcp/server/fastmcp/server.py

@@ -1282,9 +1282,14 @@ async def log(
             related_request_id=self.request_id,
         )
 
+    # TODO(maxisbey): see if this is needed otherwise remove
     @property
     def client_id(self) -> str | None:
-        """Get the client ID if available."""
+        """Get the client ID if available.
+
+        Note: this reads from the MCP request's `_meta` params, not the OAuth
+        bearer token. For that, use `get_access_token().client_id`.
+        """
         return (
             getattr(self.request_context.meta, "client_id", None) if self.request_context.meta else None
         )  # pragma: no cover

tests/server/fastmcp/auth/test_auth_integration.py

@@ -54,6 +54,7 @@ async def authorize(self, client: OAuthClientInformationFull, params: Authorizat
             redirect_uri_provided_explicitly=params.redirect_uri_provided_explicitly,
             expires_at=time.time() + 300,
             scopes=params.scopes or ["read", "write"],
+            subject="test-user",
         )
         self.auth_codes[code.code] = code
 
@@ -80,6 +81,7 @@ async def exchange_authorization_code(
             client_id=client.client_id,
             scopes=authorization_code.scopes,
             expires_at=int(time.time()) + 3600,
+            subject=authorization_code.subject,
         )
 
         self.refresh_tokens[refresh_token] = access_token
@@ -109,6 +111,7 @@ async def load_refresh_token(self, client: OAuthClientInformationFull, refresh_t
             client_id=token_info.client_id,
             scopes=token_info.scopes,
             expires_at=token_info.expires_at,
+            subject=token_info.subject,
         )
 
         return refresh_obj
@@ -142,6 +145,7 @@ async def exchange_refresh_token(
             client_id=client.client_id,
             scopes=scopes or token_info.scopes,
             expires_at=int(time.time()) + 3600,
+            subject=refresh_token.subject,
         )
 
         self.refresh_tokens[new_refresh_token] = new_access_token
@@ -170,6 +174,7 @@ async def load_access_token(self, token: str) -> AccessToken | None:
             client_id=token_info.client_id,
             scopes=token_info.scopes,
             expires_at=token_info.expires_at,
+            subject=token_info.subject,
         )
 
     async def revoke_token(self, token: AccessToken | RefreshToken) -> None:
@@ -783,6 +788,7 @@ async def test_authorization_get(
         assert auth_info.client_id == client_info["client_id"]
         assert "read" in auth_info.scopes
         assert "write" in auth_info.scopes
+        assert auth_info.subject == "test-user"
 
         # 6. Refresh the token
         response = await test_client.post(
@@ -803,6 +809,10 @@ async def test_authorization_get(
         assert new_token_response["access_token"] != access_token
         assert new_token_response["refresh_token"] != refresh_token
 
+        refreshed_auth_info = await mock_oauth_provider.load_access_token(new_token_response["access_token"])
+        assert refreshed_auth_info
+        assert refreshed_auth_info.subject == "test-user"
+
         # 7. Revoke the token
         response = await test_client.post(
             "/revoke",