ceos/fastmcp
1
0
Fork 0
mirror of https://github.com/PrefectHQ/fastmcp.git synced 2026-08-09 15:19:10 +02:00
Code Issues Projects Releases Packages Wiki Activity

Fix OAuth resource URL handling and WWW-Authenticate header (#1706)

Browse source
This commit is contained in:
Jeremiah Lowin 2025-09-02 15:14:05 -04:00 • committed by GitHub
commit d36ea42b92
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
26 changed files with 518 additions and 354 deletions
Download patch file Download diff file Expand all files Collapse all files

2
docs/python-sdk/fastmcp-server-auth-auth.mdx
View file

@ -168,7 +168,7 @@ Get OAuth authorization server routes and optional protected resource routes.
This method creates the full set of OAuth routes including:
- Standard OAuth authorization server routes (/.well-known/oauth-authorization-server, /authorize, /token, etc.)
- Optional protected resource routes if resource_server_url is configured
- Optional protected resource routes if base_url is configured
**Returns:**
- List of OAuth routes

8
docs/servers/auth/oauth-proxy.mdx
View file

@ -158,9 +158,7 @@ The `OAuthProxy` class provides the complete proxy implementation:
Optional URL to your service documentation
</ParamField>
<ParamField body="resource_server_url" type="AnyHttpUrl | str | None">
Path of the FastMCP server (defaults to base_url). **Important**: This should point to your MCP endpoint path. For example, if your MCP server is accessible at `{base_url}/mcp`, specify `https://your-server.com/mcp` here for proper RFC 8707 compliance.
</ParamField>
<ParamField body="allowed_client_redirect_uris" type="list[str] | None">
List of allowed redirect URI patterns for MCP clients. Patterns support wildcards (e.g., `"http://localhost:*"`, `"https://*.example.com/*"`).
@ -229,9 +227,7 @@ auth = OAuthProxy(
# Optional: customize callback path (defaults to "/auth/callback")
redirect_path="/auth/callback",
# Optional: specify MCP endpoint path if different from base_url
# resource_server_url="https://your-server.com/mcp"
)
mcp = FastMCP(name="My Server", auth=auth)

6
docs/servers/auth/remote-oauth.mdx
View file

@ -111,7 +111,7 @@ token_verifier = JWTVerifier(
auth = RemoteAuthProvider(
token_verifier=token_verifier,
authorization_servers=[AnyHttpUrl("https://auth.yourcompany.com")],
resource_server_url="https://api.yourcompany.com/mcp", # Point to your MCP endpoint
base_url="https://api.yourcompany.com", # Your server base URL
# Optional: customize allowed client redirect URIs (defaults to localhost only)
allowed_client_redirect_uris=["http://localhost:*", "http://127.0.0.1:*"]
)
@ -121,7 +121,7 @@ mcp = FastMCP(name="Company API", auth=auth)
This configuration creates a server that accepts tokens issued by `auth.yourcompany.com` and provides the OAuth discovery metadata that MCP clients need. The `JWTVerifier` handles token validation using your identity provider's public keys, while the `RemoteAuthProvider` generates the required OAuth endpoints.
The `authorization_servers` list tells MCP clients which identity providers you trust. The `resource_server_url` identifies your server in OAuth metadata, enabling proper token audience validation. **Important**: The `resource_server_url` should point to your actual MCP endpoint - for example, if your MCP server is accessible at `https://api.yourcompany.com/mcp`, use that full path rather than just the base URL.
The `authorization_servers` list tells MCP clients which identity providers you trust. The `base_url` identifies your server in OAuth metadata, enabling proper token audience validation. **Important**: The `base_url` should point to your server base URL - for example, if your MCP server is accessible at `https://api.yourcompany.com/mcp`, use `https://api.yourcompany.com` as the base URL.
### Custom Endpoints
@ -143,7 +143,7 @@ class CompanyAuthProvider(RemoteAuthProvider):
super().__init__(
token_verifier=token_verifier,
authorization_servers=[AnyHttpUrl("https://auth.yourcompany.com")],
resource_server_url="https://api.yourcompany.com/mcp" # Your MCP endpoint path
base_url="https://api.yourcompany.com" # Your server base URL
)
def get_routes(self) -> list[Route]:

1
examples/auth/azure_oauth/server.py
View file

@ -23,7 +23,6 @@ auth = AzureProvider(
tenant_id=os.getenv("AZURE_TENANT_ID")
or "", # Required for single-tenant apps - get from Azure Portal
base_url="http://localhost:8000",
resource_server_url="http://localhost:8000/mcp",
# redirect_path="/auth/callback", # Default path - change if using a different callback URL
)

1
examples/auth/github_oauth/server.py
View file

@ -19,7 +19,6 @@ auth = GitHubProvider(
client_id=os.getenv("FASTMCP_SERVER_AUTH_GITHUB_CLIENT_ID") or "",
client_secret=os.getenv("FASTMCP_SERVER_AUTH_GITHUB_CLIENT_SECRET") or "",
base_url="http://localhost:8000",
resource_server_url="http://localhost:8000/mcp",
# redirect_path="/auth/callback", # Default path - change if using a different callback URL
)

1
examples/auth/google_oauth/server.py
View file

@ -19,7 +19,6 @@ auth = GoogleProvider(
client_id=os.getenv("FASTMCP_SERVER_AUTH_GOOGLE_CLIENT_ID") or "",
client_secret=os.getenv("FASTMCP_SERVER_AUTH_GOOGLE_CLIENT_SECRET") or "",
base_url="http://localhost:8000",
resource_server_url="http://localhost:8000/mcp",
# redirect_path="/auth/callback", # Default path - change if using a different callback URL
# Optional: specify required scopes
# required_scopes=["openid", "https://www.googleapis.com/auth/userinfo.email"],

1
examples/auth/workos_oauth/server.py
View file

@ -21,7 +21,6 @@ auth = WorkOSProvider(
client_secret=os.getenv("WORKOS_CLIENT_SECRET") or "",
authkit_domain=os.getenv("WORKOS_AUTHKIT_DOMAIN") or "https://your-app.authkit.app",
base_url="http://localhost:8000",
resource_server_url="http://localhost:8000/mcp",
# redirect_path="/auth/callback", # Default path - change if using a different callback URL
)

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

@ -1,7 +1,13 @@
from __future__ import annotations
from typing import Any
from urllib.parse import urljoin
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 (
AccessToken as _SDKAccessToken,
)
@ -22,6 +28,8 @@ from mcp.server.auth.settings import (
RevocationOptions,
)
from pydantic import AnyHttpUrl
from starlette.middleware import Middleware
from starlette.middleware.authentication import AuthenticationMiddleware
from starlette.routing import Route
@ -40,18 +48,23 @@ class AuthProvider(TokenVerifierProtocol):
custom authentication routes.
"""
def __init__(self, resource_server_url: AnyHttpUrl | str | None = None):
def __init__(
self,
base_url: AnyHttpUrl | str | None = None,
required_scopes: list[str] | None = None,
):
"""
Initialize the auth provider.
Args:
resource_server_url: The URL of this resource server. This is used
for RFC 8707 resource indicators, including creating the WWW-Authenticate
header.
base_url: The base URL of this server (e.g., http://localhost:8000).
This is used for constructing .well-known endpoints and OAuth metadata.
required_scopes: List of OAuth scopes required for all requests.
"""
if isinstance(resource_server_url, str):
resource_server_url = AnyHttpUrl(resource_server_url)
self.resource_server_url = resource_server_url
if isinstance(base_url, str):
base_url = AnyHttpUrl(base_url)
self.base_url = base_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.
@ -66,7 +79,11 @@ class AuthProvider(TokenVerifierProtocol):
"""
raise NotImplementedError("Subclasses must implement verify_token")
def get_routes(self) -> list[Route]:
def get_routes(
self,
mcp_path: str | None = None,
mcp_endpoint: Any | None = None,
) -> list[Route]:
"""Get the routes for this authentication provider.
Each provider is responsible for creating whatever routes it needs:
@ -75,22 +92,63 @@ class AuthProvider(TokenVerifierProtocol):
- OAuthProvider: full OAuth authorization server routes
- Custom providers: whatever routes they need
Returns:
List of routes for this provider
"""
return []
Args:
mcp_path: The path where the MCP endpoint is mounted (e.g., "/mcp")
mcp_endpoint: The MCP endpoint handler to protect with auth
def get_resource_metadata_url(self) -> AnyHttpUrl | None:
"""Get the resource metadata URL for RFC 9728 compliance."""
if self.resource_server_url is None:
Returns:
List of routes for this provider, including protected MCP endpoints if provided
"""
routes = []
# Add protected MCP endpoint if provided
if mcp_path and mcp_endpoint:
resource_metadata_url = self._get_resource_url(
"/.well-known/oauth-protected-resource"
)
routes.append(
Route(
mcp_path,
endpoint=RequireAuthMiddleware(
mcp_endpoint, self.required_scopes, resource_metadata_url
),
)
)
return routes
def get_middleware(self) -> list:
"""Get HTTP application-level middleware for this auth provider.
Returns:
List of Starlette Middleware instances to apply to the HTTP app
"""
return [
Middleware(
AuthenticationMiddleware,
backend=BearerAuthBackend(self),
),
Middleware(AuthContextMiddleware),
]
def _get_resource_url(self, path: str | None = None) -> AnyHttpUrl | None:
"""Get the actual resource URL being protected.
Args:
path: The path where the resource endpoint is mounted (e.g., "/mcp")
Returns:
The full URL of the protected resource
"""
if self.base_url is None:
return None
# Add .well-known path for RFC 9728 compliance
resource_metadata_url = AnyHttpUrl(
str(self.resource_server_url).rstrip("/")
+ "/.well-known/oauth-protected-resource"
)
return resource_metadata_url
if path:
return AnyHttpUrl(urljoin(str(self.base_url), path))
return self.base_url
class TokenVerifier(AuthProvider):
@ -102,20 +160,17 @@ class TokenVerifier(AuthProvider):
def __init__(
self,
resource_server_url: AnyHttpUrl | str | None = None,
base_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. This is used
for RFC 8707 resource indicators, including creating the WWW-Authenticate
header.
base_url: The base URL of this server
required_scopes: Scopes that are required for all requests
"""
super().__init__(resource_server_url=resource_server_url)
self.required_scopes = required_scopes or []
super().__init__(base_url=base_url, required_scopes=required_scopes)
async def verify_token(self, token: str) -> AccessToken | None:
"""Verify a bearer token and return access info if valid."""
@ -135,13 +190,13 @@ class RemoteAuthProvider(AuthProvider):
the authorization servers that issue valid tokens.
"""
resource_server_url: AnyHttpUrl
base_url: AnyHttpUrl
def __init__(
self,
token_verifier: TokenVerifier,
authorization_servers: list[AnyHttpUrl],
resource_server_url: AnyHttpUrl | str,
base_url: AnyHttpUrl | str,
resource_name: str | None = None,
resource_documentation: AnyHttpUrl | None = None,
):
@ -150,11 +205,14 @@ class RemoteAuthProvider(AuthProvider):
Args:
token_verifier: TokenVerifier instance for token validation
authorization_servers: List of authorization servers that issue valid tokens
resource_server_url: URL of this resource server. This is used
for RFC 8707 resource indicators, including creating the WWW-Authenticate
header.
base_url: The base URL of this server
resource_name: Optional name for the protected resource
resource_documentation: Optional documentation URL for the protected resource
"""
super().__init__(resource_server_url=resource_server_url)
super().__init__(
base_url=base_url,
required_scopes=token_verifier.required_scopes,
)
self.token_verifier = token_verifier
self.authorization_servers = authorization_servers
self.resource_name = resource_name
@ -164,21 +222,34 @@ class RemoteAuthProvider(AuthProvider):
"""Verify token using the configured token verifier."""
return await self.token_verifier.verify_token(token)
def get_routes(self) -> list[Route]:
def get_routes(
self,
mcp_path: str | None = None,
mcp_endpoint: Any | None = None,
) -> list[Route]:
"""Get OAuth routes for this provider.
By default, returns only the standardized OAuth 2.0 Protected Resource routes.
Subclasses can override this method to add additional routes by calling
super().get_routes() and extending the returned list.
Creates protected resource metadata routes and optionally wraps MCP endpoints with auth.
"""
# Start with base routes (protected MCP endpoint)
routes = super().get_routes(mcp_path, mcp_endpoint)
return create_protected_resource_routes(
resource_url=self.resource_server_url,
authorization_servers=self.authorization_servers,
scopes_supported=self.token_verifier.required_scopes,
resource_name=self.resource_name,
resource_documentation=self.resource_documentation,
)
# Get the resource URL based on the MCP path
resource_url = self._get_resource_url(mcp_path)
if resource_url:
# Add protected resource metadata routes
routes.extend(
create_protected_resource_routes(
resource_url=resource_url,
authorization_servers=self.authorization_servers,
scopes_supported=self.token_verifier.required_scopes,
resource_name=self.resource_name,
resource_documentation=self.resource_documentation,
)
)
return routes
class OAuthProvider(
@ -200,7 +271,6 @@ class OAuthProvider(
client_registration_options: ClientRegistrationOptions | None = None,
revocation_options: RevocationOptions | None = None,
required_scopes: list[str] | None = None,
resource_server_url: AnyHttpUrl | str | None = None,
):
"""
Initialize the OAuth provider.
@ -212,14 +282,13 @@ class OAuthProvider(
client_registration_options: The client registration options.
revocation_options: The revocation options.
required_scopes: Scopes that are required for all requests.
resource_server_url: The URL of this resource server (for RFC 8707 resource indicators, defaults to base_url)
"""
super().__init__()
# Convert URLs to proper types
if isinstance(base_url, str):
base_url = AnyHttpUrl(base_url)
super().__init__(base_url=base_url, required_scopes=required_scopes)
self.base_url = base_url
if issuer_url is None:
@ -229,15 +298,6 @@ class OAuthProvider(
else:
self.issuer_url = issuer_url
# Handle our own resource_server_url and required_scopes
if resource_server_url is None:
self.resource_server_url = base_url
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 []
# Initialize OAuth Authorization Server Provider
OAuthAuthorizationServerProvider.__init__(self)
@ -263,12 +323,17 @@ class OAuthProvider(
"""
return await self.load_access_token(token)
def get_routes(self) -> list[Route]:
def get_routes(
self,
mcp_path: str | None = None,
mcp_endpoint: Any | None = None,
) -> list[Route]:
"""Get OAuth authorization server routes and optional protected resource routes.
This method creates the full set of OAuth routes including:
- Standard OAuth authorization server routes (/.well-known/oauth-authorization-server, /authorize, /token, etc.)
- Optional protected resource routes if resource_server_url is configured
- Optional protected resource routes
- Protected MCP endpoints if provided
Returns:
List of OAuth routes
@ -283,13 +348,19 @@ class OAuthProvider(
revocation_options=self.revocation_options,
)
# Get the resource URL based on the MCP path
resource_url = self._get_resource_url(mcp_path)
# Add protected resource routes if this server is also acting as a resource server
if self.resource_server_url:
if resource_url:
protected_routes = create_protected_resource_routes(
resource_url=self.resource_server_url,
resource_url=resource_url,
authorization_servers=[self.issuer_url],
scopes_supported=self.required_scopes,
)
oauth_routes.extend(protected_routes)
# Add protected MCP endpoint from base class
oauth_routes.extend(super().get_routes(mcp_path, mcp_endpoint))
return oauth_routes

15
src/fastmcp/server/auth/oauth_proxy.py
View file

@ -241,7 +241,6 @@ class OAuthProxy(OAuthProvider):
redirect_path: str = "/auth/callback",
issuer_url: AnyHttpUrl | str | None = None,
service_documentation_url: AnyHttpUrl | str | None = None,
resource_server_url: AnyHttpUrl | str | None = None,
# Client redirect URI validation
allowed_client_redirect_uris: list[str] | None = None,
):
@ -259,7 +258,6 @@ class OAuthProxy(OAuthProvider):
redirect_path: Redirect path configured in upstream OAuth app (defaults to "/auth/callback")
issuer_url: Issuer URL for OAuth metadata (defaults to base_url)
service_documentation_url: Optional service documentation URL
resource_server_url: Path of the FastMCP server.
allowed_client_redirect_uris: List of allowed redirect URI patterns for MCP clients.
Patterns support wildcards (e.g., "http://localhost:*", "https://*.example.com/*").
If None (default), only localhost redirect URIs are allowed.
@ -281,7 +279,6 @@ class OAuthProxy(OAuthProvider):
client_registration_options=client_registration_options,
revocation_options=revocation_options,
required_scopes=token_verifier.required_scopes,
resource_server_url=resource_server_url,
)
# Store upstream configuration
@ -875,14 +872,22 @@ class OAuthProxy(OAuthProvider):
except Exception as e:
logger.warning("Failed to store tokens from upstream response: %s", e)
def get_routes(self) -> list[Route]:
def get_routes(
self,
mcp_path: str | None = None,
mcp_endpoint: Any | None = None,
) -> list[Route]:
"""Get OAuth routes with custom proxy token handler.
This method creates standard OAuth routes and replaces the token endpoint
with our proxy handler that forwards requests to the upstream OAuth server.
Args:
mcp_path: The path where the MCP endpoint is mounted (e.g., "/mcp")
mcp_endpoint: The MCP endpoint handler to protect with auth
"""
# Get standard OAuth routes from parent class
routes = super().get_routes()
routes = super().get_routes(mcp_path, mcp_endpoint)
custom_routes = []
token_route_found = False

13
src/fastmcp/server/auth/providers/azure.py
View file

@ -35,7 +35,6 @@ class AzureProviderSettings(BaseSettings):
redirect_path: str | None = None
required_scopes: list[str] | None = None
timeout_seconds: int | None = None
resource_server_url: str | None = None
allowed_client_redirect_uris: list[str] | None = None
@field_validator("required_scopes", mode="before")
@ -160,7 +159,6 @@ class AzureProvider(OAuthProxy):
redirect_path: str | NotSetT = NotSet,
required_scopes: list[str] | None | NotSetT = NotSet,
timeout_seconds: int | NotSetT = NotSet,
resource_server_url: str | NotSetT = NotSet,
allowed_client_redirect_uris: list[str] | NotSetT = NotSet,
):
"""Initialize Azure OAuth provider.
@ -173,8 +171,6 @@ class AzureProvider(OAuthProxy):
redirect_path: Redirect path configured in Azure (defaults to "/auth/callback")
required_scopes: Required scopes (defaults to ["User.Read", "email", "openid", "profile"])
timeout_seconds: HTTP request timeout for Azure API calls
resource_server_url: Path of the FastMCP server (defaults to base_url). If your MCP endpoint is at
a different path like {base_url}/mcp, specify it here for RFC 8707 compliance.
allowed_client_redirect_uris: List of allowed redirect URI patterns for MCP clients.
If None (default), all URIs are allowed. If empty list, no URIs are allowed.
"""
@ -189,7 +185,6 @@ class AzureProvider(OAuthProxy):
"redirect_path": redirect_path,
"required_scopes": required_scopes,
"timeout_seconds": timeout_seconds,
"resource_server_url": resource_server_url,
"allowed_client_redirect_uris": allowed_client_redirect_uris,
}.items()
if v is not NotSet
@ -215,7 +210,7 @@ class AzureProvider(OAuthProxy):
# Apply defaults
tenant_id_final = settings.tenant_id
base_url_final = settings.base_url or "http://localhost:8000"
redirect_path_final = settings.redirect_path or "/auth/callback"
timeout_seconds_final = settings.timeout_seconds or 10
# Default scopes for Azure - User.Read gives us access to user info via Graph API
@ -225,7 +220,6 @@ class AzureProvider(OAuthProxy):
"openid",
"profile",
]
resource_server_url_final = settings.resource_server_url or base_url_final
allowed_client_redirect_uris_final = settings.allowed_client_redirect_uris
# Extract secret string from SecretStr
@ -254,11 +248,10 @@ class AzureProvider(OAuthProxy):
upstream_client_id=settings.client_id,
upstream_client_secret=client_secret_str,
token_verifier=token_verifier,
base_url=base_url_final,
base_url=settings.base_url,
redirect_path=redirect_path_final,
issuer_url=base_url_final,
issuer_url=settings.base_url,
allowed_client_redirect_uris=allowed_client_redirect_uris_final,
resource_server_url=resource_server_url_final,
)
logger.info(

16
src/fastmcp/server/auth/providers/github.py
View file

@ -50,7 +50,6 @@ class GitHubProviderSettings(BaseSettings):
redirect_path: str | None = None
required_scopes: list[str] | None = None
timeout_seconds: int | None = None
resource_server_url: AnyHttpUrl | str | None = None
allowed_client_redirect_uris: list[str] | None = None
@field_validator("required_scopes", mode="before")
@ -185,7 +184,7 @@ class GitHubProvider(OAuthProxy):
auth = GitHubProvider(
client_id="Ov23li...",
client_secret="abc123...",
base_url="https://my-server.com" # Optional, defaults to http://localhost:8000
base_url="https://my-server.com"
)
mcp = FastMCP("My App", auth=auth)
@ -201,7 +200,6 @@ class GitHubProvider(OAuthProxy):
redirect_path: str | NotSetT = NotSet,
required_scopes: list[str] | NotSetT = NotSet,
timeout_seconds: int | NotSetT = NotSet,
resource_server_url: AnyHttpUrl | str | NotSetT = NotSet,
allowed_client_redirect_uris: list[str] | NotSetT = NotSet,
):
"""Initialize GitHub OAuth provider.
@ -213,11 +211,10 @@ class GitHubProvider(OAuthProxy):
redirect_path: Redirect path configured in GitHub OAuth app (defaults to "/auth/callback")
required_scopes: Required GitHub scopes (defaults to ["user"])
timeout_seconds: HTTP request timeout for GitHub API calls
resource_server_url: Path of the FastMCP server (defaults to base_url). If your MCP endpoint is at
a different path like {base_url}/mcp, specify it here for RFC 8707 compliance.
allowed_client_redirect_uris: List of allowed redirect URI patterns for MCP clients.
If None (default), all URIs are allowed. If empty list, no URIs are allowed.
"""
settings = GitHubProviderSettings.model_validate(
{
k: v
@ -228,7 +225,6 @@ class GitHubProvider(OAuthProxy):
"redirect_path": redirect_path,
"required_scopes": required_scopes,
"timeout_seconds": timeout_seconds,
"resource_server_url": resource_server_url,
"allowed_client_redirect_uris": allowed_client_redirect_uris,
}.items()
if v is not NotSet
@ -246,11 +242,10 @@ class GitHubProvider(OAuthProxy):
)
# Apply defaults
base_url_final = settings.base_url or "http://localhost:8000"
redirect_path_final = settings.redirect_path or "/auth/callback"
timeout_seconds_final = settings.timeout_seconds or 10
required_scopes_final = settings.required_scopes or ["user"]
resource_server_url_final = settings.resource_server_url or base_url_final
allowed_client_redirect_uris_final = settings.allowed_client_redirect_uris
# Create GitHub token verifier
@ -271,11 +266,10 @@ class GitHubProvider(OAuthProxy):
upstream_client_id=settings.client_id,
upstream_client_secret=client_secret_str,
token_verifier=token_verifier,
base_url=base_url_final,
base_url=settings.base_url,
redirect_path=redirect_path_final,
issuer_url=base_url_final, # We act as the issuer for client registration
issuer_url=settings.base_url, # We act as the issuer for client registration
allowed_client_redirect_uris=allowed_client_redirect_uris_final,
resource_server_url=resource_server_url_final,
)
logger.info(

13
src/fastmcp/server/auth/providers/google.py
View file

@ -52,7 +52,6 @@ class GoogleProviderSettings(BaseSettings):
redirect_path: str | None = None
required_scopes: list[str] | None = None
timeout_seconds: int | None = None
resource_server_url: AnyHttpUrl | str | None = None
allowed_client_redirect_uris: list[str] | None = None
@field_validator("required_scopes", mode="before")
@ -201,7 +200,7 @@ class GoogleProvider(OAuthProxy):
auth = GoogleProvider(
client_id="123456789.apps.googleusercontent.com",
client_secret="GOCSPX-abc123...",
base_url="https://my-server.com" # Optional, defaults to http://localhost:8000
base_url="https://my-server.com"
)
mcp = FastMCP("My App", auth=auth)
@ -217,7 +216,6 @@ class GoogleProvider(OAuthProxy):
redirect_path: str | NotSetT = NotSet,
required_scopes: list[str] | NotSetT = NotSet,
timeout_seconds: int | NotSetT = NotSet,
resource_server_url: AnyHttpUrl | str | NotSetT = NotSet,
allowed_client_redirect_uris: list[str] | NotSetT = NotSet,
):
"""Initialize Google OAuth provider.
@ -235,6 +233,7 @@ class GoogleProvider(OAuthProxy):
allowed_client_redirect_uris: List of allowed redirect URI patterns for MCP clients.
If None (default), all URIs are allowed. If empty list, no URIs are allowed.
"""
settings = GoogleProviderSettings.model_validate(
{
k: v
@ -245,7 +244,6 @@ class GoogleProvider(OAuthProxy):
"redirect_path": redirect_path,
"required_scopes": required_scopes,
"timeout_seconds": timeout_seconds,
"resource_server_url": resource_server_url,
"allowed_client_redirect_uris": allowed_client_redirect_uris,
}.items()
if v is not NotSet
@ -263,12 +261,10 @@ class GoogleProvider(OAuthProxy):
)
# Apply defaults
base_url_final = settings.base_url or "http://localhost:8000"
redirect_path_final = settings.redirect_path or "/auth/callback"
timeout_seconds_final = settings.timeout_seconds or 10
# Google requires at least one scope - openid is the minimal OIDC scope
required_scopes_final = settings.required_scopes or ["openid"]
resource_server_url_final = settings.resource_server_url or base_url_final
allowed_client_redirect_uris_final = settings.allowed_client_redirect_uris
# Create Google token verifier
@ -289,11 +285,10 @@ class GoogleProvider(OAuthProxy):
upstream_client_id=settings.client_id,
upstream_client_secret=client_secret_str,
token_verifier=token_verifier,
base_url=base_url_final,
base_url=settings.base_url,
redirect_path=redirect_path_final,
issuer_url=base_url_final, # We act as the issuer for client registration
issuer_url=settings.base_url, # We act as the issuer for client registration
allowed_client_redirect_uris=allowed_client_redirect_uris_final,
resource_server_url=resource_server_url_final,
)
logger.info(

2
src/fastmcp/server/auth/providers/in_memory.py
View file

@ -41,7 +41,6 @@ class InMemoryOAuthProvider(OAuthProvider):
client_registration_options: ClientRegistrationOptions | None = None,
revocation_options: RevocationOptions | None = None,
required_scopes: list[str] | None = None,
resource_server_url: AnyHttpUrl | str | None = None,
):
super().__init__(
base_url=base_url or "http://fastmcp.example.com",
@ -49,7 +48,6 @@ class InMemoryOAuthProvider(OAuthProvider):
client_registration_options=client_registration_options,
revocation_options=revocation_options,
required_scopes=required_scopes,
resource_server_url=resource_server_url,
)
self.clients: dict[str, OAuthClientInformationFull] = {}
self.auth_codes: dict[str, AuthorizationCode] = {}

