Add complete support for TokenVerifier protocol (#1297)

docs/docs.json

@@ -65,10 +65,7 @@
                   {
                     "group": "Essentials",
                     "icon": "cube",
-                    "pages": [
-                      "servers/server",
-                      "deployment/running-server"
-                    ]
+                    "pages": ["servers/server", "deployment/running-server"]
                   },
                   {
                     "group": "Core Components",
@@ -96,9 +93,7 @@
                   {
                     "group": "Authentication",
                     "icon": "shield-check",
-                    "pages": [
-                      "servers/auth/bearer"
-                    ]
+                    "pages": ["servers/auth/verifiers"]
                   }
                 ]
               },
@@ -108,10 +103,7 @@
                   {
                     "group": "Essentials",
                     "icon": "cube",
-                    "pages": [
-                      "clients/client",
-                      "clients/transports"
-                    ]
+                    "pages": ["clients/client", "clients/transports"]
                   },
                   {
                     "group": "Core Operations",
@@ -137,10 +129,7 @@
                   {
                     "group": "Authentication",
                     "icon": "user-shield",
-                    "pages": [
-                      "clients/auth/oauth",
-                      "clients/auth/bearer"
-                    ]
+                    "pages": ["clients/auth/oauth", "clients/auth/bearer"]
                   }
                 ]
               },
@@ -186,17 +175,12 @@
           },
           {
             "anchor": "What's New",
-            "pages": [
-              "updates",
-              "changelog"
-            ]
+            "pages": ["updates", "changelog"]
           },
           {
             "anchor": "Community",
             "icon": "users",
-            "pages": [
-              "community/showcase"
-            ]
+            "pages": ["community/showcase"]
           }
         ]
       },

docs/integrations/anthropic.mdx

@@ -125,7 +125,7 @@ For this example, we'll quickly generate our own tokens with FastMCP's `RSAKeyPa
 We'll start by creating an RSA key pair to sign and verify tokens.
 
 ```python
-from fastmcp.server.auth.providers.bearer import RSAKeyPair
+from fastmcp.server.auth.verifiers import RSAKeyPair
 
 key_pair = RSAKeyPair.generate()
 access_token = key_pair.create_token(audience="dice-server")
@@ -135,13 +135,13 @@ access_token = key_pair.create_token(audience="dice-server")
 FastMCP's `RSAKeyPair` utility is for development and testing only.
 </Warning> 
 
-Next, we'll create a `BearerAuthProvider` to authenticate the server. 
+Next, we'll create a `JWTVerifier` to authenticate the server. 
 
 ```python
 from fastmcp import FastMCP
-from fastmcp.server.auth import BearerAuthProvider
+from fastmcp.server.auth import JWTVerifier
 
-auth = BearerAuthProvider(
+auth = JWTVerifier(
     public_key=key_pair.public_key,
     audience="dice-server",
 )
@@ -153,14 +153,14 @@ Here is a complete example that you can copy/paste. For simplicity and the purpo
 
 ```python server.py [expandable]
 from fastmcp import FastMCP
-from fastmcp.server.auth import BearerAuthProvider
-from fastmcp.server.auth.providers.bearer import RSAKeyPair
+from fastmcp.server.auth import JWTVerifier
+from fastmcp.server.auth.verifiers import RSAKeyPair
 import random
 
 key_pair = RSAKeyPair.generate()
 access_token = key_pair.create_token(audience="dice-server")
 
-auth = BearerAuthProvider(
+auth = JWTVerifier(
     public_key=key_pair.public_key,
     audience="dice-server",
 )

docs/integrations/openai.mdx

@@ -123,7 +123,7 @@ For this example, we'll quickly generate our own tokens with FastMCP's `RSAKeyPa
 We'll start by creating an RSA key pair to sign and verify tokens.
 
 ```python
-from fastmcp.server.auth.providers.bearer import RSAKeyPair
+from fastmcp.server.auth.verifiers import RSAKeyPair
 
 key_pair = RSAKeyPair.generate()
 access_token = key_pair.create_token(audience="dice-server")
@@ -133,13 +133,13 @@ access_token = key_pair.create_token(audience="dice-server")
 FastMCP's `RSAKeyPair` utility is for development and testing only.
 </Warning> 
 
-Next, we'll create a `BearerAuthProvider` to authenticate the server. 
+Next, we'll create a `JWTVerifier` to authenticate the server. 
 
 ```python
 from fastmcp import FastMCP
-from fastmcp.server.auth import BearerAuthProvider
+from fastmcp.server.auth import JWTVerifier
 
-auth = BearerAuthProvider(
+auth = JWTVerifier(
     public_key=key_pair.public_key,
     audience="dice-server",
 )
@@ -151,14 +151,14 @@ Here is a complete example that you can copy/paste. For simplicity and the purpo
 
 ```python server.py [expandable]
 from fastmcp import FastMCP
-from fastmcp.server.auth import BearerAuthProvider
-from fastmcp.server.auth.providers.bearer import RSAKeyPair
+from fastmcp.server.auth import JWTVerifier
+from fastmcp.server.auth.verifiers import RSAKeyPair
 import random
 
 key_pair = RSAKeyPair.generate()
 access_token = key_pair.create_token(audience="dice-server")
 
-auth = BearerAuthProvider(
+auth = JWTVerifier(
     public_key=key_pair.public_key,
     audience="dice-server",
 )

docs/servers/auth/{bearer.mdx → verifiers.mdx}

@@ -1,22 +1,19 @@
 ---
-title: Bearer Token Authentication
-sidebarTitle: Bearer Auth
-description: Secure your FastMCP server's HTTP endpoints by validating JWT Bearer tokens.
+title: Token Verification
+sidebarTitle: Token Verification
+description: Secure your FastMCP server's HTTP endpoints by validating JWT tokens.
 icon: key
 tag: NEW
 ---
 
 import { VersionBadge } from "/snippets/version-badge.mdx"
 
-<VersionBadge version="2.6.0" />
+<VersionBadge version="2.11.0" />
+
 <Tip>
 Authentication and authorization are only relevant for HTTP-based transports.
 </Tip>
 
-<Note>
-The [MCP specification](https://modelcontextprotocol.io/specification/2025-03-26/basic/authorization) requires servers to implement full OAuth 2.1 authorization flows with dynamic client registration, server metadata discovery, and complete token endpoints. FastMCP's Bearer Token authentication provides a simpler, more practical alternative by directly validating pre-issued JWT tokens—ideal for service-to-service communication and programmatic environments where full OAuth flows may be impractical, and in accordance with how the MCP ecosystem is pragmatically evolving. However, please note that since it doesn't implement the full OAuth 2.1 flow, this implementation does not strictly comply with the MCP specification.
-</Note>
-
 Bearer Token authentication is a common way to secure HTTP-based APIs. In this model, the client sends a token (usually a JSON Web Token or JWT) in the `Authorization` header with the "Bearer" scheme. The server then validates this token to grant or deny access.
 
 FastMCP supports Bearer Token authentication for its HTTP-based transports (`http` and `sse`), allowing you to protect your server from unauthorized access.
@@ -32,37 +29,29 @@ FastMCP uses **asymmetric encryption** for token validation, which provides a cl
 
 This design allows you to integrate FastMCP servers into existing authentication infrastructures without compromising security boundaries.
 
-## Configuration
+## Token Verification Approaches
 
-To enable Bearer Token validation on your FastMCP server, use the `BearerAuthProvider` class. This provider validates incoming JWTs by verifying signatures, checking expiration, and optionally validating claims.
+FastMCP provides three token verification approaches:
 
-<Warning>
-The `BearerAuthProvider` validates tokens; it does **not** issue them (or implement any part of an OAuth flow). You'll need to generate tokens separately, either using FastMCP utilities or an external Identity Provider (IdP) or OAuth 2.1 Authorization Server.
-</Warning>
-
-### Basic Setup
+### JWTVerifier
+Validates JWT tokens using public key cryptography. Use when you have JWT tokens issued by an external identity provider (Auth0, Okta, Keycloak, etc.) and want self-contained validation without network calls.
 
-To configure bearer token authentication, instantiate a `BearerAuthProvider` instance and pass it to the `auth` parameter of the `FastMCP` instance.
+### IntrospectionTokenVerifier
+Validates tokens by calling a remote OAuth 2.0 authorization server's introspection endpoint (RFC 7662). Use when your authorization server is separate from your FastMCP server, you're using opaque tokens, or you need real-time token revocation.
 
-The `BearerAuthProvider` requires either a static public key or a JWKS URI (but not both!) in order to verify the token's signature. All other parameters are optional -- if they are provided, they will be used as additional validation criteria.
-
-```python {2, 10}
-from fastmcp import FastMCP
-from fastmcp.server.auth import BearerAuthProvider
+### StaticTokenVerifier
+Validates tokens against a predefined dictionary. Use for development and testing only - never in production.
 
-auth = BearerAuthProvider(
-    jwks_uri="https://my-identity-provider.com/.well-known/jwks.json",
-    issuer="https://my-identity-provider.com/",
-    algorithm="RS512",
-    audience="my-mcp-server"
-)
+<Warning>
+These verifiers validate tokens; they do **not** issue them (or implement any part of an OAuth flow). You'll need to generate tokens separately, either using FastMCP utilities or an external Identity Provider (IdP) or OAuth 2.1 Authorization Server.
+</Warning>
 
-mcp = FastMCP(name="My MCP Server", auth=auth)
-```
 
 ### Configuration Parameters
 
-<Card icon="code" title="BearerAuthProvider Configuration">
+<Tabs>
+<Tab title="JWTVerifier">
+<Card icon="code" title="JWTVerifier Configuration">
 <ParamField body="public_key" type="str">
   RSA public key in PEM format for static key validation. Required if `jwks_uri` is not provided
 </ParamField>
@@ -87,13 +76,66 @@ mcp = FastMCP(name="My MCP Server", auth=auth)
   Global scopes required for all requests
 </ParamField>
 </Card>
+</Tab>
+
+<Tab title="IntrospectionTokenVerifier">
+<Card icon="code" title="IntrospectionTokenVerifier Configuration">
+<ParamField body="introspection_endpoint" type="str">
+  OAuth 2.0 Token Introspection endpoint URL (RFC 7662)
+</ParamField>
+
+<ParamField body="client_id" type="str">
+  Resource server client ID for introspection authentication
+</ParamField>
+
+<ParamField body="client_secret" type="str">
+  Resource server client secret for introspection authentication
+</ParamField>
+
+<ParamField body="required_scopes" type="list[str] | None">
+  Global scopes required for all requests
+</ParamField>
+</Card>
+</Tab>
+
+<Tab title="StaticTokenVerifier">
+<Card icon="code" title="StaticTokenVerifier Configuration">
+<ParamField body="valid_tokens" type="dict[str, dict]">
+  Mapping of valid tokens to their claims. Each token maps to a dictionary containing token metadata like `sub`, `scope`, etc.
+</ParamField>
+
+<ParamField body="required_scopes" type="list[str] | None">
+  Global scopes required for all requests
+</ParamField>
+</Card>
+</Tab>
+</Tabs>
+
+## JWT Verification
+
+The `JWTVerifier` validates JWT tokens using public key cryptography. Use this when you have JWT tokens issued by an external identity provider and want self-contained validation without network calls.
 
-#### Public Key
+```python
+from fastmcp import FastMCP
+from fastmcp.server.auth.verifiers import JWTVerifier
 
-If you have a public key in PEM format, you can provide it to the `BearerAuthProvider` as a string.
+verifier = JWTVerifier(
+    jwks_uri="https://my-identity-provider.com/.well-known/jwks.json",
+    issuer="https://my-identity-provider.com/",
+    audience="my-mcp-server"
+)
+
+mcp = FastMCP(name="My MCP Server", auth=verifier)
+```
+
+### Public Key Configuration
+
+#### Using a Static Public Key
+
+If you have a public key in PEM format, you can provide it to the `JWTVerifier` as a string.
 
 ```python {12}
-from fastmcp.server.auth import BearerAuthProvider
+from fastmcp.server.auth.verifiers import JWTVerifier
 import inspect
 
 public_key_pem = inspect.cleandoc(
@@ -104,13 +146,13 @@ public_key_pem = inspect.cleandoc(
     """
 )
 
-auth = BearerAuthProvider(public_key=public_key_pem)
+auth = JWTVerifier(public_key=public_key_pem)
 ```
 
-#### JWKS URI
+#### Using JWKS URI
 
 ```python
-provider = BearerAuthProvider(
+verifier = JWTVerifier(
     jwks_uri="https://idp.example.com/.well-known/jwks.json"
 )
 ```
@@ -119,6 +161,58 @@ provider = BearerAuthProvider(
 JWKS is recommended for production as it supports automatic key rotation and multiple signing keys.
 </Note>
 
+## OAuth 2.0 Token Introspection
+
+The `IntrospectionTokenVerifier` validates tokens by calling an OAuth 2.0 authorization server's introspection endpoint (RFC 7662). This is useful when your authorization server is separate from your FastMCP server, you're using opaque tokens, or you need real-time token validation with immediate revocation support.
+
+```python
+from fastmcp.server.auth.verifiers import IntrospectionTokenVerifier
+
+verifier = IntrospectionTokenVerifier(
+    introspection_endpoint="https://auth.company.com/oauth/introspect",
+    server_url="https://mcp.company.com",  # This server's URL
+    client_id="mcp-resource-server",
+    client_secret="your-secret",
+    required_scopes=["mcp:access"]
+)
+
+mcp = FastMCP(name="MCP Server", auth=verifier)
+```
+
+For each request, the verifier makes an HTTP call to the introspection endpoint to check if the token is valid and active. This provides real-time validation but requires network connectivity.
+
+## Static Token Verification
+
+The `StaticTokenVerifier` validates tokens against a predefined dictionary of token strings and claims. Use this for development and testing when you need predictable tokens without setting up a real OAuth server.
+
+```python
+from fastmcp.server.auth.verifiers import StaticTokenVerifier
+
+verifier = StaticTokenVerifier(
+    tokens={
+        "dev-token-123": {
+            "client_id": "dev-user",
+            "scopes": ["read", "write"],
+            "sub": "developer@example.com"
+        },
+        "readonly-token": {
+            "client_id": "readonly-user",
+            "scopes": ["read"],
+            "expires_at": 1735689600  # Optional expiration
+        }
+    },
+    required_scopes=["read"]
+)
+
+mcp = FastMCP(name="Development Server", auth=verifier)
+```
+
+Token claims can include `client_id` (required), `scopes`, `sub`, `expires_at`, and any custom metadata your application needs.
+
+<Warning>
+Never use StaticTokenVerifier in production - tokens are stored in plain text.
+</Warning>
+
 ## Generating Tokens
 
 For development and testing, FastMCP provides the `RSAKeyPair` utility class to generate tokens without needing an external OAuth provider.
@@ -130,14 +224,13 @@ The `RSAKeyPair` utility is intended for development and testing only. For produ
 
 ```python
 from fastmcp import FastMCP
-from fastmcp.server.auth import BearerAuthProvider
-from fastmcp.server.auth.providers.bearer import RSAKeyPair
+from fastmcp.server.auth.verifiers import JWTVerifier, RSAKeyPair
 
 # Generate a new key pair
 key_pair = RSAKeyPair.generate()
 
-# Configure the auth provider with the public key
-auth = BearerAuthProvider(
+# Configure the auth verifier with the public key
+auth = JWTVerifier(
     public_key=key_pair.public_key,
     issuer="https://dev.example.com",
     audience="my-dev-server"
@@ -191,6 +284,7 @@ The `create_token()` method accepts these parameters:
 </Card>
 
 
+
 ## Accessing Token Claims
 
 Once authenticated, your tools, resources, or prompts can access token information using the `get_access_token()` dependency function:

docs/servers/server.mdx

@@ -40,6 +40,10 @@ The `FastMCP` constructor accepts several arguments:
   Description of how to interact with this server. These instructions help clients understand the server's purpose and available functionality
 </ParamField>
 
+<ParamField body="auth" type="OAuthProvider | TokenVerifier | None">
+  Authentication provider for securing HTTP-based transports. See [Bearer Token Authentication](/servers/auth/bearer) for configuration options
+</ParamField>
+
 <ParamField body="lifespan" type="AsyncContextManager | None">
   An async context manager function for server startup and shutdown logic
 </ParamField>

src/fastmcp/contrib/component_manager/example.py

@@ -1,10 +1,10 @@
 from fastmcp import FastMCP
 from fastmcp.contrib.component_manager import set_up_component_manager
-from fastmcp.server.auth.providers.bearer import BearerAuthProvider, RSAKeyPair
+from fastmcp.server.auth.verifiers import JWTVerifier, RSAKeyPair
 
 key_pair = RSAKeyPair.generate()
 
-auth = BearerAuthProvider(
+auth = JWTVerifier(
     public_key=key_pair.public_key,
     issuer="https://dev.example.com",
     audience="my-dev-server",

src/fastmcp/server/auth/__init__.py

@@ -1,4 +1,20 @@
-from .providers.bearer import BearerAuthProvider
+from .auth import OAuthProvider, TokenVerifier
+from .verifiers import IntrospectionTokenVerifier, JWTVerifier, StaticTokenVerifier
 
 
-__all__ = ["BearerAuthProvider"]
+__all__ = [
+    "OAuthProvider",
+    "TokenVerifier",
+    "IntrospectionTokenVerifier",
+    "JWTVerifier",
+    "StaticTokenVerifier",
+]
+
+
+def __getattr__(name: str):
+    # Defer import because it raises a deprecation warning
+    if name == "BearerAuthProvider":
+        from .providers.bearer import BearerAuthProvider
+
+        return BearerAuthProvider
+    raise AttributeError(f"module '{__name__}' has no attribute '{name}'")

src/fastmcp/server/auth/auth.py

@@ -4,6 +4,9 @@ from mcp.server.auth.provider import (
     OAuthAuthorizationServerProvider,
     RefreshToken,
 )
+from mcp.server.auth.provider import (
+    TokenVerifier as TokenVerifierProtocol,
+)
 from mcp.server.auth.settings import (
     ClientRegistrationOptions,
     RevocationOptions,
@@ -11,6 +14,35 @@ from mcp.server.auth.settings import (
 from pydantic import AnyHttpUrl
 
 
+class TokenVerifier(TokenVerifierProtocol):
+    """Base class for token verifiers (Resource Servers)."""
+
+    def __init__(
+        self,
+        resource_server_url: AnyHttpUrl | str | None = None,
+        required_scopes: list[str] | None = None,
+    ):
+        """
+        Initialize the token verifier.
+
+        Args:
+            resource_server_url: The URL of this resource server (for RFC 8707 resource indicators)
+            required_scopes: Scopes that are required for all requests
+        """
+        self.resource_server_url: AnyHttpUrl | None
+        if resource_server_url is None:
+            self.resource_server_url = None
+        elif isinstance(resource_server_url, str):
+            self.resource_server_url = AnyHttpUrl(resource_server_url)
+        else:
+            self.resource_server_url = resource_server_url
+        self.required_scopes = required_scopes or []
+
+    async def verify_token(self, token: str) -> AccessToken | None:
+        """Verify a bearer token and return access info if valid."""
+        raise NotImplementedError("Subclasses must implement verify_token")
+
+
 class OAuthProvider(
     OAuthAuthorizationServerProvider[AuthorizationCode, RefreshToken, AccessToken]
 ):
@@ -44,21 +76,3 @@ class OAuthProvider(
         self.client_registration_options = client_registration_options
         self.revocation_options = revocation_options
         self.required_scopes = required_scopes
-        self.resource_server_url = (
-            AnyHttpUrl(resource_server_url) if resource_server_url else None
-        )
-
-    async def verify_token(self, token: str) -> AccessToken | None:
-        """
-        Verify a bearer token and return access info if valid.
-
-        This method implements the TokenVerifier protocol by delegating
-        to our existing load_access_token method.
-
-        Args:
-            token: The token string to validate
-
-        Returns:
-            AccessToken object if valid, None if invalid or expired
-        """
-        return await self.load_access_token(token)

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

@@ -1,482 +1,25 @@
-import time
-from dataclasses import dataclass
-from typing import Any
-
-import httpx
-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,
-    AuthorizationCode,
-    AuthorizationParams,
-    RefreshToken,
-)
-from mcp.shared.auth import (
-    OAuthClientInformationFull,
-    OAuthToken,
-)
-from pydantic import AnyHttpUrl, SecretStr, ValidationError
-from typing_extensions import TypedDict
-
-from fastmcp.server.auth.auth import (
-    ClientRegistrationOptions,
-    OAuthProvider,
-    RevocationOptions,
-)
-from fastmcp.utilities.logging import get_logger
-
-
-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
-    public_key: str
-
-    @classmethod
-    def generate(cls) -> "RSAKeyPair":
-        """
-        Generate an RSA key pair for testing.
-
-        Returns:
-            tuple: (private_key_pem, public_key_pem)
-        """
-        # Generate private key
-        private_key = rsa.generate_private_key(
-            public_exponent=65537,
-            key_size=2048,
-        )
-
-        # Get public key
-        public_key = private_key.public_key()
-
-        # Serialize private key to PEM format
-        private_pem = private_key.private_bytes(
-            encoding=serialization.Encoding.PEM,
-            format=serialization.PrivateFormat.PKCS8,
-            encryption_algorithm=serialization.NoEncryption(),
-        ).decode("utf-8")
-
-        # Serialize public key to PEM format
-        public_pem = public_key.public_bytes(
-            encoding=serialization.Encoding.PEM,
-            format=serialization.PublicFormat.SubjectPublicKeyInfo,
-        ).decode("utf-8")
-
-        return cls(
-            private_key=SecretStr(private_pem),
-            public_key=public_pem,
-        )
-
-    def create_token(
-        self,
-        subject: str = "fastmcp-user",
-        issuer: str = "https://fastmcp.example.com",
-        audience: str | list[str] | None = None,
-        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.
-
-        Args:
-            private_key_pem: RSA private key in PEM format
-            subject: Subject claim (usually user ID)
-            issuer: Issuer claim
-            audience: Audience claim - can be a string or list of strings (optional)
-            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
-        """
-        # TODO : Add support for configurable algorithms
-        jwt = JsonWebToken(["RS256"])
-
-        now = int(time.time())
-
-        # Build payload
-        payload = {
-            "iss": issuer,
-            "sub": subject,
-            "iat": now,
-            "exp": now + expires_in_seconds,
-        }
-
-        if audience:
-            payload["aud"] = audience
-
-        if scopes:
-            payload["scope"] = " ".join(scopes)
-
-        if additional_claims:
-            payload.update(additional_claims)
-
-        # Create header
-        header = {"alg": "RS256"}
-        if kid:
-            header["kid"] = kid
-
-        # Sign and return token
-        token_bytes = jwt.encode(
-            header,
-            payload,
-            key=self.private_key.get_secret_value(),
-        )
-        return token_bytes.decode("utf-8")
-
-
-class BearerAuthProvider(OAuthProvider):
-    """
-    Simple JWT Bearer Token validator for hosted MCP servers.
-    Uses RS256 asymmetric encryption by default but supports all JWA algorithms. Supports either static public key
-    or JWKS URI for key rotation.
-
-    Note that this provider DOES NOT permit client registration or revocation, or any OAuth flows.
-    It is intended to be used with a control plane that manages clients and tokens.
-    """
-
-    def __init__(
-        self,
-        public_key: str | None = None,
-        jwks_uri: str | None = None,
-        issuer: str | None = None,
-        algorithm: str | None = None,
-        audience: str | list[str] | None = None,
-        required_scopes: list[str] | None = None,
-        resource_server: str | None = None,
-    ):
-        """
-        Initialize the provider. Either public_key or jwks_uri must be provided.
-
-        Args:
-            public_key: RSA public key in PEM format (for static key)
-            jwks_uri: URI to fetch keys from (for key rotation)
-            issuer: Expected issuer claim (optional)
-            algorithm: Algorithm to use for verification (optional, defaults to RS256)
-            audience: Expected audience claim - can be a string or list of strings (optional)
-            required_scopes: List of required scopes for access (optional)
-        """
-        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")
-
-        if not algorithm:
-            algorithm = "RS256"
-        if algorithm not in {
-            "HS256",
-            "HS384",
-            "HS512",
-            "RS256",
-            "RS384",
-            "RS512",
-            "ES256",
-            "ES384",
-            "ES512",
-            "PS256",
-            "PS384",
-            "PS512",
-        }:
-            raise ValueError(f"Unsupported algorithm: {algorithm}.")
-
-        # Only pass issuer to parent if it's a valid URL, otherwise use default
-        # This allows the issuer claim validation to work with string issuers per RFC 7519
-        try:
-            issuer_url = AnyHttpUrl(issuer) if issuer else "https://fastmcp.example.com"
-        except ValidationError:
-            # Issuer is not a valid URL, use default for parent class
-            issuer_url = "https://fastmcp.example.com"
-
-        try:
-            resource_server_url = (
-                AnyHttpUrl(resource_server) if resource_server else None
-            )
-        except ValidationError:
-            resource_server_url = None
-
-        super().__init__(
-            issuer_url=issuer_url,
-            client_registration_options=ClientRegistrationOptions(enabled=False),
-            revocation_options=RevocationOptions(enabled=False),
-            required_scopes=required_scopes,
-            resource_server_url=resource_server_url,
-        )
-
-        self.algorithm = algorithm
-        self.issuer = issuer
-        self.audience = audience
-        self.public_key = public_key
-        self.jwks_uri = jwks_uri
-        self.jwt = JsonWebToken([self.algorithm])  # Use RS256 by default
-        self.logger = get_logger(__name__)
-
-        # 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")
-
-            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 | 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 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:
-            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")
-                jwk = JsonWebKey.import_key(key_data)
-                public_key = jwk.get_public_key()  # type: ignore
-
-                if key_kid:
-                    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
-
-            # Select the appropriate key
-            if kid:
-                if kid not in self._jwks_cache:
-                    self.logger.debug(
-                        "JWKS key lookup failed: key ID '%s' not found", kid
-                    )
-                    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:
-            self.logger.debug("JWKS fetch failed: %s", str(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)
-
-            # Extract client ID early for logging
-            client_id = claims.get("client_id") or claims.get("sub") or "unknown"
-
-            # Validate expiration
-            exp = claims.get("exp")
-            if exp and exp < time.time():
-                self.logger.debug(
-                    "Token validation failed: expired token for client %s", client_id
-                )
-                self.logger.info("Bearer token rejected for client %s", client_id)
-                return None
-
-            # Validate issuer - note we use issuer instead of issuer_url here because
-            # issuer is optional, allowing users to make this check optional
-            if self.issuer:
-                if claims.get("iss") != self.issuer:
-                    self.logger.debug(
-                        "Token validation failed: issuer mismatch for client %s",
-                        client_id,
-                    )
-                    self.logger.info("Bearer token rejected for client %s", client_id)
-                    return None
-
-            # Validate audience if configured
-            if self.audience:
-                aud = claims.get("aud")
-
-                # Handle different combinations of audience types
-                audience_valid = False
-                if isinstance(self.audience, list):
-                    # self.audience is a list - check if any expected audience is present
-                    if isinstance(aud, list):
-                        # Both are lists - check for intersection
-                        audience_valid = any(
-                            expected in aud for expected in self.audience
-                        )
-                    else:
-                        # aud is a string - check if it's in our expected list
-                        audience_valid = aud in self.audience
-                else:
-                    # self.audience is a string - use original logic
-                    if isinstance(aud, list):
-                        audience_valid = self.audience in aud
-                    else:
-                        audience_valid = aud == self.audience
-
-                if not audience_valid:
-                    self.logger.debug(
-                        "Token validation failed: audience mismatch for client %s",
-                        client_id,
-                    )
-                    self.logger.info("Bearer token rejected for client %s", client_id)
-                    return None
-
-            # Extract scopes
-            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:
-            self.logger.debug("Token validation failed: JWT signature/format invalid")
-            return None
-        except Exception as e:
-            self.logger.debug("Token validation failed: %s", str(e))
-            return None
-
-    def _extract_scopes(self, claims: dict[str, Any]) -> list[str]:
-        """
-        Extract scopes from JWT claims. Supports both 'scope' and 'scp'
-        claims.
-
-        Checks the `scope` claim first (standard OAuth2 claim), then the `scp`
-        claim (used by some Identity Providers).
-        """
-
-        for claim in ["scope", "scp"]:
-            if claim in claims:
-                if isinstance(claims[claim], str):
-                    return claims[claim].split()
-                elif isinstance(claims[claim], list):
-                    return claims[claim]
-
-        return []
-
-    async def verify_token(self, token: str) -> AccessToken | None:
-        """
-        Verify a bearer token and return access info if valid.
-
-        This method implements the TokenVerifier protocol by delegating
-        to our existing load_access_token method.
-
-        Args:
-            token: The JWT token string to validate
-
-        Returns:
-            AccessToken object if valid, None if invalid or expired
-        """
-        return await self.load_access_token(token)
-
-    # --- 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")
+"""Backwards compatibility shim for BearerAuthProvider.
+
+The BearerAuthProvider class has been moved to fastmcp.server.auth.verifiers.JWTVerifier
+for better organization. This module provides a backwards-compatible import.
+"""
+
+import warnings
+
+import fastmcp
+from fastmcp.server.auth.verifiers import JWKData, JWKSData, RSAKeyPair
+from fastmcp.server.auth.verifiers import JWTVerifier as BearerAuthProvider
+
+# Re-export for backwards compatibility
+__all__ = ["BearerAuthProvider", "RSAKeyPair", "JWKData", "JWKSData"]
+
+# Deprecated in 2.11
+if fastmcp.settings.deprecation_warnings:
+    warnings.warn(
+        "The `fastmcp.server.auth.providers.bearer` module is deprecated "
+        "and will be removed in a future version. "
+        "Please use `fastmcp.server.auth.verifiers.JWTVerifier` "
+        "instead of this module's BearerAuthProvider.",
+        DeprecationWarning,
+        stacklevel=2,
+    )

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

@@ -1,65 +0,0 @@
-from types import EllipsisType
-
-from pydantic_settings import BaseSettings, SettingsConfigDict
-
-from fastmcp.server.auth.providers.bearer import BearerAuthProvider
-
-
-class EnvBearerAuthProviderSettings(BaseSettings):
-    """Settings for the BearerAuthProvider."""
-
-    model_config = SettingsConfigDict(
-        env_prefix="FASTMCP_AUTH_BEARER_",
-        env_file=".env",
-        extra="ignore",
-    )
-
-    public_key: str | None = None
-    jwks_uri: str | None = None
-    issuer: str | None = None
-    algorithm: str | None = None
-    audience: str | None = None
-    required_scopes: list[str] | None = None
-
-
-class EnvBearerAuthProvider(BearerAuthProvider):
-    """
-    A BearerAuthProvider that loads settings from environment variables. Any
-    providing setting will always take precedence over the environment
-    variables.
-    """
-
-    def __init__(
-        self,
-        public_key: str | None | EllipsisType = ...,
-        jwks_uri: str | None | EllipsisType = ...,
-        issuer: str | None | EllipsisType = ...,
-        algorithm: str | None | EllipsisType = ...,
-        audience: str | None | EllipsisType = ...,
-        required_scopes: list[str] | None | EllipsisType = ...,
-        resource_server: str | None | EllipsisType = ...,
-    ):
-        """
-        Initialize the provider.
-
-        Args:
-            public_key: RSA public key in PEM format (for static key)
-            jwks_uri: URI to fetch keys from (for key rotation)
-            issuer: Expected issuer claim (optional)
-            algorithm: Algorithm to use for verification (optional)
-            audience: Expected audience claim (optional)
-            required_scopes: List of required scopes for access (optional)
-        """
-        kwargs = {
-            "public_key": public_key,
-            "jwks_uri": jwks_uri,
-            "issuer": issuer,
-            "algorithm": algorithm,
-            "audience": audience,
-            "required_scopes": required_scopes,
-            "resource_server": resource_server,
-        }
-        settings = EnvBearerAuthProviderSettings(
-            **{k: v for k, v in kwargs.items() if v is not ...}
-        )
-        super().__init__(**settings.model_dump())

src/fastmcp/server/auth/verifiers.py

@@ -0,0 +1,718 @@
+"""TokenVerifier implementations for FastMCP."""
+
+from __future__ import annotations
+
+import time
+from dataclasses import dataclass
+from typing import Any
+
+import httpx
+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, ValidationError
+from pydantic_settings import BaseSettings, SettingsConfigDict
+from typing_extensions import TypedDict
+
+from fastmcp.server.auth.auth import TokenVerifier
+from fastmcp.utilities.logging import get_logger
+from fastmcp.utilities.types import NotSet, NotSetT
+
+logger = get_logger(__name__)
+
+
+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:
+    """RSA key pair for JWT testing."""
+
+    private_key: SecretStr
+    public_key: str
+
+    @classmethod
+    def generate(cls) -> RSAKeyPair:
+        """
+        Generate an RSA key pair for testing.
+
+        Returns:
+            RSAKeyPair: Generated key pair
+        """
+        # Generate private key
+        private_key = rsa.generate_private_key(
+            public_exponent=65537,
+            key_size=2048,
+        )
+
+        # Serialize private key to PEM format
+        private_pem = private_key.private_bytes(
+            encoding=serialization.Encoding.PEM,
+            format=serialization.PrivateFormat.PKCS8,
+            encryption_algorithm=serialization.NoEncryption(),
+        ).decode("utf-8")
+
+        # Serialize public key to PEM format
+        public_pem = (
+            private_key.public_key()
+            .public_bytes(
+                encoding=serialization.Encoding.PEM,
+                format=serialization.PublicFormat.SubjectPublicKeyInfo,
+            )
+            .decode("utf-8")
+        )
+
+        return cls(
+            private_key=SecretStr(private_pem),
+            public_key=public_pem,
+        )
+
+    def create_token(
+        self,
+        subject: str = "fastmcp-user",
+        issuer: str = "https://fastmcp.example.com",
+        audience: str | list[str] | None = None,
+        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.
+
+        Args:
+            subject: Subject claim (usually user ID)
+            issuer: Issuer claim
+            audience: Audience claim - can be a string or list of strings (optional)
+            scopes: List of scopes to include
+            expires_in_seconds: Token expiration time in seconds
+            additional_claims: Any additional claims to include
+            kid: Key ID to include in header
+        """
+        import time
+
+        # Create header
+        header = {"alg": "RS256"}
+        if kid:
+            header["kid"] = kid
+
+        # Create payload
+        payload = {
+            "sub": subject,
+            "iss": issuer,
+            "iat": int(time.time()),
+            "exp": int(time.time()) + expires_in_seconds,
+        }
+
+        if audience:
+            payload["aud"] = audience
+
+        if scopes:
+            payload["scope"] = " ".join(scopes)
+
+        if additional_claims:
+            payload.update(additional_claims)
+
+        # Create JWT
+        jwt_lib = JsonWebToken(["RS256"])
+        token_bytes = jwt_lib.encode(
+            header, payload, self.private_key.get_secret_value()
+        )
+        return (
+            token_bytes.decode("utf-8")
+            if isinstance(token_bytes, bytes)
+            else token_bytes
+        )
+
+
+class JWTVerifier(TokenVerifier):
+    """
+    JWT token verifier using public key or JWKS.
+
+    This verifier validates JWT tokens signed by an external issuer. It's ideal for
+    scenarios where you have a centralized identity provider (like Auth0, Okta, or
+    your own OAuth server) that issues JWTs, and your FastMCP server acts as a
+    resource server validating those tokens.
+
+    Use this when:
+    - You have JWT tokens issued by an external service
+    - You want asymmetric key verification (public/private key pairs)
+    - You need JWKS support for automatic key rotation
+    - Your tokens contain standard OAuth scopes and claims
+    """
+
+    def __init__(
+        self,
+        public_key: str | None = None,
+        jwks_uri: str | None = None,
+        issuer: str | None = None,
+        audience: str | list[str] | None = None,
+        algorithm: str | None = None,
+        required_scopes: list[str] | None = None,
+        resource_server_url: AnyHttpUrl | str | None = None,
+    ):
+        """
+        Initialize the JWT token verifier.
+
+        Args:
+            public_key: PEM-encoded public key for verification
+            jwks_uri: URI to fetch JSON Web Key Set
+            issuer: Expected issuer claim
+            audience: Expected audience claim(s)
+            algorithm: JWT signing algorithm (default: RS256)
+            required_scopes: Required scopes for all tokens
+            resource_server_url: Resource server URL for TokenVerifier protocol
+        """
+        if not public_key and not 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")
+
+        if not algorithm:
+            algorithm = "RS256"
+        if algorithm not in {
+            "HS256",
+            "HS384",
+            "HS512",
+            "RS256",
+            "RS384",
+            "RS512",
+            "ES256",
+            "ES384",
+            "ES512",
+            "PS256",
+            "PS384",
+            "PS512",
+        }:
+            raise ValueError(f"Unsupported algorithm: {algorithm}.")
+
+        # Initialize parent TokenVerifier
+        super().__init__(
+            resource_server_url=resource_server_url, required_scopes=required_scopes
+        )
+
+        self.algorithm = algorithm
+        self.issuer = issuer
+        self.audience = audience
+        self.public_key = public_key
+        self.jwks_uri = jwks_uri
+        self.jwt = JsonWebToken([self.algorithm])
+        self.logger = get_logger(__name__)
+
+        # 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")
+
+            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 | 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 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:
+            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")
+                jwk = JsonWebKey.import_key(key_data)
+                public_key = jwk.get_public_key()  # type: ignore
+
+                if key_kid:
+                    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
+
+            # Select the appropriate key
+            if kid:
+                if kid not in self._jwks_cache:
+                    self.logger.debug(
+                        "JWKS key lookup failed: key ID '%s' not found", kid
+                    )
+                    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 httpx.HTTPError as e:
+            raise ValueError(f"Failed to fetch JWKS: {e}")
+        except Exception as e:
+            self.logger.debug(f"JWKS fetch failed: {e}")
+            raise ValueError(f"Failed to fetch JWKS: {e}")
+
+    def _extract_scopes(self, claims: dict[str, Any]) -> list[str]:
+        """
+        Extract scopes from JWT claims. Supports both 'scope' and 'scp'
+        claims.
+
+        Checks the `scope` claim first (standard OAuth2 claim), then the `scp`
+        claim (used by some Identity Providers).
+        """
+        for claim in ["scope", "scp"]:
+            if claim in claims:
+                if isinstance(claims[claim], str):
+                    return claims[claim].split()
+                elif isinstance(claims[claim], list):
+                    return claims[claim]
+
+        return []
+
+    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)
+
+            # Extract client ID early for logging
+            client_id = claims.get("client_id") or claims.get("sub") or "unknown"
+
+            # Validate expiration
+            exp = claims.get("exp")
+            if exp and exp < time.time():
+                self.logger.debug(
+                    "Token validation failed: expired token for client %s", client_id
+                )
+                self.logger.info("Bearer token rejected for client %s", client_id)
+                return None
+
+            # Validate issuer - note we use issuer instead of issuer_url here because
+            # issuer is optional, allowing users to make this check optional
+            if self.issuer:
+                if claims.get("iss") != self.issuer:
+                    self.logger.debug(
+                        "Token validation failed: issuer mismatch for client %s",
+                        client_id,
+                    )
+                    self.logger.info("Bearer token rejected for client %s", client_id)
+                    return None
+
+            # Validate audience if configured
+            if self.audience:
+                aud = claims.get("aud")
+
+                # Handle different combinations of audience types
+                audience_valid = False
+                if isinstance(self.audience, list):
+                    # self.audience is a list - check if any expected audience is present
+                    if isinstance(aud, list):
+                        # Both are lists - check for intersection
+                        audience_valid = any(
+                            expected in aud for expected in self.audience
+                        )
+                    else:
+                        # aud is a string - check if it's in our expected list
+                        audience_valid = aud in self.audience
+                else:
+                    # self.audience is a string - use original logic
+                    if isinstance(aud, list):
+                        audience_valid = self.audience in aud
+                    else:
+                        audience_valid = aud == self.audience
+
+                if not audience_valid:
+                    self.logger.debug(
+                        "Token validation failed: audience mismatch for client %s",
+                        client_id,
+                    )
+                    self.logger.info("Bearer token rejected for client %s", client_id)
+                    return None
+
+            # Extract scopes
+            scopes = self._extract_scopes(claims)
+
+            # Check required scopes
+            if self.required_scopes:
+                token_scopes = set(scopes)
+                required_scopes = set(self.required_scopes)
+                if not required_scopes.issubset(token_scopes):
+                    self.logger.debug(
+                        "Token missing required scopes. Has: %s, Required: %s",
+                        token_scopes,
+                        required_scopes,
+                    )
+                    self.logger.info("Bearer token rejected for client %s", client_id)
+                    return None
+
+            return AccessToken(
+                token=token,
+                client_id=str(client_id),
+                scopes=scopes,
+                expires_at=int(exp) if exp else None,
+            )
+
+        except JoseError:
+            self.logger.debug("Token validation failed: JWT signature/format invalid")
+            return None
+        except Exception as e:
+            self.logger.debug("Token validation failed: %s", str(e))
+            return None
+
+    async def verify_token(self, token: str) -> AccessToken | None:
+        """
+        Verify a bearer token and return access info if valid.
+
+        This method implements the TokenVerifier protocol by delegating
+        to our existing load_access_token method.
+
+        Args:
+            token: The JWT token string to validate
+
+        Returns:
+            AccessToken object if valid, None if invalid or expired
+        """
+        return await self.load_access_token(token)
+
+
+class JWTVerifierSettings(BaseSettings):
+    """Settings for the BearerAuthProvider."""
+
+    model_config = SettingsConfigDict(
+        env_prefix="FASTMCP_AUTH_JWT_",
+        env_file=".env",
+        extra="ignore",
+    )
+
+    public_key: str | None = None
+    jwks_uri: str | None = None
+    issuer: str | None = None
+    algorithm: str | None = None
+    audience: str | None = None
+    required_scopes: list[str] | None = None
+    resource_server_url: AnyHttpUrl | str | None = None
+
+
+class EnvJWTVerifier(JWTVerifier):
+    def __init__(
+        self,
+        public_key: str | None | NotSetT = NotSet,
+        jwks_uri: str | None | NotSetT = NotSet,
+        issuer: str | None | NotSetT = NotSet,
+        audience: str | list[str] | None | NotSetT = NotSet,
+        algorithm: str | None | NotSetT = NotSet,
+        required_scopes: list[str] | None | NotSetT = NotSet,
+        resource_server_url: AnyHttpUrl | str | None | NotSetT = NotSet,
+    ):
+        kwargs = {
+            "public_key": public_key,
+            "jwks_uri": jwks_uri,
+            "issuer": issuer,
+            "algorithm": algorithm,
+            "audience": audience,
+            "required_scopes": required_scopes,
+            "resource_server_url": resource_server_url,
+        }
+        settings = JWTVerifierSettings(
+            **{k: v for k, v in kwargs.items() if v is not NotSet}
+        )
+        super().__init__(**settings.model_dump())
+
+
+class IntrospectionTokenVerifier(TokenVerifier):
+    """
+    OAuth 2.0 Token Introspection verifier (RFC 7662).
+
+    This verifier validates tokens by making real-time calls to an OAuth 2.0
+    authorization server's introspection endpoint. Unlike JWT verification, this
+    approach works with both opaque tokens and JWTs, and provides real-time
+    validation including immediate revocation support.
+
+    Use this when:
+    - Your authorization server is separate from your FastMCP server
+    - You're using opaque (non-JWT) tokens
+    - You need real-time token validation and revocation support
+    - Your authorization server supports RFC 7662 introspection
+    - You want centralized token management without sharing secrets
+    """
+
+    def __init__(
+        self,
+        introspection_endpoint: AnyHttpUrl | str,
+        server_url: AnyHttpUrl | str,
+        client_id: str | None = None,
+        client_secret: str | None = None,
+        validate_resource: bool = False,
+        required_scopes: list[str] | None = None,
+        timeout: float = 10.0,
+    ):
+        """
+        Initialize the introspection token verifier.
+
+        Args:
+            introspection_endpoint: OAuth 2.0 introspection endpoint URL
+            server_url: This server's URL for resource validation
+            client_id: Client ID for introspection authentication
+            client_secret: Client secret for introspection authentication
+            validate_resource: Whether to validate RFC 8707 resource parameter
+            required_scopes: Required scopes for all tokens
+            timeout: HTTP request timeout in seconds
+        """
+        try:
+            self.introspection_endpoint = AnyHttpUrl(introspection_endpoint)
+            server_url_validated = AnyHttpUrl(server_url)
+        except ValidationError as e:
+            raise ValueError(f"Invalid URL provided: {e}") from e
+
+        # Basic SSRF protection - reject private/localhost URLs
+        if self._is_private_url(str(self.introspection_endpoint)):
+            raise ValueError("Introspection endpoint cannot be a private/localhost URL")
+
+        # Initialize parent TokenVerifier with the resource server URL
+        super().__init__(
+            resource_server_url=server_url_validated, required_scopes=required_scopes
+        )
+
+        self.client_id = client_id
+        self.client_secret = client_secret
+        self.validate_resource = validate_resource
+        self.timeout = timeout
+
+        # Create HTTP client with security settings
+        self._client = httpx.AsyncClient(
+            timeout=timeout,
+            verify=True,  # Always verify SSL
+            limits=httpx.Limits(max_connections=10, max_keepalive_connections=5),
+        )
+
+    @property
+    def server_url(self) -> AnyHttpUrl:
+        """The resource server URL for this verifier."""
+        if self.resource_server_url is None:
+            raise ValueError("Resource server URL not set")
+        return self.resource_server_url
+
+    def _is_private_url(self, url: str) -> bool:
+        """Check if URL points to private/localhost addresses (basic SSRF protection)."""
+        import ipaddress
+        from urllib.parse import urlparse
+
+        parsed = urlparse(url)
+        hostname = parsed.hostname
+
+        if not hostname:
+            return False
+
+        # Check for localhost
+        if hostname.lower() in ("localhost", "127.0.0.1", "::1"):
+            return True
+
+        # Check for private IP ranges
+        try:
+            ip = ipaddress.ip_address(hostname)
+            return ip.is_private or ip.is_loopback
+        except ValueError:
+            # Not an IP address, assume it's a hostname
+            return False
+
+    async def verify_token(self, token: str) -> AccessToken | None:
+        """Verify token using OAuth 2.0 introspection."""
+        try:
+            # Prepare introspection request
+            data = {"token": token}
+
+            # Add resource parameter if validation is enabled (RFC 8707)
+            if self.validate_resource and self.resource_server_url:
+                data["resource"] = str(self.resource_server_url)
+
+            # Prepare authentication and make introspection request
+            if self.client_id and self.client_secret:
+                response = await self._client.post(
+                    str(self.introspection_endpoint),
+                    data=data,
+                    auth=(self.client_id, self.client_secret),
+                    headers={"Content-Type": "application/x-www-form-urlencoded"},
+                )
+            else:
+                response = await self._client.post(
+                    str(self.introspection_endpoint),
+                    data=data,
+                    headers={"Content-Type": "application/x-www-form-urlencoded"},
+                )
+            response.raise_for_status()
+
+            introspection_response = response.json()
+
+            # Check if token is active
+            if not introspection_response.get("active", False):
+                return None
+
+            # Extract token information
+            client_id = introspection_response.get("client_id", "unknown")
+            scopes = (
+                introspection_response.get("scope", "").split()
+                if introspection_response.get("scope")
+                else []
+            )
+            exp = introspection_response.get("exp")
+
+            # Check required scopes
+            if self.required_scopes:
+                token_scopes = set(scopes)
+                required_scopes = set(self.required_scopes)
+                if not required_scopes.issubset(token_scopes):
+                    logger.debug(
+                        f"Token missing required scopes. Has: {token_scopes}, Required: {required_scopes}"
+                    )
+                    return None
+
+            return AccessToken(
+                token=token,
+                client_id=client_id,
+                scopes=scopes,
+                expires_at=exp,
+                resource=str(self.resource_server_url)
+                if self.resource_server_url
+                else None,
+            )
+
+        except Exception as e:
+            logger.debug(f"Introspection verification failed: {e}")
+            return None
+
+    async def __aenter__(self):
+        """Async context manager entry."""
+        return self
+
+    async def __aexit__(self, exc_type, exc_val, exc_tb):
+        """Async context manager exit."""
+        await self._client.aclose()
+
+
+class StaticTokenVerifier(TokenVerifier):
+    """
+    Simple static token verifier for testing and development.
+
+    This verifier validates tokens against a predefined dictionary of valid token
+    strings and their associated claims. When a token string matches a key in the
+    dictionary, the verifier returns the corresponding claims as if the token was
+    validated by a real authorization server.
+
+    Use this when:
+    - You're developing or testing locally without a real OAuth server
+    - You need predictable tokens for automated testing
+    - You want to simulate different users/scopes without complex setup
+    - You're prototyping and need simple API key-style authentication
+
+    WARNING: Never use this in production - tokens are stored in plain text!
+    """
+
+    def __init__(
+        self,
+        tokens: dict[str, dict[str, Any]],
+        required_scopes: list[str] | None = None,
+    ):
+        """
+        Initialize the static token verifier.
+
+        Args:
+            tokens: Dict mapping token strings to token metadata
+                   Each token should have: client_id, scopes, expires_at (optional)
+            required_scopes: Required scopes for all tokens
+        """
+        super().__init__(required_scopes=required_scopes)
+        self.tokens = tokens
+
+    async def verify_token(self, token: str) -> AccessToken | None:
+        """Verify token against static token dictionary."""
+        token_data = self.tokens.get(token)
+        if not token_data:
+            return None
+
+        # Check expiration if present
+        expires_at = token_data.get("expires_at")
+        if expires_at is not None and expires_at < time.time():
+            return None
+
+        scopes = token_data.get("scopes", [])
+
+        # Check required scopes
+        if self.required_scopes:
+            token_scopes = set(scopes)
+            required_scopes = set(self.required_scopes)
+            if not required_scopes.issubset(token_scopes):
+                logger.debug(
+                    f"Token missing required scopes. Has: {token_scopes}, Required: {required_scopes}"
+                )
+                return None
+
+        return AccessToken(
+            token=token,
+            client_id=token_data["client_id"],
+            scopes=scopes,
+            expires_at=expires_at,
+        )

src/fastmcp/server/http.py

@@ -3,13 +3,14 @@ from __future__ import annotations
 from collections.abc import AsyncGenerator, Callable, Generator
 from contextlib import asynccontextmanager, contextmanager
 from contextvars import ContextVar
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, cast
 
 from mcp.server.auth.middleware.auth_context import AuthContextMiddleware
 from mcp.server.auth.middleware.bearer_auth import (
     BearerAuthBackend,
     RequireAuthMiddleware,
 )
+from mcp.server.auth.provider import TokenVerifier as TokenVerifierProtocol
 from mcp.server.auth.routes import create_auth_routes
 from mcp.server.lowlevel.server import LifespanResultT
 from mcp.server.sse import SseServerTransport
@@ -24,7 +25,7 @@ from starlette.responses import Response
 from starlette.routing import BaseRoute, Mount, Route
 from starlette.types import Lifespan, Receive, Scope, Send
 
-from fastmcp.server.auth.auth import OAuthProvider
+from fastmcp.server.auth.auth import OAuthProvider, TokenVerifier
 from fastmcp.utilities.logging import get_logger
 
 if TYPE_CHECKING:
@@ -71,39 +72,45 @@ class RequestContextMiddleware:
 
 
 def setup_auth_middleware_and_routes(
-    auth: OAuthProvider,
+    auth: OAuthProvider | TokenVerifier,
 ) -> tuple[list[Middleware], list[BaseRoute], list[str]]:
     """Set up authentication middleware and routes if auth is enabled.
 
     Args:
-        auth: The OAuthProvider authorization server provider
+        auth: Either an OAuthProvider or TokenVerifier for authentication
 
     Returns:
         Tuple of (middleware, auth_routes, required_scopes)
     """
-    middleware: list[Middleware] = []
-    auth_routes: list[BaseRoute] = []
-    required_scopes: list[str] = []
-
-    middleware = [
+    middleware: list[Middleware] = [
         Middleware(
             AuthenticationMiddleware,
-            backend=BearerAuthBackend(auth),
+            backend=BearerAuthBackend(cast(TokenVerifierProtocol, auth)),
         ),
         Middleware(AuthContextMiddleware),
     ]
 
-    required_scopes = auth.required_scopes or []
+    auth_routes: list[BaseRoute] = []
+    required_scopes: list[str] = []
 
-    auth_routes.extend(
-        create_auth_routes(
-            provider=auth,
-            issuer_url=auth.issuer_url,
-            service_documentation_url=auth.service_documentation_url,
-            client_registration_options=auth.client_registration_options,
-            revocation_options=auth.revocation_options,
+    # Handle TokenVerifier vs OAuthProvider
+    # Check if it's an OAuthProvider by looking for issuer_url attribute
+    if hasattr(auth, "issuer_url"):
+        # OAuthProvider: create auth routes and get required scopes
+        # We know this is an OAuthProvider because it has issuer_url
+        auth_routes = list(
+            create_auth_routes(
+                provider=auth,  # type: ignore[arg-type]
+                issuer_url=auth.issuer_url,  # type: ignore[attr-defined]
+                service_documentation_url=auth.service_documentation_url,  # type: ignore[attr-defined]
+                client_registration_options=auth.client_registration_options,  # type: ignore[attr-defined]
+                revocation_options=auth.revocation_options,  # type: ignore[attr-defined]
+            )
         )
-    )
+        required_scopes = auth.required_scopes or []  # type: ignore[attr-defined]
+    else:
+        # TokenVerifier: no auth routes but may have required scopes
+        required_scopes = getattr(auth, "required_scopes", None) or []
 
     return middleware, auth_routes, required_scopes
 
@@ -140,7 +147,7 @@ def create_sse_app(
     server: FastMCP[LifespanResultT],
     message_path: str,
     sse_path: str,
-    auth: OAuthProvider | None = None,
+    auth: OAuthProvider | TokenVerifier | None = None,
     debug: bool = False,
     routes: list[BaseRoute] | None = None,
     middleware: list[Middleware] | None = None,
@@ -151,7 +158,7 @@ def create_sse_app(
         server: The FastMCP server instance
         message_path: Path for SSE messages
         sse_path: Path for SSE connections
-        auth: Optional auth provider
+        auth: Optional authentication provider (OAuthProvider or TokenVerifier)
         debug: Whether to enable debug mode
         routes: Optional list of custom routes
         middleware: Optional list of middleware
@@ -176,8 +183,6 @@ def create_sse_app(
         return Response()
 
     # Get auth middleware and routes
-
-    # Add SSE routes with or without auth
     if auth:
         auth_middleware, auth_routes, required_scopes = (
             setup_auth_middleware_and_routes(auth)
@@ -185,18 +190,32 @@ def create_sse_app(
 
         server_routes.extend(auth_routes)
         server_middleware.extend(auth_middleware)
+
+        # Determine resource_metadata_url for TokenVerifier
+        resource_metadata_url = None
+        if isinstance(auth, TokenVerifier) and auth.resource_server_url:
+            # Add .well-known path for RFC 9728 compliance
+            resource_metadata_url = AnyHttpUrl(
+                str(auth.resource_server_url).rstrip("/")
+                + "/.well-known/oauth-protected-resource"
+            )
+
         # Auth is enabled, wrap endpoints with RequireAuthMiddleware
         server_routes.append(
             Route(
                 sse_path,
-                endpoint=RequireAuthMiddleware(handle_sse, required_scopes),
+                endpoint=RequireAuthMiddleware(
+                    handle_sse, required_scopes, resource_metadata_url
+                ),
                 methods=["GET"],
             )
         )
         server_routes.append(
             Mount(
                 message_path,
-                app=RequireAuthMiddleware(sse.handle_post_message, required_scopes),
+                app=RequireAuthMiddleware(
+                    sse.handle_post_message, required_scopes, resource_metadata_url
+                ),
             )
         )
     else:
@@ -244,7 +263,7 @@ def create_streamable_http_app(
     server: FastMCP[LifespanResultT],
     streamable_http_path: str,
     event_store: EventStore | None = None,
-    auth: OAuthProvider | None = None,
+    auth: OAuthProvider | TokenVerifier | None = None,
     json_response: bool = False,
     stateless_http: bool = False,
     debug: bool = False,
@@ -257,7 +276,7 @@ def create_streamable_http_app(
         server: The FastMCP server instance
         streamable_http_path: Path for StreamableHTTP connections
         event_store: Optional event store for session management
-        auth: Optional auth provider
+        auth: Optional authentication provider (OAuthProvider or TokenVerifier)
         json_response: Whether to use JSON response format
         stateless_http: Whether to use stateless mode (new transport per request)
         debug: Whether to enable debug mode
@@ -310,7 +329,7 @@ def create_streamable_http_app(
     if auth:
         resource_metadata_url = None
 
-        if auth.resource_server_url:
+        if isinstance(auth, TokenVerifier) and auth.resource_server_url:
             resource_metadata_url = AnyHttpUrl(
                 str(auth.resource_server_url).rstrip("/")
                 + "/.well-known/oauth-protected-resource"
@@ -323,6 +342,15 @@ def create_streamable_http_app(
         server_routes.extend(auth_routes)
         server_middleware.extend(auth_middleware)
 
+        # Determine resource_metadata_url for TokenVerifier
+        resource_metadata_url = None
+        if isinstance(auth, TokenVerifier) and auth.resource_server_url:
+            # Add .well-known path for RFC 9728 compliance
+            resource_metadata_url = AnyHttpUrl(
+                str(auth.resource_server_url).rstrip("/")
+                + "/.well-known/oauth-protected-resource"
+            )
+
         # Auth is enabled, wrap endpoint with RequireAuthMiddleware
         server_routes.append(
             Mount(

src/fastmcp/server/server.py

@@ -50,8 +50,8 @@ from fastmcp.prompts import Prompt, PromptManager
 from fastmcp.prompts.prompt import FunctionPrompt
 from fastmcp.resources import Resource, ResourceManager
 from fastmcp.resources.template import ResourceTemplate
-from fastmcp.server.auth.auth import OAuthProvider
-from fastmcp.server.auth.providers.bearer_env import EnvBearerAuthProvider
+from fastmcp.server.auth.auth import OAuthProvider, TokenVerifier
+from fastmcp.server.auth.verifiers import EnvJWTVerifier
 from fastmcp.server.http import (
     StarletteWithLifespan,
     create_sse_app,
@@ -133,7 +133,7 @@ class FastMCP(Generic[LifespanResultT]):
         instructions: str | None = None,
         *,
         version: str | None = None,
-        auth: OAuthProvider | None = None,
+        auth: OAuthProvider | TokenVerifier | None = None,
         middleware: list[Middleware] | None = None,
         lifespan: (
             Callable[
@@ -205,8 +205,9 @@ class FastMCP(Generic[LifespanResultT]):
             lifespan=_lifespan_wrapper(self, lifespan),
         )
 
-        if auth is None and fastmcp.settings.default_auth_provider == "bearer_env":
-            auth = EnvBearerAuthProvider()
+        if auth is None and fastmcp.settings.default_auth_provider == "jwt-env":
+            auth = EnvJWTVerifier()
+
         self.auth = auth
 
         if tools:

src/fastmcp/settings.py

@@ -260,7 +260,7 @@ class Settings(BaseSettings):
 
     # Auth settings
     default_auth_provider: Annotated[
-        Literal["bearer_env"] | None,
+        Literal["jwt-env"] | None,
         Field(
             description=inspect.cleandoc(
                 """

tests/auth/providers/test_token_verifier.py

@@ -3,12 +3,12 @@
 import pytest
 from mcp.server.auth.provider import AccessToken
 
-from fastmcp.server.auth.providers.bearer import BearerAuthProvider, RSAKeyPair
 from fastmcp.server.auth.providers.in_memory import InMemoryOAuthProvider
+from fastmcp.server.auth.verifiers import JWTVerifier, RSAKeyPair
 
 
-class TestBearerAuthProviderTokenVerifier:
-    """Test that BearerAuthProvider implements TokenVerifier protocol correctly."""
+class TestJWTVerifierTokenVerifier:
+    """Test that JWTVerifier implements TokenVerifier protocol correctly."""
 
     @pytest.fixture
     def rsa_key_pair(self) -> RSAKeyPair:
@@ -16,9 +16,9 @@ class TestBearerAuthProviderTokenVerifier:
         return RSAKeyPair.generate()
 
     @pytest.fixture
-    def bearer_provider(self, rsa_key_pair: RSAKeyPair) -> BearerAuthProvider:
-        """Create BearerAuthProvider for testing."""
-        return BearerAuthProvider(
+    def jwt_verifier(self, rsa_key_pair: RSAKeyPair) -> JWTVerifier:
+        """Create JWTVerifier for testing."""
+        return JWTVerifier(
             public_key=rsa_key_pair.public_key,
             issuer="https://test.example.com",
             audience="https://api.example.com",
@@ -45,10 +45,10 @@ class TestBearerAuthProviderTokenVerifier:
         )
 
     async def test_verify_token_with_valid_token(
-        self, bearer_provider: BearerAuthProvider, valid_token: str
+        self, jwt_verifier: JWTVerifier, valid_token: str
     ):
         """Test that verify_token returns AccessToken for valid token."""
-        result = await bearer_provider.verify_token(valid_token)
+        result = await jwt_verifier.verify_token(valid_token)
 
         assert result is not None
         assert isinstance(result, AccessToken)
@@ -58,33 +58,29 @@ class TestBearerAuthProviderTokenVerifier:
         assert "write" in result.scopes
 
     async def test_verify_token_with_expired_token(
-        self, bearer_provider: BearerAuthProvider, expired_token: str
+        self, jwt_verifier: JWTVerifier, expired_token: str
     ):
         """Test that verify_token returns None for expired token."""
-        result = await bearer_provider.verify_token(expired_token)
+        result = await jwt_verifier.verify_token(expired_token)
         assert result is None
 
-    async def test_verify_token_with_invalid_token(
-        self, bearer_provider: BearerAuthProvider
-    ):
+    async def test_verify_token_with_invalid_token(self, jwt_verifier: JWTVerifier):
         """Test that verify_token returns None for invalid token."""
-        result = await bearer_provider.verify_token("invalid.token.here")
+        result = await jwt_verifier.verify_token("invalid.token.here")
         assert result is None
 
-    async def test_verify_token_with_malformed_token(
-        self, bearer_provider: BearerAuthProvider
-    ):
+    async def test_verify_token_with_malformed_token(self, jwt_verifier: JWTVerifier):
         """Test that verify_token returns None for malformed token."""
-        result = await bearer_provider.verify_token("not-a-jwt")
+        result = await jwt_verifier.verify_token("not-a-jwt")
         assert result is None
 
     async def test_verify_token_delegation_to_load_access_token(
-        self, bearer_provider: BearerAuthProvider, valid_token: str
+        self, jwt_verifier: JWTVerifier, valid_token: str
     ):
         """Test that verify_token delegates to load_access_token."""
         # Both methods should return the same result
-        verify_result = await bearer_provider.verify_token(valid_token)
-        load_result = await bearer_provider.load_access_token(valid_token)
+        verify_result = await jwt_verifier.verify_token(valid_token)
+        load_result = await jwt_verifier.load_access_token(valid_token)
 
         assert verify_result == load_result
         if verify_result is not None and load_result is not None:
@@ -162,9 +158,9 @@ class TestTokenVerifierProtocolCompliance:
     """Test that our providers properly implement the TokenVerifier protocol."""
 
     async def test_bearer_provider_implements_protocol(self):
-        """Test that BearerAuthProvider can be used as TokenVerifier."""
+        """Test that JWTVerifier can be used as TokenVerifier."""
         key_pair = RSAKeyPair.generate()
-        provider = BearerAuthProvider(public_key=key_pair.public_key)
+        provider = JWTVerifier(public_key=key_pair.public_key)
 
         # Should have the required method for TokenVerifier protocol
         assert hasattr(provider, "verify_token")

tests/auth/{providers/test_bearer_env.py → verifiers/test_env_jwt.py}

@@ -1,9 +1,8 @@
 import pytest
-from pydantic import AnyHttpUrl, ValidationError
+from pydantic import ValidationError
 
 from fastmcp import FastMCP
-from fastmcp.server.auth.providers.bearer import BearerAuthProvider
-from fastmcp.server.auth.providers.bearer_env import EnvBearerAuthProvider
+from fastmcp.server.auth.verifiers import EnvJWTVerifier, JWTVerifier
 from fastmcp.settings import Settings
 from fastmcp.utilities.tests import temporary_settings
 
@@ -12,19 +11,19 @@ def test_load_bearer_env_from_env_var(monkeypatch):
     mcp = FastMCP()
     assert mcp.auth is None
 
-    monkeypatch.setenv("FASTMCP_DEFAULT_AUTH_PROVIDER", "bearer_env")
-    monkeypatch.setenv("FASTMCP_AUTH_BEARER_PUBLIC_KEY", "test-public-key")
+    monkeypatch.setenv("FASTMCP_DEFAULT_AUTH_PROVIDER", "jwt-env")
+    monkeypatch.setenv("FASTMCP_AUTH_JWT_PUBLIC_KEY", "test-public-key")
 
     with temporary_settings(**Settings().model_dump()):
         mcp_with_auth = FastMCP()
-    assert isinstance(mcp_with_auth.auth, EnvBearerAuthProvider)
+    assert isinstance(mcp_with_auth.auth, EnvJWTVerifier)
 
 
 def test_load_bearer_env_from_env_var_requires_public_key_or_jwks_uri(monkeypatch):
     mcp = FastMCP()
     assert mcp.auth is None
 
-    monkeypatch.setenv("FASTMCP_DEFAULT_AUTH_PROVIDER", "bearer_env")
+    monkeypatch.setenv("FASTMCP_DEFAULT_AUTH_PROVIDER", "jwt-env")
 
     with temporary_settings(**Settings().model_dump()):
         with pytest.raises(
@@ -34,26 +33,25 @@ def test_load_bearer_env_from_env_var_requires_public_key_or_jwks_uri(monkeypatc
 
 
 def test_configure_bearer_env_from_env_var(monkeypatch):
-    monkeypatch.setenv("FASTMCP_DEFAULT_AUTH_PROVIDER", "bearer_env")
-    monkeypatch.setenv("FASTMCP_AUTH_BEARER_PUBLIC_KEY", "test-public-key")
-    monkeypatch.setenv("FASTMCP_AUTH_BEARER_ISSUER", "http://test-issuer")
-    monkeypatch.setenv("FASTMCP_AUTH_BEARER_AUDIENCE", "test-audience")
+    monkeypatch.setenv("FASTMCP_DEFAULT_AUTH_PROVIDER", "jwt-env")
+    monkeypatch.setenv("FASTMCP_AUTH_JWT_PUBLIC_KEY", "test-public-key")
+    monkeypatch.setenv("FASTMCP_AUTH_JWT_ISSUER", "http://test-issuer")
+    monkeypatch.setenv("FASTMCP_AUTH_JWT_AUDIENCE", "test-audience")
     monkeypatch.setenv(
-        "FASTMCP_AUTH_BEARER_REQUIRED_SCOPES", '["test-scope1", "test-scope2"]'
+        "FASTMCP_AUTH_JWT_REQUIRED_SCOPES", '["test-scope1", "test-scope2"]'
     )
 
     with temporary_settings(**Settings().model_dump()):
         mcp = FastMCP()
-    assert isinstance(mcp.auth, EnvBearerAuthProvider)
+    assert isinstance(mcp.auth, EnvJWTVerifier)
     assert mcp.auth.public_key == "test-public-key"
-    assert mcp.auth.issuer_url == AnyHttpUrl("http://test-issuer")
     assert mcp.auth.audience == "test-audience"
     assert mcp.auth.required_scopes == ["test-scope1", "test-scope2"]
 
 
 def test_list_of_scopes_must_be_a_list(monkeypatch):
-    monkeypatch.setenv("FASTMCP_DEFAULT_AUTH_PROVIDER", "bearer_env")
-    monkeypatch.setenv("FASTMCP_AUTH_BEARER_REQUIRED_SCOPES", "test-scope1")
+    monkeypatch.setenv("FASTMCP_DEFAULT_AUTH_PROVIDER", "jwt-env")
+    monkeypatch.setenv("FASTMCP_AUTH_JWT_REQUIRED_SCOPES", "test-scope1")
 
     with temporary_settings(**Settings().model_dump()):
         with pytest.raises(ValidationError, match="Input should be a valid list"):
@@ -61,19 +59,19 @@ def test_list_of_scopes_must_be_a_list(monkeypatch):
 
 
 def test_configure_bearer_env_jwks_uri_from_env_var(monkeypatch):
-    monkeypatch.setenv("FASTMCP_DEFAULT_AUTH_PROVIDER", "bearer_env")
-    monkeypatch.setenv("FASTMCP_AUTH_BEARER_JWKS_URI", "test-jwks-uri")
+    monkeypatch.setenv("FASTMCP_DEFAULT_AUTH_PROVIDER", "jwt-env")
+    monkeypatch.setenv("FASTMCP_AUTH_JWT_JWKS_URI", "test-jwks-uri")
 
     with temporary_settings(**Settings().model_dump()):
         mcp = FastMCP()
-        assert isinstance(mcp.auth, EnvBearerAuthProvider)
+        assert isinstance(mcp.auth, EnvJWTVerifier)
         assert mcp.auth.jwks_uri == "test-jwks-uri"
 
 
 def test_configure_bearer_env_public_key_and_jwks_uri_error(monkeypatch):
-    monkeypatch.setenv("FASTMCP_DEFAULT_AUTH_PROVIDER", "bearer_env")
-    monkeypatch.setenv("FASTMCP_AUTH_BEARER_PUBLIC_KEY", "test-public-key")
-    monkeypatch.setenv("FASTMCP_AUTH_BEARER_JWKS_URI", "test-jwks-uri")
+    monkeypatch.setenv("FASTMCP_DEFAULT_AUTH_PROVIDER", "jwt-env")
+    monkeypatch.setenv("FASTMCP_AUTH_JWT_PUBLIC_KEY", "test-public-key")
+    monkeypatch.setenv("FASTMCP_AUTH_JWT_JWKS_URI", "test-jwks-uri")
 
     with temporary_settings(**Settings().model_dump()):
         with pytest.raises(ValueError, match="Provide either public_key or jwks_uri"):
@@ -81,11 +79,11 @@ def test_configure_bearer_env_public_key_and_jwks_uri_error(monkeypatch):
 
 
 def test_provided_auth_takes_precedence_over_env_vars(monkeypatch):
-    monkeypatch.setenv("FASTMCP_DEFAULT_AUTH_PROVIDER", "bearer_env")
-    monkeypatch.setenv("FASTMCP_AUTH_BEARER_PUBLIC_KEY", "test-public-key")
+    monkeypatch.setenv("FASTMCP_DEFAULT_AUTH_PROVIDER", "jwt-env")
+    monkeypatch.setenv("FASTMCP_AUTH_JWT_PUBLIC_KEY", "test-public-key")
 
     with temporary_settings(**Settings().model_dump()):
-        mcp = FastMCP(auth=BearerAuthProvider(public_key="test-public-key-2"))
-        assert isinstance(mcp.auth, BearerAuthProvider)
-        assert not isinstance(mcp.auth, EnvBearerAuthProvider)
+        mcp = FastMCP(auth=JWTVerifier(public_key="test-public-key-2"))
+        assert isinstance(mcp.auth, JWTVerifier)
+        assert not isinstance(mcp.auth, EnvJWTVerifier)
         assert mcp.auth.public_key == "test-public-key-2"

tests/auth/{providers/test_bearer.py → verifiers/test_jwt_verifier.py}

@@ -7,12 +7,7 @@ from pytest_httpx import HTTPXMock
 
 from fastmcp import Client, FastMCP
 from fastmcp.client.auth.bearer import BearerAuth
-from fastmcp.server.auth.providers.bearer import (
-    BearerAuthProvider,
-    JWKData,
-    JWKSData,
-    RSAKeyPair,
-)
+from fastmcp.server.auth.verifiers import JWKData, JWKSData, JWTVerifier, RSAKeyPair
 from fastmcp.utilities.tests import run_server_in_process
 
 
@@ -31,8 +26,8 @@ def bearer_token(rsa_key_pair: RSAKeyPair) -> str:
 
 
 @pytest.fixture
-def bearer_provider(rsa_key_pair: RSAKeyPair) -> BearerAuthProvider:
-    return BearerAuthProvider(
+def bearer_provider(rsa_key_pair: RSAKeyPair) -> JWTVerifier:
+    return JWTVerifier(
         public_key=rsa_key_pair.public_key,
         issuer="https://test.example.com",
         audience="https://api.example.com",
@@ -47,7 +42,7 @@ def run_mcp_server(
     run_kwargs: dict[str, Any] | None = None,
 ) -> None:
     mcp = FastMCP(
-        auth=BearerAuthProvider(
+        auth=JWTVerifier(
             public_key=public_key,
             **auth_kwargs or {},
         )
@@ -113,9 +108,9 @@ class TestBearerTokenJWKS:
     """Tests for JWKS URI functionality."""
 
     @pytest.fixture
-    def jwks_provider(self, rsa_key_pair: RSAKeyPair) -> BearerAuthProvider:
+    def jwks_provider(self, rsa_key_pair: RSAKeyPair) -> JWTVerifier:
         """Provider configured with JWKS URI."""
-        return BearerAuthProvider(
+        return JWTVerifier(
             jwks_uri="https://test.example.com/.well-known/jwks.json",
             issuer="https://test.example.com",
             audience="https://api.example.com",
@@ -137,7 +132,7 @@ class TestBearerTokenJWKS:
     async def test_jwks_token_validation(
         self,
         rsa_key_pair: RSAKeyPair,
-        jwks_provider: BearerAuthProvider,
+        jwks_provider: JWTVerifier,
         mock_jwks_data: JWKSData,
         httpx_mock: HTTPXMock,
     ):
@@ -159,7 +154,7 @@ class TestBearerTokenJWKS:
     async def test_jwks_token_validation_with_invalid_key(
         self,
         rsa_key_pair: RSAKeyPair,
-        jwks_provider: BearerAuthProvider,
+        jwks_provider: JWTVerifier,
         mock_jwks_data: JWKSData,
         httpx_mock: HTTPXMock,
     ):
@@ -179,7 +174,7 @@ class TestBearerTokenJWKS:
     async def test_jwks_token_validation_with_kid(
         self,
         rsa_key_pair: RSAKeyPair,
-        jwks_provider: BearerAuthProvider,
+        jwks_provider: JWTVerifier,
         mock_jwks_data: JWKSData,
         httpx_mock: HTTPXMock,
     ):
@@ -202,7 +197,7 @@ class TestBearerTokenJWKS:
     async def test_jwks_token_validation_with_kid_and_no_kid_in_token(
         self,
         rsa_key_pair: RSAKeyPair,
-        jwks_provider: BearerAuthProvider,
+        jwks_provider: JWTVerifier,
         mock_jwks_data: JWKSData,
         httpx_mock: HTTPXMock,
     ):
@@ -224,7 +219,7 @@ class TestBearerTokenJWKS:
     async def test_jwks_token_validation_with_no_kid_and_kid_in_jwks(
         self,
         rsa_key_pair: RSAKeyPair,
-        jwks_provider: BearerAuthProvider,
+        jwks_provider: JWTVerifier,
         mock_jwks_data: JWKSData,
         httpx_mock: HTTPXMock,
     ):
@@ -246,7 +241,7 @@ class TestBearerTokenJWKS:
     async def test_jwks_token_validation_with_kid_mismatch(
         self,
         rsa_key_pair: RSAKeyPair,
-        jwks_provider: BearerAuthProvider,
+        jwks_provider: JWTVerifier,
         mock_jwks_data: JWKSData,
         httpx_mock: HTTPXMock,
     ):
@@ -268,7 +263,7 @@ class TestBearerTokenJWKS:
     async def test_jwks_token_validation_with_multiple_keys_and_no_kid_in_token(
         self,
         rsa_key_pair: RSAKeyPair,
-        jwks_provider: BearerAuthProvider,
+        jwks_provider: JWTVerifier,
         mock_jwks_data: JWKSData,
         httpx_mock: HTTPXMock,
     ):
@@ -300,7 +295,7 @@ class TestBearerTokenJWKS:
 class TestBearerToken:
     def test_initialization_with_public_key(self, rsa_key_pair: RSAKeyPair):
         """Test provider initialization with public key."""
-        provider = BearerAuthProvider(
+        provider = JWTVerifier(
             public_key=rsa_key_pair.public_key, issuer="https://test.example.com"
         )
 
@@ -310,7 +305,7 @@ class TestBearerToken:
 
     def test_initialization_with_jwks_uri(self):
         """Test provider initialization with JWKS URI."""
-        provider = BearerAuthProvider(
+        provider = JWTVerifier(
             jwks_uri="https://test.example.com/.well-known/jwks.json",
             issuer="https://test.example.com",
         )
@@ -324,21 +319,21 @@ class TestBearerToken:
         with pytest.raises(
             ValueError, match="Either public_key or jwks_uri must be provided"
         ):
-            BearerAuthProvider(issuer="https://test.example.com")
+            JWTVerifier(issuer="https://test.example.com")
 
     def test_initialization_rejects_both_key_and_uri(self, rsa_key_pair: RSAKeyPair):
         """Test that both public_key and jwks_uri cannot be provided."""
         with pytest.raises(
             ValueError, match="Provide either public_key or jwks_uri, not both"
         ):
-            BearerAuthProvider(
+            JWTVerifier(
                 public_key=rsa_key_pair.public_key,
                 jwks_uri="https://test.example.com/.well-known/jwks.json",
                 issuer="https://test.example.com",
             )
 
     async def test_valid_token_validation(
-        self, rsa_key_pair: RSAKeyPair, bearer_provider: BearerAuthProvider
+        self, rsa_key_pair: RSAKeyPair, bearer_provider: JWTVerifier
     ):
         """Test validation of a valid token."""
         token = rsa_key_pair.create_token(
@@ -357,7 +352,7 @@ class TestBearerToken:
         assert access_token.expires_at is not None
 
     async def test_expired_token_rejection(
-        self, rsa_key_pair: RSAKeyPair, bearer_provider: BearerAuthProvider
+        self, rsa_key_pair: RSAKeyPair, bearer_provider: JWTVerifier
     ):
         """Test rejection of expired tokens."""
         token = rsa_key_pair.create_token(
@@ -371,7 +366,7 @@ class TestBearerToken:
         assert access_token is None
 
     async def test_invalid_issuer_rejection(
-        self, rsa_key_pair: RSAKeyPair, bearer_provider: BearerAuthProvider
+        self, rsa_key_pair: RSAKeyPair, bearer_provider: JWTVerifier
     ):
         """Test rejection of tokens with invalid issuer."""
         token = rsa_key_pair.create_token(
@@ -384,7 +379,7 @@ class TestBearerToken:
         assert access_token is None
 
     async def test_invalid_audience_rejection(
-        self, rsa_key_pair: RSAKeyPair, bearer_provider: BearerAuthProvider
+        self, rsa_key_pair: RSAKeyPair, bearer_provider: JWTVerifier
     ):
         """Test rejection of tokens with invalid audience."""
         token = rsa_key_pair.create_token(
@@ -398,7 +393,7 @@ class TestBearerToken:
 
     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(
+        provider = JWTVerifier(
             public_key=rsa_key_pair.public_key,
             issuer=None,  # No issuer validation
         )
@@ -412,7 +407,7 @@ class TestBearerToken:
 
     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(
+        provider = JWTVerifier(
             public_key=rsa_key_pair.public_key,
             issuer="https://test.example.com",
             audience=None,  # No audience validation
@@ -429,7 +424,7 @@ class TestBearerToken:
 
     async def test_multiple_audiences_validation(self, rsa_key_pair: RSAKeyPair):
         """Test validation with multiple audiences in token."""
-        provider = BearerAuthProvider(
+        provider = JWTVerifier(
             public_key=rsa_key_pair.public_key,
             issuer="https://test.example.com",
             audience="https://api.example.com",
@@ -450,7 +445,7 @@ class TestBearerToken:
         self, rsa_key_pair: RSAKeyPair
     ):
         """Test provider configured with multiple expected audiences."""
-        provider = BearerAuthProvider(
+        provider = JWTVerifier(
             public_key=rsa_key_pair.public_key,
             issuer="https://test.example.com",
             audience=["https://api.example.com", "https://other-api.example.com"],
@@ -486,7 +481,7 @@ class TestBearerToken:
         assert access_token3 is None
 
     async def test_scope_extraction_string(
-        self, rsa_key_pair: RSAKeyPair, bearer_provider: BearerAuthProvider
+        self, rsa_key_pair: RSAKeyPair, bearer_provider: JWTVerifier
     ):
         """Test scope extraction from space-separated string."""
         token = rsa_key_pair.create_token(
@@ -502,7 +497,7 @@ class TestBearerToken:
         assert set(access_token.scopes) == {"read", "write", "admin"}
 
     async def test_scope_extraction_list(
-        self, rsa_key_pair: RSAKeyPair, bearer_provider: BearerAuthProvider
+        self, rsa_key_pair: RSAKeyPair, bearer_provider: JWTVerifier
     ):
         """Test scope extraction from list format."""
         token = rsa_key_pair.create_token(
@@ -518,7 +513,7 @@ class TestBearerToken:
         assert set(access_token.scopes) == {"read", "write"}
 
     async def test_no_scopes(
-        self, rsa_key_pair: RSAKeyPair, bearer_provider: BearerAuthProvider
+        self, rsa_key_pair: RSAKeyPair, bearer_provider: JWTVerifier
     ):
         """Test token with no scopes."""
         token = rsa_key_pair.create_token(
@@ -534,7 +529,7 @@ class TestBearerToken:
         assert access_token.scopes == []
 
     async def test_scp_claim_extraction_string(
-        self, rsa_key_pair: RSAKeyPair, bearer_provider: BearerAuthProvider
+        self, rsa_key_pair: RSAKeyPair, bearer_provider: JWTVerifier
     ):
         """Test scope extraction from 'scp' claim with space-separated string."""
         token = rsa_key_pair.create_token(
@@ -550,7 +545,7 @@ class TestBearerToken:
         assert set(access_token.scopes) == {"read", "write", "admin"}
 
     async def test_scp_claim_extraction_list(
-        self, rsa_key_pair: RSAKeyPair, bearer_provider: BearerAuthProvider
+        self, rsa_key_pair: RSAKeyPair, bearer_provider: JWTVerifier
     ):
         """Test scope extraction from 'scp' claim with list format."""
         token = rsa_key_pair.create_token(
@@ -568,7 +563,7 @@ class TestBearerToken:
         assert set(access_token.scopes) == {"read", "write", "admin"}
 
     async def test_scope_precedence_over_scp(
-        self, rsa_key_pair: RSAKeyPair, bearer_provider: BearerAuthProvider
+        self, rsa_key_pair: RSAKeyPair, bearer_provider: JWTVerifier
     ):
         """Test that 'scope' claim takes precedence over 'scp' claim when both are present."""
         token = rsa_key_pair.create_token(
@@ -586,7 +581,7 @@ class TestBearerToken:
         assert access_token is not None
         assert set(access_token.scopes) == {"read", "write"}  # Only 'scope' claim used
 
-    async def test_malformed_token_rejection(self, bearer_provider: BearerAuthProvider):
+    async def test_malformed_token_rejection(self, bearer_provider: JWTVerifier):
         """Test rejection of malformed tokens."""
         malformed_tokens = [
             "not.a.jwt",
@@ -601,7 +596,7 @@ class TestBearerToken:
             assert access_token is None
 
     async def test_invalid_signature_rejection(
-        self, rsa_key_pair: RSAKeyPair, bearer_provider: BearerAuthProvider
+        self, rsa_key_pair: RSAKeyPair, bearer_provider: JWTVerifier
     ):
         """Test rejection of tokens with invalid signatures."""
         # Create a token with a different key pair
@@ -616,7 +611,7 @@ class TestBearerToken:
         assert access_token is None
 
     async def test_client_id_fallback(
-        self, rsa_key_pair: RSAKeyPair, bearer_provider: BearerAuthProvider
+        self, rsa_key_pair: RSAKeyPair, bearer_provider: JWTVerifier
     ):
         """Test client_id extraction with fallback logic."""
         # Test with explicit client_id claim
@@ -634,7 +629,7 @@ class TestBearerToken:
     async def test_string_issuer_validation(self, rsa_key_pair: RSAKeyPair):
         """Test that string (non-URL) issuers are supported per RFC 7519."""
         # Create provider with string issuer
-        provider = BearerAuthProvider(
+        provider = JWTVerifier(
             public_key=rsa_key_pair.public_key,
             issuer="my-service",  # String issuer, not a URL
         )
@@ -652,7 +647,7 @@ class TestBearerToken:
     async def test_string_issuer_mismatch_rejection(self, rsa_key_pair: RSAKeyPair):
         """Test that mismatched string issuers are rejected."""
         # Create provider with one string issuer
-        provider = BearerAuthProvider(
+        provider = JWTVerifier(
             public_key=rsa_key_pair.public_key,
             issuer="my-service",
         )
@@ -669,7 +664,7 @@ class TestBearerToken:
     async def test_url_issuer_still_works(self, rsa_key_pair: RSAKeyPair):
         """Test that URL issuers still work after the fix."""
         # Create provider with URL issuer
-        provider = BearerAuthProvider(
+        provider = JWTVerifier(
             public_key=rsa_key_pair.public_key,
             issuer="https://my-auth-server.com",  # URL issuer
         )
@@ -688,9 +683,9 @@ class TestBearerToken:
 class TestFastMCPBearerAuth:
     def test_bearer_auth(self):
         mcp = FastMCP(
-            auth=BearerAuthProvider(issuer="https://test.example.com", public_key="abc")
+            auth=JWTVerifier(issuer="https://test.example.com", public_key="abc")
         )
-        assert isinstance(mcp.auth, BearerAuthProvider)
+        assert isinstance(mcp.auth, JWTVerifier)
 
     async def test_unauthorized_access(self, mcp_server_url: str):
         with pytest.raises(httpx.HTTPStatusError) as exc_info:
@@ -755,7 +750,10 @@ class TestFastMCPBearerAuth:
             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
+            # JWTVerifier returns 401 when verify_token returns None (invalid token)
+            # This is correct behavior - when TokenVerifier.verify_token returns None,
+            # it indicates the token is invalid (not just insufficient permissions)
+            assert exc_info.value.response.status_code == 401
             assert "tools" not in locals()
 
     async def test_token_with_sufficient_scopes(

tests/contrib/test_component_manager.py

@@ -4,7 +4,7 @@ from starlette.testclient import TestClient
 
 from fastmcp import FastMCP
 from fastmcp.contrib.component_manager import set_up_component_manager
-from fastmcp.server.auth.providers.bearer import BearerAuthProvider, RSAKeyPair
+from fastmcp.server.auth.verifiers import JWTVerifier, RSAKeyPair
 
 
 class TestComponentManagementRoutes:
@@ -340,7 +340,7 @@ class TestAuthComponentManagementRoutes:
         """Set up test fixtures."""
         # Generate a key pair and create an auth provider
         key_pair = RSAKeyPair.generate()
-        self.auth = BearerAuthProvider(
+        self.auth = JWTVerifier(
             public_key=key_pair.public_key,
             issuer="https://dev.example.com",
             audience="my-dev-server",
@@ -425,7 +425,7 @@ class TestAuthComponentManagementRoutes:
         assert tool.enabled is False
 
     async def test_forbidden_enable_tool(self):
-        """Test that unauthenticated requests to enable a resource are rejected."""
+        """Test that requests with insufficient scopes are rejected."""
         tool = await self.mcp._tool_manager.get_tool("test_tool")
         tool.enabled = False
 
@@ -459,7 +459,7 @@ class TestAuthComponentManagementRoutes:
         assert resource.enabled is True
 
     async def test_forbidden_enable_resource(self):
-        """Test that unauthenticated requests to enable a resource are rejected."""
+        """Test that requests with insufficient scopes are rejected."""
         resource = await self.mcp._resource_manager.get_resource("data://test_resource")
         resource.enabled = False
 
@@ -515,7 +515,7 @@ class TestAuthComponentManagementRoutes:
         assert prompt.enabled is True
 
     async def test_forbidden_disable_prompt(self):
-        """Test that unauthenticated requests to enable a resource are rejected."""
+        """Test that requests with insufficient scopes are rejected."""
         prompt = await self.mcp._prompt_manager.get_prompt("test_prompt")
         prompt.enabled = True
 
@@ -606,11 +606,10 @@ class TestComponentManagerWithPathAuth:
     def setup_method(self):
         # Generate a key pair and create an auth provider
         key_pair = RSAKeyPair.generate()
-        self.auth = BearerAuthProvider(
+        self.auth = JWTVerifier(
             public_key=key_pair.public_key,
             issuer="https://dev.example.com",
             audience="my-dev-server",
-            required_scopes=["tool:write", "tool:read"],
         )
         self.mcp = FastMCP("TestServerWithPathAuth", auth=self.auth)
         set_up_component_manager(

tests/deprecated/test_bearer_auth_provider.py

@@ -0,0 +1,13 @@
+import pytest
+
+# reset deprecation warnings for this module
+pytestmark = pytest.mark.filterwarnings("default::DeprecationWarning")
+
+
+def test_bearer_auth_provider_deprecated():
+    """Test that BearerAuthProvider import shows deprecation warning."""
+    with pytest.warns(
+        DeprecationWarning,
+        match="The `fastmcp.server.auth.providers.bearer` module is deprecated and will be removed in a future version. Please use `fastmcp.server.auth.verifiers.JWTVerifier` instead of this module's BearerAuthProvider.",
+    ):
+        from fastmcp.server.auth import BearerAuthProvider  # noqa: F401

tests/server/auth/test_token_verifier_integration.py

@@ -0,0 +1,142 @@
+"""Tests for TokenVerifier integration with FastMCP."""
+
+import httpx
+import pytest
+from mcp.server.auth.provider import AccessToken
+
+from fastmcp.server import FastMCP
+from fastmcp.server.auth.verifiers import StaticTokenVerifier
+
+
+class TestTokenVerifierIntegration:
+    """Test TokenVerifier integration with FastMCP server."""
+
+    def test_static_token_verifier_creation(self):
+        """Test creating a FastMCP server with StaticTokenVerifier."""
+        verifier = StaticTokenVerifier(
+            {"test-token": {"client_id": "test-client", "scopes": ["read", "write"]}}
+        )
+
+        server = FastMCP("TestServer", auth=verifier)
+        assert server.auth is verifier
+
+    async def test_static_token_verifier_verify_token(self):
+        """Test StaticTokenVerifier token verification."""
+        verifier = StaticTokenVerifier(
+            {
+                "valid-token": {
+                    "client_id": "test-client",
+                    "scopes": ["read", "write"],
+                    "expires_at": None,
+                },
+                "scoped-token": {"client_id": "limited-client", "scopes": ["read"]},
+            }
+        )
+
+        # Test valid token
+        result = await verifier.verify_token("valid-token")
+        assert isinstance(result, AccessToken)
+        assert result.client_id == "test-client"
+        assert result.scopes == ["read", "write"]
+        assert result.token == "valid-token"
+        assert result.expires_at is None
+
+        # Test token with different scopes
+        result = await verifier.verify_token("scoped-token")
+        assert isinstance(result, AccessToken)
+        assert result.client_id == "limited-client"
+        assert result.scopes == ["read"]
+
+        # Test invalid token
+        result = await verifier.verify_token("invalid-token")
+        assert result is None
+
+    async def test_server_with_token_verifier_http_app(self):
+        """Test that FastMCP server works with TokenVerifier for HTTP requests."""
+        verifier = StaticTokenVerifier(
+            {"test-token": {"client_id": "test-client", "scopes": ["read", "write"]}}
+        )
+
+        server = FastMCP("TestServer", auth=verifier)
+
+        @server.tool
+        def greet(name: str) -> str:
+            return f"Hello, {name}!"
+
+        # Create HTTP app
+        app = server.http_app(transport="http")
+
+        # Test unauthenticated request gets 401
+        async with httpx.AsyncClient(
+            transport=httpx.ASGITransport(app=app), base_url="http://test"
+        ) as client:
+            response = await client.post("/mcp/")
+            assert response.status_code == 401
+            assert "WWW-Authenticate" in response.headers
+
+    def test_server_rejects_both_oauth_and_token_verifier(self):
+        """Test that server raises error when both OAuth and TokenVerifier provided."""
+        from fastmcp.server.auth.providers.in_memory import InMemoryOAuthProvider
+
+        oauth_provider = InMemoryOAuthProvider("http://test.com")
+        token_verifier = StaticTokenVerifier({"token": {"client_id": "test"}})
+
+        # This should work - OAuth provider
+        server1 = FastMCP("Test1", auth=oauth_provider)
+        assert server1.auth is oauth_provider
+
+        # This should work - TokenVerifier
+        server2 = FastMCP("Test2", auth=token_verifier)
+        assert server2.auth is token_verifier
+
+
+class TestJWTVerifierImport:
+    """Test JWT token verifier can be imported and created."""
+
+    def test_jwt_verifier_requires_pyjwt(self):
+        """Test that JWTVerifier raises helpful error without PyJWT."""
+        # Since PyJWT is likely installed in test environment, we'll just test construction
+        from fastmcp.server.auth.verifiers import JWTVerifier
+
+        # This should work if PyJWT is available
+        try:
+            verifier = JWTVerifier(public_key="dummy-key")
+            assert verifier.public_key == "dummy-key"
+            assert verifier.algorithm == "RS256"
+        except ImportError as e:
+            # If PyJWT not available, should get helpful error
+            assert "PyJWT is required" in str(e)
+
+
+class TestIntrospectionTokenVerifierImport:
+    """Test introspection token verifier can be imported and created."""
+
+    def test_introspection_verifier_creation(self):
+        """Test IntrospectionTokenVerifier construction."""
+        from fastmcp.server.auth.verifiers import IntrospectionTokenVerifier
+
+        verifier = IntrospectionTokenVerifier(
+            "https://auth.example.com/introspect", "https://resource.example.com"
+        )
+
+        assert (
+            str(verifier.introspection_endpoint)
+            == "https://auth.example.com/introspect"
+        )
+        assert str(verifier.server_url) == "https://resource.example.com/"
+        assert verifier.validate_resource is False
+        assert verifier.required_scopes == []
+
+    def test_introspection_verifier_rejects_private_urls(self):
+        """Test that IntrospectionTokenVerifier rejects private URLs."""
+        from fastmcp.server.auth.verifiers import IntrospectionTokenVerifier
+
+        with pytest.raises(ValueError, match="private/localhost URL"):
+            IntrospectionTokenVerifier(
+                "http://localhost/introspect", "https://resource.example.com"
+            )
+
+        with pytest.raises(ValueError, match="private/localhost URL"):
+            IntrospectionTokenVerifier(
+                "http://127.0.0.1/introspect", "https://resource.example.com"
+            )

tests/server/http/test_auth_setup.py

@@ -6,8 +6,8 @@ from mcp.server.auth.provider import AccessToken
 from starlette.middleware import Middleware
 from starlette.middleware.authentication import AuthenticationMiddleware
 
-from fastmcp.server.auth.providers.bearer import BearerAuthProvider, RSAKeyPair
 from fastmcp.server.auth.providers.in_memory import InMemoryOAuthProvider
+from fastmcp.server.auth.verifiers import JWTVerifier, RSAKeyPair
 from fastmcp.server.http import setup_auth_middleware_and_routes
 
 
@@ -15,10 +15,10 @@ class TestSetupAuthMiddlewareAndRoutes:
     """Test setup_auth_middleware_and_routes with TokenVerifier providers."""
 
     @pytest.fixture
-    def bearer_provider(self) -> BearerAuthProvider:
-        """Create BearerAuthProvider for testing."""
+    def jwt_verifier(self) -> JWTVerifier:
+        """Create JWTVerifier for testing."""
         key_pair = RSAKeyPair.generate()
-        return BearerAuthProvider(
+        return JWTVerifier(
             public_key=key_pair.public_key,
             issuer="https://test.example.com",
             audience="https://api.example.com",
@@ -33,10 +33,10 @@ class TestSetupAuthMiddlewareAndRoutes:
             required_scopes=["user"],
         )
 
-    def test_setup_with_bearer_provider(self, bearer_provider: BearerAuthProvider):
-        """Test that setup works with BearerAuthProvider as TokenVerifier."""
+    def test_setup_with_jwt_verifier(self, jwt_verifier: JWTVerifier):
+        """Test that setup works with JWTVerifier as TokenVerifier."""
         middleware, auth_routes, required_scopes = setup_auth_middleware_and_routes(
-            bearer_provider
+            jwt_verifier
         )
 
         # Should return middleware list
@@ -51,11 +51,11 @@ class TestSetupAuthMiddlewareAndRoutes:
 
         backend = auth_middleware.kwargs["backend"]
         assert isinstance(backend, BearerAuthBackend)
-        assert backend.token_verifier is bearer_provider  # type: ignore[attr-defined]
+        assert backend.token_verifier is jwt_verifier  # type: ignore[attr-defined]
 
         # Should return auth routes
         assert isinstance(auth_routes, list)
-        assert len(auth_routes) > 0  # Should have OAuth routes
+        assert len(auth_routes) == 0  # TokenVerifier should not have OAuth routes
 
         # Should return required scopes
         assert required_scopes == ["read", "write"]
@@ -81,25 +81,23 @@ class TestSetupAuthMiddlewareAndRoutes:
         # Should return required scopes
         assert required_scopes == ["user"]
 
-    def test_setup_preserves_provider_functionality(
-        self, bearer_provider: BearerAuthProvider
-    ):
+    def test_setup_preserves_provider_functionality(self, jwt_verifier: JWTVerifier):
         """Test that setup doesn't break the provider's functionality."""
         # Setup should not modify the provider
-        original_issuer = bearer_provider.issuer
-        original_scopes = bearer_provider.required_scopes
+        original_issuer = jwt_verifier.issuer
+        original_scopes = jwt_verifier.required_scopes
 
         middleware, auth_routes, required_scopes = setup_auth_middleware_and_routes(
-            bearer_provider
+            jwt_verifier
         )
 
         # Provider should be unchanged
-        assert bearer_provider.issuer == original_issuer
-        assert bearer_provider.required_scopes == original_scopes
+        assert jwt_verifier.issuer == original_issuer
+        assert jwt_verifier.required_scopes == original_scopes
 
         # Provider should still work as TokenVerifier
-        assert hasattr(bearer_provider, "verify_token")
-        assert callable(bearer_provider.verify_token)
+        assert hasattr(jwt_verifier, "verify_token")
+        assert callable(jwt_verifier.verify_token)
 
 
 class MockOAuthProvider:

tests/server/http/test_bearer_auth_backend.py

@@ -5,7 +5,7 @@ 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.providers.bearer import BearerAuthProvider, RSAKeyPair
+from fastmcp.server.auth.verifiers import JWTVerifier, RSAKeyPair
 
 
 class TestBearerAuthBackendTokenVerifierIntegration:
@@ -17,9 +17,9 @@ class TestBearerAuthBackendTokenVerifierIntegration:
         return RSAKeyPair.generate()
 
     @pytest.fixture
-    def bearer_provider(self, rsa_key_pair: RSAKeyPair) -> BearerAuthProvider:
-        """Create BearerAuthProvider for testing."""
-        return BearerAuthProvider(
+    def jwt_verifier(self, rsa_key_pair: RSAKeyPair) -> JWTVerifier:
+        """Create JWTVerifier for testing."""
+        return JWTVerifier(
             public_key=rsa_key_pair.public_key,
             issuer="https://test.example.com",
             audience="https://api.example.com",
@@ -36,18 +36,18 @@ class TestBearerAuthBackendTokenVerifierIntegration:
         )
 
     def test_bearer_auth_backend_constructor_accepts_token_verifier(
-        self, bearer_provider: BearerAuthProvider
+        self, jwt_verifier: JWTVerifier
     ):
         """Test that BearerAuthBackend constructor accepts TokenVerifier."""
         # This should not raise an error
-        backend = BearerAuthBackend(bearer_provider)
-        assert backend.token_verifier is bearer_provider  # type: ignore[attr-defined]
+        backend = BearerAuthBackend(jwt_verifier)
+        assert backend.token_verifier is jwt_verifier  # type: ignore[attr-defined]
 
     async def test_bearer_auth_backend_authenticate_with_valid_token(
-        self, bearer_provider: BearerAuthProvider, valid_token: str
+        self, jwt_verifier: JWTVerifier, valid_token: str
     ):
         """Test BearerAuthBackend authentication with valid token."""
-        backend = BearerAuthBackend(bearer_provider)
+        backend = BearerAuthBackend(jwt_verifier)
 
         # Create mock HTTPConnection with Authorization header
         scope = {
@@ -66,10 +66,10 @@ class TestBearerAuthBackendTokenVerifierIntegration:
         assert user.access_token.token == valid_token
 
     async def test_bearer_auth_backend_authenticate_with_invalid_token(
-        self, bearer_provider: BearerAuthProvider
+        self, jwt_verifier: JWTVerifier
     ):
         """Test BearerAuthBackend authentication with invalid token."""
-        backend = BearerAuthBackend(bearer_provider)
+        backend = BearerAuthBackend(jwt_verifier)
 
         # Create mock HTTPConnection with invalid Authorization header
         scope = {
@@ -82,10 +82,10 @@ class TestBearerAuthBackendTokenVerifierIntegration:
         assert result is None
 
     async def test_bearer_auth_backend_authenticate_with_no_header(
-        self, bearer_provider: BearerAuthProvider
+        self, jwt_verifier: JWTVerifier
     ):
         """Test BearerAuthBackend authentication with no Authorization header."""
-        backend = BearerAuthBackend(bearer_provider)
+        backend = BearerAuthBackend(jwt_verifier)
 
         # Create mock HTTPConnection without Authorization header
         scope = {
@@ -98,10 +98,10 @@ class TestBearerAuthBackendTokenVerifierIntegration:
         assert result is None
 
     async def test_bearer_auth_backend_authenticate_with_non_bearer_token(
-        self, bearer_provider: BearerAuthProvider
+        self, jwt_verifier: JWTVerifier
     ):
         """Test BearerAuthBackend authentication with non-Bearer token."""
-        backend = BearerAuthBackend(bearer_provider)
+        backend = BearerAuthBackend(jwt_verifier)
 
         # Create mock HTTPConnection with Basic auth header
         scope = {

tests/server/http/test_http_auth_middleware.py

@@ -3,7 +3,7 @@ from mcp.server.auth.middleware.bearer_auth import RequireAuthMiddleware
 from starlette.routing import Mount
 
 from fastmcp.server import FastMCP
-from fastmcp.server.auth.providers.bearer import BearerAuthProvider, RSAKeyPair
+from fastmcp.server.auth.verifiers import JWTVerifier, RSAKeyPair
 from fastmcp.server.http import create_streamable_http_app
 
 
@@ -17,11 +17,11 @@ class TestStreamableHTTPAppResourceMetadataURL:
 
     @pytest.fixture
     def bearer_auth_provider(self, rsa_key_pair):
-        provider = BearerAuthProvider(
+        provider = JWTVerifier(
             public_key=rsa_key_pair.public_key,
             issuer="https://issuer",
             audience="https://audience",
-            resource_server="https://resource.example.com",
+            resource_server_url="https://resource.example.com",
         )
         return provider
 
@@ -45,11 +45,11 @@ class TestStreamableHTTPAppResourceMetadataURL:
         )
 
     def test_trailing_slash_handling_in_resource_server_url(self, rsa_key_pair):
-        provider = BearerAuthProvider(
+        provider = JWTVerifier(
             public_key=rsa_key_pair.public_key,
             issuer="https://issuer",
             audience="https://audience",
-            resource_server="https://resource.example.com/",
+            resource_server_url="https://resource.example.com/",
         )
         server = FastMCP(name="TestServer")
         app = create_streamable_http_app(