From 920cb47778f16aca8113bfdef4103999b3472610 Mon Sep 17 00:00:00 2001
From: Jeremiah Lowin <153965+jlowin@users.noreply.github.com>
Date: Mon, 27 Jul 2026 12:36:58 -0400
Subject: [PATCH] Add require_roles auth check (#4656)
* Add require_roles auth check
* Make role docs runnable standalone and fully annotated
* Treat a scalar role claim as one role; correct step-up docs
* Add v4 version badge to require_roles docs
---
docs/servers/authorization.mdx | 74 +++++++-
fastmcp_slim/fastmcp/server/auth/__init__.py | 2 +
.../fastmcp/server/auth/authorization.py | 2 +
.../fastmcp/utilities/authorization.py | 89 +++++++++-
tests/server/auth/test_authorization.py | 158 +++++++++++++++++-
5 files changed, 319 insertions(+), 6 deletions(-)
diff --git a/docs/servers/authorization.mdx b/docs/servers/authorization.mdx
index 5481cbc2a..68bac9181 100644
--- a/docs/servers/authorization.mdx
+++ b/docs/servers/authorization.mdx
@@ -58,6 +58,75 @@ def read_write_operation() -> str:
return "Read/write action completed"
```
+### require_roles
+
+
+
+Scopes are standardized, so `require_scopes` works the same everywhere. Roles and groups are not part of OIDC, so every identity provider puts them under a different claim. `require_roles` handles the comparison and takes an `extract` callable that tells it where to look.
+
+```python
+from fastmcp import FastMCP
+from fastmcp.server.auth import require_roles
+
+def keycloak_roles(claims: dict) -> list[str]:
+ return claims["realm_access"]["roles"]
+
+mcp = FastMCP("Role Server")
+
+@mcp.tool(auth=require_roles("admin", extract=keycloak_roles))
+def admin_operation() -> str:
+ """Requires the 'admin' role."""
+ return "Admin action completed"
+
+@mcp.tool(auth=require_roles("admin", "auditor", extract=keycloak_roles))
+def audited_admin_operation() -> str:
+ """Requires both the 'admin' AND 'auditor' roles."""
+ return "Audited admin action"
+```
+
+Multiple roles are required together, matching `require_scopes`. A token whose claims lack the path entirely is denied rather than raising, so the extractor can index directly.
+
+Keeping the claim path at the call site means any provider works, including ones with unusual shapes. Common locations:
+
+| Provider | Extractor |
+| --- | --- |
+| Keycloak | `lambda c: c["realm_access"]["roles"]` |
+| Microsoft Entra | `lambda c: c["roles"]` |
+| AWS Cognito | `lambda c: c["cognito:groups"]` |
+| Auth0 | `lambda c: c["permissions"]` |
+
+Verify the claim against your own tenant before relying on it. Auth0's namespaced custom claims are configured per tenant, and Entra emits `roles` or `groups` depending on the app manifest.
+
+
+`require_roles` cannot signal a scope shortfall, because OAuth has no way to request a role. A role denial surfaces as a plain `AuthorizationError` rather than one of the `insufficient_scope` challenges described in [Signaling Scope Shortfalls](#signaling-scope-shortfalls), and it suppresses any scope shortfall raised alongside it — a caller blocked by their role should not be told to go obtain a scope that would not help them. Combining `require_roles` with `require_scopes` is otherwise fine: whenever the role check passes, a scope shortfall is reported as usual.
+
+
+### Checking Other Claims
+
+`require_roles` is a convenience for the common case. `AccessToken.claims` holds every claim from the token, so gating on anything else needs no special API — just an auth check that reads it.
+
+```python
+from fastmcp import FastMCP
+from fastmcp.server.auth import AuthCheck, AuthContext
+
+mcp = FastMCP("Claim Server")
+
+def require_tenant(tenant_id: str) -> AuthCheck:
+ """Require the token to come from a specific tenant."""
+ def check(ctx: AuthContext) -> bool:
+ if ctx.token is None:
+ return False
+ return ctx.token.claims.get("tid") == tenant_id
+ return check
+
+@mcp.tool(auth=require_tenant("acme"))
+def tenant_operation() -> str:
+ """Only callable by tokens issued for the acme tenant."""
+ return "Tenant action completed"
+```
+
+The same caveat applies: a check like this is opaque, so it suppresses scope disclosure for its siblings.
+
### restrict_tag
Tag-based restrictions apply scope requirements conditionally. If a component has the specified tag, the token must have the required scopes. Components without the tag are unaffected.
@@ -107,7 +176,7 @@ Any callable that accepts `AuthContext` and returns `bool` can serve as an auth
```python
from fastmcp import FastMCP
-from fastmcp.server.auth import AuthContext
+from fastmcp.server.auth import AuthCheck, AuthContext
mcp = FastMCP("Custom Auth Server")
@@ -117,7 +186,7 @@ def require_premium_user(ctx: AuthContext) -> bool:
return False
return ctx.token.claims.get("premium", False) is True
-def require_access_level(minimum_level: int):
+def require_access_level(minimum_level: int) -> AuthCheck:
"""Factory function for level-based authorization."""
def check(ctx: AuthContext) -> bool:
if ctx.token is None:
@@ -417,6 +486,7 @@ from fastmcp.server.auth import (
AuthContext, # Context with .token, .component
AuthCheck, # Type alias: sync or async Callable[[AuthContext], bool]
require_scopes, # Built-in: requires specific scopes
+ require_roles, # Built-in: requires roles read from token claims
restrict_tag, # Built-in: tag-based scope requirements
run_auth_checks, # Utility: run checks with AND logic
)
diff --git a/fastmcp_slim/fastmcp/server/auth/__init__.py b/fastmcp_slim/fastmcp/server/auth/__init__.py
index 67ee25915..d0a2953e9 100644
--- a/fastmcp_slim/fastmcp/server/auth/__init__.py
+++ b/fastmcp_slim/fastmcp/server/auth/__init__.py
@@ -11,6 +11,7 @@ from .auth import (
from .authorization import (
AuthCheck,
AuthContext,
+ require_roles,
require_scopes,
restrict_tag,
run_auth_checks,
@@ -75,6 +76,7 @@ __all__ = [
"RemoteAuthProvider",
"StaticTokenVerifier",
"TokenVerifier",
+ "require_roles",
"require_scopes",
"restrict_tag",
"run_auth_checks",
diff --git a/fastmcp_slim/fastmcp/server/auth/authorization.py b/fastmcp_slim/fastmcp/server/auth/authorization.py
index 236db2051..b6e731a2e 100644
--- a/fastmcp_slim/fastmcp/server/auth/authorization.py
+++ b/fastmcp_slim/fastmcp/server/auth/authorization.py
@@ -3,6 +3,7 @@
from fastmcp.utilities.authorization import (
AuthCheck,
AuthContext,
+ require_roles,
require_scopes,
restrict_tag,
run_auth_checks,
@@ -13,6 +14,7 @@ from fastmcp.utilities.authorization import (
__all__ = [
"AuthCheck",
"AuthContext",
+ "require_roles",
"require_scopes",
"restrict_tag",
"run_auth_checks",
diff --git a/fastmcp_slim/fastmcp/utilities/authorization.py b/fastmcp_slim/fastmcp/utilities/authorization.py
index 433644c9e..5ae4ae184 100644
--- a/fastmcp_slim/fastmcp/utilities/authorization.py
+++ b/fastmcp_slim/fastmcp/utilities/authorization.py
@@ -9,9 +9,9 @@ from __future__ import annotations
import inspect
import logging
-from collections.abc import Awaitable, Callable
+from collections.abc import Awaitable, Callable, Iterable
from dataclasses import dataclass
-from typing import TYPE_CHECKING, cast
+from typing import TYPE_CHECKING, Any, cast
from fastmcp.exceptions import AuthorizationError
@@ -104,11 +104,96 @@ class _RestrictTag(_ScopeAwareCheck):
return set(self.required_scopes) - set(ctx.token.scopes)
+class _RequireRoles:
+ """Callable auth check requiring all of a fixed set of roles.
+
+ Deliberately not a :class:`_ScopeAwareCheck`. Roles are not scopes and
+ cannot be requested through OAuth, so a shortfall has no spec-correct
+ step-up representation.
+ """
+
+ def __init__(
+ self,
+ roles: tuple[str, ...],
+ extract: Callable[[dict[str, Any]], Iterable[str]],
+ ) -> None:
+ self.required_roles: frozenset[str] = frozenset(roles)
+ self._extract = extract
+
+ def __call__(self, ctx: AuthContext) -> bool:
+ if ctx.token is None:
+ return False
+ try:
+ extracted = self._extract(ctx.token.claims)
+ # A provider that stores a single role as a bare string satisfies
+ # `Iterable[str]`, but iterating it yields characters: "admin"
+ # would deny an "admin" requirement and grant an "a" one. Treat a
+ # string as the single role it plainly means.
+ if isinstance(extracted, str):
+ extracted = [extracted]
+ granted = set(extracted)
+ except (KeyError, IndexError, TypeError):
+ # A caller whose token simply lacks the claim is an ordinary
+ # denial, not a broken check: return False rather than letting
+ # `_evaluate_check` log a warning on every unauthorized request.
+ return False
+ return self.required_roles.issubset(granted)
+
+
def require_scopes(*scopes: str) -> AuthCheck:
"""Require all of the given OAuth scopes."""
return _RequireScopes(scopes)
+def require_roles(
+ *roles: str,
+ extract: Callable[[dict[str, Any]], Iterable[str]],
+) -> AuthCheck:
+ """Require all of the given roles, read from the token's claims.
+
+ Roles and groups are not part of OIDC, so every identity provider puts them
+ somewhere different: `realm_access.roles` on Keycloak, `roles` on Microsoft
+ Entra, `cognito:groups` on AWS Cognito, `permissions` or a namespaced custom
+ claim on Auth0. `extract` receives the token's claims and returns the
+ caller's roles, which keeps that provider-specific knowledge at the call
+ site instead of guessing it here.
+
+ ```python
+ from fastmcp.server.auth import require_roles
+
+ keycloak = require_roles("admin", extract=lambda c: c["realm_access"]["roles"])
+ cognito = require_roles("admins", extract=lambda c: c["cognito:groups"])
+ ```
+
+ A token missing the claim entirely is denied rather than treated as an
+ error, so `extract` may index into the claims without guarding. An
+ extractor returning a bare string is treated as one role, since a provider
+ that stores a single role as a scalar is common.
+
+ Unlike `require_scopes`, this check cannot signal a shortfall: OAuth has no
+ way to request a role, so there is no `insufficient_scope` challenge to
+ emit. A role denial is therefore reported as a plain `AuthorizationError`,
+ and it suppresses any scope shortfall alongside it — a caller blocked by
+ their role must not be told to go obtain a scope that would not help.
+ Scope shortfalls are still reported normally whenever the role check
+ passes.
+
+ Args:
+ *roles: Roles the caller must hold. All are required (AND logic).
+ extract: Callable mapping the token's claims to the caller's roles.
+
+ Raises:
+ ValueError: If no roles are given, which would allow any authenticated
+ caller and is more likely a mistake than an intent.
+ """
+ if not roles:
+ raise ValueError(
+ "require_roles() needs at least one role; a check with no roles "
+ "would admit any authenticated caller."
+ )
+ return _RequireRoles(roles, extract)
+
+
def restrict_tag(tag: str, *, scopes: list[str]) -> AuthCheck:
"""Require scopes when the accessed component has a specific tag."""
return _RestrictTag(tag, scopes)
diff --git a/tests/server/auth/test_authorization.py b/tests/server/auth/test_authorization.py
index 7060a978c..b3f268806 100644
--- a/tests/server/auth/test_authorization.py
+++ b/tests/server/auth/test_authorization.py
@@ -12,6 +12,7 @@ from fastmcp.exceptions import AuthorizationError, InsufficientScopeError
from fastmcp.server.auth import (
AccessToken,
AuthContext,
+ require_roles,
require_scopes,
restrict_tag,
run_auth_checks,
@@ -19,6 +20,7 @@ from fastmcp.server.auth import (
from fastmcp.server.middleware import AuthMiddleware
from fastmcp.server.transforms import ToolTransform
from fastmcp.tools.tool_transform import ToolTransformConfig, TransformedTool
+from fastmcp.utilities.authorization import scope_requirements
from fastmcp.utilities.versions import VersionSpec
# =============================================================================
@@ -26,14 +28,17 @@ from fastmcp.utilities.versions import VersionSpec
# =============================================================================
-def make_token(scopes: list[str] | None = None) -> AccessToken:
+def make_token(
+ scopes: list[str] | None = None,
+ claims: dict | None = None,
+) -> AccessToken:
"""Create a test access token."""
return AccessToken(
token="test-token",
client_id="test-client",
scopes=scopes or [],
expires_at=None,
- claims={},
+ claims=claims or {},
)
@@ -86,6 +91,155 @@ class TestRequireScopes:
assert check(ctx) is False
+# =============================================================================
+# Tests for require_roles
+# =============================================================================
+
+
+KEYCLOAK = {"realm_access": {"roles": ["admin", "viewer"]}}
+
+
+def keycloak_roles(claims: dict) -> list[str]:
+ return claims["realm_access"]["roles"]
+
+
+class TestRequireRoles:
+ @pytest.mark.parametrize(
+ "claims, extract",
+ [
+ (KEYCLOAK, keycloak_roles),
+ ({"roles": ["admin"]}, lambda c: c["roles"]),
+ ({"cognito:groups": ["admin"]}, lambda c: c["cognito:groups"]),
+ ({"permissions": ["admin"]}, lambda c: c["permissions"]),
+ (
+ {"https://app.example.com/roles": ["admin"]},
+ lambda c: c["https://app.example.com/roles"],
+ ),
+ ],
+ )
+ def test_reads_roles_from_provider_specific_claim(self, claims, extract):
+ ctx = AuthContext(token=make_token(claims=claims), component=make_tool())
+ assert require_roles("admin", extract=extract)(ctx) is True
+
+ def test_requires_all_roles(self):
+ ctx = AuthContext(token=make_token(claims=KEYCLOAK), component=make_tool())
+ check = require_roles("admin", "viewer", extract=keycloak_roles)
+ assert check(ctx) is True
+
+ def test_denies_when_one_role_missing(self):
+ ctx = AuthContext(token=make_token(claims=KEYCLOAK), component=make_tool())
+ check = require_roles("admin", "auditor", extract=keycloak_roles)
+ assert check(ctx) is False
+
+ def test_denies_without_token(self):
+ ctx = AuthContext(token=None, component=make_tool())
+ assert require_roles("admin", extract=keycloak_roles)(ctx) is False
+
+ @pytest.mark.parametrize(
+ "claims",
+ [{}, {"realm_access": {}}, {"realm_access": None}, {"realm_access": []}],
+ )
+ def test_denies_when_claim_absent_or_malformed(self, claims):
+ """A token without the claim is an ordinary denial, not a broken check."""
+ ctx = AuthContext(token=make_token(claims=claims), component=make_tool())
+ assert require_roles("admin", extract=keycloak_roles)(ctx) is False
+
+ def test_rejects_empty_role_list(self):
+ """A check with no roles would admit any authenticated caller."""
+ with pytest.raises(ValueError, match="at least one role"):
+ require_roles(extract=keycloak_roles)
+
+ def test_is_opaque_to_scope_shortfall(self):
+ """Roles cannot be requested via OAuth, so they yield no step-up."""
+ ctx = AuthContext(token=make_token(claims=KEYCLOAK), component=make_tool())
+ check = require_roles("auditor", extract=keycloak_roles)
+ assert scope_requirements(check, ctx) is None
+
+ def test_suppresses_shortfall_disclosure_of_sibling_scope_checks(self):
+ """One opaque check withholds the whole list's scope requirements."""
+ token = make_token(scopes=["read"], claims=KEYCLOAK)
+ ctx = AuthContext(token=token, component=make_tool())
+ checks = [
+ require_scopes("write"),
+ require_roles("admin", extract=keycloak_roles),
+ ]
+ assert scope_requirements(checks, ctx) is None
+ assert scope_requirements([require_scopes("write")], ctx) == ["write"]
+
+ @pytest.mark.parametrize(
+ "required, expected",
+ [("admin", True), ("a", False), ("dmin", False)],
+ )
+ def test_scalar_role_claim_is_one_role(self, required: str, expected: bool):
+ """A provider storing one role as a string must not be iterated.
+
+ `str` satisfies `Iterable[str]`, so a bare "admin" would otherwise
+ become the character set {a, d, m, i, n} — denying the "admin" it
+ plainly grants and granting any single character it contains.
+ """
+ ctx = AuthContext(
+ token=make_token(claims={"role": "admin"}), component=make_tool()
+ )
+ check = require_roles(required, extract=lambda c: c["role"])
+ assert check(ctx) is expected
+
+ async def test_role_denial_suppresses_scope_challenge(self):
+ """A caller blocked by their role is not told to obtain a scope."""
+ mcp = FastMCP(
+ middleware=[
+ AuthMiddleware(
+ auth=[
+ require_scopes("api"),
+ require_roles("admin", extract=keycloak_roles),
+ ]
+ )
+ ]
+ )
+
+ @mcp.tool
+ def t() -> str:
+ return "ok"
+
+ token = make_token(scopes=["read"], claims={"realm_access": {"roles": ["v"]}})
+ tok = set_token(token)
+ try:
+ with pytest.raises(AuthorizationError) as exc_info:
+ await mcp.call_tool("t", {})
+ finally:
+ auth_context_var.reset(tok)
+
+ assert not isinstance(exc_info.value, InsufficientScopeError)
+
+ async def test_passing_role_still_allows_scope_challenge(self):
+ """Mixing the two checks does not disable step-up on its own."""
+ mcp = FastMCP(
+ middleware=[
+ AuthMiddleware(
+ auth=[
+ require_scopes("api"),
+ require_roles("admin", extract=keycloak_roles),
+ ]
+ )
+ ]
+ )
+
+ @mcp.tool
+ def t() -> str:
+ return "ok"
+
+ token = make_token(
+ scopes=["read"], claims={"realm_access": {"roles": ["admin"]}}
+ )
+ tok = set_token(token)
+ try:
+ with pytest.raises(InsufficientScopeError) as exc_info:
+ await mcp.call_tool("t", {})
+ finally:
+ auth_context_var.reset(tok)
+
+ assert exc_info.value.required_scopes == ["api"]
+
+
# =============================================================================
# Tests for restrict_tag
# =============================================================================