10
src/fastmcp/server/auth/providers/jwt.py
View file

@ -153,7 +153,7 @@ class JWTVerifierSettings(BaseSettings):
algorithm: str | None = None
audience: str | list[str] | None = None
required_scopes: list[str] | None = None
resource_server_url: AnyHttpUrl | str | None = None
base_url: AnyHttpUrl | str | None = None
@field_validator("required_scopes", mode="before")
@classmethod
@ -189,7 +189,7 @@ class JWTVerifier(TokenVerifier):
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,
base_url: AnyHttpUrl | str | None | NotSetT = NotSet,
):
"""
Initialize the JWT token verifier.
@ -204,7 +204,7 @@ class JWTVerifier(TokenVerifier):
- Asymmetric: RS256/384/512, ES256/384/512, PS256/384/512 (default: RS256)
- Symmetric: HS256, HS384, HS512
required_scopes: Required scopes for all tokens
resource_server_url: Resource server URL for TokenVerifier protocol
base_url: Base URL for TokenVerifier protocol
"""
settings = JWTVerifierSettings.model_validate(
{
@ -216,7 +216,7 @@ class JWTVerifier(TokenVerifier):
"audience": audience,
"algorithm": algorithm,
"required_scopes": required_scopes,
"resource_server_url": resource_server_url,
"base_url": base_url,
}.items()
if v is not NotSet
}
@ -247,7 +247,7 @@ class JWTVerifier(TokenVerifier):
# Initialize parent TokenVerifier
super().__init__(
resource_server_url=settings.resource_server_url,
base_url=settings.base_url,
required_scopes=settings.required_scopes,
)

