Merge branch 'main' into claude/issue-3049-20260131-2232

This commit is contained in:
Bill Easton 2026-02-02 19:08:11 -06:00 committed by GitHub
commit 32017f7fb1
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
7 changed files with 396 additions and 3 deletions

View file

@ -282,3 +282,47 @@ Parameters (`jwt_signing_key` and `client_storage`) work together to ensure toke
For complete details on these parameters, see the [OAuth Proxy documentation](/servers/auth/oauth-proxy#configuration-parameters).
</Note>
## Token Verification Only (Managed Identity)
<VersionBadge version="2.15.0" />
For deployments where your server only needs to **validate incoming tokens** — such as Azure Container Apps with Managed Identity — use `AzureJWTVerifier` with `RemoteAuthProvider` instead of the full `AzureProvider`.
This pattern is ideal when:
- Your infrastructure handles authentication (e.g., Managed Identity)
- You don't need the OAuth proxy flow (no `client_secret` required)
- You just need to verify that incoming Azure AD tokens are valid
```python server.py
from fastmcp import FastMCP
from fastmcp.server.auth import RemoteAuthProvider
from fastmcp.server.auth.providers.azure import AzureJWTVerifier
from pydantic import AnyHttpUrl
tenant_id = "your-tenant-id"
client_id = "your-client-id"
# AzureJWTVerifier auto-configures JWKS, issuer, and audience
verifier = AzureJWTVerifier(
client_id=client_id,
tenant_id=tenant_id,
required_scopes=["access_as_user"], # Scope names from Azure Portal
)
auth = RemoteAuthProvider(
token_verifier=verifier,
authorization_servers=[
AnyHttpUrl(f"https://login.microsoftonline.com/{tenant_id}/v2.0")
],
base_url="https://your-container-app.azurecontainerapps.io",
)
mcp = FastMCP(name="Azure MI App", auth=auth)
```
`AzureJWTVerifier` handles Azure's scope format automatically. You write scope names exactly as they appear in Azure Portal under **Expose an API** (e.g., `access_as_user`). The verifier validates tokens using the short-form scopes that Azure puts in the `scp` claim, while advertising the full URI scopes (e.g., `api://your-client-id/access_as_user`) in OAuth metadata so MCP clients know what to request.
<Note>
For Azure Government, pass `base_authority="login.microsoftonline.us"` to `AzureJWTVerifier`.
</Note>

View file

@ -127,6 +127,21 @@ This configuration creates a server that accepts tokens issued by `auth.yourcomp
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.
### Overriding Advertised Scopes
Some identity providers use different scope formats for authorization requests versus token claims. For example, Azure AD requires clients to request full URI scopes like `api://client-id/read`, but the token's `scp` claim contains just `read`. The `scopes_supported` parameter lets you advertise the full-form scopes in metadata while validating against the short form:
```python
auth = RemoteAuthProvider(
token_verifier=token_verifier,
authorization_servers=[AnyHttpUrl("https://auth.example.com")],
base_url="https://api.example.com",
scopes_supported=["api://my-api/read", "api://my-api/write"],
)
```
When not set, `scopes_supported` defaults to the token verifier's `required_scopes`. For Azure AD specifically, see the [AzureJWTVerifier](/integrations/azure#token-verification-only-managed-identity) which handles this automatically.
### Custom Endpoints
You can extend `RemoteAuthProvider` to add additional endpoints beyond the standard OAuth protected resource metadata. These don't have to be OAuth-specific - you can add any endpoints your authentication integration requires.

View file

@ -274,6 +274,17 @@ class TokenVerifier(AuthProvider):
"""
super().__init__(base_url=base_url, required_scopes=required_scopes)
@property
def scopes_supported(self) -> list[str]:
"""Scopes to advertise in OAuth metadata.
Defaults to required_scopes. Override in subclasses when the
advertised scopes differ from the validation scopes (e.g., Azure AD
where tokens contain short-form scopes but clients request full URI
scopes).
"""
return self.required_scopes or []
async def verify_token(self, token: str) -> AccessToken | None:
"""Verify a bearer token and return access info if valid."""
raise NotImplementedError("Subclasses must implement verify_token")
@ -299,6 +310,7 @@ class RemoteAuthProvider(AuthProvider):
token_verifier: TokenVerifier,
authorization_servers: list[AnyHttpUrl],
base_url: AnyHttpUrl | str,
scopes_supported: list[str] | None = None,
resource_name: str | None = None,
resource_documentation: AnyHttpUrl | None = None,
):
@ -308,6 +320,10 @@ class RemoteAuthProvider(AuthProvider):
token_verifier: TokenVerifier instance for token validation
authorization_servers: List of authorization servers that issue valid tokens
base_url: The base URL of this server
scopes_supported: Scopes to advertise in OAuth metadata. If None,
uses the token verifier's scopes_supported property. Use this
when the scopes clients request differ from the scopes that
appear in tokens (e.g., Azure AD full URI scopes vs short-form).
resource_name: Optional name for the protected resource
resource_documentation: Optional documentation URL for the protected resource
"""
@ -317,6 +333,7 @@ class RemoteAuthProvider(AuthProvider):
)
self.token_verifier = token_verifier
self.authorization_servers = authorization_servers
self._scopes_supported = scopes_supported
self.resource_name = resource_name
self.resource_documentation = resource_documentation
@ -343,7 +360,11 @@ class RemoteAuthProvider(AuthProvider):
create_protected_resource_routes(
resource_url=resource_url,
authorization_servers=self.authorization_servers,
scopes_supported=self.token_verifier.required_scopes,
scopes_supported=(
self._scopes_supported
if self._scopes_supported is not None
else self.token_verifier.scopes_supported
),
resource_name=self.resource_name,
resource_documentation=self.resource_documentation,
)

View file

@ -452,3 +452,103 @@ class AzureProvider(OAuthProxy):
except Exception as e:
logger.debug("Failed to extract Azure claims: %s", e)
return None
class AzureJWTVerifier(JWTVerifier):
"""JWT verifier pre-configured for Azure AD / Microsoft Entra ID.
Auto-configures JWKS URI, issuer, audience, and scope handling from your
Azure app registration details. Designed for Managed Identity and other
token-verification-only scenarios where AzureProvider's full OAuth proxy
isn't needed.
Handles Azure's scope format automatically:
- Validates tokens using short-form scopes (what Azure puts in ``scp`` claims)
- Advertises full-URI scopes in OAuth metadata (what clients need to request)
Example::
from fastmcp.server.auth import RemoteAuthProvider
from fastmcp.server.auth.providers.azure import AzureJWTVerifier
from pydantic import AnyHttpUrl
verifier = AzureJWTVerifier(
client_id="your-client-id",
tenant_id="your-tenant-id",
required_scopes=["access_as_user"],
)
auth = RemoteAuthProvider(
token_verifier=verifier,
authorization_servers=[
AnyHttpUrl("https://login.microsoftonline.com/your-tenant-id/v2.0")
],
base_url="https://my-server.com",
)
"""
def __init__(
self,
*,
client_id: str,
tenant_id: str,
required_scopes: list[str] | None = None,
identifier_uri: str | None = None,
base_authority: str = "login.microsoftonline.com",
):
"""Initialize Azure JWT verifier.
Args:
client_id: Azure application (client) ID from your App registration
tenant_id: Azure tenant ID (specific tenant GUID, "organizations", or "consumers").
For multi-tenant apps ("organizations" or "consumers"), issuer validation
is skipped since Azure tokens carry the actual tenant GUID as issuer.
required_scopes: Scope names as they appear in Azure Portal under "Expose an API"
(e.g., ["access_as_user", "read"]). These are validated against
the short-form scopes in token ``scp`` claims, and automatically
prefixed with identifier_uri for OAuth metadata.
identifier_uri: Application ID URI (defaults to ``api://{client_id}``).
Used to prefix scopes in OAuth metadata so clients know the full
scope URIs to request from Azure.
base_authority: Azure authority base URL (defaults to "login.microsoftonline.com").
For Azure Government, use "login.microsoftonline.us".
"""
self._identifier_uri = identifier_uri or f"api://{client_id}"
# For multi-tenant apps, Azure tokens carry the actual tenant GUID as
# issuer, not the literal "organizations" or "consumers" string. Skip
# issuer validation for these — audience still protects against wrong-app tokens.
multi_tenant_values = {"organizations", "consumers", "common"}
issuer: str | None = (
None
if tenant_id in multi_tenant_values
else f"https://{base_authority}/{tenant_id}/v2.0"
)
super().__init__(
jwks_uri=f"https://{base_authority}/{tenant_id}/discovery/v2.0/keys",
issuer=issuer,
audience=client_id,
algorithm="RS256",
required_scopes=required_scopes,
)
@property
def scopes_supported(self) -> list[str]:
"""Return scopes with Azure URI prefix for OAuth metadata.
Azure tokens contain short-form scopes (e.g., ``read``) in the ``scp``
claim, but clients must request full URI scopes (e.g.,
``api://client-id/read``) from the Azure authorization endpoint. This
property returns the full-URI form for OAuth metadata while
``required_scopes`` retains the short form for token validation.
"""
if not self.required_scopes:
return []
prefixed = []
for scope in self.required_scopes:
if scope in OIDC_SCOPES or "://" in scope or "/" in scope:
prefixed.append(scope)
else:
prefixed.append(f"{self._identifier_uri}/{scope}")
return prefixed

View file

@ -6,8 +6,12 @@ from mcp.server.auth.provider import AuthorizationParams
from mcp.shared.auth import OAuthClientInformationFull
from pydantic import AnyUrl
from fastmcp.server.auth.providers.azure import OIDC_SCOPES, AzureProvider
from fastmcp.server.auth.providers.jwt import JWTVerifier
from fastmcp.server.auth.providers.azure import (
OIDC_SCOPES,
AzureJWTVerifier,
AzureProvider,
)
from fastmcp.server.auth.providers.jwt import JWTVerifier, RSAKeyPair
class TestAzureProvider:
@ -983,3 +987,138 @@ class TestAzureTokenExchangeScopes:
["read", "write"]
)
assert len(refresh_scopes) > 0
class TestAzureJWTVerifier:
"""Tests for AzureJWTVerifier pre-configured JWT verifier."""
def test_auto_configures_from_client_and_tenant(self):
verifier = AzureJWTVerifier(
client_id="my-client-id",
tenant_id="my-tenant-id",
required_scopes=["access_as_user"],
)
assert (
verifier.jwks_uri
== "https://login.microsoftonline.com/my-tenant-id/discovery/v2.0/keys"
)
assert verifier.issuer == "https://login.microsoftonline.com/my-tenant-id/v2.0"
assert verifier.audience == "my-client-id"
assert verifier.algorithm == "RS256"
assert verifier.required_scopes == ["access_as_user"]
async def test_validates_short_form_scopes(self):
key_pair = RSAKeyPair.generate()
verifier = AzureJWTVerifier(
client_id="my-client-id",
tenant_id="my-tenant-id",
required_scopes=["access_as_user"],
)
# Override to use our test key instead of JWKS
verifier.public_key = key_pair.public_key
verifier.jwks_uri = None
token = key_pair.create_token(
subject="test-user",
issuer="https://login.microsoftonline.com/my-tenant-id/v2.0",
audience="my-client-id",
additional_claims={"scp": "access_as_user"},
)
result = await verifier.load_access_token(token)
assert result is not None
assert "access_as_user" in result.scopes
def test_scopes_supported_returns_prefixed_form(self):
verifier = AzureJWTVerifier(
client_id="my-client-id",
tenant_id="my-tenant-id",
required_scopes=["read", "write"],
)
assert verifier.scopes_supported == [
"api://my-client-id/read",
"api://my-client-id/write",
]
def test_already_prefixed_scopes_pass_through(self):
verifier = AzureJWTVerifier(
client_id="my-client-id",
tenant_id="my-tenant-id",
required_scopes=["api://my-client-id/read"],
)
assert verifier.scopes_supported == ["api://my-client-id/read"]
def test_oidc_scopes_not_prefixed(self):
verifier = AzureJWTVerifier(
client_id="my-client-id",
tenant_id="my-tenant-id",
required_scopes=["openid", "read"],
)
assert verifier.scopes_supported == ["openid", "api://my-client-id/read"]
def test_custom_identifier_uri(self):
verifier = AzureJWTVerifier(
client_id="my-client-id",
tenant_id="my-tenant-id",
required_scopes=["read"],
identifier_uri="api://custom-uri",
)
assert verifier.scopes_supported == ["api://custom-uri/read"]
def test_custom_base_authority_for_gov_cloud(self):
verifier = AzureJWTVerifier(
client_id="my-client-id",
tenant_id="my-tenant-id",
required_scopes=["read"],
base_authority="login.microsoftonline.us",
)
assert (
verifier.jwks_uri
== "https://login.microsoftonline.us/my-tenant-id/discovery/v2.0/keys"
)
assert verifier.issuer == "https://login.microsoftonline.us/my-tenant-id/v2.0"
def test_scopes_supported_empty_when_no_required_scopes(self):
verifier = AzureJWTVerifier(
client_id="my-client-id",
tenant_id="my-tenant-id",
)
assert verifier.scopes_supported == []
def test_default_identifier_uri_uses_client_id(self):
verifier = AzureJWTVerifier(
client_id="abc-123",
tenant_id="my-tenant-id",
required_scopes=["read"],
)
assert verifier.scopes_supported == ["api://abc-123/read"]
def test_multi_tenant_organizations_skips_issuer(self):
verifier = AzureJWTVerifier(
client_id="my-client-id",
tenant_id="organizations",
)
assert verifier.issuer is None
def test_multi_tenant_consumers_skips_issuer(self):
verifier = AzureJWTVerifier(
client_id="my-client-id",
tenant_id="consumers",
)
assert verifier.issuer is None
def test_multi_tenant_common_skips_issuer(self):
verifier = AzureJWTVerifier(
client_id="my-client-id",
tenant_id="common",
)
assert verifier.issuer is None
def test_specific_tenant_sets_issuer(self):
verifier = AzureJWTVerifier(
client_id="my-client-id",
tenant_id="12345678-1234-1234-1234-123456789012",
)
assert (
verifier.issuer
== "https://login.microsoftonline.com/12345678-1234-1234-1234-123456789012/v2.0"
)

