From eebdc8c031a96a033d5dff263d0ffed363ce375f Mon Sep 17 00:00:00 2001
From: Carlos Rian <65134623+carlos-rian@users.noreply.github.com>
Date: Wed, 22 Apr 2026 10:24:09 -0300
Subject: [PATCH] feat: add AzureB2CProvider for Azure AD B2C user flows
(#3995)
---
docs/integrations/azure.mdx | 82 +++++
src/fastmcp/server/auth/providers/azure.py | 91 +++++-
tests/server/auth/providers/test_azure.py | 361 +++++++++++++++++++++
3 files changed, 533 insertions(+), 1 deletion(-)
diff --git a/docs/integrations/azure.mdx b/docs/integrations/azure.mdx
index 4376a38ce..cba92349e 100644
--- a/docs/integrations/azure.mdx
+++ b/docs/integrations/azure.mdx
@@ -458,3 +458,85 @@ For advanced OBO scenarios, use `CurrentAccessToken()` to get the user's token,
For a complete working example of Azure OBO with FastMCP, see [Pamela Fox's blog post on OBO flow for Entra-based MCP servers](https://blog.pamelafox.org/2026/01/using-on-behalf-of-flow-for-entra-based.html).
+
+## Azure AD B2C
+
+
+
+Azure AD B2C (Business-to-Consumer) uses different endpoints, scope URIs, and
+token issuers than standard Microsoft Entra ID. The `AzureProvider.from_b2c()`
+factory handles all of these differences automatically.
+
+
+Azure AD B2C does **not** support the On-Behalf-Of (OBO) flow. If you need
+OBO for downstream API calls, use `AzureProvider` with standard Entra ID
+instead.
+
+
+### Quick Start
+
+```python server.py
+from fastmcp import FastMCP
+from fastmcp.server.auth.providers.azure import AzureProvider
+
+auth = AzureProvider.from_b2c(
+ tenant_name="mytenant",
+ policy_name="B2C_1_susi",
+ client_id="00000000-0000-0000-0000-000000000000",
+ client_secret="my-secret",
+ required_scopes=["mcp-access"],
+ base_url="https://myserver.com",
+)
+
+mcp = FastMCP("My App", auth=auth)
+```
+
+`from_b2c()` derives the following values automatically:
+
+| Derived value | Formula |
+|---|---|
+| Authority host | `{tenant_name}.b2clogin.com` |
+| Authorization endpoint | `https://{tenant_name}.b2clogin.com/{tenant_name}.onmicrosoft.com/{policy_name}/oauth2/v2.0/authorize` |
+| Token endpoint | `https://{tenant_name}.b2clogin.com/{tenant_name}.onmicrosoft.com/{policy_name}/oauth2/v2.0/token` |
+| Scope identifier URI | `https://{tenant_name}.onmicrosoft.com/{client_id}` |
+
+### Token Issuer Validation
+
+B2C access tokens carry the **tenant GUID** (not the `.onmicrosoft.com` name)
+in the `iss` claim, and the exact format varies by policy and custom-domain
+configuration. `from_b2c()` therefore **disables issuer validation by
+default**; **audience validation still enforces that tokens target the correct
+application**.
+
+Once you have confirmed a successful end-to-end login, read the actual `iss`
+value from the decoded claims and enable strict validation:
+
+```python
+auth = AzureProvider.from_b2c(
+ tenant_name="mytenant",
+ policy_name="B2C_1_susi",
+ client_id="00000000-0000-0000-0000-000000000000",
+ client_secret="my-secret",
+ required_scopes=["mcp-access"],
+ base_url="https://myserver.com",
+ token_issuer="https://mytenant.b2clogin.com/11111111-2222-3333-4444-555555555555/v2.0/",
+)
+```
+
+### Custom Domains
+
+If your B2C tenant uses a [custom domain](https://learn.microsoft.com/en-us/azure/active-directory-b2c/custom-domain)
+(e.g. `auth.mycompany.com` instead of `mytenant.b2clogin.com`), pass it via
+`custom_domain`:
+
+```python
+auth = AzureProvider.from_b2c(
+ tenant_name="mytenant",
+ policy_name="B2C_1_susi",
+ client_id="00000000-0000-0000-0000-000000000000",
+ client_secret="my-secret",
+ required_scopes=["mcp-access"],
+ base_url="https://myserver.com",
+ custom_domain="auth.mycompany.com",
+)
+```
diff --git a/src/fastmcp/server/auth/providers/azure.py b/src/fastmcp/server/auth/providers/azure.py
index 716fff04b..11d75db98 100644
--- a/src/fastmcp/server/auth/providers/azure.py
+++ b/src/fastmcp/server/auth/providers/azure.py
@@ -117,6 +117,7 @@ class AzureProvider(OAuthProxy):
forward_resource: bool = True,
fallback_refresh_token_expiry_seconds: int | None = None,
base_authority: str = "login.microsoftonline.com",
+ token_issuer: str | None = None,
http_client: httpx.AsyncClient | None = None,
enable_cimd: bool = True,
) -> None:
@@ -141,6 +142,9 @@ class AzureProvider(OAuthProxy):
redirect_path: Redirect path configured in Azure App registration (defaults to "/auth/callback")
base_authority: Azure authority base URL (defaults to "login.microsoftonline.com").
For Azure Government, use "login.microsoftonline.us".
+ token_issuer: Override the expected `iss` claim value for JWT validation.
+ Defaults to the standard Entra ID issuer derived from `base_authority`
+ and `tenant_id`. Pass an explicit string to enforce a specific issuer.
required_scopes: Custom API scope names WITHOUT prefix (e.g., ["read", "write"]).
- Automatically prefixed with identifier_uri during initialization
- Validated on all tokens
@@ -197,13 +201,14 @@ class AzureProvider(OAuthProxy):
# to avoid redundant OBO exchanges for the same user + scopes.
self._obo_credentials: OrderedDict[str, OnBehalfOfCredential] = OrderedDict()
self._obo_max_credentials: int = 128
+ self._obo_supported = True
# Apply defaults
self.identifier_uri = identifier_uri or f"api://{client_id}"
self.additional_authorize_scopes: list[str] = parsed_additional_scopes
# Always validate tokens against the app's API client ID using JWT
- issuer = f"https://{base_authority}/{tenant_id}/v2.0"
+ issuer = token_issuer or f"https://{base_authority}/{tenant_id}/v2.0"
jwks_uri = f"https://{base_authority}/{tenant_id}/discovery/v2.0/keys"
# Azure access tokens only include custom API scopes in the `scp` claim,
@@ -272,6 +277,84 @@ class AzureProvider(OAuthProxy):
authority_info,
)
+ @classmethod
+ def from_b2c(
+ cls,
+ *,
+ tenant_name: str,
+ policy_name: str,
+ client_id: str,
+ client_secret: str | None = None,
+ required_scopes: list[str],
+ base_url: str,
+ custom_domain: str | None = None,
+ identifier_uri: str | None = None,
+ token_issuer: str | None = None,
+ **kwargs: Any,
+ ) -> AzureProvider:
+ """Create an AzureProvider pre-configured for Azure AD B2C.
+
+ Derives authority host, tenant path, and identifier URI from
+ `tenant_name` and `policy_name`, then delegates to the standard
+ constructor. Returns a plain `AzureProvider` instance.
+
+ B2C issuer validation is disabled by default (`token_issuer=None`)
+ because B2C issuers embed the tenant GUID. Pass an explicit
+ `token_issuer` string once you know the real `iss` value.
+
+ Azure AD B2C does **not** support OBO.
+
+ Args:
+ tenant_name: Short B2C tenant name without `.onmicrosoft.com`
+ (e.g. `"mytenant"`).
+ policy_name: User-flow or custom-policy name
+ (e.g. `"B2C_1_susi"`).
+ client_id: Application (client) ID from the B2C app registration.
+ client_secret: Client secret from the B2C app registration.
+ required_scopes: Custom API scope names without prefix
+ (e.g. `["mcp-access"]`).
+ base_url: Public base URL of this server.
+ custom_domain: Custom domain for the B2C authority
+ (e.g. `"auth.mycompany.com"`). Defaults to
+ `{tenant_name}.b2clogin.com`.
+ identifier_uri: Application ID URI. Defaults to
+ `https://{tenant_name}.onmicrosoft.com/{client_id}`.
+ token_issuer: Expected `iss` claim. `None` (default) disables
+ issuer validation.
+ **kwargs: Forwarded to `AzureProvider.__init__`.
+ """
+ if ".onmicrosoft.com" in tenant_name:
+ raise ValueError(
+ f"tenant_name should be the short name without the "
+ f".onmicrosoft.com suffix (e.g. 'mytenant'), got {tenant_name!r}"
+ )
+
+ if custom_domain is not None:
+ custom_domain = (
+ custom_domain.removeprefix("https://")
+ .removeprefix("http://")
+ .rstrip("/")
+ )
+
+ authority = custom_domain or f"{tenant_name}.b2clogin.com"
+ tenant_path = f"{tenant_name}.onmicrosoft.com/{policy_name}"
+ uri = identifier_uri or f"https://{tenant_name}.onmicrosoft.com/{client_id}"
+
+ provider = cls(
+ client_id=client_id,
+ client_secret=client_secret,
+ tenant_id=tenant_path,
+ required_scopes=required_scopes,
+ base_url=base_url,
+ base_authority=authority,
+ identifier_uri=uri,
+ token_issuer=token_issuer,
+ **kwargs,
+ )
+ provider._token_validator.issuer = token_issuer # type: ignore[union-attr]
+ provider._obo_supported = False
+ return provider
+
async def authorize(
self,
client: OAuthClientInformationFull,
@@ -512,8 +595,14 @@ class AzureProvider(OAuthProxy):
A configured OnBehalfOfCredential ready for get_token() calls.
Raises:
+ NotImplementedError: If OBO is not supported (e.g. Azure AD B2C).
ImportError: If azure-identity is not installed (requires fastmcp[azure]).
"""
+ if not self._obo_supported:
+ raise NotImplementedError(
+ "Azure AD B2C does not support the On-Behalf-Of (OBO) flow. "
+ "Use AzureProvider with standard Entra ID for OBO scenarios."
+ )
_require_azure_identity("OBO token exchange")
from azure.identity.aio import OnBehalfOfCredential
diff --git a/tests/server/auth/providers/test_azure.py b/tests/server/auth/providers/test_azure.py
index 54b119ce9..880e72558 100644
--- a/tests/server/auth/providers/test_azure.py
+++ b/tests/server/auth/providers/test_azure.py
@@ -822,3 +822,364 @@ class TestAzureProvider:
# Should have 3 items (read deduplicated, plus offline_access)
assert len(result) == 3
assert result.count("api://my-api/read") == 1
+
+
+class TestAzureProviderTokenIssuer:
+ """Tests for the token_issuer parameter on AzureProvider."""
+
+ def test_default_issuer_is_derived(self, memory_storage: MemoryStore):
+ """Without token_issuer, the issuer is derived from base_authority/tenant_id."""
+ provider = AzureProvider(
+ client_id="test_client",
+ client_secret="test_secret",
+ tenant_id="my-tenant",
+ base_url="https://myserver.com",
+ required_scopes=["read"],
+ jwt_signing_key="test-secret",
+ client_storage=memory_storage,
+ )
+
+ assert isinstance(provider._token_validator, JWTVerifier)
+ assert (
+ provider._token_validator.issuer
+ == "https://login.microsoftonline.com/my-tenant/v2.0"
+ )
+
+ def test_explicit_token_issuer_is_used(self, memory_storage: MemoryStore):
+ """An explicit token_issuer string is passed to the verifier."""
+ custom_issuer = "https://custom.issuer.com/v2.0"
+ provider = AzureProvider(
+ client_id="test_client",
+ client_secret="test_secret",
+ tenant_id="my-tenant",
+ base_url="https://myserver.com",
+ required_scopes=["read"],
+ token_issuer=custom_issuer,
+ jwt_signing_key="test-secret",
+ client_storage=memory_storage,
+ )
+
+ assert isinstance(provider._token_validator, JWTVerifier)
+ assert provider._token_validator.issuer == custom_issuer
+
+ async def test_explicit_issuer_enforced(self, memory_storage: MemoryStore):
+ """With an explicit token_issuer, wrong issuers are rejected."""
+ key_pair = RSAKeyPair.generate()
+ expected = "https://expected.issuer.com/v2.0"
+ provider = AzureProvider(
+ client_id="test_client",
+ client_secret="test_secret",
+ tenant_id="my-tenant",
+ base_url="https://myserver.com",
+ required_scopes=["read"],
+ token_issuer=expected,
+ jwt_signing_key="test-secret",
+ client_storage=memory_storage,
+ )
+
+ assert isinstance(provider._token_validator, JWTVerifier)
+ verifier = provider._token_validator
+ verifier.public_key = key_pair.public_key
+ verifier.jwks_uri = None
+
+ good_token = key_pair.create_token(
+ subject="test-user",
+ issuer=expected,
+ audience="test_client",
+ additional_claims={"scp": "read"},
+ )
+ assert await verifier.load_access_token(good_token) is not None
+
+ bad_token = key_pair.create_token(
+ subject="test-user",
+ issuer="https://wrong.issuer.com/v2.0",
+ audience="test_client",
+ additional_claims={"scp": "read"},
+ )
+ assert await verifier.load_access_token(bad_token) is None
+
+
+class TestAzureProviderFromB2C:
+ """Tests for the AzureProvider.from_b2c() classmethod factory."""
+
+ def test_b2c_endpoints_derived_correctly(self, memory_storage: MemoryStore):
+ """from_b2c() produces correct B2C authority and tenant path."""
+ provider = AzureProvider.from_b2c(
+ tenant_name="mytenant",
+ policy_name="B2C_1_susi",
+ client_id="client-id",
+ client_secret="secret",
+ required_scopes=["mcp-access"],
+ base_url="https://myserver.com",
+ jwt_signing_key="test-secret",
+ client_storage=memory_storage,
+ )
+
+ assert provider._upstream_authorization_endpoint == (
+ "https://mytenant.b2clogin.com"
+ "/mytenant.onmicrosoft.com/B2C_1_susi"
+ "/oauth2/v2.0/authorize"
+ )
+ assert provider._upstream_token_endpoint == (
+ "https://mytenant.b2clogin.com"
+ "/mytenant.onmicrosoft.com/B2C_1_susi"
+ "/oauth2/v2.0/token"
+ )
+
+ def test_b2c_identifier_uri_uses_https(self, memory_storage: MemoryStore):
+ """from_b2c() sets identifier_uri with https:// scheme, not api://."""
+ provider = AzureProvider.from_b2c(
+ tenant_name="mytenant",
+ policy_name="B2C_1_susi",
+ client_id="00000000-0000-0000-0000-000000000001",
+ client_secret="secret",
+ required_scopes=["mcp-access"],
+ base_url="https://myserver.com",
+ jwt_signing_key="test-secret",
+ client_storage=memory_storage,
+ )
+
+ assert provider.identifier_uri == (
+ "https://mytenant.onmicrosoft.com/00000000-0000-0000-0000-000000000001"
+ )
+ assert provider.identifier_uri.startswith("https://")
+ assert not provider.identifier_uri.startswith("api://")
+
+ def test_b2c_issuer_disabled_by_default(self, memory_storage: MemoryStore):
+ """from_b2c() disables issuer validation by default (token_issuer=None)."""
+ provider = AzureProvider.from_b2c(
+ tenant_name="mytenant",
+ policy_name="B2C_1_susi",
+ client_id="client-id",
+ client_secret="secret",
+ required_scopes=["mcp-access"],
+ base_url="https://myserver.com",
+ jwt_signing_key="test-secret",
+ client_storage=memory_storage,
+ )
+
+ assert isinstance(provider._token_validator, JWTVerifier)
+ assert provider._token_validator.issuer is None
+
+ def test_b2c_explicit_token_issuer(self, memory_storage: MemoryStore):
+ """from_b2c() forwards an explicit token_issuer to the verifier."""
+ explicit_issuer = (
+ "https://mytenant.b2clogin.com/11111111-2222-3333-4444-555555555555/v2.0/"
+ )
+ provider = AzureProvider.from_b2c(
+ tenant_name="mytenant",
+ policy_name="B2C_1_susi",
+ client_id="client-id",
+ client_secret="secret",
+ required_scopes=["mcp-access"],
+ base_url="https://myserver.com",
+ token_issuer=explicit_issuer,
+ jwt_signing_key="test-secret",
+ client_storage=memory_storage,
+ )
+
+ assert isinstance(provider._token_validator, JWTVerifier)
+ assert provider._token_validator.issuer == explicit_issuer
+
+ def test_b2c_custom_domain(self, memory_storage: MemoryStore):
+ """from_b2c() uses custom_domain in place of {tenant}.b2clogin.com."""
+ provider = AzureProvider.from_b2c(
+ tenant_name="mytenant",
+ policy_name="B2C_1_susi",
+ client_id="client-id",
+ client_secret="secret",
+ required_scopes=["mcp-access"],
+ base_url="https://myserver.com",
+ custom_domain="auth.mycompany.com",
+ jwt_signing_key="test-secret",
+ client_storage=memory_storage,
+ )
+
+ assert "auth.mycompany.com" in provider._upstream_authorization_endpoint
+ assert "auth.mycompany.com" in provider._upstream_token_endpoint
+ assert "mytenant.b2clogin.com" not in provider._upstream_authorization_endpoint
+
+ def test_b2c_custom_domain_with_scheme_normalized(
+ self, memory_storage: MemoryStore
+ ):
+ """from_b2c() strips scheme and trailing slash from custom_domain."""
+ provider = AzureProvider.from_b2c(
+ tenant_name="mytenant",
+ policy_name="B2C_1_susi",
+ client_id="client-id",
+ client_secret="secret",
+ required_scopes=["mcp-access"],
+ base_url="https://myserver.com",
+ custom_domain="https://auth.mycompany.com/",
+ jwt_signing_key="test-secret",
+ client_storage=memory_storage,
+ )
+
+ assert "auth.mycompany.com" in provider._upstream_authorization_endpoint
+ assert "https://https://" not in provider._upstream_authorization_endpoint
+
+ def test_b2c_custom_identifier_uri(self, memory_storage: MemoryStore):
+ """from_b2c() respects an explicit identifier_uri override."""
+ custom_uri = "https://mycompany.com/api/mcp"
+ provider = AzureProvider.from_b2c(
+ tenant_name="mytenant",
+ policy_name="B2C_1_susi",
+ client_id="client-id",
+ client_secret="secret",
+ required_scopes=["mcp-access"],
+ base_url="https://myserver.com",
+ identifier_uri=custom_uri,
+ jwt_signing_key="test-secret",
+ client_storage=memory_storage,
+ )
+
+ assert provider.identifier_uri == custom_uri
+
+ def test_b2c_scope_prefix_uses_https(self, memory_storage: MemoryStore):
+ """from_b2c() scopes are prefixed with the https:// identifier URI."""
+ provider = AzureProvider.from_b2c(
+ tenant_name="mytenant",
+ policy_name="B2C_1_susi",
+ client_id="aabbccdd",
+ client_secret="secret",
+ required_scopes=["mcp-access"],
+ base_url="https://myserver.com",
+ jwt_signing_key="test-secret",
+ client_storage=memory_storage,
+ )
+
+ result = provider._prefix_scopes_for_azure(["mcp-access"])
+ assert result == ["https://mytenant.onmicrosoft.com/aabbccdd/mcp-access"]
+
+ def test_b2c_returns_azure_provider_instance(self, memory_storage: MemoryStore):
+ """from_b2c() returns an AzureProvider, not a subclass."""
+ provider = AzureProvider.from_b2c(
+ tenant_name="mytenant",
+ policy_name="B2C_1_susi",
+ client_id="client-id",
+ client_secret="secret",
+ required_scopes=["mcp-access"],
+ base_url="https://myserver.com",
+ jwt_signing_key="test-secret",
+ client_storage=memory_storage,
+ )
+
+ assert type(provider) is AzureProvider
+
+ def test_b2c_custom_policy_name(self, memory_storage: MemoryStore):
+ """from_b2c() accepts custom policy names (B2C_1A_*)."""
+ provider = AzureProvider.from_b2c(
+ tenant_name="mytenant",
+ policy_name="B2C_1A_SIGNUP_SIGNIN",
+ client_id="client-id",
+ client_secret="secret",
+ required_scopes=["mcp-access"],
+ base_url="https://myserver.com",
+ jwt_signing_key="test-secret",
+ client_storage=memory_storage,
+ )
+
+ assert "B2C_1A_SIGNUP_SIGNIN" in provider._upstream_authorization_endpoint
+ assert "B2C_1A_SIGNUP_SIGNIN" in provider._upstream_token_endpoint
+
+ async def test_b2c_token_accepted_with_any_issuer(
+ self, memory_storage: MemoryStore
+ ):
+ """B2C provider (issuer=None) accepts tokens from any issuer."""
+ key_pair = RSAKeyPair.generate()
+ provider = AzureProvider.from_b2c(
+ tenant_name="mytenant",
+ policy_name="B2C_1_susi",
+ client_id="my-client-id",
+ client_secret="secret",
+ required_scopes=["mcp-access"],
+ base_url="https://myserver.com",
+ jwt_signing_key="test-secret",
+ client_storage=memory_storage,
+ )
+
+ assert isinstance(provider._token_validator, JWTVerifier)
+ verifier = provider._token_validator
+ verifier.public_key = key_pair.public_key
+ verifier.jwks_uri = None
+
+ token = key_pair.create_token(
+ subject="test-user",
+ issuer="https://mytenant.b2clogin.com/11111111-guid/v2.0/",
+ audience="my-client-id",
+ additional_claims={"scp": "mcp-access"},
+ )
+ result = await verifier.load_access_token(token)
+ assert result is not None
+
+ async def test_b2c_token_rejected_with_wrong_audience(
+ self, memory_storage: MemoryStore
+ ):
+ """B2C provider still rejects tokens with wrong audience."""
+ key_pair = RSAKeyPair.generate()
+ provider = AzureProvider.from_b2c(
+ tenant_name="mytenant",
+ policy_name="B2C_1_susi",
+ client_id="my-client-id",
+ client_secret="secret",
+ required_scopes=["mcp-access"],
+ base_url="https://myserver.com",
+ jwt_signing_key="test-secret",
+ client_storage=memory_storage,
+ )
+
+ assert isinstance(provider._token_validator, JWTVerifier)
+ verifier = provider._token_validator
+ verifier.public_key = key_pair.public_key
+ verifier.jwks_uri = None
+
+ token = key_pair.create_token(
+ subject="test-user",
+ issuer="https://mytenant.b2clogin.com/11111111-guid/v2.0/",
+ audience="wrong-app-id",
+ additional_claims={"scp": "mcp-access"},
+ )
+ result = await verifier.load_access_token(token)
+ assert result is None
+
+ async def test_b2c_obo_raises_not_implemented(self, memory_storage: MemoryStore):
+ """from_b2c() providers must reject OBO calls with NotImplementedError."""
+ provider = AzureProvider.from_b2c(
+ tenant_name="mytenant",
+ policy_name="B2C_1_susi",
+ client_id="client-id",
+ client_secret="secret",
+ required_scopes=["mcp-access"],
+ base_url="https://myserver.com",
+ jwt_signing_key="test-secret",
+ client_storage=memory_storage,
+ )
+
+ with pytest.raises(NotImplementedError, match="OBO"):
+ await provider.get_obo_credential(user_assertion="fake-token")
+
+
+class TestAzureProviderFromB2CInputValidation:
+ """Input validation for from_b2c() parameters."""
+
+ @pytest.mark.parametrize(
+ "tenant_name",
+ [
+ "mytenant.onmicrosoft.com",
+ "my.onmicrosoft.com.tenant",
+ ],
+ )
+ def test_tenant_name_with_onmicrosoft_suffix_rejected(
+ self, memory_storage: MemoryStore, tenant_name: str
+ ):
+ with pytest.raises(ValueError, match="onmicrosoft.com"):
+ AzureProvider.from_b2c(
+ tenant_name=tenant_name,
+ policy_name="B2C_1_susi",
+ client_id="client-id",
+ client_secret="secret",
+ required_scopes=["mcp-access"],
+ base_url="https://myserver.com",
+ jwt_signing_key="test-secret",
+ client_storage=memory_storage,
+ )