mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-08-22 05:24:18 +02:00
Merge branch 'main' into switch-kvstore
This commit is contained in:
William Easton
GitHub
commit
7ca6bd422f
18 changed files with 1006 additions and 16 deletions
|
|
@ -168,6 +168,7 @@
|
|||
"pages": [
|
||||
"integrations/auth0",
|
||||
"integrations/authkit",
|
||||
"integrations/aws-cognito",
|
||||
"integrations/azure",
|
||||
"integrations/descope",
|
||||
"integrations/github",
|
||||
|
|
|
|||
322
docs/integrations/aws-cognito.mdx
Normal file
322
docs/integrations/aws-cognito.mdx
Normal file
|
|
@ -0,0 +1,322 @@
|
|||
---
|
||||
title: AWS Cognito OAuth 🤝 FastMCP
|
||||
sidebarTitle: AWS Cognito
|
||||
description: Secure your FastMCP server with AWS Cognito user pools
|
||||
icon: aws
|
||||
tag: NEW
|
||||
---
|
||||
|
||||
import { VersionBadge } from "/snippets/version-badge.mdx"
|
||||
|
||||
<VersionBadge version="2.12.4" />
|
||||
|
||||
This guide shows you how to secure your FastMCP server using **AWS Cognito user pools**. Since AWS Cognito doesn't support Dynamic Client Registration, this integration uses the [**OAuth Proxy**](/servers/auth/oauth-proxy) pattern to bridge AWS Cognito's traditional OAuth with MCP's authentication requirements. It also includes robust JWT token validation, ensuring enterprise-grade authentication.
|
||||
|
||||
## Configuration
|
||||
|
||||
### Prerequisites
|
||||
|
||||
Before you begin, you will need:
|
||||
1. An **[AWS Account](https://aws.amazon.com/)** with access to create AWS Cognito user pools
|
||||
2. Basic familiarity with AWS Cognito concepts (user pools, app clients)
|
||||
3. Your FastMCP server's URL (can be localhost for development, e.g., `http://localhost:8000`)
|
||||
|
||||
### Step 1: Create an AWS Cognito User Pool and App Client
|
||||
|
||||
Set up AWS Cognito user pool with an app client to get the credentials needed for authentication:
|
||||
|
||||
<Steps>
|
||||
<Step title="Navigate to AWS Cognito">
|
||||
Go to the **[AWS Cognito Console](https://console.aws.amazon.com/cognito/)** and ensure you're in your desired AWS region.
|
||||
|
||||
Select **"User pools"** from the side navigation (click on the hamburger icon at the top left in case you don't see any), and click **"Create user pool"** to create a new user pool.
|
||||
</Step>
|
||||
|
||||
<Step title="Define Your Application">
|
||||
AWS Cognito now provides a streamlined setup experience:
|
||||
|
||||
1. **Application type**: Select **"Traditional web application"** (this is the correct choice for FastMCP server-side authentication)
|
||||
2. **Name your application**: Enter a descriptive name (e.g., `FastMCP Server`)
|
||||
|
||||
The traditional web application type automatically configures:
|
||||
- Server-side authentication with client secrets
|
||||
- Authorization code grant flow
|
||||
- Appropriate security settings for confidential clients
|
||||
|
||||
<Info>
|
||||
Choose "Traditional web application" rather than SPA, Mobile app, or Machine-to-machine options. This ensures proper OAuth 2.0 configuration for FastMCP.
|
||||
</Info>
|
||||
</Step>
|
||||
|
||||
<Step title="Configure Options">
|
||||
AWS will guide you through configuration options:
|
||||
|
||||
- **Sign-in identifiers**: Choose how users will sign in (email, username, or phone)
|
||||
- **Required attributes**: Select any additional user information you need
|
||||
- **Return URL**: Add your callback URL (e.g., `http://localhost:8000/auth/callback` for development)
|
||||
|
||||
<Tip>
|
||||
The simplified interface handles most OAuth security settings automatically based on your application type selection.
|
||||
</Tip>
|
||||
</Step>
|
||||
|
||||
<Step title="Review and Create">
|
||||
Review your configuration and click **"Create user pool"**.
|
||||
|
||||
After creation, you'll see your user pool details. Save these important values:
|
||||
- **User pool ID** (format: `eu-central-1_XXXXXXXXX`)
|
||||
- **Client ID** (found under → "Applications" → "App clients" in the side navigation → \<Your application name, e.g., `FastMCP Server`\> → "App client information")
|
||||
- **Client Secret** (found under → "Applications" → "App clients" in the side navigation → \<Your application name, e.g., `FastMCP Server`\> → "App client information")
|
||||
|
||||
<Tip>
|
||||
The user pool ID and app client credentials are all you need for FastMCP configuration.
|
||||
</Tip>
|
||||
</Step>
|
||||
|
||||
<Step title="Configure OAuth Settings">
|
||||
Under "Login pages" in your app client's settings, you can double check and adjust the OAuth configuration:
|
||||
|
||||
- **Allowed callback URLs**: Add your server URL + `/auth/callback` (e.g., `http://localhost:8000/auth/callback`)
|
||||
- **Allowed sign-out URLs**: Optional, for logout functionality
|
||||
- **OAuth 2.0 grant types**: Ensure "Authorization code grant" is selected
|
||||
- **OpenID Connect scopes**: Select scopes your application needs (e.g., `openid`, `email`, `profile`)
|
||||
|
||||
<Tip>
|
||||
For local development, you can use `http://localhost` URLs. For production, you must use HTTPS.
|
||||
</Tip>
|
||||
</Step>
|
||||
|
||||
<Step title="Pick Up AWS Cognito Domain">
|
||||
Navigate to **"Branding" → "Domain"** in the side navigation to find or configure Your AWS Cognito domain:
|
||||
|
||||
**Option 1: Use Auto-Generated Domain**
|
||||
- If AWS has already created a domain automatically, note the **domain prefix** (the part before `.auth.region.amazoncognito.com`)
|
||||
- This prefix is what you'll use in your FastMCP configuration
|
||||
|
||||
**Option 2: Create a Custom Domain Prefix**
|
||||
- If no domain exists or you want a better name, delete the existing domain and create a new one using the **"Actions"** menu
|
||||
- Under **"Domain"** → **"Cognito domain"** in the **"Create Cognito domain"** dialog, enter a meaningful prefix (e.g., `my-app`) that is available in the AWS region you are in
|
||||
- Just note the **domain prefix** you entered (e.g., `my-fastmcp-app`) - this is what you'll use in your FastMCP configuration
|
||||
|
||||
<Info>
|
||||
The FastMCP AWS Cognito provider automatically constructs the full domain from your prefix and region, simplifying configuration.
|
||||
</Info>
|
||||
</Step>
|
||||
|
||||
<Step title="Save Your Credentials">
|
||||
After setup, you'll have:
|
||||
|
||||
- **User Pool ID**: Format like `eu-central-1_XXXXXXXXX`
|
||||
- **Client ID**: Your application's client identifier
|
||||
- **Client Secret**: Generated client secret (keep secure)
|
||||
- **Domain Prefix**: The prefix of Your AWS Cognito domain
|
||||
- **AWS Region**: Where Your AWS Cognito user pool is located
|
||||
|
||||
<Tip>
|
||||
Store these credentials securely. Never commit them to version control. Use environment variables or AWS Secrets Manager in production.
|
||||
</Tip>
|
||||
</Step>
|
||||
</Steps>
|
||||
|
||||
### Step 2: FastMCP Configuration
|
||||
|
||||
Create your FastMCP server using the `AWSCognitoProvider`, which handles AWS Cognito's JWT tokens and user claims automatically:
|
||||
|
||||
```python server.py
|
||||
from fastmcp import FastMCP
|
||||
from fastmcp.server.auth.providers.aws import AWSCognitoProvider
|
||||
from fastmcp.server.dependencies import get_access_token
|
||||
|
||||
# The AWSCognitoProvider handles JWT validation and user claims
|
||||
auth_provider = AWSCognitoProvider(
|
||||
user_pool_id="eu-central-1_XXXXXXXXX", # Your AWS Cognito user pool ID
|
||||
aws_region="eu-central-1", # AWS region (defaults to eu-central-1)
|
||||
client_id="your-app-client-id", # Your app client ID
|
||||
client_secret="your-app-client-secret", # Your app client Secret
|
||||
base_url="http://localhost:8000", # Must match your callback URL
|
||||
# redirect_path="/auth/callback" # Default value, customize if needed
|
||||
)
|
||||
|
||||
mcp = FastMCP(name="AWS Cognito Secured App", auth=auth_provider)
|
||||
|
||||
# Add a protected tool to test authentication
|
||||
@mcp.tool
|
||||
async def get_access_token_claims() -> dict:
|
||||
"""Get the authenticated user's access token claims."""
|
||||
token = get_access_token()
|
||||
return {
|
||||
"sub": token.claims.get("sub"),
|
||||
"username": token.claims.get("username"),
|
||||
"cognito:groups": token.claims.get("cognito:groups", []),
|
||||
}
|
||||
```
|
||||
|
||||
## Testing
|
||||
|
||||
### Running the Server
|
||||
|
||||
Start your FastMCP server with HTTP transport to enable OAuth flows:
|
||||
|
||||
```bash
|
||||
fastmcp run server.py --transport http --port 8000
|
||||
```
|
||||
|
||||
Your server is now running and protected by AWS Cognito OAuth authentication.
|
||||
|
||||
### Testing with a Client
|
||||
|
||||
Create a test client that authenticates with Your AWS Cognito-protected server:
|
||||
|
||||
```python test_client.py
|
||||
from fastmcp import Client
|
||||
import asyncio
|
||||
|
||||
async def main():
|
||||
# The client will automatically handle AWS Cognito OAuth
|
||||
async with Client("http://localhost:8000/mcp/", auth="oauth") as client:
|
||||
# First-time connection will open AWS Cognito login in your browser
|
||||
print("✓ Authenticated with AWS Cognito!")
|
||||
|
||||
# Test the protected tool
|
||||
print("Calling protected tool: get_access_token_claims")
|
||||
result = await client.call_tool("get_access_token_claims")
|
||||
user_data = result.data
|
||||
print("Available access token claims:")
|
||||
print(f"- sub: {user_data.get('sub', 'N/A')}")
|
||||
print(f"- username: {user_data.get('username', 'N/A')}")
|
||||
print(f"- cognito:groups: {user_data.get('cognito:groups', [])}")
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
```
|
||||
|
||||
When you run the client for the first time:
|
||||
1. Your browser will open to AWS Cognito's hosted UI login page
|
||||
2. After you sign in (or sign up), you'll be redirected back to your MCP server
|
||||
3. The client receives the JWT token and can make authenticated requests
|
||||
|
||||
<Info>
|
||||
The client caches tokens locally, so you won't need to re-authenticate for subsequent runs unless the token expires or you explicitly clear the cache.
|
||||
</Info>
|
||||
|
||||
## Environment Variables
|
||||
|
||||
For production deployments, use environment variables instead of hardcoding credentials.
|
||||
|
||||
### Provider Selection
|
||||
|
||||
Setting this environment variable allows the AWS Cognito provider to be used automatically without explicitly instantiating it in code.
|
||||
|
||||
<Card>
|
||||
<ParamField path="FASTMCP_SERVER_AUTH" default="Not set">
|
||||
Set to `fastmcp.server.auth.providers.aws.AWSCognitoProvider` to use AWS Cognito authentication.
|
||||
</ParamField>
|
||||
</Card>
|
||||
|
||||
### AWS Cognito-Specific Configuration
|
||||
|
||||
These environment variables provide default values for the AWS Cognito provider, whether it's instantiated manually or configured via `FASTMCP_SERVER_AUTH`.
|
||||
|
||||
<Card>
|
||||
<ParamField path="FASTMCP_SERVER_AUTH_AWS_COGNITO_USER_POOL_ID" required>
|
||||
Your AWS Cognito user pool ID (e.g., `eu-central-1_XXXXXXXXX`)
|
||||
</ParamField>
|
||||
|
||||
<ParamField path="FASTMCP_SERVER_AUTH_AWS_COGNITO_AWS_REGION" default="eu-central-1">
|
||||
AWS region where your AWS Cognito user pool is located
|
||||
</ParamField>
|
||||
|
||||
<ParamField path="FASTMCP_SERVER_AUTH_AWS_COGNITO_CLIENT_ID" required>
|
||||
Your AWS Cognito app client ID
|
||||
</ParamField>
|
||||
|
||||
<ParamField path="FASTMCP_SERVER_AUTH_AWS_COGNITO_CLIENT_SECRET" required>
|
||||
Your AWS Cognito app client secret
|
||||
</ParamField>
|
||||
|
||||
<ParamField path="FASTMCP_SERVER_AUTH_AWS_COGNITO_BASE_URL" default="http://localhost:8000">
|
||||
Public URL of your FastMCP server for OAuth callbacks
|
||||
</ParamField>
|
||||
|
||||
<ParamField path="FASTMCP_SERVER_AUTH_AWS_COGNITO_REDIRECT_PATH" default="/auth/callback">
|
||||
One of the redirect paths configured in your AWS Cognito app client
|
||||
</ParamField>
|
||||
|
||||
<ParamField path="FASTMCP_SERVER_AUTH_AWS_COGNITO_REQUIRED_SCOPES" default='["openid"]'>
|
||||
Comma-, space-, or JSON-separated list of required OAuth scopes (e.g., `openid email` or `["openid","email","profile"]`)
|
||||
</ParamField>
|
||||
</Card>
|
||||
|
||||
Example `.env` file:
|
||||
```bash
|
||||
# Use the AWS Cognito provider
|
||||
FASTMCP_SERVER_AUTH=fastmcp.server.auth.providers.aws.AWSCognitoProvider
|
||||
|
||||
# AWS Cognito credentials
|
||||
FASTMCP_SERVER_AUTH_AWS_COGNITO_USER_POOL_ID=eu-central-1_XXXXXXXXX
|
||||
FASTMCP_SERVER_AUTH_AWS_COGNITO_AWS_REGION=eu-central-1
|
||||
FASTMCP_SERVER_AUTH_AWS_COGNITO_CLIENT_ID=your-app-client-id
|
||||
FASTMCP_SERVER_AUTH_AWS_COGNITO_CLIENT_SECRET=your-app-client-secret
|
||||
FASTMCP_SERVER_AUTH_AWS_COGNITO_BASE_URL=https://your-server.com
|
||||
FASTMCP_SERVER_AUTH_AWS_COGNITO_REQUIRED_SCOPES=openid,email,profile
|
||||
```
|
||||
|
||||
With environment variables set, your server code simplifies to:
|
||||
|
||||
```python server.py
|
||||
from fastmcp import FastMCP
|
||||
from fastmcp.server.dependencies import get_access_token
|
||||
|
||||
# Authentication is automatically configured from environment
|
||||
mcp = FastMCP(name="AWS Cognito Secured App")
|
||||
|
||||
@mcp.tool
|
||||
async def get_access_token_claims() -> dict:
|
||||
"""Get the authenticated user's access token claims."""
|
||||
token = get_access_token()
|
||||
return {
|
||||
"sub": token.claims.get("sub"),
|
||||
"username": token.claims.get("username"),
|
||||
"cognito:groups": token.claims.get("cognito:groups", []),
|
||||
}
|
||||
```
|
||||
|
||||
## Features
|
||||
|
||||
### JWT Token Validation
|
||||
|
||||
The AWS Cognito provider includes robust JWT token validation:
|
||||
|
||||
- **Signature Verification**: Validates tokens against AWS Cognito's public keys (JWKS)
|
||||
- **Expiration Checking**: Automatically rejects expired tokens
|
||||
- **Issuer Validation**: Ensures tokens come from your specific AWS Cognito user pool
|
||||
- **Scope Enforcement**: Verifies required OAuth scopes are present
|
||||
|
||||
### User Claims and Groups
|
||||
|
||||
Access rich user information from AWS Cognito JWT tokens:
|
||||
|
||||
```python
|
||||
from fastmcp.server.dependencies import get_access_token
|
||||
|
||||
@mcp.tool
|
||||
async def admin_only_tool() -> str:
|
||||
"""A tool only available to admin users."""
|
||||
token = get_access_token()
|
||||
user_groups = token.claims.get("cognito:groups", [])
|
||||
|
||||
if "admin" not in user_groups:
|
||||
raise ValueError("This tool requires admin access")
|
||||
|
||||
return "Admin access granted!"
|
||||
```
|
||||
|
||||
### Enterprise Integration
|
||||
|
||||
Perfect for enterprise environments with:
|
||||
|
||||
- **Single Sign-On (SSO)**: Integrate with corporate identity providers
|
||||
- **Multi-Factor Authentication (MFA)**: Leverage AWS Cognito's built-in MFA
|
||||
- **User Groups**: Role-based access control through AWS Cognito groups
|
||||
- **Custom Attributes**: Access custom user attributes defined in your AWS Cognito user pool
|
||||
- **Compliance**: Meet enterprise security and compliance requirements
|
||||
|
|
@ -2,7 +2,7 @@
|
|||
title: Descope 🤝 FastMCP
|
||||
sidebarTitle: Descope
|
||||
description: Secure your FastMCP server with Descope
|
||||
icon: globe
|
||||
icon: shield-check
|
||||
tag: NEW
|
||||
---
|
||||
|
||||
|
|
|
|||
|
|
@ -374,7 +374,7 @@ containing the prompt messages and any additional metadata.
|
|||
#### `complete_mcp` <sup><a href="https://github.com/jlowin/fastmcp/blob/main/src/fastmcp/client/client.py#L744" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>
|
||||
|
||||
```python
|
||||
complete_mcp(self, ref: mcp.types.ResourceTemplateReference | mcp.types.PromptReference, argument: dict[str, str]) -> mcp.types.CompleteResult
|
||||
complete_mcp(self, ref: mcp.types.ResourceTemplateReference | mcp.types.PromptReference, argument: dict[str, str], context_arguments: dict[str, Any] | None = None) -> mcp.types.CompleteResult
|
||||
```
|
||||
|
||||
Send a completion request and return the complete MCP protocol result.
|
||||
|
|
@ -382,6 +382,7 @@ Send a completion request and return the complete MCP protocol result.
|
|||
**Args:**
|
||||
- `ref`: The reference to complete.
|
||||
- `argument`: Arguments to pass to the completion request.
|
||||
- `context_arguments`: Optional context arguments to include with the completion request. Defaults to None.
|
||||
|
||||
**Returns:**
|
||||
- mcp.types.CompleteResult: The complete response object from the protocol,
|
||||
|
|
@ -391,10 +392,10 @@ containing the completion and any additional metadata.
|
|||
- `RuntimeError`: If called while the client is not connected.
|
||||
|
||||
|
||||
#### `complete` <sup><a href="https://github.com/jlowin/fastmcp/blob/main/src/fastmcp/client/client.py#L767" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>
|
||||
#### `complete` <sup><a href="https://github.com/jlowin/fastmcp/blob/main/src/fastmcp/client/client.py#L772" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a></sup>
|
||||
|
||||
```python
|
||||
complete(self, ref: mcp.types.ResourceTemplateReference | mcp.types.PromptReference, argument: dict[str, str]) -> mcp.types.Completion
|
||||
complete(self, ref: mcp.types.ResourceTemplateReference | mcp.types.PromptReference, argument: dict[str, str], context_arguments: dict[str, Any] | None = None) -> mcp.types.Completion
|
||||
```
|
||||
|
||||
Send a completion request to the server.
|
||||
|
|
@ -402,6 +403,7 @@ Send a completion request to the server.
|
|||
**Args:**
|
||||
- `ref`: The reference to complete.
|
||||
- `argument`: Arguments to pass to the completion request.
|
||||
- `context_arguments`: Optional context arguments to include with the completion request. Defaults to None.
|
||||
|
||||
**Returns:**
|
||||
- mcp.types.Completion: The completion object.
|
||||
|
|
|
|||
|
|
@ -127,7 +127,7 @@ This example uses WorkOS AuthKit as the external identity provider. The `AuthKit
|
|||
|
||||
<VersionBadge version="2.12.0" />
|
||||
|
||||
`OAuthProxy` enables authentication with OAuth providers that **don't support Dynamic Client Registration (DCR)**, such as GitHub, Google, Azure, and most traditional enterprise identity systems.
|
||||
`OAuthProxy` enables authentication with OAuth providers that **don't support Dynamic Client Registration (DCR)**, such as GitHub, Google, Azure, AWS, and most traditional enterprise identity systems.
|
||||
|
||||
When identity providers require manual app registration and fixed credentials, `OAuthProxy` bridges the gap. It presents a DCR-compliant interface to MCP clients (accepting any registration request) while using your pre-registered credentials with the upstream provider. The proxy handles the complexity of callback forwarding, enabling dynamic client callbacks to work with providers that require fixed redirect URIs.
|
||||
|
||||
|
|
@ -256,7 +256,7 @@ This approach simplifies deployment pipelines and follows twelve-factor app prin
|
|||
|
||||
The authentication approach you choose depends on your existing infrastructure, security requirements, and operational constraints.
|
||||
|
||||
**For OAuth providers without DCR support (GitHub, Google, Azure, most enterprise systems), use OAuth Proxy.** These providers require manual app registration through their developer consoles. OAuth Proxy bridges the gap by presenting a DCR-compliant interface to MCP clients while using your fixed credentials with the provider. The proxy's callback forwarding pattern enables dynamic client ports to work with providers that require fixed redirect URIs.
|
||||
**For OAuth providers without DCR support (GitHub, Google, Azure, AWS, most enterprise systems), use OAuth Proxy.** These providers require manual app registration through their developer consoles. OAuth Proxy bridges the gap by presenting a DCR-compliant interface to MCP clients while using your fixed credentials with the provider. The proxy's callback forwarding pattern enables dynamic client ports to work with providers that require fixed redirect URIs.
|
||||
|
||||
**For identity providers with DCR support (Descope, WorkOS AuthKit, modern auth platforms), use RemoteAuthProvider.** These providers allow clients to dynamically register and obtain credentials without manual configuration. This enables the fully automated authentication flow that MCP is designed for, providing the best user experience and simplest implementation.
|
||||
|
||||
|
|
|
|||
|
|
@ -10,7 +10,7 @@ import { VersionBadge } from "/snippets/version-badge.mdx";
|
|||
|
||||
<VersionBadge version="2.12.0" />
|
||||
|
||||
OAuth Proxy enables FastMCP servers to authenticate with OAuth providers that **don't support Dynamic Client Registration (DCR)**. This includes virtually all traditional OAuth providers: GitHub, Google, Azure, Discord, Facebook, and most enterprise identity systems. For providers that do support DCR (like Descope and WorkOS AuthKit), use [`RemoteAuthProvider`](/servers/auth/remote-oauth) instead.
|
||||
OAuth Proxy enables FastMCP servers to authenticate with OAuth providers that **don't support Dynamic Client Registration (DCR)**. This includes virtually all traditional OAuth providers: GitHub, Google, Azure, AWS, Discord, Facebook, and most enterprise identity systems. For providers that do support DCR (like Descope and WorkOS AuthKit), use [`RemoteAuthProvider`](/servers/auth/remote-oauth) instead.
|
||||
|
||||
MCP clients expect to register automatically and obtain credentials on the fly, but traditional providers require manual app registration through their developer consoles. OAuth Proxy bridges this gap by presenting a DCR-compliant interface to MCP clients while using your pre-registered credentials with the upstream provider. When a client attempts to register, the proxy returns your fixed credentials. When a client initiates authorization, the proxy handles the complexity of callback forwarding—storing the client's dynamic callback URL, using its own fixed callback with the provider, then forwarding back to the client after token exchange.
|
||||
|
||||
|
|
@ -136,7 +136,7 @@ mcp = FastMCP(name="My Server", auth=auth)
|
|||
PKCE parameters to send upstream while separately validating the client's
|
||||
PKCE. This ensures end-to-end PKCE security at both layers (client-to-proxy
|
||||
and proxy-to-upstream). - `True` (default): Forward PKCE for providers that
|
||||
support it (Google, Azure, GitHub, etc.) - `False`: Disable only if upstream
|
||||
support it (Google, Azure, AWS, GitHub, etc.) - `False`: Disable only if upstream
|
||||
provider doesn't support PKCE
|
||||
</ParamField>
|
||||
|
||||
|
|
|
|||
|
|
@ -10,7 +10,7 @@ import { VersionBadge } from "/snippets/version-badge.mdx";
|
|||
|
||||
<VersionBadge version="2.12.4" />
|
||||
|
||||
OIDC Proxy enables FastMCP servers to authenticate with OIDC providers that **don't support Dynamic Client Registration (DCR)** out of the box. This includes OAuth providers like: Auth0, Google, Azure, etc. For providers that do support DCR (like WorkOS AuthKit), use [`RemoteAuthProvider`](/servers/auth/remote-oauth) instead.
|
||||
OIDC Proxy enables FastMCP servers to authenticate with OIDC providers that **don't support Dynamic Client Registration (DCR)** out of the box. This includes OAuth providers like: Auth0, Google, Azure, AWS, etc. For providers that do support DCR (like WorkOS AuthKit), use [`RemoteAuthProvider`](/servers/auth/remote-oauth) instead.
|
||||
|
||||
The OIDC Proxy is built upon [`OAuthProxy`](/servers/auth/oauth-proxy) so it has all the same functionality under the covers.
|
||||
|
||||
|
|
|
|||
|
|
@ -15,7 +15,7 @@ Remote OAuth integration allows your FastMCP server to leverage external identit
|
|||
<Tip>
|
||||
**When to use RemoteAuthProvider vs OAuth Proxy:**
|
||||
- **RemoteAuthProvider**: For providers WITH Dynamic Client Registration (Descope, WorkOS AuthKit, modern OIDC providers)
|
||||
- **OAuth Proxy**: For providers WITHOUT Dynamic Client Registration (GitHub, Google, Azure, Discord, etc.)
|
||||
- **OAuth Proxy**: For providers WITHOUT Dynamic Client Registration (GitHub, Google, Azure, AWS, Discord, etc.)
|
||||
|
||||
RemoteAuthProvider requires DCR support for fully automated client registration and authentication.
|
||||
</Tip>
|
||||
|
|
|
|||
47
examples/auth/aws_oauth/README.md
Normal file
47
examples/auth/aws_oauth/README.md
Normal file
|
|
@ -0,0 +1,47 @@
|
|||
# AWS Cognito OAuth Example
|
||||
|
||||
Demonstrates FastMCP server protection with AWS Cognito OAuth.
|
||||
|
||||
## Setup
|
||||
|
||||
1. Create an AWS Cognito User Pool and App Client:
|
||||
- Go to [AWS Cognito Console](https://console.aws.amazon.com/cognito/)
|
||||
- Create a new User Pool or use an existing one
|
||||
- Create an App Client in your User Pool
|
||||
- Configure the App Client settings:
|
||||
- Enable "Authorization code grant" flow
|
||||
- Add Callback URL: `http://localhost:8000/auth/callback`
|
||||
- Configure OAuth scopes (at minimum: `openid`)
|
||||
- Note your User Pool ID, App Client ID, Client Secret, and Cognito Domain Prefix
|
||||
|
||||
2. Set environment variables:
|
||||
|
||||
```bash
|
||||
export FASTMCP_SERVER_AUTH_AWS_COGNITO_USER_POOL_ID="your-user-pool-id"
|
||||
export FASTMCP_SERVER_AUTH_AWS_COGNITO_AWS_REGION="your-aws-region"
|
||||
export FASTMCP_SERVER_AUTH_AWS_COGNITO_CLIENT_ID="your-app-client-id"
|
||||
export FASTMCP_SERVER_AUTH_AWS_COGNITO_CLIENT_SECRET="your-app-client-secret"
|
||||
```
|
||||
|
||||
Or create a `.env` file:
|
||||
|
||||
```env
|
||||
FASTMCP_SERVER_AUTH_AWS_COGNITO_USER_POOL_ID=your-user-pool-id
|
||||
FASTMCP_SERVER_AUTH_AWS_COGNITO_AWS_REGION=your-aws-region
|
||||
FASTMCP_SERVER_AUTH_AWS_COGNITO_CLIENT_ID=your-app-client-id
|
||||
FASTMCP_SERVER_AUTH_AWS_COGNITO_CLIENT_SECRET=your-app-client-secret
|
||||
```
|
||||
|
||||
3. Run the server:
|
||||
|
||||
```bash
|
||||
python server.py
|
||||
```
|
||||
|
||||
4. In another terminal, run the client:
|
||||
|
||||
```bash
|
||||
python client.py
|
||||
```
|
||||
|
||||
The client will open your browser for AWS Cognito authentication.
|
||||
42
examples/auth/aws_oauth/client.py
Normal file
42
examples/auth/aws_oauth/client.py
Normal file
|
|
@ -0,0 +1,42 @@
|
|||
"""OAuth client example for connecting to FastMCP servers.
|
||||
|
||||
This example demonstrates how to connect to an OAuth-protected FastMCP server.
|
||||
|
||||
To run:
|
||||
python client.py
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
|
||||
from fastmcp.client import Client
|
||||
|
||||
SERVER_URL = "http://localhost:8000/mcp"
|
||||
|
||||
|
||||
async def main():
|
||||
try:
|
||||
async with Client(SERVER_URL, auth="oauth") as client:
|
||||
assert await client.ping()
|
||||
print("✅ Successfully authenticated!")
|
||||
|
||||
tools = await client.list_tools()
|
||||
print(f"🔧 Available tools ({len(tools)}):")
|
||||
for tool in tools:
|
||||
print(f" - {tool.name}: {tool.description}")
|
||||
|
||||
# Test the protected tool
|
||||
print("🔒 Calling protected tool: get_access_token_claims")
|
||||
result = await client.call_tool("get_access_token_claims")
|
||||
user_data = result.data
|
||||
print("📄 Available access token claims:")
|
||||
print(f" - sub: {user_data.get('sub', 'N/A')}")
|
||||
print(f" - username: {user_data.get('username', 'N/A')}")
|
||||
print(f" - cognito:groups: {user_data.get('cognito:groups', [])}")
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ Authentication failed: {e}")
|
||||
raise
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
2
examples/auth/aws_oauth/requirements.txt
Normal file
2
examples/auth/aws_oauth/requirements.txt
Normal file
|
|
@ -0,0 +1,2 @@
|
|||
fastmcp
|
||||
python-dotenv
|
||||
59
examples/auth/aws_oauth/server.py
Normal file
59
examples/auth/aws_oauth/server.py
Normal file
|
|
@ -0,0 +1,59 @@
|
|||
"""AWS Cognito OAuth server example for FastMCP.
|
||||
|
||||
This example demonstrates how to protect a FastMCP server with AWS Cognito.
|
||||
|
||||
Required environment variables:
|
||||
- FASTMCP_SERVER_AUTH_AWS_COGNITO_USER_POOL_ID: Your AWS Cognito User Pool ID
|
||||
- FASTMCP_SERVER_AUTH_AWS_COGNITO_AWS_REGION: Your AWS region (optional, defaults to eu-central-1)
|
||||
- FASTMCP_SERVER_AUTH_AWS_COGNITO_CLIENT_ID: Your Cognito app client ID
|
||||
- FASTMCP_SERVER_AUTH_AWS_COGNITO_CLIENT_SECRET: Your Cognito app client secret
|
||||
|
||||
To run:
|
||||
python server.py
|
||||
"""
|
||||
|
||||
import logging
|
||||
import os
|
||||
|
||||
from dotenv import load_dotenv
|
||||
|
||||
from fastmcp import FastMCP
|
||||
from fastmcp.server.auth.providers.aws import AWSCognitoProvider
|
||||
from fastmcp.server.dependencies import get_access_token
|
||||
|
||||
logging.basicConfig(level=logging.DEBUG)
|
||||
|
||||
load_dotenv(".env", override=True)
|
||||
|
||||
auth = AWSCognitoProvider(
|
||||
user_pool_id=os.getenv("FASTMCP_SERVER_AUTH_AWS_COGNITO_USER_POOL_ID") or "",
|
||||
aws_region=os.getenv("FASTMCP_SERVER_AUTH_AWS_COGNITO_AWS_REGION")
|
||||
or "eu-central-1",
|
||||
client_id=os.getenv("FASTMCP_SERVER_AUTH_AWS_COGNITO_CLIENT_ID") or "",
|
||||
client_secret=os.getenv("FASTMCP_SERVER_AUTH_AWS_COGNITO_CLIENT_SECRET") or "",
|
||||
base_url="http://localhost:8000",
|
||||
# redirect_path="/custom/callback"
|
||||
)
|
||||
|
||||
mcp = FastMCP("AWS Cognito OAuth Example Server", auth=auth)
|
||||
|
||||
|
||||
@mcp.tool
|
||||
def echo(message: str) -> str:
|
||||
"""Echo the provided message."""
|
||||
return message
|
||||
|
||||
|
||||
@mcp.tool
|
||||
async def get_access_token_claims() -> dict:
|
||||
"""Get the authenticated user's access token claims."""
|
||||
token = get_access_token()
|
||||
return {
|
||||
"sub": token.claims.get("sub"),
|
||||
"username": token.claims.get("username"),
|
||||
"cognito:groups": token.claims.get("cognito:groups", []),
|
||||
}
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
mcp.run(transport="http", port=8000)
|
||||
|
|
@ -745,12 +745,15 @@ class Client(Generic[ClientTransportT]):
|
|||
self,
|
||||
ref: mcp.types.ResourceTemplateReference | mcp.types.PromptReference,
|
||||
argument: dict[str, str],
|
||||
context_arguments: dict[str, Any] | None = None,
|
||||
) -> mcp.types.CompleteResult:
|
||||
"""Send a completion request and return the complete MCP protocol result.
|
||||
|
||||
Args:
|
||||
ref (mcp.types.ResourceTemplateReference | mcp.types.PromptReference): The reference to complete.
|
||||
argument (dict[str, str]): Arguments to pass to the completion request.
|
||||
context_arguments (dict[str, Any] | None, optional): Optional context arguments to
|
||||
include with the completion request. Defaults to None.
|
||||
|
||||
Returns:
|
||||
mcp.types.CompleteResult: The complete response object from the protocol,
|
||||
|
|
@ -761,19 +764,24 @@ class Client(Generic[ClientTransportT]):
|
|||
"""
|
||||
logger.debug(f"[{self.name}] called complete: {ref}")
|
||||
|
||||
result = await self.session.complete(ref=ref, argument=argument)
|
||||
result = await self.session.complete(
|
||||
ref=ref, argument=argument, context_arguments=context_arguments
|
||||
)
|
||||
return result
|
||||
|
||||
async def complete(
|
||||
self,
|
||||
ref: mcp.types.ResourceTemplateReference | mcp.types.PromptReference,
|
||||
argument: dict[str, str],
|
||||
context_arguments: dict[str, Any] | None = None,
|
||||
) -> mcp.types.Completion:
|
||||
"""Send a completion request to the server.
|
||||
|
||||
Args:
|
||||
ref (mcp.types.ResourceTemplateReference | mcp.types.PromptReference): The reference to complete.
|
||||
argument (dict[str, str]): Arguments to pass to the completion request.
|
||||
context_arguments (dict[str, Any] | None, optional): Optional context arguments to
|
||||
include with the completion request. Defaults to None.
|
||||
|
||||
Returns:
|
||||
mcp.types.Completion: The completion object.
|
||||
|
|
@ -781,7 +789,9 @@ class Client(Generic[ClientTransportT]):
|
|||
Raises:
|
||||
RuntimeError: If called while the client is not connected.
|
||||
"""
|
||||
result = await self.complete_mcp(ref=ref, argument=argument)
|
||||
result = await self.complete_mcp(
|
||||
ref=ref, argument=argument, context_arguments=context_arguments
|
||||
)
|
||||
return result.completion
|
||||
|
||||
# --- Tools ---
|
||||
|
|
|
|||
|
|
@ -1,7 +1,6 @@
|
|||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
from urllib.parse import urljoin
|
||||
|
||||
from mcp.server.auth.middleware.auth_context import AuthContextMiddleware
|
||||
from mcp.server.auth.middleware.bearer_auth import (
|
||||
|
|
@ -146,8 +145,9 @@ class AuthProvider(TokenVerifierProtocol):
|
|||
return None
|
||||
|
||||
if path:
|
||||
return AnyHttpUrl(urljoin(str(self.base_url), path))
|
||||
|
||||
prefix = str(self.base_url).rstrip("/")
|
||||
suffix = path.lstrip("/")
|
||||
return AnyHttpUrl(f"{prefix}/{suffix}")
|
||||
return self.base_url
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -274,7 +274,7 @@ class OAuthProxy(OAuthProvider):
|
|||
valid_scopes: List of all the possible valid scopes for a client.
|
||||
These are advertised to clients through the `/.well-known` endpoints. Defaults to `required_scopes` if not provided.
|
||||
forward_pkce: Whether to forward PKCE to upstream server (default True).
|
||||
Enable for providers that support/require PKCE (Google, Azure, etc.).
|
||||
Enable for providers that support/require PKCE (Google, Azure, AWS, etc.).
|
||||
Disable only if upstream provider doesn't support PKCE.
|
||||
token_endpoint_auth_method: Token endpoint authentication method for upstream server.
|
||||
Common values: "client_secret_basic", "client_secret_post", "none".
|
||||
|
|
|
|||
237
src/fastmcp/server/auth/providers/aws.py
Normal file
237
src/fastmcp/server/auth/providers/aws.py
Normal file
|
|
@ -0,0 +1,237 @@
|
|||
"""AWS Cognito OAuth provider for FastMCP.
|
||||
|
||||
This module provides a complete AWS Cognito OAuth integration that's ready to use
|
||||
with a user pool ID, domain prefix, client ID and client secret. It handles all
|
||||
the complexity of AWS Cognito's OAuth flow, token validation, and user management.
|
||||
|
||||
Example:
|
||||
```python
|
||||
from fastmcp import FastMCP
|
||||
from fastmcp.server.auth.providers.aws_cognito import AWSCognitoProvider
|
||||
|
||||
# Simple AWS Cognito OAuth protection
|
||||
auth = AWSCognitoProvider(
|
||||
user_pool_id="your-user-pool-id",
|
||||
aws_region="eu-central-1",
|
||||
client_id="your-cognito-client-id",
|
||||
client_secret="your-cognito-client-secret"
|
||||
)
|
||||
|
||||
mcp = FastMCP("My Protected Server", auth=auth)
|
||||
```
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from pydantic import AnyHttpUrl, SecretStr, field_validator
|
||||
from pydantic_settings import BaseSettings, SettingsConfigDict
|
||||
|
||||
from fastmcp.server.auth import TokenVerifier
|
||||
from fastmcp.server.auth.auth import AccessToken
|
||||
from fastmcp.server.auth.oidc_proxy import OIDCProxy
|
||||
from fastmcp.server.auth.providers.jwt import JWTVerifier
|
||||
from fastmcp.utilities.auth import parse_scopes
|
||||
from fastmcp.utilities.logging import get_logger
|
||||
from fastmcp.utilities.types import NotSet, NotSetT
|
||||
|
||||
logger = get_logger(__name__)
|
||||
|
||||
|
||||
class AWSCognitoProviderSettings(BaseSettings):
|
||||
"""Settings for AWS Cognito OAuth provider."""
|
||||
|
||||
model_config = SettingsConfigDict(
|
||||
env_prefix="FASTMCP_SERVER_AUTH_AWS_COGNITO_",
|
||||
env_file=".env",
|
||||
extra="ignore",
|
||||
)
|
||||
|
||||
user_pool_id: str | None = None
|
||||
aws_region: str | None = None
|
||||
client_id: str | None = None
|
||||
client_secret: SecretStr | None = None
|
||||
base_url: AnyHttpUrl | str | None = None
|
||||
redirect_path: str | None = None
|
||||
required_scopes: list[str] | None = None
|
||||
allowed_client_redirect_uris: list[str] | None = None
|
||||
|
||||
@field_validator("required_scopes", mode="before")
|
||||
@classmethod
|
||||
def _parse_scopes(cls, v):
|
||||
return parse_scopes(v)
|
||||
|
||||
|
||||
class AWSCognitoTokenVerifier(JWTVerifier):
|
||||
"""Token verifier that filters claims to Cognito-specific subset."""
|
||||
|
||||
async def verify_token(self, token: str) -> AccessToken | None:
|
||||
"""Verify token and filter claims to Cognito-specific subset."""
|
||||
# Use base JWT verification
|
||||
access_token = await super().verify_token(token)
|
||||
if not access_token:
|
||||
return None
|
||||
|
||||
# Filter claims to Cognito-specific subset
|
||||
cognito_claims = {
|
||||
"sub": access_token.claims.get("sub"),
|
||||
"username": access_token.claims.get("username"),
|
||||
"cognito:groups": access_token.claims.get("cognito:groups", []),
|
||||
}
|
||||
|
||||
# Return new AccessToken with filtered claims
|
||||
return AccessToken(
|
||||
token=access_token.token,
|
||||
client_id=access_token.client_id,
|
||||
scopes=access_token.scopes,
|
||||
expires_at=access_token.expires_at,
|
||||
claims=cognito_claims,
|
||||
)
|
||||
|
||||
|
||||
class AWSCognitoProvider(OIDCProxy):
|
||||
"""Complete AWS Cognito OAuth provider for FastMCP.
|
||||
|
||||
This provider makes it trivial to add AWS Cognito OAuth protection to any
|
||||
FastMCP server using OIDC Discovery. Just provide your Cognito User Pool details,
|
||||
client credentials, and a base URL, and you're ready to go.
|
||||
|
||||
Features:
|
||||
- Automatic OIDC Discovery from AWS Cognito User Pool
|
||||
- Automatic JWT token validation via Cognito's public keys
|
||||
- Cognito-specific claim filtering (sub, username, cognito:groups)
|
||||
- Support for Cognito User Pools
|
||||
|
||||
Example:
|
||||
```python
|
||||
from fastmcp import FastMCP
|
||||
from fastmcp.server.auth.providers.aws_cognito import AWSCognitoProvider
|
||||
|
||||
auth = AWSCognitoProvider(
|
||||
user_pool_id="eu-central-1_XXXXXXXXX",
|
||||
aws_region="eu-central-1",
|
||||
client_id="your-cognito-client-id",
|
||||
client_secret="your-cognito-client-secret",
|
||||
base_url="https://my-server.com",
|
||||
redirect_path="/custom/callback",
|
||||
)
|
||||
|
||||
mcp = FastMCP("My App", auth=auth)
|
||||
```
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
*,
|
||||
user_pool_id: str | NotSetT = NotSet,
|
||||
aws_region: str | NotSetT = NotSet,
|
||||
client_id: str | NotSetT = NotSet,
|
||||
client_secret: str | NotSetT = NotSet,
|
||||
base_url: AnyHttpUrl | str | NotSetT = NotSet,
|
||||
redirect_path: str | NotSetT = NotSet,
|
||||
required_scopes: list[str] | NotSetT = NotSet,
|
||||
allowed_client_redirect_uris: list[str] | NotSetT = NotSet,
|
||||
):
|
||||
"""Initialize AWS Cognito OAuth provider.
|
||||
|
||||
Args:
|
||||
user_pool_id: Your Cognito User Pool ID (e.g., "eu-central-1_XXXXXXXXX")
|
||||
aws_region: AWS region where your User Pool is located (defaults to "eu-central-1")
|
||||
client_id: Cognito app client ID
|
||||
client_secret: Cognito app client secret
|
||||
base_url: Public URL of your FastMCP server (for OAuth callbacks)
|
||||
redirect_path: Redirect path configured in Cognito app (defaults to "/auth/callback")
|
||||
required_scopes: Required Cognito scopes (defaults to ["openid"])
|
||||
allowed_client_redirect_uris: List of allowed redirect URI patterns for MCP clients.
|
||||
If None (default), all URIs are allowed. If empty list, no URIs are allowed.
|
||||
"""
|
||||
|
||||
settings = AWSCognitoProviderSettings.model_validate(
|
||||
{
|
||||
k: v
|
||||
for k, v in {
|
||||
"user_pool_id": user_pool_id,
|
||||
"aws_region": aws_region,
|
||||
"client_id": client_id,
|
||||
"client_secret": client_secret,
|
||||
"base_url": base_url,
|
||||
"redirect_path": redirect_path,
|
||||
"required_scopes": required_scopes,
|
||||
"allowed_client_redirect_uris": allowed_client_redirect_uris,
|
||||
}.items()
|
||||
if v is not NotSet
|
||||
}
|
||||
)
|
||||
|
||||
# Validate required settings
|
||||
if not settings.user_pool_id:
|
||||
raise ValueError(
|
||||
"user_pool_id is required - set via parameter or FASTMCP_SERVER_AUTH_AWS_COGNITO_USER_POOL_ID"
|
||||
)
|
||||
if not settings.client_id:
|
||||
raise ValueError(
|
||||
"client_id is required - set via parameter or FASTMCP_SERVER_AUTH_AWS_COGNITO_CLIENT_ID"
|
||||
)
|
||||
if not settings.client_secret:
|
||||
raise ValueError(
|
||||
"client_secret is required - set via parameter or FASTMCP_SERVER_AUTH_AWS_COGNITO_CLIENT_SECRET"
|
||||
)
|
||||
|
||||
# Apply defaults
|
||||
required_scopes_final = settings.required_scopes or ["openid"]
|
||||
allowed_client_redirect_uris_final = settings.allowed_client_redirect_uris
|
||||
aws_region_final = settings.aws_region or "eu-central-1"
|
||||
redirect_path_final = settings.redirect_path or "/auth/callback"
|
||||
|
||||
# Construct OIDC discovery URL
|
||||
config_url = f"https://cognito-idp.{aws_region_final}.amazonaws.com/{settings.user_pool_id}/.well-known/openid-configuration"
|
||||
|
||||
# Extract secret string from SecretStr
|
||||
client_secret_str = (
|
||||
settings.client_secret.get_secret_value() if settings.client_secret else ""
|
||||
)
|
||||
|
||||
# Store Cognito-specific info for claim filtering
|
||||
self.user_pool_id = settings.user_pool_id
|
||||
self.aws_region = aws_region_final
|
||||
|
||||
# Initialize OIDC proxy with Cognito discovery
|
||||
super().__init__(
|
||||
config_url=config_url,
|
||||
client_id=settings.client_id,
|
||||
client_secret=client_secret_str,
|
||||
algorithm="RS256",
|
||||
required_scopes=required_scopes_final,
|
||||
base_url=settings.base_url,
|
||||
redirect_path=redirect_path_final,
|
||||
allowed_client_redirect_uris=allowed_client_redirect_uris_final,
|
||||
)
|
||||
|
||||
logger.info(
|
||||
"Initialized AWS Cognito OAuth provider for client %s with scopes: %s",
|
||||
settings.client_id,
|
||||
required_scopes_final,
|
||||
)
|
||||
|
||||
def get_token_verifier(
|
||||
self,
|
||||
*,
|
||||
algorithm: str | None = None,
|
||||
audience: str | None = None,
|
||||
required_scopes: list[str] | None = None,
|
||||
timeout_seconds: int | None = None,
|
||||
) -> TokenVerifier:
|
||||
"""Creates a Cognito-specific token verifier with claim filtering.
|
||||
|
||||
Args:
|
||||
algorithm: Optional token verifier algorithm
|
||||
audience: Optional token verifier audience
|
||||
required_scopes: Optional token verifier required_scopes
|
||||
timeout_seconds: HTTP request timeout in seconds
|
||||
"""
|
||||
return AWSCognitoTokenVerifier(
|
||||
issuer=str(self.oidc_config.issuer),
|
||||
audience=audience,
|
||||
algorithm=algorithm,
|
||||
jwks_uri=str(self.oidc_config.jwks_uri),
|
||||
required_scopes=required_scopes,
|
||||
)
|
||||
245
tests/server/auth/providers/test_aws.py
Normal file
245
tests/server/auth/providers/test_aws.py
Normal file
|
|
@ -0,0 +1,245 @@
|
|||
"""Unit tests for AWS Cognito OAuth provider."""
|
||||
|
||||
import os
|
||||
from contextlib import contextmanager
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
|
||||
from fastmcp.server.auth.providers.aws import (
|
||||
AWSCognitoProvider,
|
||||
AWSCognitoProviderSettings,
|
||||
)
|
||||
|
||||
|
||||
@contextmanager
|
||||
def mock_cognito_oidc_discovery():
|
||||
"""Context manager to mock AWS Cognito OIDC discovery endpoint."""
|
||||
mock_oidc_config = {
|
||||
"issuer": "https://cognito-idp.us-east-1.amazonaws.com/us-east-1_XXXXXXXXX",
|
||||
"authorization_endpoint": "https://test.auth.us-east-1.amazoncognito.com/oauth2/authorize",
|
||||
"token_endpoint": "https://test.auth.us-east-1.amazoncognito.com/oauth2/token",
|
||||
"jwks_uri": "https://cognito-idp.us-east-1.amazonaws.com/us-east-1_XXXXXXXXX/.well-known/jwks.json",
|
||||
"userinfo_endpoint": "https://test.auth.us-east-1.amazoncognito.com/oauth2/userInfo",
|
||||
"response_types_supported": ["code", "token"],
|
||||
"subject_types_supported": ["public"],
|
||||
"id_token_signing_alg_values_supported": ["RS256"],
|
||||
"scopes_supported": ["openid", "email", "phone", "profile"],
|
||||
"token_endpoint_auth_methods_supported": [
|
||||
"client_secret_basic",
|
||||
"client_secret_post",
|
||||
],
|
||||
}
|
||||
|
||||
with patch("httpx.get") as mock_get:
|
||||
mock_response = mock_get.return_value
|
||||
mock_response.raise_for_status.return_value = None
|
||||
mock_response.json.return_value = mock_oidc_config
|
||||
yield
|
||||
|
||||
|
||||
class TestAWSCognitoProviderSettings:
|
||||
"""Test settings for AWS Cognito OAuth provider."""
|
||||
|
||||
def test_settings_from_env_vars(self):
|
||||
"""Test that settings can be loaded from environment variables."""
|
||||
with patch.dict(
|
||||
os.environ,
|
||||
{
|
||||
"FASTMCP_SERVER_AUTH_AWS_COGNITO_USER_POOL_ID": "us-east-1_XXXXXXXXX",
|
||||
"FASTMCP_SERVER_AUTH_AWS_COGNITO_AWS_REGION": "us-east-1",
|
||||
"FASTMCP_SERVER_AUTH_AWS_COGNITO_CLIENT_ID": "env_client_id",
|
||||
"FASTMCP_SERVER_AUTH_AWS_COGNITO_CLIENT_SECRET": "env_secret",
|
||||
"FASTMCP_SERVER_AUTH_AWS_COGNITO_BASE_URL": "https://example.com",
|
||||
"FASTMCP_SERVER_AUTH_AWS_COGNITO_REDIRECT_PATH": "/custom/callback",
|
||||
},
|
||||
):
|
||||
settings = AWSCognitoProviderSettings()
|
||||
|
||||
assert settings.user_pool_id == "us-east-1_XXXXXXXXX"
|
||||
assert settings.aws_region == "us-east-1"
|
||||
assert settings.client_id == "env_client_id"
|
||||
assert (
|
||||
settings.client_secret
|
||||
and settings.client_secret.get_secret_value() == "env_secret"
|
||||
)
|
||||
assert settings.base_url == "https://example.com"
|
||||
assert settings.redirect_path == "/custom/callback"
|
||||
|
||||
def test_settings_explicit_override_env(self):
|
||||
"""Test that explicit settings override environment variables."""
|
||||
with patch.dict(
|
||||
os.environ,
|
||||
{
|
||||
"FASTMCP_SERVER_AUTH_AWS_COGNITO_USER_POOL_ID": "env_pool_id",
|
||||
"FASTMCP_SERVER_AUTH_AWS_COGNITO_CLIENT_ID": "env_client_id",
|
||||
"FASTMCP_SERVER_AUTH_AWS_COGNITO_CLIENT_SECRET": "env_secret",
|
||||
},
|
||||
):
|
||||
settings = AWSCognitoProviderSettings.model_validate(
|
||||
{
|
||||
"user_pool_id": "explicit_pool_id",
|
||||
"client_id": "explicit_client_id",
|
||||
"client_secret": "explicit_secret",
|
||||
}
|
||||
)
|
||||
|
||||
assert settings.user_pool_id == "explicit_pool_id"
|
||||
assert settings.client_id == "explicit_client_id"
|
||||
assert (
|
||||
settings.client_secret
|
||||
and settings.client_secret.get_secret_value() == "explicit_secret"
|
||||
)
|
||||
|
||||
|
||||
class TestAWSCognitoProvider:
|
||||
"""Test AWSCognitoProvider initialization."""
|
||||
|
||||
def test_init_with_explicit_params(self):
|
||||
"""Test initialization with explicit parameters."""
|
||||
with mock_cognito_oidc_discovery():
|
||||
provider = AWSCognitoProvider(
|
||||
user_pool_id="us-east-1_XXXXXXXXX",
|
||||
aws_region="us-east-1",
|
||||
client_id="test_client",
|
||||
client_secret="test_secret",
|
||||
base_url="https://example.com",
|
||||
redirect_path="/custom/callback",
|
||||
required_scopes=["openid", "email"],
|
||||
)
|
||||
|
||||
# Check that the provider was initialized correctly
|
||||
assert provider._upstream_client_id == "test_client"
|
||||
assert provider._upstream_client_secret.get_secret_value() == "test_secret"
|
||||
assert (
|
||||
str(provider.base_url) == "https://example.com/"
|
||||
) # URLs get normalized with trailing slash
|
||||
assert provider._redirect_path == "/custom/callback"
|
||||
# OIDC provider should have discovered the endpoints automatically
|
||||
assert (
|
||||
provider._upstream_authorization_endpoint
|
||||
== "https://test.auth.us-east-1.amazoncognito.com/oauth2/authorize"
|
||||
)
|
||||
assert (
|
||||
provider._upstream_token_endpoint
|
||||
== "https://test.auth.us-east-1.amazoncognito.com/oauth2/token"
|
||||
)
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"scopes_env",
|
||||
[
|
||||
"openid,email",
|
||||
'["openid", "email"]',
|
||||
],
|
||||
)
|
||||
def test_init_with_env_vars(self, scopes_env):
|
||||
"""Test initialization with environment variables."""
|
||||
with patch.dict(
|
||||
os.environ,
|
||||
{
|
||||
"FASTMCP_SERVER_AUTH_AWS_COGNITO_USER_POOL_ID": "us-east-1_XXXXXXXXX",
|
||||
"FASTMCP_SERVER_AUTH_AWS_COGNITO_AWS_REGION": "us-east-1",
|
||||
"FASTMCP_SERVER_AUTH_AWS_COGNITO_CLIENT_ID": "env_client_id",
|
||||
"FASTMCP_SERVER_AUTH_AWS_COGNITO_CLIENT_SECRET": "env_secret",
|
||||
"FASTMCP_SERVER_AUTH_AWS_COGNITO_BASE_URL": "https://env-example.com",
|
||||
"FASTMCP_SERVER_AUTH_AWS_COGNITO_REQUIRED_SCOPES": scopes_env,
|
||||
},
|
||||
):
|
||||
with mock_cognito_oidc_discovery():
|
||||
provider = AWSCognitoProvider()
|
||||
|
||||
assert provider._upstream_client_id == "env_client_id"
|
||||
assert (
|
||||
provider._upstream_client_secret.get_secret_value() == "env_secret"
|
||||
)
|
||||
assert str(provider.base_url) == "https://env-example.com/"
|
||||
assert provider._token_validator.required_scopes == ["openid", "email"]
|
||||
|
||||
def test_init_explicit_overrides_env(self):
|
||||
"""Test that explicit parameters override environment variables."""
|
||||
with patch.dict(
|
||||
os.environ,
|
||||
{
|
||||
"FASTMCP_SERVER_AUTH_AWS_COGNITO_USER_POOL_ID": "env_pool_id",
|
||||
"FASTMCP_SERVER_AUTH_AWS_COGNITO_CLIENT_ID": "env_client_id",
|
||||
"FASTMCP_SERVER_AUTH_AWS_COGNITO_CLIENT_SECRET": "env_secret",
|
||||
},
|
||||
):
|
||||
with mock_cognito_oidc_discovery():
|
||||
provider = AWSCognitoProvider(
|
||||
user_pool_id="explicit_pool_id",
|
||||
client_id="explicit_client",
|
||||
client_secret="explicit_secret",
|
||||
base_url="https://example.com",
|
||||
)
|
||||
|
||||
assert provider._upstream_client_id == "explicit_client"
|
||||
assert (
|
||||
provider._upstream_client_secret.get_secret_value()
|
||||
== "explicit_secret"
|
||||
)
|
||||
# OIDC discovery should have configured the endpoints automatically
|
||||
assert provider._upstream_authorization_endpoint is not None
|
||||
|
||||
def test_init_missing_user_pool_id_raises_error(self):
|
||||
"""Test that missing user_pool_id raises ValueError."""
|
||||
with patch.dict(os.environ, {}, clear=True):
|
||||
with pytest.raises(ValueError, match="user_pool_id is required"):
|
||||
AWSCognitoProvider(
|
||||
client_id="test_client",
|
||||
client_secret="test_secret",
|
||||
)
|
||||
|
||||
def test_init_missing_client_id_raises_error(self):
|
||||
"""Test that missing client_id raises ValueError."""
|
||||
with patch.dict(os.environ, {}, clear=True):
|
||||
with pytest.raises(ValueError, match="client_id is required"):
|
||||
AWSCognitoProvider(
|
||||
user_pool_id="us-east-1_XXXXXXXXX",
|
||||
client_secret="test_secret",
|
||||
)
|
||||
|
||||
def test_init_missing_client_secret_raises_error(self):
|
||||
"""Test that missing client_secret raises ValueError."""
|
||||
with patch.dict(os.environ, {}, clear=True):
|
||||
with pytest.raises(ValueError, match="client_secret is required"):
|
||||
AWSCognitoProvider(
|
||||
user_pool_id="us-east-1_XXXXXXXXX",
|
||||
client_id="test_client",
|
||||
)
|
||||
|
||||
def test_init_defaults(self):
|
||||
"""Test that default values are applied correctly."""
|
||||
with mock_cognito_oidc_discovery():
|
||||
provider = AWSCognitoProvider(
|
||||
user_pool_id="us-east-1_XXXXXXXXX",
|
||||
client_id="test_client",
|
||||
client_secret="test_secret",
|
||||
base_url="https://example.com",
|
||||
)
|
||||
|
||||
# Check defaults
|
||||
assert str(provider.base_url) == "https://example.com/"
|
||||
assert provider._redirect_path == "/auth/callback"
|
||||
assert provider._token_validator.required_scopes == ["openid"]
|
||||
assert provider.aws_region == "eu-central-1"
|
||||
|
||||
def test_oidc_discovery_integration(self):
|
||||
"""Test that OIDC discovery endpoints are used correctly."""
|
||||
with mock_cognito_oidc_discovery():
|
||||
provider = AWSCognitoProvider(
|
||||
user_pool_id="us-west-2_YYYYYYYY",
|
||||
aws_region="us-west-2",
|
||||
client_id="test_client",
|
||||
client_secret="test_secret",
|
||||
base_url="https://example.com",
|
||||
)
|
||||
|
||||
# OIDC discovery should have configured the endpoints automatically
|
||||
assert provider._upstream_authorization_endpoint is not None
|
||||
assert provider._upstream_token_endpoint is not None
|
||||
assert "amazoncognito.com" in provider._upstream_authorization_endpoint
|
||||
|
||||
|
||||
# Token verification functionality is now tested as part of the OIDC provider integration
|
||||
# The CognitoTokenVerifier class is an internal implementation detail
|
||||
|
|
@ -105,6 +105,28 @@ class TestRemoteAuthProvider:
|
|||
"https://api.example.com/.well-known/oauth-protected-resource"
|
||||
)
|
||||
|
||||
def test_get_resource_url_with_nested_base_url(self):
|
||||
"""Test _get_resource_url returns correct URL for .well-known path with nested base_url."""
|
||||
tokens = {
|
||||
"test_token": {
|
||||
"client_id": "test-client",
|
||||
"scopes": ["read"],
|
||||
}
|
||||
}
|
||||
token_verifier = StaticTokenVerifier(tokens=tokens)
|
||||
provider = RemoteAuthProvider(
|
||||
token_verifier=token_verifier,
|
||||
authorization_servers=[AnyHttpUrl("https://auth.example.com")],
|
||||
base_url="https://api.example.com/v1/",
|
||||
)
|
||||
|
||||
metadata_url = provider._get_resource_url(
|
||||
"/.well-known/oauth-protected-resource"
|
||||
)
|
||||
assert metadata_url == AnyHttpUrl(
|
||||
"https://api.example.com/v1/.well-known/oauth-protected-resource"
|
||||
)
|
||||
|
||||
def test_get_resource_url_handles_trailing_slash(self):
|
||||
"""Test _get_resource_url handles trailing slash correctly."""
|
||||
tokens = {
|
||||
|
|
@ -216,6 +238,7 @@ class TestRemoteAuthProviderIntegration:
|
|||
[
|
||||
("https://api.example.com", "https://api.example.com/mcp"),
|
||||
("https://api.example.com/", "https://api.example.com/mcp"),
|
||||
("https://api.example.com/v1/", "https://api.example.com/v1/mcp"),
|
||||
],
|
||||
)
|
||||
async def test_base_url_configurations(self, base_url: str, expected_resource: str):
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue