mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-08-23 05:54:19 +02:00
Improve DescopeProvider scope discovery and well-known URL support (#4489)
* Improve DescopeProvider scope discovery and well-known URL support Co-authored-by: Cursor <cursoragent@cursor.com> * Simplify DescopeProvider scope and URL handling Co-authored-by: Cursor <cursoragent@cursor.com> * Make DescopeProvider scope discovery async and lazy Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> 🤖 Generated with Claude Code * Use generic scope in Descope tests Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> 🤖 Generated with Claude Code * Address Descope discovery edge cases * Deduplicate Descope metadata fallback --------- Co-authored-by: Cursor <cursoragent@cursor.com> Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com> Co-authored-by: Jeremiah Lowin <153965+jlowin@users.noreply.github.com>
This commit is contained in:
parent
c241cd4698
commit
3fdeedb567
3 changed files with 525 additions and 108 deletions
|
|
@ -23,11 +23,10 @@ Before you begin, you will need:
|
|||
### Step 1: Configure Descope
|
||||
|
||||
<Steps>
|
||||
<Step title="Create an MCP Server">
|
||||
1. Go to the [MCP Servers page](https://app.descope.com/mcp-servers) of the Descope Console, and create a new MCP Server.
|
||||
2. Give the MCP server a name and description.
|
||||
3. Ensure that **Dynamic Client Registration (DCR)** is enabled. Then click **Create**.
|
||||
4. Once you've created the MCP Server, note your Well-Known URL.
|
||||
<Step title="Configure a Descope application">
|
||||
You can use either a resource-specific Descope MCP Server or a project-level inbound app.
|
||||
|
||||
To create an MCP Server, go to the [MCP Servers page](https://app.descope.com/mcp-servers) of the Descope Console, create a server, and enable **Dynamic Client Registration (DCR)**.
|
||||
|
||||
|
||||
<Warning>
|
||||
|
|
@ -35,10 +34,17 @@ Before you begin, you will need:
|
|||
</Warning>
|
||||
</Step>
|
||||
|
||||
<Step title="Note Your Well-Known URL">
|
||||
Save your Well-Known URL from [MCP Server Settings](https://app.descope.com/mcp-servers):
|
||||
<Step title="Copy the Well-Known URL">
|
||||
`DescopeProvider` accepts both resource-specific MCP Server URLs:
|
||||
|
||||
```
|
||||
Well-Known URL: https://.../v1/apps/agentic/P.../M.../.well-known/openid-configuration
|
||||
https://api.descope.com/v1/apps/agentic/P.../M.../.well-known/openid-configuration
|
||||
```
|
||||
|
||||
and project-level inbound app URLs:
|
||||
|
||||
```
|
||||
https://api.descope.com/v1/apps/P.../.well-known/openid-configuration
|
||||
```
|
||||
</Step>
|
||||
</Steps>
|
||||
|
|
@ -48,7 +54,7 @@ Before you begin, you will need:
|
|||
Create a `.env` file with your Descope configuration:
|
||||
|
||||
```bash
|
||||
DESCOPE_CONFIG_URL=https://.../v1/apps/agentic/P.../M.../.well-known/openid-configuration # Your Descope Well-Known URL
|
||||
DESCOPE_CONFIG_URL=https://api.descope.com/v1/apps/P.../.well-known/openid-configuration
|
||||
SERVER_URL=http://localhost:3000 # Your server's base URL
|
||||
```
|
||||
|
||||
|
|
@ -60,18 +66,35 @@ Create your FastMCP server file and use the DescopeProvider to handle all the OA
|
|||
from fastmcp import FastMCP
|
||||
from fastmcp.server.auth.providers.descope import DescopeProvider
|
||||
|
||||
# The DescopeProvider automatically discovers Descope endpoints
|
||||
# and configures JWT token validation
|
||||
# DescopeProvider accepts either supported Well-Known URL format.
|
||||
auth_provider = DescopeProvider(
|
||||
config_url="https://.../.well-known/openid-configuration", # Your MCP Server .well-known URL
|
||||
base_url=SERVER_URL, # Your server's public URL
|
||||
config_url="https://api.descope.com/v1/apps/P.../.well-known/openid-configuration",
|
||||
base_url="https://your-fastmcp-server.com",
|
||||
)
|
||||
|
||||
# Create FastMCP server with auth
|
||||
mcp = FastMCP(name="My Descope Protected Server", auth=auth_provider)
|
||||
|
||||
```
|
||||
|
||||
### Scope discovery and validation
|
||||
|
||||
When both `scopes_supported` and `required_scopes` are omitted, `DescopeProvider` discovers `scopes_supported` lazily from the OpenID configuration and advertises them to MCP clients. Provider construction remains network-free, and a transient discovery failure is retried on a later metadata request.
|
||||
|
||||
Set both options when clients should request a broader set of scopes than the server requires on every token:
|
||||
|
||||
```python
|
||||
from fastmcp.server.auth.providers.descope import DescopeProvider
|
||||
|
||||
auth_provider = DescopeProvider(
|
||||
config_url="https://api.descope.com/v1/apps/P.../.well-known/openid-configuration",
|
||||
base_url="https://your-fastmcp-server.com",
|
||||
scopes_supported=["mcp:read", "mcp:write"],
|
||||
required_scopes=["mcp:read"],
|
||||
)
|
||||
```
|
||||
|
||||
`scopes_supported` controls what the protected resource metadata advertises. `required_scopes` controls what the JWT verifier requires during token validation. When only `required_scopes` is set, those scopes are also advertised to clients.
|
||||
|
||||
## Testing
|
||||
|
||||
To test your server, you can use the `fastmcp` CLI to run it locally. Assuming you've saved the above code to `server.py` (after replacing the environment variables with your actual values!), you can run the following command:
|
||||
|
|
|
|||
|
|
@ -7,11 +7,16 @@ for seamless MCP client authentication.
|
|||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
from urllib.parse import urlparse
|
||||
|
||||
import httpx2
|
||||
from mcp.server.auth.json_response import PydanticJSONResponse
|
||||
from mcp.server.auth.routes import build_resource_metadata_url, cors_middleware
|
||||
from mcp.shared.auth import ProtectedResourceMetadata
|
||||
from pydantic import AnyHttpUrl
|
||||
from starlette.responses import JSONResponse
|
||||
from starlette.requests import Request
|
||||
from starlette.responses import JSONResponse, Response
|
||||
from starlette.routing import Route
|
||||
|
||||
from fastmcp.server.auth import RemoteAuthProvider, TokenVerifier
|
||||
|
|
@ -21,43 +26,87 @@ from fastmcp.utilities.logging import get_logger
|
|||
|
||||
logger = get_logger(__name__)
|
||||
|
||||
_OPENID_WK = "/.well-known/openid-configuration"
|
||||
_OAUTH_WK = "/.well-known/oauth-authorization-server"
|
||||
|
||||
|
||||
def _parse_descope_config_url(config_url: str) -> tuple[str, str, str, str]:
|
||||
openid_url = config_url.rstrip("/")
|
||||
if not openid_url.endswith(_OPENID_WK):
|
||||
openid_url = f"{openid_url}{_OPENID_WK}"
|
||||
|
||||
issuer_url = openid_url[: -len(_OPENID_WK)]
|
||||
parsed = urlparse(issuer_url)
|
||||
parts = parsed.path.strip("/").split("/")
|
||||
descope_base_url = f"{parsed.scheme}://{parsed.netloc}".rstrip("/")
|
||||
|
||||
if "agentic" in parts:
|
||||
index = parts.index("agentic") + 1
|
||||
project_id = parts[index] if index < len(parts) else ""
|
||||
elif "apps" in parts:
|
||||
index = parts.index("apps") + 1
|
||||
project_id = parts[index] if index < len(parts) else ""
|
||||
if project_id == "agentic":
|
||||
project_id = ""
|
||||
else:
|
||||
project_id = ""
|
||||
|
||||
if not project_id:
|
||||
raise ValueError(f"Could not extract project_id from config_url: {issuer_url}")
|
||||
|
||||
return descope_base_url, project_id, issuer_url, openid_url
|
||||
|
||||
|
||||
async def _discover_scopes(openid_configuration_url: str) -> list[str] | None:
|
||||
try:
|
||||
async with httpx2.AsyncClient() as client:
|
||||
response = await client.get(openid_configuration_url, timeout=10.0)
|
||||
response.raise_for_status()
|
||||
scopes = response.json().get("scopes_supported")
|
||||
if isinstance(scopes, list):
|
||||
parsed = [scope for scope in scopes if isinstance(scope, str)]
|
||||
if not scopes or parsed:
|
||||
return parsed
|
||||
except Exception:
|
||||
logger.warning(
|
||||
"Failed to fetch Descope OpenID configuration from %s",
|
||||
openid_configuration_url,
|
||||
exc_info=True,
|
||||
)
|
||||
return None
|
||||
|
||||
|
||||
class DescopeProvider(RemoteAuthProvider):
|
||||
"""Descope metadata provider for DCR (Dynamic Client Registration).
|
||||
"""Descope metadata provider for Dynamic Client Registration (DCR).
|
||||
|
||||
This provider implements Descope integration using metadata forwarding.
|
||||
This is the recommended approach for Descope DCR
|
||||
as it allows Descope to handle the OAuth flow directly while FastMCP acts
|
||||
as a resource server.
|
||||
The provider accepts either a resource-specific Descope MCP Server URL such
|
||||
as `/v1/apps/agentic/P.../M.../.well-known/openid-configuration` or a
|
||||
project-level inbound app URL such as
|
||||
`/v1/apps/P.../.well-known/openid-configuration`.
|
||||
|
||||
IMPORTANT SETUP REQUIREMENTS:
|
||||
|
||||
1. Create an MCP Server in Descope Console:
|
||||
- Go to the [MCP Servers page](https://app.descope.com/mcp-servers) of the Descope Console
|
||||
- Create a new MCP Server
|
||||
- Ensure that **Dynamic Client Registration (DCR)** is enabled
|
||||
- Note your Well-Known URL
|
||||
|
||||
2. Note your Well-Known URL:
|
||||
- Save your Well-Known URL from [MCP Server Settings](https://app.descope.com/mcp-servers)
|
||||
- Format: ``https://.../v1/apps/agentic/P.../M.../.well-known/openid-configuration``
|
||||
|
||||
For detailed setup instructions, see:
|
||||
https://docs.descope.com/identity-federation/inbound-apps/creating-inbound-apps#method-2-dynamic-client-registration-dcr
|
||||
When neither `scopes_supported` nor `required_scopes` is provided, advertised
|
||||
scopes are discovered lazily from the OpenID configuration. Use
|
||||
`scopes_supported` and `required_scopes` together when the scopes clients
|
||||
should request differ from the scopes enforced during token validation.
|
||||
|
||||
Example:
|
||||
```python
|
||||
from fastmcp import FastMCP
|
||||
from fastmcp.server.auth.providers.descope import DescopeProvider
|
||||
|
||||
# Create Descope metadata provider (JWT verifier created automatically)
|
||||
descope_auth = DescopeProvider(
|
||||
config_url="https://.../v1/apps/agentic/P.../M.../.well-known/openid-configuration",
|
||||
auth = DescopeProvider(
|
||||
config_url=(
|
||||
"https://api.descope.com/v1/apps/P.../"
|
||||
".well-known/openid-configuration"
|
||||
),
|
||||
base_url="https://your-fastmcp-server.com",
|
||||
)
|
||||
|
||||
# Use with FastMCP
|
||||
mcp = FastMCP("My App", auth=descope_auth)
|
||||
mcp = FastMCP("My App", auth=auth)
|
||||
```
|
||||
|
||||
See [Descope's inbound app documentation](https://docs.descope.com/identity-federation/inbound-apps/creating-inbound-apps#method-2-dynamic-client-registration-dcr)
|
||||
for DCR setup instructions.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
|
|
@ -73,121 +122,182 @@ class DescopeProvider(RemoteAuthProvider):
|
|||
resource_documentation: AnyHttpUrl | None = None,
|
||||
token_verifier: TokenVerifier | None = None,
|
||||
):
|
||||
"""Initialize Descope metadata provider.
|
||||
"""Initialize the Descope provider.
|
||||
|
||||
Args:
|
||||
base_url: Public URL of this FastMCP server
|
||||
config_url: Your Descope Well-Known URL (e.g., "https://.../v1/apps/agentic/P.../M.../.well-known/openid-configuration")
|
||||
This is the new recommended way. If provided, project_id and descope_base_url are ignored.
|
||||
project_id: Your Descope Project ID (e.g., "P2abc123"). Used with descope_base_url for backwards compatibility.
|
||||
descope_base_url: Your Descope base URL (e.g., "https://api.descope.com"). Used with project_id for backwards compatibility.
|
||||
required_scopes: Optional list of scopes that must be present in validated tokens.
|
||||
These scopes will be included in the protected resource metadata.
|
||||
scopes_supported: Optional list of scopes to advertise in OAuth metadata.
|
||||
If None, uses required_scopes. Use this when the scopes clients should
|
||||
request differ from the scopes enforced on tokens.
|
||||
resource_name: Optional name for the protected resource metadata.
|
||||
resource_documentation: Optional documentation URL for the protected resource.
|
||||
token_verifier: Optional token verifier. If None, creates JWT verifier for Descope
|
||||
base_url: Public URL of this FastMCP server.
|
||||
config_url: A resource-specific or project-level Descope OpenID
|
||||
configuration URL. When provided, `project_id` and
|
||||
`descope_base_url` are ignored.
|
||||
project_id: Descope project ID. Used with `descope_base_url` for
|
||||
backwards compatibility.
|
||||
descope_base_url: Descope API base URL. Used with `project_id` for
|
||||
backwards compatibility.
|
||||
required_scopes: Scopes required during token validation. When
|
||||
`scopes_supported` is omitted, these are also advertised to clients.
|
||||
scopes_supported: Scopes advertised to OAuth clients. When both this
|
||||
and `required_scopes` are omitted, scopes are discovered lazily
|
||||
from `config_url`.
|
||||
resource_name: Optional protected resource name.
|
||||
resource_documentation: Optional protected resource documentation URL.
|
||||
token_verifier: Optional custom token verifier. A Descope JWT verifier
|
||||
is created when omitted.
|
||||
"""
|
||||
self.base_url = AnyHttpUrl(str(base_url).rstrip("/"))
|
||||
|
||||
# Parse scopes if provided as string
|
||||
parsed_scopes = (
|
||||
parsed_required_scopes = (
|
||||
parse_scopes(required_scopes) if required_scopes is not None else None
|
||||
)
|
||||
parsed_scopes_supported = (
|
||||
parse_scopes(scopes_supported) if scopes_supported is not None else None
|
||||
)
|
||||
|
||||
# Determine which API is being used
|
||||
if config_url is not None:
|
||||
# New API: use config_url
|
||||
# Strip /.well-known/openid-configuration from config_url if present
|
||||
issuer_url = str(config_url)
|
||||
if issuer_url.endswith("/.well-known/openid-configuration"):
|
||||
issuer_url = issuer_url[: -len("/.well-known/openid-configuration")]
|
||||
|
||||
# Parse the issuer URL to extract descope_base_url and project_id for other uses
|
||||
parsed_url = urlparse(issuer_url)
|
||||
path_parts = parsed_url.path.strip("/").split("/")
|
||||
|
||||
# Extract project_id from path (format: /v1/apps/agentic/P.../M...)
|
||||
if "agentic" in path_parts:
|
||||
agentic_index = path_parts.index("agentic")
|
||||
if agentic_index + 1 < len(path_parts):
|
||||
self.project_id = path_parts[agentic_index + 1]
|
||||
else:
|
||||
raise ValueError(
|
||||
f"Could not extract project_id from config_url: {issuer_url}"
|
||||
)
|
||||
else:
|
||||
raise ValueError(
|
||||
f"Could not find 'agentic' in config_url path: {issuer_url}"
|
||||
)
|
||||
|
||||
# Extract descope_base_url (scheme + netloc)
|
||||
self.descope_base_url = f"{parsed_url.scheme}://{parsed_url.netloc}".rstrip(
|
||||
"/"
|
||||
)
|
||||
(
|
||||
self.descope_base_url,
|
||||
self.project_id,
|
||||
issuer_url,
|
||||
self.openid_configuration_url,
|
||||
) = _parse_descope_config_url(str(config_url))
|
||||
elif project_id is not None and descope_base_url is not None:
|
||||
# Old API: use project_id and descope_base_url
|
||||
self.project_id = project_id
|
||||
descope_base_url_str = str(descope_base_url).rstrip("/")
|
||||
# Ensure descope_base_url has a scheme
|
||||
if not descope_base_url_str.startswith(("http://", "https://")):
|
||||
descope_base_url_str = f"https://{descope_base_url_str}"
|
||||
self.descope_base_url = descope_base_url_str
|
||||
# Old issuer format
|
||||
issuer_url = f"{self.descope_base_url}/v1/apps/{self.project_id}"
|
||||
self.openid_configuration_url = f"{issuer_url}{_OPENID_WK}"
|
||||
else:
|
||||
raise ValueError(
|
||||
"Either config_url (new API) or both project_id and descope_base_url (old API) must be provided"
|
||||
)
|
||||
|
||||
# Create default JWT verifier if none provided
|
||||
self.oauth_authorization_server_metadata_url = (
|
||||
self.openid_configuration_url.replace(_OPENID_WK, _OAUTH_WK)
|
||||
)
|
||||
|
||||
# Advertised scopes are discovered from Descope's OpenID configuration
|
||||
# only when the caller supplied neither explicit advertised scopes nor
|
||||
# required scopes. Discovery is deferred to the first protected resource
|
||||
# metadata request (see get_routes) so construction never performs I/O
|
||||
# and a transient failure can be retried instead of being frozen for the
|
||||
# provider's lifetime.
|
||||
custom_verifier_scopes = (
|
||||
token_verifier.scopes_supported if token_verifier is not None else []
|
||||
)
|
||||
self._scopes_discovery_enabled = (
|
||||
parsed_scopes_supported is None
|
||||
and parsed_required_scopes is None
|
||||
and not custom_verifier_scopes
|
||||
)
|
||||
self._discovered_scopes: list[str] | None = None
|
||||
self._scopes_discovered = False
|
||||
self._scopes_discovery_lock = asyncio.Lock()
|
||||
|
||||
if token_verifier is None:
|
||||
token_verifier = JWTVerifier(
|
||||
jwks_uri=f"{self.descope_base_url}/{self.project_id}/.well-known/jwks.json",
|
||||
issuer=issuer_url,
|
||||
algorithm="RS256",
|
||||
audience=self.project_id,
|
||||
required_scopes=parsed_scopes,
|
||||
required_scopes=parsed_required_scopes,
|
||||
)
|
||||
|
||||
# Initialize RemoteAuthProvider with Descope as the authorization server
|
||||
super().__init__(
|
||||
token_verifier=token_verifier,
|
||||
authorization_servers=[AnyHttpUrl(issuer_url)],
|
||||
base_url=self.base_url,
|
||||
scopes_supported=scopes_supported,
|
||||
scopes_supported=parsed_scopes_supported,
|
||||
resource_name=resource_name,
|
||||
resource_documentation=resource_documentation,
|
||||
)
|
||||
|
||||
async def _get_scopes_supported(self) -> list[str] | None:
|
||||
"""Return the advertised scopes, discovering them lazily if enabled.
|
||||
|
||||
The result of a successful discovery is cached for the provider's
|
||||
lifetime. Transient failures return ``None`` without caching, so the
|
||||
next protected resource metadata request retries discovery.
|
||||
"""
|
||||
if self._scopes_discovered:
|
||||
return self._discovered_scopes
|
||||
|
||||
async with self._scopes_discovery_lock:
|
||||
if self._scopes_discovered:
|
||||
return self._discovered_scopes
|
||||
|
||||
scopes = await _discover_scopes(self.openid_configuration_url)
|
||||
if scopes is not None:
|
||||
self._discovered_scopes = scopes
|
||||
self._scopes_discovered = True
|
||||
return scopes
|
||||
|
||||
def _create_protected_resource_route(self, resource_url: AnyHttpUrl) -> Route:
|
||||
"""Build a protected resource metadata route that discovers scopes lazily.
|
||||
|
||||
Mirrors ``create_protected_resource_routes`` (RFC 9728) but resolves
|
||||
``scopes_supported`` per request so the value can be discovered from
|
||||
Descope after construction.
|
||||
"""
|
||||
|
||||
async def protected_resource_metadata(request: Request) -> Response:
|
||||
scopes_supported = await self._get_scopes_supported()
|
||||
metadata = ProtectedResourceMetadata(
|
||||
resource=resource_url,
|
||||
authorization_servers=self.authorization_servers,
|
||||
scopes_supported=scopes_supported,
|
||||
resource_name=self.resource_name,
|
||||
resource_documentation=self.resource_documentation,
|
||||
)
|
||||
cache_control = (
|
||||
"public, max-age=3600" if self._scopes_discovered else "no-store"
|
||||
)
|
||||
return PydanticJSONResponse(
|
||||
content=metadata,
|
||||
headers={"Cache-Control": cache_control},
|
||||
)
|
||||
|
||||
well_known_path = urlparse(str(build_resource_metadata_url(resource_url))).path
|
||||
return Route(
|
||||
well_known_path,
|
||||
endpoint=cors_middleware(protected_resource_metadata, ["GET", "OPTIONS"]),
|
||||
methods=["GET", "OPTIONS"],
|
||||
)
|
||||
|
||||
def get_routes(
|
||||
self,
|
||||
mcp_path: str | None = None,
|
||||
) -> list[Route]:
|
||||
"""Get OAuth routes including Descope authorization server metadata forwarding.
|
||||
|
||||
This returns the standard protected resource routes plus an authorization server
|
||||
metadata endpoint that forwards Descope's OAuth metadata to clients.
|
||||
|
||||
Args:
|
||||
mcp_path: The path where the MCP endpoint is mounted (e.g., "/mcp")
|
||||
This is used to advertise the resource URL in metadata.
|
||||
"""
|
||||
# Get the standard protected resource routes from RemoteAuthProvider
|
||||
routes = super().get_routes(mcp_path)
|
||||
if self._scopes_discovery_enabled:
|
||||
# Serve protected resource metadata from an async handler that
|
||||
# discovers scopes_supported lazily. The parent's static route would
|
||||
# freeze the (as-yet-unknown) scopes at startup.
|
||||
self.set_mcp_path(mcp_path)
|
||||
routes = []
|
||||
resource_url = self._get_resource_url(mcp_path)
|
||||
if resource_url:
|
||||
routes.append(self._create_protected_resource_route(resource_url))
|
||||
else:
|
||||
# Advertised scopes are already known; the parent builds the static
|
||||
# protected resource metadata route with no network access.
|
||||
routes = super().get_routes(mcp_path)
|
||||
|
||||
async def oauth_authorization_server_metadata(request):
|
||||
"""Forward Descope OAuth authorization server metadata with FastMCP customizations."""
|
||||
metadata_urls = [self.oauth_authorization_server_metadata_url]
|
||||
project_metadata_url = (
|
||||
f"{self.descope_base_url}/v1/apps/{self.project_id}{_OAUTH_WK}"
|
||||
)
|
||||
if project_metadata_url != self.oauth_authorization_server_metadata_url:
|
||||
metadata_urls.append(project_metadata_url)
|
||||
try:
|
||||
async with httpx2.AsyncClient() as client:
|
||||
response = await client.get(
|
||||
f"{self.descope_base_url}/v1/apps/{self.project_id}/.well-known/oauth-authorization-server"
|
||||
)
|
||||
response.raise_for_status()
|
||||
metadata = response.json()
|
||||
return JSONResponse(metadata)
|
||||
for metadata_url in metadata_urls:
|
||||
try:
|
||||
response = await client.get(metadata_url)
|
||||
response.raise_for_status()
|
||||
return JSONResponse(response.json())
|
||||
except Exception:
|
||||
continue
|
||||
except Exception as e:
|
||||
return JSONResponse(
|
||||
{
|
||||
|
|
@ -197,7 +307,14 @@ class DescopeProvider(RemoteAuthProvider):
|
|||
status_code=500,
|
||||
)
|
||||
|
||||
# Add Descope authorization server metadata forwarding
|
||||
return JSONResponse(
|
||||
{
|
||||
"error": "server_error",
|
||||
"error_description": "Failed to fetch Descope metadata",
|
||||
},
|
||||
status_code=500,
|
||||
)
|
||||
|
||||
routes.append(
|
||||
Route(
|
||||
"/.well-known/oauth-authorization-server",
|
||||
|
|
|
|||
|
|
@ -1,10 +1,12 @@
|
|||
"""Tests for Descope OAuth provider."""
|
||||
|
||||
import os
|
||||
from unittest.mock import patch
|
||||
from unittest.mock import AsyncMock, patch
|
||||
|
||||
import httpx2
|
||||
import pytest
|
||||
from mcp import MCPError
|
||||
from starlette.requests import Request
|
||||
|
||||
from fastmcp import Client, FastMCP
|
||||
from fastmcp.client.transports import StreamableHttpTransport
|
||||
|
|
@ -12,6 +14,12 @@ from fastmcp.server.auth.providers.descope import DescopeProvider
|
|||
from fastmcp.server.auth.providers.jwt import JWTVerifier
|
||||
from fastmcp.utilities.tests import HeadlessOAuth, run_server_async
|
||||
|
||||
PROJECT_LEVEL_OPENID_CONFIGURATION = {
|
||||
"issuer": "https://api.descope.com/v1/apps/P2v9EBlmO4XTrOwMRfsY1jeUONxU",
|
||||
"scopes_supported": ["mcp:read"],
|
||||
"jwks_uri": "https://api.descope.com/P2v9EBlmO4XTrOwMRfsY1jeUONxU/.well-known/jwks.json",
|
||||
}
|
||||
|
||||
|
||||
class TestDescopeProvider:
|
||||
"""Test Descope OAuth provider functionality."""
|
||||
|
|
@ -66,6 +74,229 @@ class TestDescopeProvider:
|
|||
assert str(provider3.descope_base_url) == "https://api.descope.com"
|
||||
assert provider3.project_id == "P2abc123"
|
||||
|
||||
def test_project_level_config_url_parsing(self):
|
||||
"""Test project-level well-known URLs without the agentic path segment."""
|
||||
provider = DescopeProvider(
|
||||
config_url="https://api.descope.com/v1/apps/P2v9EBlmO4XTrOwMRfsY1jeUONxU/.well-known/openid-configuration",
|
||||
base_url="https://myserver.com",
|
||||
)
|
||||
|
||||
assert provider.project_id == "P2v9EBlmO4XTrOwMRfsY1jeUONxU"
|
||||
assert str(provider.descope_base_url) == "https://api.descope.com"
|
||||
assert isinstance(provider.token_verifier, JWTVerifier)
|
||||
assert (
|
||||
provider.token_verifier.issuer
|
||||
== "https://api.descope.com/v1/apps/P2v9EBlmO4XTrOwMRfsY1jeUONxU"
|
||||
)
|
||||
assert provider.openid_configuration_url == (
|
||||
"https://api.descope.com/v1/apps/P2v9EBlmO4XTrOwMRfsY1jeUONxU/.well-known/openid-configuration"
|
||||
)
|
||||
assert provider.oauth_authorization_server_metadata_url == (
|
||||
"https://api.descope.com/v1/apps/P2v9EBlmO4XTrOwMRfsY1jeUONxU/.well-known/oauth-authorization-server"
|
||||
)
|
||||
|
||||
def test_construction_is_network_free(self):
|
||||
"""Construction must not perform discovery I/O, even when it is enabled."""
|
||||
config_url = "https://api.descope.com/v1/apps/P2v9EBlmO4XTrOwMRfsY1jeUONxU/.well-known/openid-configuration"
|
||||
|
||||
no_network = AsyncMock(side_effect=AssertionError("no network during init"))
|
||||
with patch("httpx2.AsyncClient.get", new=no_network):
|
||||
provider = DescopeProvider(
|
||||
config_url=config_url,
|
||||
base_url="https://myserver.com",
|
||||
)
|
||||
|
||||
# Discovery is enabled but deferred; nothing has been fetched yet.
|
||||
assert provider._scopes_discovery_enabled is True
|
||||
assert provider._scopes_supported is None
|
||||
assert provider._discovered_scopes is None
|
||||
assert provider._scopes_discovered is False
|
||||
assert provider.token_verifier.required_scopes == []
|
||||
|
||||
async def test_discover_scopes_supported_lazily(self):
|
||||
"""scopes_supported are discovered lazily and cached after first fetch."""
|
||||
config_url = "https://api.descope.com/v1/apps/P2v9EBlmO4XTrOwMRfsY1jeUONxU/.well-known/openid-configuration"
|
||||
provider = DescopeProvider(
|
||||
config_url=config_url,
|
||||
base_url="https://myserver.com",
|
||||
)
|
||||
|
||||
mock_response = httpx2.Response(
|
||||
200,
|
||||
json=PROJECT_LEVEL_OPENID_CONFIGURATION,
|
||||
request=httpx2.Request("GET", config_url),
|
||||
)
|
||||
|
||||
with patch(
|
||||
"httpx2.AsyncClient.get", new=AsyncMock(return_value=mock_response)
|
||||
) as mock_get:
|
||||
scopes = await provider._get_scopes_supported()
|
||||
# A second call returns the cached result without another fetch.
|
||||
scopes_again = await provider._get_scopes_supported()
|
||||
|
||||
assert scopes == ["mcp:read"]
|
||||
assert scopes_again == ["mcp:read"]
|
||||
assert provider._discovered_scopes == ["mcp:read"]
|
||||
assert provider._scopes_discovered is True
|
||||
mock_get.assert_awaited_once()
|
||||
|
||||
async def test_scope_discovery_retries_after_transient_failure(self):
|
||||
"""A transient discovery failure is not cached and is retried."""
|
||||
config_url = "https://api.descope.com/v1/apps/P2v9EBlmO4XTrOwMRfsY1jeUONxU/.well-known/openid-configuration"
|
||||
provider = DescopeProvider(
|
||||
config_url=config_url,
|
||||
base_url="https://myserver.com",
|
||||
)
|
||||
|
||||
failing = AsyncMock(side_effect=httpx2.ConnectError("boom"))
|
||||
with patch("httpx2.AsyncClient.get", new=failing):
|
||||
first = await provider._get_scopes_supported()
|
||||
|
||||
# Failure yields no scopes and is not frozen for the provider's lifetime.
|
||||
assert first is None
|
||||
assert provider._scopes_discovered is False
|
||||
assert provider._discovered_scopes is None
|
||||
|
||||
mock_response = httpx2.Response(
|
||||
200,
|
||||
json=PROJECT_LEVEL_OPENID_CONFIGURATION,
|
||||
request=httpx2.Request("GET", config_url),
|
||||
)
|
||||
with patch("httpx2.AsyncClient.get", new=AsyncMock(return_value=mock_response)):
|
||||
second = await provider._get_scopes_supported()
|
||||
|
||||
assert second == ["mcp:read"]
|
||||
assert provider._scopes_discovered is True
|
||||
|
||||
async def test_empty_scope_discovery_is_cached_as_success(self):
|
||||
"""An explicitly empty Descope scope list is a successful discovery."""
|
||||
config_url = (
|
||||
"https://api.descope.com/v1/apps/P2abc123/.well-known/openid-configuration"
|
||||
)
|
||||
provider = DescopeProvider(
|
||||
config_url=config_url,
|
||||
base_url="https://myserver.com",
|
||||
)
|
||||
mock_response = httpx2.Response(
|
||||
200,
|
||||
json={"scopes_supported": []},
|
||||
request=httpx2.Request("GET", config_url),
|
||||
)
|
||||
|
||||
with patch(
|
||||
"httpx2.AsyncClient.get", new=AsyncMock(return_value=mock_response)
|
||||
) as mock_get:
|
||||
first = await provider._get_scopes_supported()
|
||||
second = await provider._get_scopes_supported()
|
||||
|
||||
assert first == []
|
||||
assert second == []
|
||||
assert provider._scopes_discovered is True
|
||||
assert provider._discovered_scopes == []
|
||||
mock_get.assert_awaited_once()
|
||||
|
||||
def test_get_routes_is_network_free(self):
|
||||
"""get_routes must not perform discovery I/O when building routes."""
|
||||
config_url = "https://api.descope.com/v1/apps/P2v9EBlmO4XTrOwMRfsY1jeUONxU/.well-known/openid-configuration"
|
||||
provider = DescopeProvider(
|
||||
config_url=config_url,
|
||||
base_url="https://myserver.com",
|
||||
)
|
||||
|
||||
no_network = AsyncMock(side_effect=AssertionError("no network at get_routes"))
|
||||
with patch("httpx2.AsyncClient.get", new=no_network):
|
||||
routes = provider.get_routes("/mcp")
|
||||
|
||||
paths = [route.path for route in routes]
|
||||
assert any("oauth-protected-resource" in path for path in paths)
|
||||
assert any("oauth-authorization-server" in path for path in paths)
|
||||
|
||||
async def test_project_level_metadata_failure_is_not_retried(self):
|
||||
"""Identical project-level primary and fallback URLs are fetched once."""
|
||||
provider = DescopeProvider(
|
||||
config_url="https://api.descope.com/v1/apps/P2abc123/.well-known/openid-configuration",
|
||||
base_url="https://myserver.com",
|
||||
)
|
||||
metadata_route = next(
|
||||
route
|
||||
for route in provider.get_routes("/mcp")
|
||||
if route.path == "/.well-known/oauth-authorization-server"
|
||||
)
|
||||
request = Request(
|
||||
{
|
||||
"type": "http",
|
||||
"method": "GET",
|
||||
"path": metadata_route.path,
|
||||
"headers": [],
|
||||
}
|
||||
)
|
||||
|
||||
failing = AsyncMock(side_effect=httpx2.ConnectError("boom"))
|
||||
with patch("httpx2.AsyncClient.get", new=failing):
|
||||
response = await metadata_route.endpoint(request)
|
||||
|
||||
assert response.status_code == 500
|
||||
failing.assert_awaited_once_with(
|
||||
provider.oauth_authorization_server_metadata_url
|
||||
)
|
||||
|
||||
def test_scopes_supported_and_required_scopes_can_differ(self):
|
||||
"""Test that scopes_supported and required_scopes can be configured independently."""
|
||||
provider = DescopeProvider(
|
||||
config_url="https://api.descope.com/v1/apps/agentic/P2abc123/M123/.well-known/openid-configuration",
|
||||
base_url="https://myserver.com",
|
||||
scopes_supported=["mcp:read", "mcp:write"],
|
||||
required_scopes=["mcp:read"],
|
||||
)
|
||||
|
||||
assert provider._scopes_supported == ["mcp:read", "mcp:write"]
|
||||
assert provider.token_verifier.required_scopes == ["mcp:read"]
|
||||
|
||||
def test_explicit_required_scopes_skip_discovery(self):
|
||||
"""Test that explicit required_scopes disable well-known discovery."""
|
||||
config_url = "https://api.descope.com/v1/apps/P2v9EBlmO4XTrOwMRfsY1jeUONxU/.well-known/openid-configuration"
|
||||
|
||||
provider = DescopeProvider(
|
||||
config_url=config_url,
|
||||
base_url="https://myserver.com",
|
||||
required_scopes=["custom:scope"],
|
||||
)
|
||||
|
||||
assert provider._scopes_discovery_enabled is False
|
||||
assert provider.token_verifier.required_scopes == ["custom:scope"]
|
||||
assert provider._scopes_supported is None
|
||||
|
||||
def test_explicit_scopes_supported_skip_discovery(self):
|
||||
"""Test that explicit scopes_supported disable well-known discovery."""
|
||||
config_url = "https://api.descope.com/v1/apps/P2v9EBlmO4XTrOwMRfsY1jeUONxU/.well-known/openid-configuration"
|
||||
|
||||
provider = DescopeProvider(
|
||||
config_url=config_url,
|
||||
base_url="https://myserver.com",
|
||||
scopes_supported=["custom:advertised"],
|
||||
)
|
||||
|
||||
assert provider._scopes_discovery_enabled is False
|
||||
assert provider._scopes_supported == ["custom:advertised"]
|
||||
assert provider.token_verifier.required_scopes == []
|
||||
|
||||
def test_custom_token_verifier_scopes_skip_discovery(self):
|
||||
"""Scopes supplied by a custom verifier retain the parent behavior."""
|
||||
token_verifier = JWTVerifier(
|
||||
public_key="secret",
|
||||
algorithm="HS256",
|
||||
required_scopes=["custom:scope"],
|
||||
)
|
||||
provider = DescopeProvider(
|
||||
config_url="https://api.descope.com/v1/apps/P2abc123/.well-known/openid-configuration",
|
||||
base_url="https://myserver.com",
|
||||
token_verifier=token_verifier,
|
||||
)
|
||||
|
||||
assert provider._scopes_discovery_enabled is False
|
||||
assert provider._scopes_supported is None
|
||||
assert provider.token_verifier.scopes_supported == ["custom:scope"]
|
||||
|
||||
def test_requires_config_url_or_project_id_and_descope_base_url(self):
|
||||
"""Test that either config_url or both project_id and descope_base_url are required."""
|
||||
# Should raise error when neither API is provided
|
||||
|
|
@ -221,6 +452,52 @@ def client_with_headless_oauth(mcp_server_url: str) -> Client:
|
|||
|
||||
|
||||
class TestDescopeProviderIntegration:
|
||||
async def test_protected_resource_metadata_serves_discovered_scopes(self):
|
||||
"""The protected resource metadata endpoint advertises discovered scopes."""
|
||||
provider = DescopeProvider(
|
||||
config_url="https://api.descope.com/v1/apps/agentic/P2test123/M123/.well-known/openid-configuration",
|
||||
base_url="http://localhost:4321",
|
||||
)
|
||||
mcp = FastMCP(auth=provider)
|
||||
|
||||
with patch(
|
||||
"fastmcp.server.auth.providers.descope._discover_scopes",
|
||||
new=AsyncMock(return_value=["mcp:read"]),
|
||||
):
|
||||
async with run_server_async(mcp, transport="http") as url:
|
||||
metadata_url = url.replace(
|
||||
"/mcp", "/.well-known/oauth-protected-resource/mcp"
|
||||
)
|
||||
async with httpx2.AsyncClient() as client:
|
||||
response = await client.get(metadata_url)
|
||||
|
||||
response.raise_for_status()
|
||||
assert response.json()["scopes_supported"] == ["mcp:read"]
|
||||
assert response.headers["cache-control"] == "public, max-age=3600"
|
||||
|
||||
async def test_failed_scope_discovery_is_not_cached(self):
|
||||
"""Clients can retry metadata discovery immediately after a failure."""
|
||||
provider = DescopeProvider(
|
||||
config_url="https://api.descope.com/v1/apps/agentic/P2test123/M123/.well-known/openid-configuration",
|
||||
base_url="http://localhost:4321",
|
||||
)
|
||||
mcp = FastMCP(auth=provider)
|
||||
|
||||
with patch(
|
||||
"fastmcp.server.auth.providers.descope._discover_scopes",
|
||||
new=AsyncMock(return_value=None),
|
||||
):
|
||||
async with run_server_async(mcp, transport="http") as url:
|
||||
metadata_url = url.replace(
|
||||
"/mcp", "/.well-known/oauth-protected-resource/mcp"
|
||||
)
|
||||
async with httpx2.AsyncClient() as client:
|
||||
response = await client.get(metadata_url)
|
||||
|
||||
response.raise_for_status()
|
||||
assert "scopes_supported" not in response.json()
|
||||
assert response.headers["cache-control"] == "no-store"
|
||||
|
||||
async def test_unauthorized_access(self, mcp_server_url: str):
|
||||
# SDK v2 surfaces the server's 401 as a generic MCPError at the client
|
||||
# boundary rather than re-raising httpx2.HTTPStatusError.
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue