Add subject and claims to AccessToken

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

@@ -120,6 +120,7 @@ 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
             }
         )
 

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

@@ -181,6 +181,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
 
@@ -219,6 +220,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,16 @@ 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."""
 
 
 RegistrationErrorCode = Literal[

src/mcp/server/mcpserver/context.py

@@ -5,6 +5,7 @@
 
 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,
@@ -210,9 +211,40 @@ async def log(
 
     @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 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
+        """
         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,6 +1,46 @@
 """Tests for mcp.server.auth.provider module."""
 
-from mcp.server.auth.provider import construct_redirect_uri
+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"
 
 
 def test_construct_redirect_uri_no_existing_params():

tests/server/mcpserver/auth/test_context_subject.py

@@ -0,0 +1,66 @@
+"""Context.subject and Context.claims read from the request's access token."""
+
+from mcp.server.auth.middleware.auth_context import auth_context_var
+from mcp.server.auth.middleware.bearer_auth import AuthenticatedUser
+from mcp.server.auth.provider import AccessToken
+from mcp.server.mcpserver import Context
+
+
+def test_subject_is_none_when_unauthenticated():
+    assert Context().subject is None
+
+
+def test_subject_is_none_when_token_has_no_subject():
+    user = AuthenticatedUser(AccessToken(token="t", client_id="c", scopes=[]))
+    cv_token = auth_context_var.set(user)
+    try:
+        assert Context().subject is None
+    finally:
+        auth_context_var.reset(cv_token)
+
+
+def test_subject_reads_from_access_token():
+    user = AuthenticatedUser(AccessToken(token="t", client_id="c", scopes=[], subject="user-123"))
+    cv_token = auth_context_var.set(user)
+    try:
+        assert Context().subject == "user-123"
+    finally:
+        auth_context_var.reset(cv_token)
+
+
+def test_subject_tracks_current_auth_context():
+    ctx = Context()
+    assert ctx.subject is None
+
+    alice = AuthenticatedUser(AccessToken(token="a", client_id="c", scopes=[], subject="alice"))
+    cv_token = auth_context_var.set(alice)
+    try:
+        assert ctx.subject == "alice"
+    finally:
+        auth_context_var.reset(cv_token)
+
+    assert ctx.subject is None
+
+
+def test_claims_is_none_when_unauthenticated():
+    assert Context().claims is None
+
+
+def test_claims_is_none_when_token_has_no_claims():
+    user = AuthenticatedUser(AccessToken(token="t", client_id="c", scopes=[]))
+    cv_token = auth_context_var.set(user)
+    try:
+        assert Context().claims is None
+    finally:
+        auth_context_var.reset(cv_token)
+
+
+def test_claims_reads_from_access_token():
+    user = AuthenticatedUser(
+        AccessToken(token="t", client_id="c", scopes=[], claims={"iss": "https://auth.example.com"})
+    )
+    cv_token = auth_context_var.set(user)
+    try:
+        assert Context().claims == {"iss": "https://auth.example.com"}
+    finally:
+        auth_context_var.reset(cv_token)