29
src/fastmcp/server/auth/providers/workos.py
View file

@ -10,6 +10,8 @@ Choose based on your WorkOS setup and authentication requirements.
from __future__ import annotations
from typing import Any
import httpx
from pydantic import AnyHttpUrl, SecretStr, field_validator
from pydantic_settings import BaseSettings, SettingsConfigDict
@ -42,7 +44,6 @@ class WorkOSProviderSettings(BaseSettings):
redirect_path: str | None = None
required_scopes: list[str] | None = None
timeout_seconds: int | None = None
resource_server_url: AnyHttpUrl | str | None = None
allowed_client_redirect_uris: list[str] | None = None
@field_validator("required_scopes", mode="before")
@ -167,7 +168,6 @@ class WorkOSProvider(OAuthProxy):
redirect_path: str | NotSetT = NotSet,
required_scopes: list[str] | None | NotSetT = NotSet,
timeout_seconds: int | NotSetT = NotSet,
resource_server_url: AnyHttpUrl | str | NotSetT = NotSet,
allowed_client_redirect_uris: list[str] | NotSetT = NotSet,
):
"""Initialize WorkOS OAuth provider.
@ -180,11 +180,10 @@ class WorkOSProvider(OAuthProxy):
redirect_path: Redirect path configured in WorkOS (defaults to "/auth/callback")
required_scopes: Required OAuth scopes (no default)
timeout_seconds: HTTP request timeout for WorkOS API calls
resource_server_url: Path of the FastMCP server (defaults to base_url). If your MCP endpoint is at
a different path like {base_url}/mcp, specify it here for RFC 8707 compliance.
allowed_client_redirect_uris: List of allowed redirect URI patterns for MCP clients.
If None (default), all URIs are allowed. If empty list, no URIs are allowed.
"""
settings = WorkOSProviderSettings.model_validate(
{
k: v
@ -196,7 +195,6 @@ class WorkOSProvider(OAuthProxy):
"redirect_path": redirect_path,
"required_scopes": required_scopes,
"timeout_seconds": timeout_seconds,
"resource_server_url": resource_server_url,
"allowed_client_redirect_uris": allowed_client_redirect_uris,
}.items()
if v is not NotSet
@ -222,11 +220,9 @@ class WorkOSProvider(OAuthProxy):
if not authkit_domain_str.startswith(("http://", "https://")):
authkit_domain_str = f"https://{authkit_domain_str}"
authkit_domain_final = authkit_domain_str.rstrip("/")
base_url_final = settings.base_url or "http://localhost:8000"
redirect_path_final = settings.redirect_path or "/auth/callback"
timeout_seconds_final = settings.timeout_seconds or 10
scopes_final = settings.required_scopes or []
resource_server_url_final = settings.resource_server_url or base_url_final
allowed_client_redirect_uris_final = settings.allowed_client_redirect_uris
# Extract secret string from SecretStr
@ -248,11 +244,10 @@ class WorkOSProvider(OAuthProxy):
upstream_client_id=settings.client_id,
upstream_client_secret=client_secret_str,
token_verifier=token_verifier,
base_url=base_url_final,
base_url=settings.base_url,
redirect_path=redirect_path_final,
issuer_url=base_url_final,
issuer_url=settings.base_url,
allowed_client_redirect_uris=allowed_client_redirect_uris_final,
resource_server_url=resource_server_url_final,
)
logger.info(
@ -359,17 +354,25 @@ class AuthKitProvider(RemoteAuthProvider):
super().__init__(
token_verifier=token_verifier,
authorization_servers=[AnyHttpUrl(self.authkit_domain)],
resource_server_url=self.base_url,
base_url=self.base_url,
)
def get_routes(self) -> list[Route]:
def get_routes(
self,
mcp_path: str | None = None,
mcp_endpoint: Any | None = None,
) -> list[Route]:
"""Get OAuth routes including AuthKit authorization server metadata forwarding.
This returns the standard protected resource routes plus an authorization server
metadata endpoint that forwards AuthKit's OAuth metadata to clients.
Args:
mcp_path: The path where the MCP endpoint is mounted (e.g., "/mcp")
mcp_endpoint: The MCP endpoint handler to protect with auth
"""
# Get the standard protected resource routes from RemoteAuthProvider
routes = super().get_routes()
routes = super().get_routes(mcp_path, mcp_endpoint)
async def oauth_authorization_server_metadata(request):
"""Forward AuthKit OAuth authorization server metadata with FastMCP customizations."""

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

@ -3,21 +3,15 @@ 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, cast
from typing import TYPE_CHECKING
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.middleware.bearer_auth import RequireAuthMiddleware
from mcp.server.lowlevel.server import LifespanResultT
from mcp.server.sse import SseServerTransport
from mcp.server.streamable_http import EventStore
from mcp.server.streamable_http_manager import StreamableHTTPSessionManager
from starlette.applications import Starlette
from starlette.middleware import Middleware
from starlette.middleware.authentication import AuthenticationMiddleware
from starlette.requests import Request
from starlette.responses import Response
from starlette.routing import BaseRoute, Mount, Route
@ -170,40 +164,26 @@ def create_sse_app(
# Set up auth if enabled
if auth:
# Create auth middleware
auth_middleware = [
Middleware(
AuthenticationMiddleware,
backend=BearerAuthBackend(auth),
),
Middleware(AuthContextMiddleware),
]
# Get auth middleware from the provider
auth_middleware = auth.get_middleware()
# Get auth routes and scopes
auth_routes = auth.get_routes()
required_scopes = getattr(auth, "required_scopes", None) or []
# Get resource metadata URL for WWW-Authenticate header
resource_metadata_url = auth.get_resource_metadata_url()
# Get auth routes including protected MCP endpoint
auth_routes = auth.get_routes(
mcp_path=sse_path,
mcp_endpoint=handle_sse,
)
server_routes.extend(auth_routes)
server_middleware.extend(auth_middleware)
# Auth is enabled, wrap endpoints with RequireAuthMiddleware
server_routes.append(
Route(
sse_path,
endpoint=RequireAuthMiddleware(
handle_sse, required_scopes, resource_metadata_url
),
methods=["GET"],
)
)
# Manually wrap the SSE message endpoint with RequireAuthMiddleware
server_routes.append(
Mount(
message_path,
app=RequireAuthMiddleware(
sse.handle_post_message, required_scopes, resource_metadata_url
sse.handle_post_message,
auth.required_scopes,
auth._get_resource_url("/.well-known/oauth-protected-resource"),
),
)
)
@ -291,34 +271,17 @@ def create_streamable_http_app(
# Add StreamableHTTP routes with or without auth
if auth:
# Create auth middleware
auth_middleware = [
Middleware(
AuthenticationMiddleware,
backend=BearerAuthBackend(cast(TokenVerifierProtocol, auth)),
),
Middleware(AuthContextMiddleware),
]
# Get auth middleware from the provider
auth_middleware = auth.get_middleware()
# Get auth routes and scopes
auth_routes = auth.get_routes()
required_scopes = getattr(auth, "required_scopes", None) or []
# Get resource metadata URL for WWW-Authenticate header
resource_metadata_url = auth.get_resource_metadata_url()
# Get auth routes including protected MCP endpoint
auth_routes = auth.get_routes(
mcp_path=streamable_http_path,
mcp_endpoint=streamable_http_app,
)
server_routes.extend(auth_routes)
server_middleware.extend(auth_middleware)
# Auth is enabled, wrap endpoint with RequireAuthMiddleware
server_routes.append(
Route(
streamable_http_path,
endpoint=RequireAuthMiddleware(
streamable_http_app, required_scopes, resource_metadata_url
),
)
)
else:
# No auth required
server_routes.append(

2
tests/server/auth/providers/test_azure.py
View file

@ -98,7 +98,7 @@ class TestAzureProvider:
)
# Check defaults
assert str(provider.base_url) == "http://localhost:8000/"
assert provider.base_url is None
assert provider._redirect_path == "/auth/callback"
# Azure provider defaults are set but we can't easily verify them without accessing internals

2
tests/server/auth/providers/test_github.py
View file

@ -149,7 +149,7 @@ class TestGitHubProvider:
)
# Check defaults
assert str(provider.base_url) == "http://localhost:8000/"
assert provider.base_url is None
assert provider._redirect_path == "/auth/callback"
# The required_scopes should be passed to the token verifier
assert provider._token_validator.required_scopes == ["user"]

2
tests/server/auth/providers/test_google.py
View file

@ -76,7 +76,7 @@ class TestGoogleProvider:
)
# Check defaults
assert str(provider.base_url) == "http://localhost:8000/"
assert provider.base_url is None
assert provider._redirect_path == "/auth/callback"
# Google provider has ["openid"] as default but we can't easily verify without accessing internals

58
tests/server/auth/providers/test_workos.py
View file

@ -1,12 +1,17 @@
"""Tests for WorkOS OAuth provider."""
import os
from collections.abc import Generator
from unittest.mock import patch
from urllib.parse import urlparse
import httpx
import pytest
from fastmcp.server.auth.providers.workos import WorkOSProvider
from fastmcp import Client, FastMCP
from fastmcp.client.transports import StreamableHttpTransport
from fastmcp.server.auth.providers.workos import AuthKitProvider, WorkOSProvider
from fastmcp.utilities.tests import HeadlessOAuth, run_server_in_process
class TestWorkOSProvider:
@ -126,7 +131,7 @@ class TestWorkOSProvider:
)
# Check defaults
assert str(provider.base_url) == "http://localhost:8000/"
assert provider.base_url is None
assert provider._redirect_path == "/auth/callback"
# WorkOS provider has no default scopes but we can't easily verify without accessing internals
@ -150,3 +155,52 @@ class TestWorkOSProvider:
assert (
provider._upstream_revocation_endpoint is None
) # WorkOS doesn't support revocation
def run_mcp_server(host: str, port: int) -> None:
mcp = FastMCP(
auth=AuthKitProvider(
authkit_domain="https://respectful-lullaby-34-staging.authkit.app",
base_url="http://localhost:4321",
)
)
@mcp.tool
def add(a: int, b: int) -> int:
return a + b
mcp.run(host=host, port=port, transport="http")
@pytest.fixture(scope="module")
def mcp_server_url() -> Generator[str]:
with run_server_in_process(run_mcp_server) as url:
yield f"{url}/mcp"
@pytest.fixture()
def client_with_headless_oauth(
mcp_server_url: str,
) -> Generator[Client, None, None]:
"""Client with headless OAuth that bypasses browser interaction."""
client = Client(
transport=StreamableHttpTransport(mcp_server_url),
auth=HeadlessOAuth(mcp_url=mcp_server_url),
)
yield client
class TestAuthKitProvider:
async def test_unauthorized_access(self, mcp_server_url: str):
with pytest.raises(httpx.HTTPStatusError) as exc_info:
async with Client(mcp_server_url) as client:
tools = await client.list_tools() # noqa: F841
assert exc_info.value.response.status_code == 401
assert "tools" not in locals()
# async def test_authorized_access(self, client_with_headless_oauth: Client):
# async with client_with_headless_oauth:
# tools = await client_with_headless_oauth.list_tools()
# assert tools is not None
# assert len(tools) > 0
# assert "add" in tools

