Scope down to model fields and add subject to auth integration test

- Drop Context.subject/Context.claims; handlers use get_access_token()
  directly for now. A designed Context auth surface can follow separately.
- Collapse AccessToken.subject/claims docstrings to one-line comments to
  match surrounding fields.
- Move the client_id TODO to a # comment and tighten its docstring note.
- Replace the trivial model and contextvar unit tests with subject
  propagation through MockOAuthProvider in test_auth_integration.py, so
  test_authorization_get now asserts subject survives the full
  /authorize -> /token -> refresh flow over HTTP.

Author: maxisbey

src/mcp/server/auth/provider.py

@@ -42,16 +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
-    """The resource owner this token was issued on behalf of — typically the
-    `sub` from a JWT (RFC 9068) or introspection response (RFC 7662). Token
-    verifiers should populate this whenever an end-user is involved so request
-    handlers and transports can distinguish users that share an OAuth client.
-    For `client_credentials` grants there is no end-user; `sub` may then
-    identify the client itself (RFC 9068 §2.2) or be absent, depending on the
-    authorization server."""
-    claims: dict[str, Any] | None = None
-    """Additional verified claims (e.g. `iss`, `act`) for request handlers."""
+    subject: str | None = None  # RFC 7662/9068 `sub`: resource owner the token was issued for
+    claims: dict[str, Any] | None = None  # additional verified claims (e.g. `iss`, `act`)
 
 
 RegistrationErrorCode = Literal[

src/mcp/server/mcpserver/context.py

@@ -5,7 +5,6 @@
 
 from pydantic import AnyUrl, BaseModel
 
-from mcp.server.auth.middleware.auth_context import get_access_token
 from mcp.server.context import LifespanContextT, RequestT, ServerRequestContext
 from mcp.server.elicitation import (
     ElicitationResult,
@@ -209,42 +208,16 @@ 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.
 
-        Note: this reads from the MCP request's `_meta` params, not from the
-        OAuth bearer token. It is unrelated to `subject` and `claims` below,
-        which come from the authenticated `AccessToken`. For the OAuth
-        `client_id`, use `get_access_token().client_id`.
-
-        TODO(maxisbey): see if this is needed otherwise remove
+        Note: this reads from the MCP request's `_meta` params, not the OAuth
+        bearer token. For that, use `get_access_token().client_id`.
         """
         return self.request_context.meta.get("client_id") if self.request_context.meta else None  # pragma: no cover
 
-    @property
-    def subject(self) -> str | None:
-        """The authenticated resource owner (`sub`) for this request, if any.
-
-        Returns `AccessToken.subject` from the bearer token that authenticated
-        the current request, or `None` when the request is unauthenticated or
-        the token verifier did not populate a subject.
-        """
-        token = get_access_token()
-        return token.subject if token is not None else None
-
-    @property
-    def claims(self) -> dict[str, Any] | None:
-        """Additional verified claims from the bearer token, if any.
-
-        Returns `AccessToken.claims` for the current request so handlers can
-        read values like `iss` or `act` without importing `get_access_token`
-        and handling the raw token directly. `None` when unauthenticated or
-        when the token verifier did not populate claims.
-        """
-        token = get_access_token()
-        return token.claims if token is not None else None
-
     @property
     def request_id(self) -> str:
         """Get the unique ID for this request."""

tests/server/auth/test_provider.py

@@ -1,46 +1,6 @@
 """Tests for mcp.server.auth.provider module."""
 
-from pydantic import AnyUrl
-
-from mcp.server.auth.provider import AccessToken, AuthorizationCode, RefreshToken, construct_redirect_uri
-
-
-def test_access_token_subject_and_claims_default_to_none():
-    token = AccessToken(token="t", client_id="c", scopes=["read"])
-    assert token.subject is None
-    assert token.claims is None
-
-
-def test_access_token_carries_subject_and_claims():
-    token = AccessToken(
-        token="t",
-        client_id="c",
-        scopes=["read"],
-        subject="user-123",
-        claims={"iss": "https://auth.example.com", "act": {"sub": "gateway"}},
-    )
-    assert token.subject == "user-123"
-    assert token.claims is not None
-    assert token.claims["iss"] == "https://auth.example.com"
-
-
-def test_authorization_code_carries_subject():
-    code = AuthorizationCode(
-        code="x",
-        scopes=["read"],
-        expires_at=0.0,
-        client_id="c",
-        code_challenge="cc",
-        redirect_uri=AnyUrl("https://example.com/cb"),
-        redirect_uri_provided_explicitly=True,
-        subject="user-123",
-    )
-    assert code.subject == "user-123"
-
-
-def test_refresh_token_carries_subject():
-    refresh = RefreshToken(token="r", client_id="c", scopes=["read"], subject="user-123")
-    assert refresh.subject == "user-123"
+from mcp.server.auth.provider import construct_redirect_uri
 
 
 def test_construct_redirect_uri_no_existing_params():

tests/server/mcpserver/auth/test_auth_integration.py

@@ -53,6 +53,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
 
@@ -79,6 +80,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
@@ -108,6 +110,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
@@ -141,6 +144,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
@@ -169,6 +173,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:
@@ -832,6 +837,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(
@@ -852,6 +858,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",