AWS Cognito OAuth 🤝 FastMCP

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", []),
    }

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())

import os
from fastmcp import FastMCP
from fastmcp.server.auth.providers.aws import AWSCognitoProvider
from key_value.aio.stores.redis import RedisStore
from key_value.aio.wrappers.encryption import FernetEncryptionWrapper
from cryptography.fernet import Fernet

# Production setup with encrypted persistent token storage
auth_provider = AWSCognitoProvider(
    user_pool_id="eu-central-1_XXXXXXXXX",
    aws_region="eu-central-1",
    client_id="your-app-client-id",
    client_secret="your-app-client-secret",
    base_url="https://your-production-domain.com",

    # Production token management
    jwt_signing_key=os.environ["JWT_SIGNING_KEY"],
    client_storage=FernetEncryptionWrapper(
        key_value=RedisStore(
            host=os.environ["REDIS_HOST"],
            port=int(os.environ["REDIS_PORT"])
        ),
        fernet=Fernet(os.environ["STORAGE_ENCRYPTION_KEY"])
    )
)

mcp = FastMCP(name="Production AWS Cognito App", auth=auth_provider)

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", []),
    }