102
tests/server/auth/test_auth_provider.py Normal file
View file

@ -0,0 +1,102 @@
import re
import httpx
import pytest
from pydantic import AnyHttpUrl
from fastmcp import FastMCP
from fastmcp.server.auth import RemoteAuthProvider
from fastmcp.server.auth.providers.jwt import StaticTokenVerifier
class TestAuthProviderBase:
"""Test suite for base AuthProvider behaviors that apply to all auth providers."""
@pytest.fixture
def basic_remote_provider(self):
"""Basic RemoteAuthProvider fixture for testing base AuthProvider behaviors."""
# Create a static token verifier with a test token
tokens = {
"test_token": {
"client_id": "test-client",
"scopes": ["read", "write"],
}
}
token_verifier = StaticTokenVerifier(tokens=tokens)
return RemoteAuthProvider(
token_verifier=token_verifier,
authorization_servers=[AnyHttpUrl("https://auth.example.com")],
base_url="https://my-server.com",
)
async def test_www_authenticate_header_points_to_base_url(
self, basic_remote_provider
):
"""Test that WWW-Authenticate header always points to base URL's .well-known.
This test verifies the fix for issue #1685 where the WWW-Authenticate header
was incorrectly including the MCP path in the .well-known URL.
"""
mcp = FastMCP("test-server", auth=basic_remote_provider)
# Mount MCP at a non-root path
mcp_http_app = mcp.http_app(path="/api/v1/mcp")
async with httpx.AsyncClient(
transport=httpx.ASGITransport(app=mcp_http_app),
base_url="https://my-server.com",
) as client:
# Make unauthorized request to MCP endpoint
response = await client.get("/api/v1/mcp")
assert response.status_code == 401
www_auth = response.headers.get("www-authenticate", "")
assert "resource_metadata=" in www_auth
# Extract the metadata URL from the header
match = re.search(r'resource_metadata="([^"]+)"', www_auth)
assert match is not None
metadata_url = match.group(1)
# Should point to base URL, not include /api/v1/mcp
assert (
metadata_url
== "https://my-server.com/.well-known/oauth-protected-resource"
)
async def test_automatic_resource_url_capture(self, basic_remote_provider):
"""Test that resource URL is automatically captured from MCP path.
This test verifies PR #1682 functionality where the resource URL
should be automatically set based on the MCP endpoint path.
"""
mcp = FastMCP("test-server", auth=basic_remote_provider)
# Mount MCP at a specific path
mcp_http_app = mcp.http_app(path="/mcp")
async with httpx.AsyncClient(
transport=httpx.ASGITransport(app=mcp_http_app),
base_url="https://my-server.com",
) as client:
# Get the .well-known metadata
response = await client.get("/.well-known/oauth-protected-resource")
assert response.status_code == 200
data = response.json()
# The resource URL should be automatically set to the MCP path
assert data.get("resource") == "https://my-server.com/mcp"
async def test_automatic_resource_url_with_nested_path(self, basic_remote_provider):
"""Test automatic resource URL capture with deeply nested MCP path."""
mcp = FastMCP("test-server", auth=basic_remote_provider)
mcp_http_app = mcp.http_app(path="/api/v2/services/mcp")
async with httpx.AsyncClient(
transport=httpx.ASGITransport(app=mcp_http_app),
base_url="https://my-server.com",
) as client:
response = await client.get("/.well-known/oauth-protected-resource")
assert response.status_code == 200
data = response.json()
# Should automatically capture the nested path
assert data.get("resource") == "https://my-server.com/api/v2/services/mcp"

