Add jwks tests

pyproject.toml

@@ -52,6 +52,7 @@ dev = [
     "pytest-cov>=6.1.1",
     "pytest-env>=1.1.5",
     "pytest-flakefinder",
+    "pytest-httpx>=0.35.0",
     "pytest-report>=0.2.1",
     "pytest-timeout>=2.4.0",
     "pytest-xdist>=3.6.1",

src/fastmcp/server/auth/bearer.py

@@ -1,256 +0,0 @@
-"""
-Simple JWT Bearer Token validation for hosted MCP servers.
-
-Uses RS256 (asymmetric) where your control plane signs with a private key
-and hosted MCP servers validate with the corresponding public key.
-
-Example usage:
-# Static public key
-provider = BearerTokenValidatorProvider(
-    public_key='''-----BEGIN PUBLIC KEY-----
-    MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEA...
-    -----END PUBLIC KEY-----''',
-    issuer="https://auth.yourservice.com"
-)
-
-# Or JWKS URI (recommended for production - allows key rotation)
-provider = BearerTokenValidatorProvider(
-    jwks_uri="https://auth.yourservice.com/.well-known/jwks.json",
-    issuer="https://auth.yourservice.com"
-)
-"""
-
-import time
-from typing import Any
-
-import httpx
-from authlib.jose import JsonWebKey, JsonWebToken
-from authlib.jose.errors import JoseError
-from mcp.server.auth.provider import (
-    AccessToken,
-    AuthorizationCode,
-    AuthorizationParams,
-    RefreshToken,
-)
-from mcp.shared.auth import (
-    OAuthClientInformationFull,
-    OAuthToken,
-)
-
-from fastmcp.server.auth.auth import (
-    ClientRegistrationOptions,
-    OAuthProvider,
-    RevocationOptions,
-)
-
-
-class BearerTokenValidatorProvider(OAuthProvider):
-    """
-    Simple JWT Bearer Token validator for hosted MCP servers.
-    Uses RS256 asymmetric encryption. Supports either static public key
-    or JWKS URI for key rotation.
-    """
-
-    def __init__(
-        self,
-        issuer: str,
-        public_key: str | None = None,
-        jwks_uri: str | None = None,
-        audience: str | None = None,
-        required_scopes: list[str] | None = None,
-    ):
-        """
-        Initialize the provider.
-
-        Args:
-            issuer: Expected issuer claim (your control plane)
-            public_key: RSA public key in PEM format (for static key)
-            jwks_uri: URI to fetch keys from (for key rotation)
-            audience: Expected audience claim (optional)
-            required_scopes: List of required scopes for access
-        """
-        if not (public_key or jwks_uri):
-            raise ValueError("Either public_key or jwks_uri must be provided")
-        if public_key and jwks_uri:
-            raise ValueError("Provide either public_key or jwks_uri, not both")
-
-        super().__init__(
-            issuer_url=issuer,
-            client_registration_options=ClientRegistrationOptions(enabled=False),
-            revocation_options=RevocationOptions(enabled=False),
-            required_scopes=required_scopes,
-        )
-
-        self.issuer = issuer
-        self.audience = audience
-        self.public_key = public_key
-        self.jwks_uri = jwks_uri
-        self.jwt = JsonWebToken(["RS256"])
-
-        # Simple JWKS cache
-        self._jwks_cache: dict[str, str] = {}
-        self._jwks_cache_time: float = 0
-        self._cache_ttl = 3600  # 1 hour
-
-    async def _get_verification_key(self, token: str) -> str:
-        """Get the verification key for the token."""
-        if self.public_key:
-            return self.public_key
-
-        # Extract kid from token header for JWKS lookup
-        try:
-            import base64
-            import json
-
-            header_b64 = token.split(".")[0]
-            header_b64 += "=" * (4 - len(header_b64) % 4)  # Add padding
-            header = json.loads(base64.urlsafe_b64decode(header_b64))
-            kid = header.get("kid")
-
-            if not kid:
-                raise ValueError("Token missing key ID (kid)")
-
-            return await self._get_jwks_key(kid)
-
-        except Exception as e:
-            raise ValueError(f"Failed to extract key ID from token: {e}")
-
-    async def _get_jwks_key(self, kid: str) -> str:
-        """Fetch key from JWKS with simple caching."""
-        if not self.jwks_uri:
-            raise ValueError("JWKS URI not configured")
-
-        current_time = time.time()
-
-        # Check cache
-        if (
-            current_time - self._jwks_cache_time < self._cache_ttl
-            and kid in self._jwks_cache
-        ):
-            return self._jwks_cache[kid]
-
-        # Fetch JWKS
-        try:
-            async with httpx.AsyncClient() as client:
-                response = await client.get(self.jwks_uri)
-                response.raise_for_status()
-                jwks_data = response.json()
-
-            # Cache all keys
-            self._jwks_cache = {}
-            for key_data in jwks_data.get("keys", []):
-                key_kid = key_data.get("kid")
-                if key_kid:
-                    jwk = JsonWebKey.import_key(key_data)
-                    self._jwks_cache[key_kid] = jwk.get_public_key()
-
-            self._jwks_cache_time = current_time
-
-            if kid not in self._jwks_cache:
-                raise ValueError(f"Key ID '{kid}' not found in JWKS")
-
-            return self._jwks_cache[kid]
-
-        except Exception as e:
-            raise ValueError(f"Failed to fetch JWKS: {e}")
-
-    async def load_access_token(self, token: str) -> AccessToken | None:
-        """
-        Validates the provided JWT bearer token.
-
-        Args:
-            token: The JWT token string to validate
-
-        Returns:
-            AccessToken object if valid, None if invalid or expired
-        """
-        try:
-            # Get verification key (static or from JWKS)
-            verification_key = await self._get_verification_key(token)
-
-            # Decode and verify the JWT token
-            claims = self.jwt.decode(token, verification_key)
-
-            # Validate expiration
-            exp = claims.get("exp")
-            if exp and exp < time.time():
-                return None
-
-            # Validate issuer
-            if claims.get("iss") != self.issuer:
-                return None
-
-            # Validate audience if configured
-            if self.audience:
-                aud = claims.get("aud")
-                if isinstance(aud, list):
-                    if self.audience not in aud:
-                        return None
-                elif aud != self.audience:
-                    return None
-
-            # Extract claims
-            client_id = claims.get("sub") or claims.get("client_id") or "unknown"
-            scopes = self._extract_scopes(claims)
-
-            return AccessToken(
-                token=token,
-                client_id=str(client_id),
-                scopes=scopes,
-                expires_at=int(exp) if exp else None,
-            )
-
-        except JoseError:
-            return None
-        except Exception:
-            return None
-
-    def _extract_scopes(self, claims: dict[str, Any]) -> list[str]:
-        """Extract scopes from JWT claims."""
-        scope_claim = claims.get("scope", "")
-        if isinstance(scope_claim, str):
-            return scope_claim.split()
-        elif isinstance(scope_claim, list):
-            return scope_claim
-        return []
-
-    # --- Unused OAuth server methods ---
-    async def get_client(self, client_id: str) -> OAuthClientInformationFull | None:
-        raise NotImplementedError("Client management not supported")
-
-    async def register_client(self, client_info: OAuthClientInformationFull) -> None:
-        raise NotImplementedError("Client registration not supported")
-
-    async def authorize(
-        self, client: OAuthClientInformationFull, params: AuthorizationParams
-    ) -> str:
-        raise NotImplementedError("Authorization flow not supported")
-
-    async def load_authorization_code(
-        self, client: OAuthClientInformationFull, authorization_code: str
-    ) -> AuthorizationCode | None:
-        raise NotImplementedError("Authorization code flow not supported")
-
-    async def exchange_authorization_code(
-        self, client: OAuthClientInformationFull, authorization_code: AuthorizationCode
-    ) -> OAuthToken:
-        raise NotImplementedError("Authorization code exchange not supported")
-
-    async def load_refresh_token(
-        self, client: OAuthClientInformationFull, refresh_token: str
-    ) -> RefreshToken | None:
-        raise NotImplementedError("Refresh token flow not supported")
-
-    async def exchange_refresh_token(
-        self,
-        client: OAuthClientInformationFull,
-        refresh_token: RefreshToken,
-        scopes: list[str],
-    ) -> OAuthToken:
-        raise NotImplementedError("Refresh token exchange not supported")
-
-    async def revoke_token(
-        self,
-        token: AccessToken | RefreshToken,
-    ) -> None:
-        raise NotImplementedError("Token revocation not supported")

src/fastmcp/server/auth/providers/bearer.py

@@ -6,7 +6,7 @@ and hosted MCP servers validate with the corresponding public key.
 
 Example usage:
 # Static public key
-provider = BearerTokenValidatorProvider(
+provider = BearerAuthProvider(
     public_key='''-----BEGIN PUBLIC KEY-----
     MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEA...
     -----END PUBLIC KEY-----''',
@@ -14,7 +14,7 @@ provider = BearerTokenValidatorProvider(
 )
 
 # Or JWKS URI (recommended for production - allows key rotation)
-provider = BearerTokenValidatorProvider(
+provider = Bear(
     jwks_uri="https://auth.yourservice.com/.well-known/jwks.json",
     issuer="https://auth.yourservice.com"
 )
@@ -22,7 +22,7 @@ provider = BearerTokenValidatorProvider(
 
 import time
 from dataclasses import dataclass
-from typing import Any
+from typing import Any, TypedDict
 
 import httpx
 from authlib.jose import JsonWebKey, JsonWebToken
@@ -48,6 +48,25 @@ from fastmcp.server.auth.auth import (
 )
 
 
+class JWKData(TypedDict, total=False):
+    """JSON Web Key data structure."""
+
+    kty: str  # Key type (e.g., "RSA") - required
+    kid: str  # Key ID (optional but recommended)
+    use: str  # Usage (e.g., "sig")
+    alg: str  # Algorithm (e.g., "RS256")
+    n: str  # Modulus (for RSA keys)
+    e: str  # Exponent (for RSA keys)
+    x5c: list[str]  # X.509 certificate chain (for JWKs)
+    x5t: str  # X.509 certificate thumbprint (for JWKs)
+
+
+class JWKSData(TypedDict):
+    """JSON Web Key Set data structure."""
+
+    keys: list[JWKData]
+
+
 @dataclass(frozen=True, kw_only=True, repr=False)
 class RSAKeyPair:
     private_key: SecretStr
@@ -96,6 +115,7 @@ class RSAKeyPair:
         scopes: list[str] | None = None,
         expires_in_seconds: int = 3600,
         additional_claims: dict[str, Any] | None = None,
+        kid: str | None = None,
     ) -> str:
         """
         Generate a test JWT token for testing purposes.
@@ -108,6 +128,7 @@ class RSAKeyPair:
             scopes: List of scopes to include
             expires_in_seconds: Token expiration time in seconds
             additional_claims: Any additional claims to include
+            kid: Key ID for JWKS lookup (optional)
 
         Returns:
             Signed JWT token string
@@ -135,6 +156,8 @@ class RSAKeyPair:
 
         # Create header
         header = {"alg": "RS256"}
+        if kid:
+            header["kid"] = kid
 
         # Sign and return token
         token_bytes = jwt.encode(
@@ -209,27 +232,25 @@ class BearerAuthProvider(OAuthProvider):
             header = json.loads(base64.urlsafe_b64decode(header_b64))
             kid = header.get("kid")
 
-            if not kid:
-                raise ValueError("Token missing key ID (kid)")
-
             return await self._get_jwks_key(kid)
 
         except Exception as e:
             raise ValueError(f"Failed to extract key ID from token: {e}")
 
-    async def _get_jwks_key(self, kid: str) -> str:
+    async def _get_jwks_key(self, kid: str | None) -> str:
         """Fetch key from JWKS with simple caching."""
         if not self.jwks_uri:
             raise ValueError("JWKS URI not configured")
 
         current_time = time.time()
 
-        # Check cache
-        if (
-            current_time - self._jwks_cache_time < self._cache_ttl
-            and kid in self._jwks_cache
-        ):
-            return self._jwks_cache[kid]
+        # Check cache first
+        if current_time - self._jwks_cache_time < self._cache_ttl:
+            if kid and kid in self._jwks_cache:
+                return self._jwks_cache[kid]
+            elif not kid and len(self._jwks_cache) == 1:
+                # If no kid but only one key cached, use it
+                return next(iter(self._jwks_cache.values()))
 
         # Fetch JWKS
         try:
@@ -242,16 +263,32 @@ class BearerAuthProvider(OAuthProvider):
             self._jwks_cache = {}
             for key_data in jwks_data.get("keys", []):
                 key_kid = key_data.get("kid")
+                jwk = JsonWebKey.import_key(key_data)
+                public_key = jwk.get_public_key()
+
                 if key_kid:
-                    jwk = JsonWebKey.import_key(key_data)
-                    self._jwks_cache[key_kid] = jwk.get_public_key()
+                    self._jwks_cache[key_kid] = public_key
+                else:
+                    # Key without kid - use a default identifier
+                    self._jwks_cache["_default"] = public_key
 
             self._jwks_cache_time = current_time
 
-            if kid not in self._jwks_cache:
-                raise ValueError(f"Key ID '{kid}' not found in JWKS")
-
-            return self._jwks_cache[kid]
+            # Select the appropriate key
+            if kid:
+                if kid not in self._jwks_cache:
+                    raise ValueError(f"Key ID '{kid}' not found in JWKS")
+                return self._jwks_cache[kid]
+            else:
+                # No kid in token - only allow if there's exactly one key
+                if len(self._jwks_cache) == 1:
+                    return next(iter(self._jwks_cache.values()))
+                elif len(self._jwks_cache) > 1:
+                    raise ValueError(
+                        "Multiple keys in JWKS but no key ID (kid) in token"
+                    )
+                else:
+                    raise ValueError("No keys found in JWKS")
 
         except Exception as e:
             raise ValueError(f"Failed to fetch JWKS: {e}")

tests/auth/providers/test_bearer.py

@@ -3,10 +3,15 @@ from typing import Any
 
 import httpx
 import pytest
+from pytest_httpx import HTTPXMock
 
 from fastmcp import Client, FastMCP
 from fastmcp.client.auth import BearerAuth
-from fastmcp.server.auth.providers.bearer import BearerAuthProvider, RSAKeyPair
+from fastmcp.server.auth.providers.bearer import (
+    BearerAuthProvider,
+    JWKSData,
+    RSAKeyPair,
+)
 from fastmcp.utilities.tests import run_server_in_process
 
 
@@ -103,6 +108,194 @@ class TestRSAKeyPair:
         # We'll validate the scopes in the BearerToken tests
 
 
+class TestBearerTokenJWKS:
+    """Tests for JWKS URI functionality."""
+
+    @pytest.fixture
+    def jwks_provider(self, rsa_key_pair: RSAKeyPair) -> BearerAuthProvider:
+        """Provider configured with JWKS URI."""
+        return BearerAuthProvider(
+            jwks_uri="https://test.example.com/.well-known/jwks.json",
+            issuer="https://test.example.com",
+            audience="https://api.example.com",
+        )
+
+    @pytest.fixture
+    def mock_jwks_data(self, rsa_key_pair: RSAKeyPair) -> JWKSData:
+        """Create mock JWKS data from RSA key pair."""
+        from authlib.jose import JsonWebKey
+
+        # Create JWK from the RSA public key
+        jwk = JsonWebKey.import_key(rsa_key_pair.public_key)
+        jwk_data = jwk.as_dict()
+        jwk_data["kid"] = "test-key-1"
+        jwk_data["alg"] = "RS256"
+
+        return {"keys": [jwk_data]}
+
+    async def test_jwks_token_validation(
+        self,
+        rsa_key_pair: RSAKeyPair,
+        jwks_provider: BearerAuthProvider,
+        mock_jwks_data: JWKSData,
+        httpx_mock: HTTPXMock,
+    ):
+        """Test token validation using JWKS URI."""
+        httpx_mock.add_response(
+            url="https://test.example.com/.well-known/jwks.json",
+            json=mock_jwks_data,
+        )
+        token = rsa_key_pair.create_token(
+            subject="test-user",
+            issuer="https://test.example.com",
+            audience="https://api.example.com",
+        )
+
+        access_token = await jwks_provider.load_access_token(token)
+        assert access_token is not None
+        assert access_token.client_id == "test-user"
+
+    async def test_jwks_token_validation_with_invalid_key(
+        self,
+        rsa_key_pair: RSAKeyPair,
+        jwks_provider: BearerAuthProvider,
+        mock_jwks_data: JWKSData,
+        httpx_mock: HTTPXMock,
+    ):
+        httpx_mock.add_response(
+            url="https://test.example.com/.well-known/jwks.json",
+            json=mock_jwks_data,
+        )
+        token = RSAKeyPair.generate().create_token(
+            subject="test-user",
+            issuer="https://test.example.com",
+            audience="https://api.example.com",
+        )
+
+        access_token = await jwks_provider.load_access_token(token)
+        assert access_token is None
+
+    async def test_jwks_token_validation_with_kid(
+        self,
+        rsa_key_pair: RSAKeyPair,
+        jwks_provider: BearerAuthProvider,
+        mock_jwks_data: JWKSData,
+        httpx_mock: HTTPXMock,
+    ):
+        mock_jwks_data["keys"][0]["kid"] = "test-key-1"
+        httpx_mock.add_response(
+            url="https://test.example.com/.well-known/jwks.json",
+            json=mock_jwks_data,
+        )
+        token = rsa_key_pair.create_token(
+            subject="test-user",
+            issuer="https://test.example.com",
+            audience="https://api.example.com",
+            kid="test-key-1",
+        )
+
+        access_token = await jwks_provider.load_access_token(token)
+        assert access_token is not None
+        assert access_token.client_id == "test-user"
+
+    async def test_jwks_token_validation_with_kid_and_no_kid_in_token(
+        self,
+        rsa_key_pair: RSAKeyPair,
+        jwks_provider: BearerAuthProvider,
+        mock_jwks_data: JWKSData,
+        httpx_mock: HTTPXMock,
+    ):
+        mock_jwks_data["keys"][0]["kid"] = "test-key-1"
+        httpx_mock.add_response(
+            url="https://test.example.com/.well-known/jwks.json",
+            json=mock_jwks_data,
+        )
+        token = rsa_key_pair.create_token(
+            subject="test-user",
+            issuer="https://test.example.com",
+            audience="https://api.example.com",
+        )
+
+        access_token = await jwks_provider.load_access_token(token)
+        assert access_token is not None
+        assert access_token.client_id == "test-user"
+
+    async def test_jwks_token_validation_with_no_kid_and_kid_in_jwks(
+        self,
+        rsa_key_pair: RSAKeyPair,
+        jwks_provider: BearerAuthProvider,
+        mock_jwks_data: JWKSData,
+        httpx_mock: HTTPXMock,
+    ):
+        mock_jwks_data["keys"][0]["kid"] = "test-key-1"
+        httpx_mock.add_response(
+            url="https://test.example.com/.well-known/jwks.json",
+            json=mock_jwks_data,
+        )
+        token = rsa_key_pair.create_token(
+            subject="test-user",
+            issuer="https://test.example.com",
+            audience="https://api.example.com",
+        )
+
+        access_token = await jwks_provider.load_access_token(token)
+        assert access_token is not None
+        assert access_token.client_id == "test-user"
+
+    async def test_jwks_token_validation_with_kid_mismatch(
+        self,
+        rsa_key_pair: RSAKeyPair,
+        jwks_provider: BearerAuthProvider,
+        mock_jwks_data: JWKSData,
+        httpx_mock: HTTPXMock,
+    ):
+        mock_jwks_data["keys"][0]["kid"] = "test-key-1"
+        httpx_mock.add_response(
+            url="https://test.example.com/.well-known/jwks.json",
+            json=mock_jwks_data,
+        )
+        token = rsa_key_pair.create_token(
+            subject="test-user",
+            issuer="https://test.example.com",
+            audience="https://api.example.com",
+            kid="test-key-2",
+        )
+
+        access_token = await jwks_provider.load_access_token(token)
+        assert access_token is None
+
+    async def test_jwks_token_validation_with_multiple_keys_and_no_kid_in_token(
+        self,
+        rsa_key_pair: RSAKeyPair,
+        jwks_provider: BearerAuthProvider,
+        mock_jwks_data: JWKSData,
+        httpx_mock: HTTPXMock,
+    ):
+        mock_jwks_data["keys"] = [
+            {
+                "kid": "test-key-1",
+                "alg": "RS256",
+            },
+            {
+                "kid": "test-key-2",
+                "alg": "RS256",
+            },
+        ]
+
+        httpx_mock.add_response(
+            url="https://test.example.com/.well-known/jwks.json",
+            json=mock_jwks_data,
+        )
+        token = rsa_key_pair.create_token(
+            subject="test-user",
+            issuer="https://test.example.com",
+            audience="https://api.example.com",
+        )
+
+        access_token = await jwks_provider.load_access_token(token)
+        assert access_token is None
+
+
 class TestBearerToken:
     def test_initialization_with_public_key(self, rsa_key_pair: RSAKeyPair):
         """Test provider initialization with public key."""
@@ -143,7 +336,6 @@ class TestBearerToken:
                 issuer="https://test.example.com",
             )
 
-    @pytest.mark.asyncio
     async def test_valid_token_validation(
         self, rsa_key_pair: RSAKeyPair, bearer_provider: BearerAuthProvider
     ):
@@ -163,7 +355,6 @@ class TestBearerToken:
         assert "write" in access_token.scopes
         assert access_token.expires_at is not None
 
-    @pytest.mark.asyncio
     async def test_expired_token_rejection(
         self, rsa_key_pair: RSAKeyPair, bearer_provider: BearerAuthProvider
     ):
@@ -178,7 +369,6 @@ class TestBearerToken:
         access_token = await bearer_provider.load_access_token(token)
         assert access_token is None
 
-    @pytest.mark.asyncio
     async def test_invalid_issuer_rejection(
         self, rsa_key_pair: RSAKeyPair, bearer_provider: BearerAuthProvider
     ):
@@ -192,7 +382,6 @@ class TestBearerToken:
         access_token = await bearer_provider.load_access_token(token)
         assert access_token is None
 
-    @pytest.mark.asyncio
     async def test_invalid_audience_rejection(
         self, rsa_key_pair: RSAKeyPair, bearer_provider: BearerAuthProvider
     ):
@@ -206,7 +395,6 @@ class TestBearerToken:
         access_token = await bearer_provider.load_access_token(token)
         assert access_token is None
 
-    @pytest.mark.asyncio
     async def test_no_issuer_validation_when_none(self, rsa_key_pair: RSAKeyPair):
         """Test that issuer validation is skipped when provider has no issuer configured."""
         provider = BearerAuthProvider(
@@ -221,7 +409,6 @@ class TestBearerToken:
         access_token = await provider.load_access_token(token)
         assert access_token is not None
 
-    @pytest.mark.asyncio
     async def test_no_audience_validation_when_none(self, rsa_key_pair: RSAKeyPair):
         """Test that audience validation is skipped when provider has no audience configured."""
         provider = BearerAuthProvider(
@@ -239,7 +426,6 @@ class TestBearerToken:
         access_token = await provider.load_access_token(token)
         assert access_token is not None
 
-    @pytest.mark.asyncio
     async def test_multiple_audiences_validation(self, rsa_key_pair: RSAKeyPair):
         """Test validation with multiple audiences in token."""
         provider = BearerAuthProvider(
@@ -259,7 +445,6 @@ class TestBearerToken:
         access_token = await provider.load_access_token(token)
         assert access_token is not None
 
-    @pytest.mark.asyncio
     async def test_scope_extraction_string(
         self, rsa_key_pair: RSAKeyPair, bearer_provider: BearerAuthProvider
     ):
@@ -276,7 +461,6 @@ class TestBearerToken:
         assert access_token is not None
         assert set(access_token.scopes) == {"read", "write", "admin"}
 
-    @pytest.mark.asyncio
     async def test_scope_extraction_list(
         self, rsa_key_pair: RSAKeyPair, bearer_provider: BearerAuthProvider
     ):
@@ -293,7 +477,6 @@ class TestBearerToken:
         assert access_token is not None
         assert set(access_token.scopes) == {"read", "write"}
 
-    @pytest.mark.asyncio
     async def test_no_scopes(
         self, rsa_key_pair: RSAKeyPair, bearer_provider: BearerAuthProvider
     ):
@@ -310,7 +493,6 @@ class TestBearerToken:
         assert access_token is not None
         assert access_token.scopes == []
 
-    @pytest.mark.asyncio
     async def test_malformed_token_rejection(self, bearer_provider: BearerAuthProvider):
         """Test rejection of malformed tokens."""
         malformed_tokens = [
@@ -325,7 +507,6 @@ class TestBearerToken:
             access_token = await bearer_provider.load_access_token(token)
             assert access_token is None
 
-    @pytest.mark.asyncio
     async def test_invalid_signature_rejection(
         self, rsa_key_pair: RSAKeyPair, bearer_provider: BearerAuthProvider
     ):
@@ -341,7 +522,6 @@ class TestBearerToken:
         access_token = await bearer_provider.load_access_token(token)
         assert access_token is None
 
-    @pytest.mark.asyncio
     async def test_client_id_fallback(
         self, rsa_key_pair: RSAKeyPair, bearer_provider: BearerAuthProvider
     ):
@@ -367,9 +547,10 @@ class TestFastMCPBearerAuth:
         assert isinstance(mcp.auth, BearerAuthProvider)
 
     async def test_unauthorized_access(self, mcp_server_url: str):
-        with pytest.raises(httpx.HTTPStatusError, match="401"):
+        with pytest.raises(httpx.HTTPStatusError) as exc_info:
             async with Client(mcp_server_url) as client:
                 tools = await client.list_tools()  # noqa: F841
+        assert exc_info.value.response.status_code == 401
         assert "tools" not in locals()
 
     async def test_authorized_access(self, mcp_server_url: str, bearer_token):
@@ -378,9 +559,10 @@ class TestFastMCPBearerAuth:
         assert tools
 
     async def test_invalid_token_raises_401(self, mcp_server_url: str):
-        with pytest.raises(httpx.HTTPStatusError, match="401"):
+        with pytest.raises(httpx.HTTPStatusError) as exc_info:
             async with Client(mcp_server_url, auth=BearerAuth("invalid")) as client:
                 tools = await client.list_tools()  # noqa: F841
+        assert exc_info.value.response.status_code == 401
         assert "tools" not in locals()
 
     async def test_expired_token(self, mcp_server_url: str, rsa_key_pair: RSAKeyPair):
@@ -391,16 +573,62 @@ class TestFastMCPBearerAuth:
             expires_in_seconds=-3600,
         )
 
-        with pytest.raises(httpx.HTTPStatusError, match="401"):
+        with pytest.raises(httpx.HTTPStatusError) as exc_info:
             async with Client(mcp_server_url, auth=BearerAuth(token)) as client:
                 tools = await client.list_tools()  # noqa: F841
+        assert exc_info.value.response.status_code == 401
         assert "tools" not in locals()
 
     async def test_token_with_bad_signature(self, mcp_server_url: str):
         rsa_key_pair = RSAKeyPair.generate()
         token = rsa_key_pair.create_token()
 
-        with pytest.raises(httpx.HTTPStatusError, match="401"):
+        with pytest.raises(httpx.HTTPStatusError) as exc_info:
             async with Client(mcp_server_url, auth=BearerAuth(token)) as client:
                 tools = await client.list_tools()  # noqa: F841
+        assert exc_info.value.response.status_code == 401
         assert "tools" not in locals()
+
+    async def test_token_with_insufficient_scopes(
+        self, mcp_server_url: str, rsa_key_pair: RSAKeyPair
+    ):
+        token = rsa_key_pair.create_token(
+            subject="test-user",
+            issuer="https://test.example.com",
+            audience="https://api.example.com",
+            scopes=["read"],
+        )
+
+        with run_server_in_process(
+            run_mcp_server,
+            public_key=rsa_key_pair.public_key,
+            auth_kwargs=dict(required_scopes=["read", "write"]),
+            run_kwargs=dict(transport="streamable-http"),
+        ) as url:
+            mcp_server_url = f"{url}/mcp"
+            with pytest.raises(httpx.HTTPStatusError) as exc_info:
+                async with Client(mcp_server_url, auth=BearerAuth(token)) as client:
+                    tools = await client.list_tools()  # noqa: F841
+            assert exc_info.value.response.status_code == 403
+            assert "tools" not in locals()
+
+    async def test_token_with_sufficient_scopes(
+        self, mcp_server_url: str, rsa_key_pair: RSAKeyPair
+    ):
+        token = rsa_key_pair.create_token(
+            subject="test-user",
+            issuer="https://test.example.com",
+            audience="https://api.example.com",
+            scopes=["read", "write"],
+        )
+
+        with run_server_in_process(
+            run_mcp_server,
+            public_key=rsa_key_pair.public_key,
+            auth_kwargs=dict(required_scopes=["read", "write"]),
+            run_kwargs=dict(transport="streamable-http"),
+        ) as url:
+            mcp_server_url = f"{url}/mcp"
+            async with Client(mcp_server_url, auth=BearerAuth(token)) as client:
+                tools = await client.list_tools()
+        assert tools

uv.lock

@@ -456,6 +456,7 @@ dev = [
     { name = "pytest-cov" },
     { name = "pytest-env" },
     { name = "pytest-flakefinder" },
+    { name = "pytest-httpx" },
     { name = "pytest-report" },
     { name = "pytest-timeout" },
     { name = "pytest-xdist" },
@@ -490,6 +491,7 @@ dev = [
     { name = "pytest-cov", specifier = ">=6.1.1" },
     { name = "pytest-env", specifier = ">=1.1.5" },
     { name = "pytest-flakefinder" },
+    { name = "pytest-httpx", specifier = ">=0.35.0" },
     { name = "pytest-report", specifier = ">=0.2.1" },
     { name = "pytest-timeout", specifier = ">=2.4.0" },
     { name = "pytest-xdist", specifier = ">=3.6.1" },
@@ -1160,6 +1162,19 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/33/8b/06787150d0fd0cbd3a8054262b56f91631c7778c1bc91bf4637e47f909ad/pytest_flakefinder-1.1.0-py2.py3-none-any.whl", hash = "sha256:741e0e8eea427052f5b8c89c2b3c3019a50c39a59ce4df6a305a2c2d9ba2bd13", size = 4644, upload-time = "2022-10-26T18:27:52.128Z" },
 ]
 
+[[package]]
+name = "pytest-httpx"
+version = "0.35.0"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "httpx" },
+    { name = "pytest" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/1f/89/5b12b7b29e3d0af3a4b9c071ee92fa25a9017453731a38f08ba01c280f4c/pytest_httpx-0.35.0.tar.gz", hash = "sha256:d619ad5d2e67734abfbb224c3d9025d64795d4b8711116b1a13f72a251ae511f", size = 54146, upload-time = "2024-11-28T19:16:54.237Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/b0/ed/026d467c1853dd83102411a78126b4842618e86c895f93528b0528c7a620/pytest_httpx-0.35.0-py3-none-any.whl", hash = "sha256:ee11a00ffcea94a5cbff47af2114d34c5b231c326902458deed73f9c459fd744", size = 19442, upload-time = "2024-11-28T19:16:52.787Z" },
+]
+
 [[package]]
 name = "pytest-report"
 version = "0.2.1"