mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-08-13 17:19:10 +02:00
502 lines
20 KiB
Text
502 lines
20 KiB
Text
---
|
|
title: Authorization
|
|
sidebarTitle: Authorization
|
|
description: Control access to components using callable-based authorization checks that filter visibility and enforce permissions.
|
|
icon: shield-halved
|
|
tag: NEW
|
|
---
|
|
|
|
import { VersionBadge } from "/snippets/version-badge.mdx"
|
|
|
|
<VersionBadge version="3.0.0" />
|
|
|
|
Authorization controls what authenticated users can do with your FastMCP server. While [authentication](/servers/auth/authentication) verifies identity (who you are), authorization determines access (what you can do). FastMCP provides a callable-based authorization system that works at both the component level and globally via middleware.
|
|
|
|
The authorization model centers on a simple concept: callable functions that receive context about the current request and return `True` to allow access or `False` to deny it. Multiple checks combine with AND logic, meaning all checks must pass for access to be granted.
|
|
|
|
<Note>
|
|
Authorization relies on OAuth tokens which are only available with HTTP transports (SSE, Streamable HTTP). In STDIO mode, there's no OAuth mechanism, so `get_access_token()` returns `None` and all auth checks are skipped.
|
|
</Note>
|
|
|
|
<Note>
|
|
When an `AuthProvider` is configured, all requests to the MCP endpoint must carry a valid token—unauthenticated requests are rejected at the transport level before any auth checks run. Authorization checks therefore differentiate between authenticated users based on their scopes and claims, not between authenticated and unauthenticated users.
|
|
</Note>
|
|
|
|
## Auth Checks
|
|
|
|
An auth check is any callable that accepts an `AuthContext` and returns a boolean. Auth checks can be synchronous or asynchronous, so checks that need to perform async operations (like reading server state or calling external services) work naturally.
|
|
|
|
```python
|
|
from fastmcp.server.auth import AuthContext
|
|
|
|
def my_custom_check(ctx: AuthContext) -> bool:
|
|
# ctx.token is AccessToken | None
|
|
# ctx.component is the Tool, Resource, or Prompt being accessed
|
|
return ctx.token is not None and "special" in ctx.token.scopes
|
|
```
|
|
|
|
FastMCP provides two built-in auth checks that cover common authorization patterns.
|
|
|
|
### require_scopes
|
|
|
|
Scope-based authorization checks that the token contains all specified OAuth scopes. When multiple scopes are provided, all must be present (AND logic).
|
|
|
|
```python
|
|
from fastmcp import FastMCP
|
|
from fastmcp.server.auth import require_scopes
|
|
|
|
mcp = FastMCP("Scoped Server")
|
|
|
|
@mcp.tool(auth=require_scopes("admin"))
|
|
def admin_operation() -> str:
|
|
"""Requires the 'admin' scope."""
|
|
return "Admin action completed"
|
|
|
|
@mcp.tool(auth=require_scopes("read", "write"))
|
|
def read_write_operation() -> str:
|
|
"""Requires both 'read' AND 'write' scopes."""
|
|
return "Read/write action completed"
|
|
```
|
|
|
|
### require_roles
|
|
|
|
<VersionBadge version="4.0.0" />
|
|
|
|
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.
|
|
|
|
<Note>
|
|
`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.
|
|
</Note>
|
|
|
|
### 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.
|
|
|
|
```python
|
|
from fastmcp import FastMCP
|
|
from fastmcp.server.auth import restrict_tag
|
|
from fastmcp.server.middleware import AuthMiddleware
|
|
|
|
mcp = FastMCP(
|
|
"Tagged Server",
|
|
middleware=[
|
|
AuthMiddleware(auth=restrict_tag("admin", scopes=["admin"]))
|
|
]
|
|
)
|
|
|
|
@mcp.tool(tags={"admin"})
|
|
def admin_tool() -> str:
|
|
"""Tagged 'admin', so requires 'admin' scope."""
|
|
return "Admin only"
|
|
|
|
@mcp.tool(tags={"public"})
|
|
def public_tool() -> str:
|
|
"""Not tagged 'admin', so no scope required by the restriction."""
|
|
return "Anyone can access"
|
|
```
|
|
|
|
### Combining Checks
|
|
|
|
Multiple auth checks can be combined by passing a list. All checks must pass for authorization to succeed (AND logic).
|
|
|
|
```python
|
|
from fastmcp import FastMCP
|
|
from fastmcp.server.auth import require_scopes
|
|
|
|
mcp = FastMCP("Combined Auth Server")
|
|
|
|
@mcp.tool(auth=[require_scopes("admin"), require_scopes("write")])
|
|
def secure_admin_action() -> str:
|
|
"""Requires both 'admin' AND 'write' scopes."""
|
|
return "Secure admin action"
|
|
```
|
|
|
|
### Custom Auth Checks
|
|
|
|
Any callable that accepts `AuthContext` and returns `bool` can serve as an auth check. This enables authorization logic based on token claims, component metadata, or external systems.
|
|
|
|
```python
|
|
from fastmcp import FastMCP
|
|
from fastmcp.server.auth import AuthCheck, AuthContext
|
|
|
|
mcp = FastMCP("Custom Auth Server")
|
|
|
|
def require_premium_user(ctx: AuthContext) -> bool:
|
|
"""Check for premium user status in token claims."""
|
|
if ctx.token is None:
|
|
return False
|
|
return ctx.token.claims.get("premium", False) is True
|
|
|
|
def require_access_level(minimum_level: int) -> AuthCheck:
|
|
"""Factory function for level-based authorization."""
|
|
def check(ctx: AuthContext) -> bool:
|
|
if ctx.token is None:
|
|
return False
|
|
user_level = ctx.token.claims.get("level", 0)
|
|
return user_level >= minimum_level
|
|
return check
|
|
|
|
@mcp.tool(auth=require_premium_user)
|
|
def premium_feature() -> str:
|
|
"""Only for premium users."""
|
|
return "Premium content"
|
|
|
|
@mcp.tool(auth=require_access_level(5))
|
|
def advanced_feature() -> str:
|
|
"""Requires access level 5 or higher."""
|
|
return "Advanced feature"
|
|
```
|
|
|
|
### Async Auth Checks
|
|
|
|
Auth checks can be `async` functions, which is useful when the authorization decision depends on asynchronous operations like reading server state or querying external services.
|
|
|
|
```python
|
|
from fastmcp import FastMCP
|
|
from fastmcp.server.auth import AuthContext
|
|
|
|
mcp = FastMCP("Async Auth Server")
|
|
|
|
async def check_user_permissions(ctx: AuthContext) -> bool:
|
|
"""Async auth check that reads server state."""
|
|
if ctx.token is None:
|
|
return False
|
|
user_id = ctx.token.claims.get("sub")
|
|
# Async operations work naturally in auth checks
|
|
permissions = await fetch_user_permissions(user_id)
|
|
return "admin" in permissions
|
|
|
|
@mcp.tool(auth=check_user_permissions)
|
|
def admin_tool() -> str:
|
|
return "Admin action completed"
|
|
```
|
|
|
|
Sync and async checks can be freely combined in a list — each check is handled according to its type.
|
|
|
|
### Error Handling
|
|
|
|
Auth checks can raise exceptions for explicit denial with custom messages:
|
|
|
|
- **`AuthorizationError`**: Propagates with its custom message, useful for explaining why access was denied
|
|
- **`InsufficientScopeError`**: A subclass of `AuthorizationError` raised by `AuthMiddleware` when the denial is a missing scope; it [names the scopes the caller needs](#signaling-scope-shortfalls)
|
|
- **Other exceptions**: Masked for security (logged internally, treated as denial)
|
|
|
|
```python
|
|
from fastmcp.server.auth import AuthContext
|
|
from fastmcp.exceptions import AuthorizationError
|
|
|
|
def require_verified_email(ctx: AuthContext) -> bool:
|
|
"""Require verified email with explicit denial message."""
|
|
if ctx.token is None:
|
|
raise AuthorizationError("Authentication required")
|
|
if not ctx.token.claims.get("email_verified"):
|
|
raise AuthorizationError("Email verification required")
|
|
return True
|
|
```
|
|
|
|
## Component-Level Authorization
|
|
|
|
The `auth` parameter on decorators controls visibility and access for individual components. When auth checks fail for the current request, the component is hidden from list responses and direct access returns not-found.
|
|
|
|
```python
|
|
from fastmcp import FastMCP
|
|
from fastmcp.server.auth import require_scopes
|
|
|
|
mcp = FastMCP("Component Auth Server")
|
|
|
|
@mcp.tool(auth=require_scopes("write"))
|
|
def write_tool() -> str:
|
|
"""Only visible to users with 'write' scope."""
|
|
return "Written"
|
|
|
|
@mcp.resource("secret://data", auth=require_scopes("read"))
|
|
def secret_resource() -> str:
|
|
"""Only visible to users with 'read' scope."""
|
|
return "Secret data"
|
|
|
|
@mcp.prompt(auth=require_scopes("admin"))
|
|
def admin_prompt() -> str:
|
|
"""Only visible to users with 'admin' scope."""
|
|
return "Admin prompt content"
|
|
```
|
|
|
|
<Note>
|
|
Component-level `auth` controls both visibility (list filtering) and access (direct lookups return not-found for unauthorized requests). Additionally use `AuthMiddleware` to apply server-wide authorization rules and get explicit `AuthorizationError` responses on unauthorized execution attempts.
|
|
</Note>
|
|
|
|
## Server-Level Authorization
|
|
|
|
For server-wide authorization enforcement, use `AuthMiddleware`. This middleware applies auth checks globally to all components—filtering list responses and blocking unauthorized execution with explicit `AuthorizationError` responses. When the denial is specifically a missing scope, the error [names the scopes the caller needs](#signaling-scope-shortfalls).
|
|
|
|
```python
|
|
from fastmcp import FastMCP
|
|
from fastmcp.server.auth import require_scopes
|
|
from fastmcp.server.middleware import AuthMiddleware
|
|
|
|
mcp = FastMCP(
|
|
"Enforced Auth Server",
|
|
middleware=[AuthMiddleware(auth=require_scopes("api"))]
|
|
)
|
|
|
|
@mcp.tool
|
|
def any_tool() -> str:
|
|
"""Requires 'api' scope to see AND call."""
|
|
return "Protected"
|
|
```
|
|
|
|
### Component Auth + Middleware
|
|
|
|
Component-level `auth` and `AuthMiddleware` work together as complementary layers. The middleware applies server-wide rules to all components, while component-level auth adds per-component requirements. Both layers are checked—all checks must pass.
|
|
|
|
```python
|
|
from fastmcp import FastMCP
|
|
from fastmcp.server.auth import require_scopes, restrict_tag
|
|
from fastmcp.server.middleware import AuthMiddleware
|
|
|
|
mcp = FastMCP(
|
|
"Layered Auth Server",
|
|
middleware=[
|
|
AuthMiddleware(auth=restrict_tag("admin", scopes=["admin"]))
|
|
]
|
|
)
|
|
|
|
# Requires "write" scope (component-level)
|
|
# Also requires "admin" scope if tagged "admin" (middleware-level)
|
|
@mcp.tool(auth=require_scopes("write"), tags={"admin"})
|
|
def admin_write() -> str:
|
|
"""Requires both 'write' AND 'admin' scopes."""
|
|
return "Admin write"
|
|
|
|
# Requires "write" scope (component-level only)
|
|
@mcp.tool(auth=require_scopes("write"))
|
|
def user_write() -> str:
|
|
"""Requires 'write' scope."""
|
|
return "User write"
|
|
```
|
|
|
|
### Tag-Based Global Authorization
|
|
|
|
A common pattern uses `restrict_tag` with `AuthMiddleware` to apply scope requirements based on component tags.
|
|
|
|
```python
|
|
from fastmcp import FastMCP
|
|
from fastmcp.server.auth import restrict_tag
|
|
from fastmcp.server.middleware import AuthMiddleware
|
|
|
|
mcp = FastMCP(
|
|
"Tag-Based Auth Server",
|
|
middleware=[
|
|
AuthMiddleware(auth=restrict_tag("admin", scopes=["admin"])),
|
|
AuthMiddleware(auth=restrict_tag("write", scopes=["write"])),
|
|
]
|
|
)
|
|
|
|
@mcp.tool(tags={"admin"})
|
|
def delete_all_data() -> str:
|
|
"""Requires 'admin' scope."""
|
|
return "Deleted"
|
|
|
|
@mcp.tool(tags={"write"})
|
|
def update_record(id: str, data: str) -> str:
|
|
"""Requires 'write' scope."""
|
|
return f"Updated {id}"
|
|
|
|
@mcp.tool
|
|
def read_record(id: str) -> str:
|
|
"""No tag restrictions, accessible to all."""
|
|
return f"Record {id}"
|
|
```
|
|
|
|
### Signaling Scope Shortfalls
|
|
|
|
<VersionBadge version="4.0.0" />
|
|
|
|
A denial is more useful when it says what would fix it. When `AuthMiddleware` blocks a call because the token is missing scopes — rather than because some other policy rejected it — it raises `InsufficientScopeError`, which carries the specific scopes the caller needs in its `required_scopes` attribute. An agent that reads the error knows exactly which scopes to re-authorize for, instead of retrying blindly against an opaque refusal.
|
|
|
|
`InsufficientScopeError` subclasses `AuthorizationError`, so existing handlers that catch `AuthorizationError` keep catching it and nothing about your error handling has to change to adopt this.
|
|
|
|
Only the scopes the token *lacks* are named, so re-authorizing accumulates permissions rather than replacing them. A caller holding `read` that needs `read` and `write` is told to obtain `write` alone, and keeps `read` through the re-authorization. When several scope requirements fail at once, every unmet scope is reported together — a caller granted them all in one round succeeds on the retry, instead of discovering the next missing scope only after obtaining the first.
|
|
|
|
```python
|
|
from fastmcp import FastMCP
|
|
from fastmcp.exceptions import InsufficientScopeError
|
|
from fastmcp.server.auth import require_scopes
|
|
from fastmcp.server.middleware import AuthMiddleware
|
|
|
|
mcp = FastMCP(
|
|
"Step-Up Server",
|
|
middleware=[AuthMiddleware(auth=require_scopes("read", "write"))],
|
|
)
|
|
|
|
@mcp.tool
|
|
def update_record(id: str) -> str:
|
|
"""Requires both 'read' and 'write'."""
|
|
return f"Updated {id}"
|
|
|
|
# A token holding only "read" is denied with:
|
|
# InsufficientScopeError(required_scopes=["write"])
|
|
```
|
|
|
|
This holds across several `AuthMiddleware` instances too, not just several checks within one. In the [tag-based configuration](#tag-based-global-authorization) each middleware contributes its own requirement, and the first to find a shortfall reports the requirements of the others alongside its own — so one re-authorization covers the whole chain rather than one layer at a time.
|
|
|
|
A shortfall is reported only when the scope requirement is what actually caused the denial. If you [combine checks](#combining-checks) and a non-scope check rejects the request first — a tenant policy, say — the denial stays a plain `AuthorizationError` and names no scopes at all. Disclosing a scope requirement for a component the caller could not reach anyway would leak information about components they are not authorized to see.
|
|
|
|
That rule also bounds what gets aggregated. Combining requirements only reaches as far down the chain as the request itself would have gone: it stops at the first layer holding a custom check, since whether that layer would admit the caller is unknown until it runs, and running it early would trigger authorization logic the request had not reached yet. Requirements at or beyond that point sit behind an unverified gate and are left out.
|
|
|
|
So a custom check early in the chain makes the reported set partial, and a caller may need more than one round to satisfy everything. The reported set is complete when the layers ahead are scope-only and conservative otherwise: it may name fewer scopes than the full chain requires, but it never names scopes behind a policy that might reject the caller regardless.
|
|
|
|
<Note>
|
|
This names the missing scopes in the error rather than emitting an HTTP `403` challenge. A per-tool denial is a JSON-RPC error carried inside a `200` response, so there is no HTTP status at that layer to attach a `WWW-Authenticate` header to. Token-level scope failures — where the token does not satisfy the server's own `required_scopes` — are a separate concern handled by the transport middleware, which does return a spec-correct `403` with an `insufficient_scope` challenge.
|
|
</Note>
|
|
|
|
## Accessing Tokens in Tools
|
|
|
|
Tools can access the current authentication token using `get_access_token()` from `fastmcp.server.dependencies`. This enables tools to make decisions based on user identity or permissions beyond simple authorization checks.
|
|
|
|
```python
|
|
from fastmcp import FastMCP
|
|
from fastmcp.server.dependencies import get_access_token
|
|
|
|
mcp = FastMCP("Token Access Server")
|
|
|
|
@mcp.tool
|
|
def personalized_greeting() -> str:
|
|
"""Greet the user based on their token claims."""
|
|
token = get_access_token()
|
|
|
|
if token is None:
|
|
return "Hello, guest!"
|
|
|
|
name = token.claims.get("name", "user")
|
|
return f"Hello, {name}!"
|
|
|
|
@mcp.tool
|
|
def user_dashboard() -> dict:
|
|
"""Return user-specific data based on token."""
|
|
token = get_access_token()
|
|
|
|
if token is None:
|
|
return {"error": "Not authenticated"}
|
|
|
|
return {
|
|
"client_id": token.client_id,
|
|
"scopes": token.scopes,
|
|
"claims": token.claims,
|
|
}
|
|
```
|
|
|
|
## Reference
|
|
|
|
### AccessToken
|
|
|
|
The `AccessToken` object contains information extracted from the OAuth token.
|
|
|
|
| Property | Type | Description |
|
|
|----------|------|-------------|
|
|
| `token` | `str` | The raw token string |
|
|
| `client_id` | `str \| None` | OAuth client identifier |
|
|
| `scopes` | `list[str]` | Granted OAuth scopes |
|
|
| `expires_at` | `datetime \| None` | Token expiration time |
|
|
| `claims` | `dict[str, Any]` | All JWT claims or custom token data |
|
|
|
|
### AuthContext
|
|
|
|
The `AuthContext` dataclass is passed to all auth check functions.
|
|
|
|
| Property | Type | Description |
|
|
|----------|------|-------------|
|
|
| `token` | `AccessToken \| None` | Current access token, or `None` if unauthenticated |
|
|
| `component` | `Tool \| Resource \| Prompt` | The component being accessed |
|
|
|
|
Access to the component object enables authorization decisions based on metadata like tags, name, or custom properties.
|
|
|
|
```python
|
|
from fastmcp.server.auth import AuthContext
|
|
|
|
def require_matching_tag(ctx: AuthContext) -> bool:
|
|
"""Require a scope matching each of the component's tags."""
|
|
if ctx.token is None:
|
|
return False
|
|
user_scopes = set(ctx.token.scopes)
|
|
return ctx.component.tags.issubset(user_scopes)
|
|
```
|
|
|
|
### Imports
|
|
|
|
```python
|
|
from fastmcp.server.auth import (
|
|
AccessToken, # Token with .token, .client_id, .scopes, .expires_at, .claims
|
|
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
|
|
)
|
|
|
|
from fastmcp.exceptions import (
|
|
AuthorizationError, # Denial with a custom message
|
|
InsufficientScopeError, # Subclass of AuthorizationError; has .required_scopes
|
|
)
|
|
|
|
from fastmcp.server.middleware import AuthMiddleware
|
|
```
|