4
tests/server/auth/test_jwt_provider.py
View file

@ -137,6 +137,10 @@ def mcp_server_url(rsa_key_pair: RSAKeyPair) -> Generator[str]:
with run_server_in_process(
run_mcp_server,
public_key=rsa_key_pair.public_key,
auth_kwargs=dict(
issuer="https://test.example.com",
audience="https://api.example.com",
),
run_kwargs=dict(transport="http"),
) as url:
yield f"{url}/mcp"

4
tests/server/auth/test_oauth_proxy.py
View file

@ -49,14 +49,12 @@ class TestOAuthProxyComprehensive:
base_url="https://api.example.com", # String instead of AnyHttpUrl
issuer_url="https://issuer.example.com", # String
service_documentation_url="https://docs.example.com", # String
resource_server_url="https://resources.example.com", # String
)
# Should work fine and convert internally to AnyHttpUrl
assert str(proxy.base_url) == "https://api.example.com/"
assert str(proxy.issuer_url) == "https://issuer.example.com/"
assert str(proxy.service_documentation_url) == "https://docs.example.com/"
assert str(proxy.resource_server_url) == "https://resources.example.com/"
def test_initialization_with_all_parameters(self, jwt_verifier):
"""Test OAuthProxy initialization with all optional parameters."""
@ -71,7 +69,6 @@ class TestOAuthProxyComprehensive:
redirect_path="/auth/callback",
issuer_url="https://issuer.example.com",
service_documentation_url="https://docs.example.com",
resource_server_url="https://resources.example.com",
)
# Verify all parameters are set correctly
@ -86,7 +83,6 @@ class TestOAuthProxyComprehensive:
assert proxy._redirect_path == "/auth/callback"
assert str(proxy.issuer_url) == "https://issuer.example.com/"
assert str(proxy.service_documentation_url) == "https://docs.example.com/"
assert str(proxy.resource_server_url) == "https://resources.example.com/"
def test_redirect_path_normalization(self, jwt_verifier):
"""Test that redirect_path is normalized to start with /."""

248
tests/server/auth/test_remote_auth_provider.py
View file

@ -3,68 +3,75 @@ import pytest
from pydantic import AnyHttpUrl
from fastmcp import FastMCP
from fastmcp.server.auth import AccessToken, RemoteAuthProvider, TokenVerifier
from fastmcp.server.auth import RemoteAuthProvider
from fastmcp.server.auth.providers.jwt import StaticTokenVerifier
class SimpleTokenVerifier(TokenVerifier):
"""Simple token verifier for testing."""
def __init__(self, valid_tokens: dict[str, AccessToken] | None = None):
super().__init__()
self.valid_tokens = valid_tokens or {}
async def verify_token(self, token: str) -> AccessToken | None:
return self.valid_tokens.get(token)
@pytest.fixture
def test_tokens():
"""Standard test tokens fixture for all auth tests."""
return {
"test_token": {
"client_id": "test-client",
"scopes": ["read", "write"],
}
}
class TestRemoteAuthProvider:
"""Test suite for RemoteAuthProvider."""
def test_init(self):
def test_init(self, test_tokens):
"""Test RemoteAuthProvider initialization."""
token_verifier = SimpleTokenVerifier()
token_verifier = StaticTokenVerifier(tokens=test_tokens)
auth_servers = [AnyHttpUrl("https://auth.example.com")]
provider = RemoteAuthProvider(
token_verifier=token_verifier,
authorization_servers=auth_servers,
resource_server_url="https://api.example.com",
base_url="https://api.example.com",
)
assert provider.token_verifier is token_verifier
assert provider.authorization_servers == auth_servers
assert provider.resource_server_url == AnyHttpUrl("https://api.example.com")
assert provider.base_url == AnyHttpUrl("https://api.example.com/")
async def test_verify_token_delegates_to_verifier(self):
async def test_verify_token_delegates_to_verifier(self, test_tokens):
"""Test that verify_token delegates to the token verifier."""
access_token = AccessToken(
token="valid_token", client_id="test-client", scopes=[]
)
token_verifier = SimpleTokenVerifier({"valid_token": access_token})
# Use a different token for this specific test
tokens = {
"valid_token": {
"client_id": "test-client",
"scopes": [],
}
}
token_verifier = StaticTokenVerifier(tokens=tokens)
provider = RemoteAuthProvider(
token_verifier=token_verifier,
authorization_servers=[AnyHttpUrl("https://auth.example.com")],
resource_server_url="https://api.example.com",
base_url="https://api.example.com",
)
# Valid token
result = await provider.verify_token("valid_token")
assert result is access_token
assert result is not None
assert result.token == "valid_token"
assert result.client_id == "test-client"
# Invalid token
result = await provider.verify_token("invalid_token")
assert result is None
def test_get_routes_creates_protected_resource_routes(self):
def test_get_routes_creates_protected_resource_routes(self, test_tokens):
"""Test that get_routes creates protected resource routes."""
token_verifier = SimpleTokenVerifier()
token_verifier = StaticTokenVerifier(tokens=test_tokens)
auth_servers = [AnyHttpUrl("https://auth.example.com")]
provider = RemoteAuthProvider(
token_verifier=token_verifier,
authorization_servers=auth_servers,
resource_server_url="https://api.example.com",
base_url="https://api.example.com",
)
routes = provider.get_routes()
@ -76,28 +83,46 @@ class TestRemoteAuthProvider:
assert route.methods is not None
assert "GET" in route.methods
def test_get_resource_metadata_url(self):
"""Test get_resource_metadata_url returns correct URL."""
def test_get_resource_url_with_well_known_path(self):
"""Test _get_resource_url returns correct URL for .well-known path."""
tokens = {
"test_token": {
"client_id": "test-client",
"scopes": ["read"],
}
}
token_verifier = StaticTokenVerifier(tokens=tokens)
provider = RemoteAuthProvider(
token_verifier=SimpleTokenVerifier(),
token_verifier=token_verifier,
authorization_servers=[AnyHttpUrl("https://auth.example.com")],
resource_server_url="https://api.example.com",
base_url="https://api.example.com",
)
metadata_url = provider.get_resource_metadata_url()
metadata_url = provider._get_resource_url(
"/.well-known/oauth-protected-resource"
)
assert metadata_url == AnyHttpUrl(
"https://api.example.com/.well-known/oauth-protected-resource"
)
def test_get_resource_metadata_url_handles_trailing_slash(self):
"""Test get_resource_metadata_url handles trailing slash correctly."""
def test_get_resource_url_handles_trailing_slash(self):
"""Test _get_resource_url handles trailing slash correctly."""
tokens = {
"test_token": {
"client_id": "test-client",
"scopes": ["read"],
}
}
token_verifier = StaticTokenVerifier(tokens=tokens)
provider = RemoteAuthProvider(
token_verifier=SimpleTokenVerifier(),
token_verifier=token_verifier,
authorization_servers=[AnyHttpUrl("https://auth.example.com")],
resource_server_url="https://api.example.com/",
base_url="https://api.example.com/",
)
metadata_url = provider.get_resource_metadata_url()
metadata_url = provider._get_resource_url(
"/.well-known/oauth-protected-resource"
)
assert metadata_url == AnyHttpUrl(
"https://api.example.com/.well-known/oauth-protected-resource"
)
@ -106,16 +131,42 @@ class TestRemoteAuthProvider:
class TestRemoteAuthProviderIntegration:
"""Integration tests for RemoteAuthProvider with FastMCP server."""
async def test_protected_resource_metadata_endpoint_status_code(self):
"""Test that the protected resource metadata endpoint returns 200."""
token_verifier = SimpleTokenVerifier()
auth_provider = RemoteAuthProvider(
@pytest.fixture
def basic_auth_provider(self, test_tokens):
"""Basic RemoteAuthProvider fixture for testing."""
token_verifier = StaticTokenVerifier(tokens=test_tokens)
return RemoteAuthProvider(
token_verifier=token_verifier,
authorization_servers=[AnyHttpUrl("https://auth.example.com")],
resource_server_url="https://api.example.com/mcp",
base_url="https://api.example.com",
)
mcp = FastMCP("test-server", auth=auth_provider)
def _create_test_auth_provider(
self, base_url="https://api.example.com", test_tokens=None, **kwargs
):
"""Helper to create a test RemoteAuthProvider with StaticTokenVerifier."""
tokens = kwargs.get(
"tokens",
test_tokens
or {
"test_token": {
"client_id": "test-client",
"scopes": ["read", "write"],
}
},
)
token_verifier = StaticTokenVerifier(tokens=tokens)
return RemoteAuthProvider(
token_verifier=token_verifier,
authorization_servers=[AnyHttpUrl("https://auth.example.com")],
base_url=base_url,
)
async def test_protected_resource_metadata_endpoint_status_code(
self, basic_auth_provider
):
"""Test that the protected resource metadata endpoint returns 200."""
mcp = FastMCP("test-server", auth=basic_auth_provider)
mcp_http_app = mcp.http_app()
async with httpx.AsyncClient(
@ -127,12 +178,7 @@ class TestRemoteAuthProviderIntegration:
async def test_protected_resource_metadata_endpoint_resource_field(self):
"""Test that the protected resource metadata endpoint returns correct resource field."""
token_verifier = SimpleTokenVerifier()
auth_provider = RemoteAuthProvider(
token_verifier=token_verifier,
authorization_servers=[AnyHttpUrl("https://auth.example.com")],
resource_server_url="https://api.example.com/mcp",
)
auth_provider = self._create_test_auth_provider()
mcp = FastMCP("test-server", auth=auth_provider)
mcp_http_app = mcp.http_app()
@ -151,12 +197,7 @@ class TestRemoteAuthProviderIntegration:
self,
):
"""Test that the protected resource metadata endpoint returns correct authorization_servers field."""
token_verifier = SimpleTokenVerifier()
auth_provider = RemoteAuthProvider(
token_verifier=token_verifier,
authorization_servers=[AnyHttpUrl("https://auth.example.com")],
resource_server_url="https://api.example.com/mcp",
)
auth_provider = self._create_test_auth_provider()
mcp = FastMCP("test-server", auth=auth_provider)
mcp_http_app = mcp.http_app()
@ -171,24 +212,15 @@ class TestRemoteAuthProviderIntegration:
assert data["authorization_servers"] == ["https://auth.example.com/"]
@pytest.mark.parametrize(
"resource_server_url,expected_resource",
"base_url,expected_resource",
[
("https://api.example.com", "https://api.example.com/"),
("https://api.example.com/", "https://api.example.com/"),
("https://api.example.com/mcp", "https://api.example.com/mcp"),
("https://api.example.com/mcp/", "https://api.example.com/mcp/"),
("https://api.example.com", "https://api.example.com/mcp"),
("https://api.example.com/", "https://api.example.com/mcp"),
],
)
async def test_resource_server_url_configurations(
self, resource_server_url: str, expected_resource: str
):
"""Test different resource_server_url configurations."""
token_verifier = SimpleTokenVerifier()
auth_provider = RemoteAuthProvider(
token_verifier=token_verifier,
authorization_servers=[AnyHttpUrl("https://auth.example.com")],
resource_server_url=resource_server_url,
)
async def test_base_url_configurations(self, base_url: str, expected_resource: str):
"""Test different base_url configurations."""
auth_provider = self._create_test_auth_provider(base_url=base_url)
mcp = FastMCP("test-server", auth=auth_provider)
mcp_http_app = mcp.http_app()
@ -204,17 +236,14 @@ class TestRemoteAuthProviderIntegration:
async def test_multiple_authorization_servers_resource_field(self):
"""Test resource field with multiple authorization servers."""
token_verifier = SimpleTokenVerifier()
auth_servers = [
AnyHttpUrl("https://auth1.example.com"),
AnyHttpUrl("https://auth2.example.com"),
]
auth_provider = RemoteAuthProvider(
token_verifier=token_verifier,
authorization_servers=auth_servers,
resource_server_url="https://api.example.com/mcp",
)
auth_provider = self._create_test_auth_provider()
# Override the authorization servers
auth_provider.authorization_servers = auth_servers
mcp = FastMCP("test-server", auth=auth_provider)
mcp_http_app = mcp.http_app()
@ -230,17 +259,14 @@ class TestRemoteAuthProviderIntegration:
async def test_multiple_authorization_servers_list(self):
"""Test authorization_servers field with multiple authorization servers."""
token_verifier = SimpleTokenVerifier()
auth_servers = [
AnyHttpUrl("https://auth1.example.com"),
AnyHttpUrl("https://auth2.example.com"),
]
auth_provider = RemoteAuthProvider(
token_verifier=token_verifier,
authorization_servers=auth_servers,
resource_server_url="https://api.example.com/mcp",
)
auth_provider = self._create_test_auth_provider()
# Override the authorization servers
auth_provider.authorization_servers = auth_servers
mcp = FastMCP("test-server", auth=auth_provider)
mcp_http_app = mcp.http_app()
@ -264,35 +290,43 @@ class TestRemoteAuthProviderIntegration:
# endpoint correctly reports the resource server URL, which is tested above
# This is primarily testing that the token verifier integration works
access_token = AccessToken(
token="valid_token", client_id="test-client", scopes=[]
)
token_verifier = SimpleTokenVerifier({"valid_token": access_token})
tokens = {
"valid_token": {
"client_id": "test-client",
"scopes": [],
}
}
token_verifier = StaticTokenVerifier(tokens=tokens)
provider = RemoteAuthProvider(
token_verifier=token_verifier,
authorization_servers=[AnyHttpUrl("https://auth.example.com")],
resource_server_url="https://api.example.com/mcp",
base_url="https://api.example.com",
)
# Test that the provider correctly delegates to the token verifier
result = await provider.verify_token("valid_token")
assert result is access_token
assert result is not None
assert result.token == "valid_token"
assert result.client_id == "test-client"
result = await provider.verify_token("invalid_token")
assert result is None
async def test_token_verification_with_invalid_auth_fails(self):
"""Test that the provider correctly rejects invalid tokens."""
access_token = AccessToken(
token="valid_token", client_id="test-client", scopes=[]
)
token_verifier = SimpleTokenVerifier({"valid_token": access_token})
tokens = {
"valid_token": {
"client_id": "test-client",
"scopes": [],
}
}
token_verifier = StaticTokenVerifier(tokens=tokens)
provider = RemoteAuthProvider(
token_verifier=token_verifier,
authorization_servers=[AnyHttpUrl("https://auth.example.com")],
resource_server_url="https://api.example.com/mcp",
base_url="https://api.example.com",
)
# Test that invalid tokens are rejected
@ -303,13 +337,19 @@ class TestRemoteAuthProviderIntegration:
"""Test that RemoteAuthProvider correctly returns the full MCP endpoint URL.
This test confirms that RemoteAuthProvider works correctly and returns
the exact resource_server_url specified, including full paths like /mcp/.
the resource URL with the MCP path appended to the base URL.
"""
token_verifier = SimpleTokenVerifier()
tokens = {
"test_token": {
"client_id": "test-client",
"scopes": ["read"],
}
}
token_verifier = StaticTokenVerifier(tokens=tokens)
auth_provider = RemoteAuthProvider(
token_verifier=token_verifier,
authorization_servers=[AnyHttpUrl("https://accounts.google.com")],
resource_server_url="https://my-server.com/mcp/",
base_url="https://my-server.com",
)
mcp = FastMCP("test-server", auth=auth_provider)
@ -325,7 +365,7 @@ class TestRemoteAuthProviderIntegration:
data = response.json()
# The RemoteAuthProvider correctly returns the full MCP endpoint URL
assert data["resource"] == "https://my-server.com/mcp/"
assert data["resource"] == "https://my-server.com/mcp"
assert data["authorization_servers"] == ["https://accounts.google.com/"]
async def test_resource_name_field(self):
@ -334,11 +374,17 @@ class TestRemoteAuthProviderIntegration:
This test confirms that RemoteAuthProvider works correctly and returns
the exact resource_name specified.
"""
token_verifier = SimpleTokenVerifier()
tokens = {
"test_token": {
"client_id": "test-client",
"scopes": ["read"],
}
}
token_verifier = StaticTokenVerifier(tokens=tokens)
auth_provider = RemoteAuthProvider(
token_verifier=token_verifier,
authorization_servers=[AnyHttpUrl("https://accounts.google.com")],
resource_server_url="https://my-server.com/mcp/",
base_url="https://my-server.com",
resource_name="My Test Resource",
)
@ -363,11 +409,17 @@ class TestRemoteAuthProviderIntegration:
This test confirms that RemoteAuthProvider works correctly and returns
the exact resource_documentation specified.
"""
token_verifier = SimpleTokenVerifier()
tokens = {
"test_token": {
"client_id": "test-client",
"scopes": ["read"],
}
}
token_verifier = StaticTokenVerifier(tokens=tokens)
auth_provider = RemoteAuthProvider(
token_verifier=token_verifier,
authorization_servers=[AnyHttpUrl("https://accounts.google.com")],
resource_server_url="https://my-server.com/mcp/",
base_url="https://my-server.com",
resource_documentation=AnyHttpUrl(
"https://doc.my-server.com/resource-docs"
),

58
tests/server/auth/test_workos.py
View file

@ -1,58 +0,0 @@
from collections.abc import Generator
import httpx
import pytest
from fastmcp import Client, FastMCP
from fastmcp.client.transports import StreamableHttpTransport
from fastmcp.server.auth.providers.workos import AuthKitProvider
from fastmcp.utilities.tests import HeadlessOAuth, run_server_in_process
def run_mcp_server(host: str, port: int) -> None:
mcp = FastMCP(
auth=AuthKitProvider(
authkit_domain="https://respectful-lullaby-34-staging.authkit.app",
base_url="http://localhost:4321",
)
)
@mcp.tool
def add(a: int, b: int) -> int:
return a + b
mcp.run(host=host, port=port, transport="http")
@pytest.fixture(scope="module")
def mcp_server_url() -> Generator[str]:
with run_server_in_process(run_mcp_server) as url:
yield f"{url}/mcp"
@pytest.fixture()
def client_with_headless_oauth(
mcp_server_url: str,
) -> Generator[Client, None, None]:
"""Client with headless OAuth that bypasses browser interaction."""
client = Client(
transport=StreamableHttpTransport(mcp_server_url),
auth=HeadlessOAuth(mcp_url=mcp_server_url),
)
yield client
class TestAuthKitProvider:
async def test_unauthorized_access(self, mcp_server_url: str):
with pytest.raises(httpx.HTTPStatusError) as exc_info:
async with Client(mcp_server_url) as client:
tools = await client.list_tools() # noqa: F841
assert exc_info.value.response.status_code == 401
assert "tools" not in locals()
# async def test_authorized_access(self, client_with_headless_oauth: Client):
# async with client_with_headless_oauth:
# tools = await client_with_headless_oauth.list_tools()
# assert tools is not None
# assert len(tools) > 0
# assert "add" in tools

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

@ -21,7 +21,7 @@ class TestStreamableHTTPAppResourceMetadataURL:
public_key=rsa_key_pair.public_key,
issuer="https://issuer",
audience="https://audience",
resource_server_url="https://resource.example.com",
base_url="https://resource.example.com",
)
return provider
@ -49,7 +49,7 @@ class TestStreamableHTTPAppResourceMetadataURL:
public_key=rsa_key_pair.public_key,
issuer="https://issuer",
audience="https://audience",
resource_server_url="https://resource.example.com/",
base_url="https://resource.example.com/",
)
server = FastMCP(name="TestServer")
app = create_streamable_http_app(