View file

@ -1099,3 +1099,20 @@ class TestJWTVerifierImport:
except ImportError as e:
# If PyJWT not available, should get helpful error
assert "PyJWT is required" in str(e)
class TestScopesSupported:
"""Tests for the scopes_supported property on TokenVerifier."""
def test_defaults_to_required_scopes(self, rsa_key_pair: RSAKeyPair):
provider = JWTVerifier(
public_key=rsa_key_pair.public_key,
required_scopes=["read", "write"],
)
assert provider.scopes_supported == ["read", "write"]
def test_empty_when_no_required_scopes(self, rsa_key_pair: RSAKeyPair):
provider = JWTVerifier(
public_key=rsa_key_pair.public_key,
)
assert provider.scopes_supported == []

View file

@ -483,3 +483,60 @@ class TestRemoteAuthProviderIntegration:
data["resource_documentation"]
== "https://doc.my-server.com/resource-docs"
)
async def test_scopes_supported_overrides_metadata(self):
"""Test that scopes_supported parameter overrides what's in metadata."""
token_verifier = StaticTokenVerifier(
tokens={
"test": {"client_id": "c", "scopes": ["read"]},
},
required_scopes=["read"],
)
provider = RemoteAuthProvider(
token_verifier=token_verifier,
authorization_servers=[AnyHttpUrl("https://auth.example.com")],
base_url="https://my-server.com",
scopes_supported=["api://my-api/read"],
)
mcp = FastMCP("test-server", auth=provider)
mcp_http_app = mcp.http_app()
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/mcp")
assert response.status_code == 200
data = response.json()
assert data["scopes_supported"] == ["api://my-api/read"]
async def test_scopes_supported_defaults_to_verifier(self):
"""Test that metadata uses verifier scopes_supported when parameter not set."""
token_verifier = StaticTokenVerifier(
tokens={
"test": {"client_id": "c", "scopes": ["read"]},
},
required_scopes=["read"],
)
provider = RemoteAuthProvider(
token_verifier=token_verifier,
authorization_servers=[AnyHttpUrl("https://auth.example.com")],
base_url="https://my-server.com",
)
mcp = FastMCP("test-server", auth=provider)
mcp_http_app = mcp.http_app()
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/mcp")
assert response.status_code == 200
data = response.json()
assert data["scopes_supported"] == ["read"]