diff --git a/docs/integrations/descope.mdx b/docs/integrations/descope.mdx index bfb6cd9c8..33786b6e2 100644 --- a/docs/integrations/descope.mdx +++ b/docs/integrations/descope.mdx @@ -23,11 +23,10 @@ Before you begin, you will need: ### Step 1: Configure Descope - - 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. + + 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)**. @@ -35,10 +34,17 @@ Before you begin, you will need: - - Save your Well-Known URL from [MCP Server Settings](https://app.descope.com/mcp-servers): + + `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 ``` @@ -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: diff --git a/fastmcp_slim/fastmcp/server/auth/providers/descope.py b/fastmcp_slim/fastmcp/server/auth/providers/descope.py index 3741aeefe..aa1255ad1 100644 --- a/fastmcp_slim/fastmcp/server/auth/providers/descope.py +++ b/fastmcp_slim/fastmcp/server/auth/providers/descope.py @@ -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", diff --git a/tests/server/auth/providers/test_descope.py b/tests/server/auth/providers/test_descope.py index cbf4607f2..b35b2dd00 100644 --- a/tests/server/auth/providers/test_descope.py +++ b/tests/server/auth/providers/test_descope.py @@ -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.