ceos/fastmcp
1
0
Fork 0
mirror of https://github.com/PrefectHQ/fastmcp.git synced 2026-08-23 22:14:18 +02:00
Code Issues Projects Releases Packages Wiki Activity

Add complete support for TokenVerifier protocol (#1297)

Browse source
This commit is contained in:
Jeremiah Lowin 2025-07-29 17:01:40 -07:00 • committed by GitHub
commit 60ca72bb0b
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
24 changed files with 1365 additions and 884 deletions
Download patch file Download diff file Expand all files Collapse all files

28
docs/docs.json
View file

@ -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"]
}
]
},

14
docs/integrations/anthropic.mdx
View file

@ -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",
)

14
docs/integrations/openai.mdx
View file

@ -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",
)

176
docs/servers/auth/bearer.mdx → docs/servers/auth/verifiers.mdx
View file

@ -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:
### 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.
### 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.
### StaticTokenVerifier
Validates tokens against a predefined dictionary. Use for development and testing only - never in production.
<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.
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>
### Basic Setup
To configure bearer token authentication, instantiate a `BearerAuthProvider` instance and pass it to the `auth` parameter of the `FastMCP` instance.
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
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"
)
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>
#### Public Key
<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>
If you have a public key in PEM format, you can provide it to the `BearerAuthProvider` as a string.
<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.
```python
from fastmcp import FastMCP
from fastmcp.server.auth.verifiers import JWTVerifier
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:

4
docs/servers/server.mdx
View file

@ -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>

4
src/fastmcp/contrib/component_manager/example.py
View file

@ -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",

20
src/fastmcp/server/auth/__init__.py
View file

@ -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}'")

50
src/fastmcp/server/auth/auth.py
View file

@ -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)

507
src/fastmcp/server/auth/providers/bearer.py
View file

@ -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,
)

65
src/fastmcp/server/auth/providers/bearer_env.py
View file

@ -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())

718
src/fastmcp/server/auth/verifiers.py Normal file
View file

@ -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,
)

84
src/fastmcp/server/http.py
View file

@ -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(

11
src/fastmcp/server/server.py
View file

@ -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:

2
src/fastmcp/settings.py
View file

@ -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(
"""

91
tests/auth/providers/test_bearer_env.py
View file

@ -1,91 +0,0 @@
import pytest
from pydantic import AnyHttpUrl, ValidationError
from fastmcp import FastMCP
from fastmcp.server.auth.providers.bearer import BearerAuthProvider
from fastmcp.server.auth.providers.bearer_env import EnvBearerAuthProvider
from fastmcp.settings import Settings
from fastmcp.utilities.tests import temporary_settings
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")
with temporary_settings(**Settings().model_dump()):
mcp_with_auth = FastMCP()
assert isinstance(mcp_with_auth.auth, EnvBearerAuthProvider)
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")
with temporary_settings(**Settings().model_dump()):
with pytest.raises(
ValueError, match="Either public_key or jwks_uri must be provided"
):
FastMCP()
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_AUTH_BEARER_REQUIRED_SCOPES", '["test-scope1", "test-scope2"]'
)
with temporary_settings(**Settings().model_dump()):
mcp = FastMCP()
assert isinstance(mcp.auth, EnvBearerAuthProvider)
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")
with temporary_settings(**Settings().model_dump()):
with pytest.raises(ValidationError, match="Input should be a valid list"):
FastMCP()
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")
with temporary_settings(**Settings().model_dump()):
mcp = FastMCP()
assert isinstance(mcp.auth, EnvBearerAuthProvider)
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")
with temporary_settings(**Settings().model_dump()):
with pytest.raises(ValueError, match="Provide either public_key or jwks_uri"):
FastMCP()
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")
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)
assert mcp.auth.public_key == "test-public-key-2"

42
tests/auth/providers/test_token_verifier.py
View file

@ -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")

89
tests/auth/verifiers/test_env_jwt.py Normal file
View file

@ -0,0 +1,89 @@
import pytest
from pydantic import ValidationError
from fastmcp import FastMCP
from fastmcp.server.auth.verifiers import EnvJWTVerifier, JWTVerifier
from fastmcp.settings import Settings
from fastmcp.utilities.tests import temporary_settings
def test_load_bearer_env_from_env_var(monkeypatch):
mcp = FastMCP()
assert mcp.auth is None
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, 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", "jwt-env")
with temporary_settings(**Settings().model_dump()):
with pytest.raises(
ValueError, match="Either public_key or jwks_uri must be provided"
):
FastMCP()
def test_configure_bearer_env_from_env_var(monkeypatch):
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_JWT_REQUIRED_SCOPES", '["test-scope1", "test-scope2"]'
)
with temporary_settings(**Settings().model_dump()):
mcp = FastMCP()
assert isinstance(mcp.auth, EnvJWTVerifier)
assert mcp.auth.public_key == "test-public-key"
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", "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"):
FastMCP()
def test_configure_bearer_env_jwks_uri_from_env_var(monkeypatch):
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, 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", "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"):
FastMCP()
def test_provided_auth_takes_precedence_over_env_vars(monkeypatch):
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=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"

88
tests/auth/providers/test_bearer.py → tests/auth/verifiers/test_jwt_verifier.py
View file

@ -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(

13
tests/contrib/test_component_manager.py
View file

@ -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(

13
tests/deprecated/test_bearer_auth_provider.py Normal file
View file

@ -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

142
tests/server/auth/test_token_verifier_integration.py Normal file
View file

@ -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"
)

36
tests/server/http/test_auth_setup.py
View file

@ -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:

30
tests/server/http/test_bearer_auth_backend.py
View file

@ -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 = {

10
tests/server/http/test_http_auth_middleware.py
View file

@ -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(