fixes #1398: Add JWT claims to AccessToken (#1399)

Co-authored-by: Jeremiah Lowin <153965+jlowin@users.noreply.github.com>
This commit is contained in:
Panos Argyrakis 2025-08-09 20:58:09 -04:00 committed by GitHub
commit c3f4623690
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
7 changed files with 63 additions and 12 deletions

View file

@ -1,7 +1,11 @@
from __future__ import annotations
from typing import Any
from mcp.server.auth.provider import (
AccessToken as _SDKAccessToken,
)
from mcp.server.auth.provider import (
AccessToken,
AuthorizationCode,
OAuthAuthorizationServerProvider,
RefreshToken,
@ -21,6 +25,12 @@ from pydantic import AnyHttpUrl
from starlette.routing import Route
class AccessToken(_SDKAccessToken):
"""AccessToken that includes all JWT claims."""
claims: dict[str, Any] = {}
class AuthProvider(TokenVerifierProtocol):
"""Base class for all FastMCP authentication providers.

View file

@ -11,12 +11,12 @@ from authlib.jose import JsonWebKey, JsonWebToken
from authlib.jose.errors import JoseError
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric import rsa
from mcp.server.auth.provider import AccessToken
from pydantic import AnyHttpUrl, SecretStr
from pydantic_settings import BaseSettings, SettingsConfigDict
from typing_extensions import TypedDict
from fastmcp.server.auth import TokenVerifier
from fastmcp.server.auth.auth import AccessToken
from fastmcp.server.auth.registry import register_provider
from fastmcp.utilities.logging import get_logger
from fastmcp.utilities.types import NotSet, NotSetT
@ -448,6 +448,7 @@ class JWTVerifier(TokenVerifier):
client_id=str(client_id),
scopes=scopes,
expires_at=int(exp) if exp else None,
claims=claims,
)
except JoseError:
@ -535,4 +536,5 @@ class StaticTokenVerifier(TokenVerifier):
client_id=token_data["client_id"],
scopes=scopes,
expires_at=expires_at,
claims=token_data,
)

View file

@ -2,10 +2,13 @@ from __future__ import annotations
from typing import TYPE_CHECKING, ParamSpec, TypeVar
from mcp.server.auth.middleware.auth_context import get_access_token
from mcp.server.auth.provider import AccessToken
from mcp.server.auth.middleware.auth_context import (
get_access_token as _sdk_get_access_token,
)
from starlette.requests import Request
from fastmcp.server.auth.auth import AccessToken
if TYPE_CHECKING:
from fastmcp.server.context import Context
@ -94,3 +97,30 @@ def get_http_headers(include_all: bool = False) -> dict[str, str]:
return headers
except RuntimeError:
return {}
# --- Access Token ---
def get_access_token() -> AccessToken | None:
"""
Get the FastMCP access token from the current context.
Returns:
The access token if an authenticated user is available, None otherwise.
"""
#
obj = _sdk_get_access_token()
if obj is None or isinstance(obj, AccessToken):
return obj
# If the object is not a FastMCP AccessToken, convert it to one if the fields are compatible
# This is a workaround for the case where the SDK returns a different type
# If it fails, it will raise a TypeError
try:
return AccessToken(**obj.model_dump())
except Exception as e:
raise TypeError(
f"Expected fastmcp.server.auth.auth.AccessToken, got {type(obj).__name__}. "
"Ensure the SDK is using the correct AccessToken type."
) from e

View file

@ -141,15 +141,25 @@ class TestBearerTokenJWKS:
url="https://test.example.com/.well-known/jwks.json",
json=mock_jwks_data,
)
username = "test-user"
issuer = "https://test.example.com"
audience = "https://api.example.com"
token = rsa_key_pair.create_token(
subject="test-user",
issuer="https://test.example.com",
audience="https://api.example.com",
subject=username,
issuer=issuer,
audience=audience,
)
access_token = await jwks_provider.load_access_token(token)
assert access_token is not None
assert access_token.client_id == "test-user"
assert access_token.client_id == username
# ensure the raw claims are present - #1398
assert access_token.claims.get("sub") == username
assert access_token.claims.get("iss") == issuer
assert access_token.claims.get("aud") == audience
async def test_jwks_token_validation_with_invalid_key(
self,

View file

@ -1,10 +1,9 @@
import httpx
import pytest
from mcp.server.auth.provider import AccessToken
from pydantic import AnyHttpUrl
from fastmcp import FastMCP
from fastmcp.server.auth.auth import RemoteAuthProvider, TokenVerifier
from fastmcp.server.auth.auth import AccessToken, RemoteAuthProvider, TokenVerifier
class SimpleTokenVerifier(TokenVerifier):

View file

@ -1,9 +1,9 @@
"""Tests for StaticTokenVerifier integration with FastMCP."""
import httpx
from mcp.server.auth.provider import AccessToken
from fastmcp.server import FastMCP
from fastmcp.server.auth.auth import AccessToken
from fastmcp.server.auth.providers.jwt import StaticTokenVerifier

View file

@ -2,9 +2,9 @@
import pytest
from mcp.server.auth.middleware.bearer_auth import BearerAuthBackend
from mcp.server.auth.provider import AccessToken
from starlette.requests import HTTPConnection
from fastmcp.server.auth.auth import AccessToken
from fastmcp.server.auth.providers.jwt import JWTVerifier, RSAKeyPair