adds fastmcp mcp example

Author: patrickkang

.gitignore

@@ -18,7 +18,6 @@ dist
 docs
 
 #testfile
-server.py
 setup.py
 test.py
 test-script.py

examples/example-fastmcp-mcp/.env.example

@@ -0,0 +1,5 @@
+# Auth0 Configuration
+AUTH0_DOMAIN=your-tenant.auth0.com
+AUTH0_AUDIENCE=https://api.example.com
+MCP_SERVER_URL=http://localhost:3001
+PORT=3001

examples/example-fastmcp-mcp/README.md

@@ -0,0 +1,43 @@
+# Example FastMCP MCP Server with Auth0 Integration
+
+This example demonstrates how to create a FastMCP MCP server that uses Auth0 for authentication using the `auth0-api-python` library.
+
+## Install dependencies
+
+```
+poetry install
+```
+
+## Auth0 Tenant Setup
+
+For detailed instructions on setting up your Auth0 tenant for MCP server integration, please refer to the [Auth0 Tenant Setup guide](https://github.com/auth0/auth0-auth-js/blob/main/examples/example-fastmcp-mcp/README.md#auth0-tenant-setup).
+
+## Configuration
+
+Rename `.env.example` to `.env` and configure the domain and audience:
+
+```
+# Auth0 tenant domain
+AUTH0_DOMAIN=example-tenant.us.auth0.com
+
+# Auth0 API Identifier
+AUTH0_AUDIENCE=http://localhost:3001
+```
+
+With the configuration in place, the example can be started by running:
+
+```bash
+poetry run python -m src.server
+```
+
+## Testing
+
+Use an MCP client like [MCP Inspector](https://github.com/modelcontextprotocol/inspector) to test your server interactively:
+
+```bash
+npx @modelcontextprotocol/inspector
+```
+
+The server will start up and the UI will be accessible at http://localhost:6274.
+
+In the MCP Inspector, select `Streamable HTTP` as the `Transport Type` and enter `http://localhost:3001/mcp` as the URL.

examples/example-fastmcp-mcp/poetry.lock

@@ -0,0 +1,19 @@
+[tool.poetry]
+name = "example-fastmcp-mcp"
+version = "0.1.0"
+description = ""
+authors = ["Auth0 <support@auth0.com>"]
+readme = "README.md"
+package-mode = false
+
+[tool.poetry.dependencies]
+python = "^3.10"
+python-dotenv = "^1.1.1"
+mcp = "^1.14.1"
+auth0-api-python = {path = "../..", develop = true}
+starlette = "^0.48.0"
+uvicorn = "^0.36.0"
+
+[build-system]
+requires = ["poetry-core>=2.0.0,<3.0.0"]
+build-backend = "poetry.core.masonry.api"

examples/example-fastmcp-mcp/pyproject.toml

@@ -0,0 +1,55 @@
+"""
+Auth0 integration for MCP server.
+
+This module provides Auth0 authentication and authorization for MCP servers,
+including token verification, middleware, and scoped tool decorators.
+"""
+
+import os
+from typing import List
+from pydantic import AnyHttpUrl
+from dotenv import load_dotenv
+from mcp.server.auth.routes import create_protected_resource_routes
+from starlette.routing import Router, Route
+from starlette.middleware import Middleware
+from .middeware import Auth0Middleware
+
+# Load environment variables
+load_dotenv()
+
+class Auth0Mcp:
+    def __init__(self, name: str):
+        self.name = name
+        self.audience = os.getenv("AUTH0_AUDIENCE", "https://api.example.com")
+        self.domain = os.getenv("AUTH0_DOMAIN", "your-tenant.auth0.com")
+
+    def auth_metadata_router(self) -> Router:
+        """
+        Returns a router that serves the OAuth Protected Resource Metadata
+        at the standard endpoint: /.well-known/oauth-protected-resource
+        """
+        routes: List[Route] = []
+
+        routes = create_protected_resource_routes(
+            resource_url=AnyHttpUrl(self.audience),
+            authorization_servers=[AnyHttpUrl(f"https://{self.domain}")],
+            scopes_supported=[
+                "openid",
+                "profile",
+                "email",
+            ],
+            resource_name=self.name,
+        )
+
+        return Router(routes=routes)
+    
+    def auth_middleware(self) -> list[Middleware]:
+        middleware: list[Middleware] = []
+
+        middleware.append(
+            Middleware(
+                Auth0Middleware
+            )
+        )
+
+        return middleware

examples/example-fastmcp-mcp/src/__init__.py

@@ -0,0 +1,95 @@
+import logging
+import os
+from starlette.middleware.base import BaseHTTPMiddleware
+from starlette.requests import Request
+from starlette.responses import JSONResponse
+from starlette.types import ASGIApp
+
+from auth0_api_python import ApiClient, ApiClientOptions
+from auth0_api_python.errors import VerifyAccessTokenError
+
+logger = logging.getLogger(__name__)
+
+class Auth0Middleware(BaseHTTPMiddleware):
+    """
+    Middleware that requires a valid Bearer token in the Authorization header.
+    This will validate the token using Auth0 SDK Client and add the auth info to request.scope["auth"].
+    """
+
+    def __init__(self, app: ASGIApp):
+        super().__init__(app)
+        self.client = ApiClient(ApiClientOptions(
+            domain=os.getenv("AUTH0_DOMAIN", "your-tenant.auth0.com"),
+            audience=os.getenv("AUTH0_AUDIENCE", "https://api.example.com")
+        ))
+
+    async def dispatch(self, request: Request, call_next):
+        # Extract Authorization header
+        auth_header = request.headers.get("authorization")
+        if not auth_header:
+            return self._return_auth_error_response(status_code=401, error="Authentication required", description="Missing Authorization header")
+        if not auth_header.lower().startswith("bearer "):
+            return self._return_auth_error_response(
+                status_code=401,
+                error="Authentication required",
+                description="Invalid Authorization header format"
+            )
+
+        # Extract and verify token
+        token = auth_header[7:] # Remove "Bearer " prefix
+        try:
+            decoded_and_verified_token = await self.client.verify_access_token(
+                token,
+                required_claims=["sub"]
+            )
+
+            # Check for client_id or azp
+            clientId = decoded_and_verified_token.get('client_id') or decoded_and_verified_token.get('azp')
+            if not clientId:
+                raise VerifyAccessTokenError("Token is missing 'client_id' or 'azp' claim")
+
+            # Set up authentication context
+            auth_data = {
+                "token": token,
+                "client_id": clientId,
+                "scopes": decoded_and_verified_token.get("scope", "").split() 
+                         if decoded_and_verified_token.get("scope") else []
+            }
+
+            if decoded_and_verified_token.get('exp'):
+                auth_data["expiresAt"] = decoded_and_verified_token.get('exp')
+
+            extra = {"sub": decoded_and_verified_token.get('sub'), "client_id": clientId}
+
+            for field in ['azp', 'name', 'email']:
+                if decoded_and_verified_token.get(field):
+                    extra[field] = decoded_and_verified_token.get(field)
+
+            auth_data["extra"] = extra
+            request.scope["auth"] = auth_data
+
+            return await call_next(request)
+        except VerifyAccessTokenError as e:
+            logger.error(f"Token verification failed: {str(e)}")
+            return self._return_auth_error_response(
+                status_code=401,
+                error="Authentication failed",
+                description="Invalid token"
+            )
+        except Exception as e:
+            logger.error(f"Unexpected error in middleware: {str(e)}")
+            return self._return_auth_error_response(
+                status_code=500,
+                error="Internal Server Error",
+                description="Internal Server Error"
+            )
+
+    def _return_auth_error_response(self, status_code: int, error: str, description: str) -> JSONResponse:
+        www_auth_parts = [f'error="{error}"', f'error_description="{description}"', f'resource_metadata="{os.getenv("MCP_SERVER_URL")}"']
+        www_authenticate = f"Bearer {', '.join(www_auth_parts)}"
+
+        return JSONResponse(
+            status_code=status_code,
+            content={"error": error, "error_description": description},
+            headers={"WWW-Authenticate": www_authenticate}
+        )

examples/example-fastmcp-mcp/src/auth0/__init__.py

@@ -0,0 +1,89 @@
+"""
+Scope-based auth decorators for MCP tools.
+
+Provides a decorator with Auth0 scope checking for MCP tools.
+"""
+from functools import wraps
+from typing import List, Callable
+import asyncio
+
+def create_scoped_tool_decorator(mcp_server):
+    """Factory function to create a scoped_tool decorator bound to a MCP server instance."""
+    
+    def scoped_tool(
+        required_scopes: List[str],
+        **tool_kwargs
+    ):
+        """
+        Decorator that combines FastMCP tool registration with Auth0 scope checking.
+        
+        Args:
+            required_scopes: List of scopes required to use this tool
+            **tool_kwargs: Additional parameters passed to @mcp.tool()
+        
+        Example:
+            @scoped_tool(required_scopes=["read:data", "write:data"])
+            def sensitive_tool(data: str, ctx: Context) -> str:
+                return f"Processing: {data}"
+        """
+        def decorator(func: Callable) -> Callable:
+            @wraps(func)
+            async def scope_checked_wrapper(*args, **kwargs):
+                # Find the Context parameter in kwargs
+                ctx = None
+                for key, value in kwargs.items():
+                    if hasattr(value, 'request_context') and hasattr(value, 'fastmcp'):
+                        ctx = value
+                        break
+
+                if not ctx:
+                    raise Exception(f"Tool '{func.__name__}' requires a Context parameter for scope checking")
+
+                # Get auth info and check scopes
+                try:
+                    request = ctx.request_context.request
+                    auth_info = get_auth_info(request)
+
+                    if not auth_info or auth_info == {}:
+                        raise Exception("Authentication required to use this tool")
+
+                    user_scopes = auth_info.get("scopes", [])
+                    client_id = auth_info.get("client_id", "unknown")
+
+                    # Check if user has all required scopes
+                    missing_scopes = [scope for scope in required_scopes if scope not in user_scopes]
+                    if missing_scopes:
+                        await ctx.error(f"Access denied: missing required scopes {missing_scopes}")
+                        raise Exception(
+                            f"Missing required scopes for tool '{func.__name__}': {missing_scopes}."
+                        )
+
+                    # Log successful scope check
+                    await ctx.info(f"Tool '{func.__name__}' authorized for client '{client_id}' with scopes: {user_scopes}")
+                    
+                except Exception as e:
+                    # Log other unexpected errors and wrap them
+                    await ctx.error(f"Authorization check failed for tool '{func.__name__}': {str(e)}")
+                    raise Exception(f"Authorization check failed: {str(e)}")
+
+                # Call the original function
+                if asyncio.iscoroutinefunction(func):
+                    return await func(*args, **kwargs)
+                else:
+                    return func(*args, **kwargs)
+
+            # Register the wrapped function as an MCP tool
+            mcp_server.add_tool(
+                scope_checked_wrapper,
+                **tool_kwargs
+            )
+            return scope_checked_wrapper
+
+        return decorator
+
+    return scoped_tool
+
+
+def get_auth_info(request) -> dict:
+    """Get authentication info from request."""
+    return request.scope.get("auth", {})

examples/example-fastmcp-mcp/src/auth0/middeware.py

@@ -0,0 +1,10 @@
+"""
+FastMCP server instance for Auth0 protected MCP server.
+"""
+
+from mcp.server.fastmcp import FastMCP
+
+mcp = FastMCP(
+    name="Auth0 Protected MCP Server",
+    stateless_http=True,
+)