Skip to content

feat(example): adds fastmcp mcp example#35

Closed
patrickkang wants to merge 14 commits intoauth0:mainfrom
patrickkang:add-example-fastmcp-mcp
Closed

feat(example): adds fastmcp mcp example#35
patrickkang wants to merge 14 commits intoauth0:mainfrom
patrickkang:add-example-fastmcp-mcp

Conversation

@patrickkang
Copy link
Copy Markdown
Contributor

@patrickkang patrickkang commented Sep 24, 2025

Changes

  • Adds a new fastmcp mcp example.

References

Testing

Follow README.md#Auth0 Tenant Setup
to set up your auth0 tenant and use a MCP client like MCP inspector to test your MCP server.

  • This change adds unit test coverage
  • This change adds integration test coverage
  • This change has been tested on the latest version of the platform/language or why not

Checklist

@patrickkang patrickkang marked this pull request as ready for review September 25, 2025 15:51
@patrickkang patrickkang requested a review from a team as a code owner September 25, 2025 15:51
Comment on lines +13 to +14
def create_scoped_tool_decorator(auth0Mcp):
"""Factory function to create a scoped_tool decorator bound to a MCP server instance."""
Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thoughts on something like this instead?

# authz.py
from __future__ import annotations
from functools import wraps
from typing import Iterable
from mcp.server.fastmcp import Context

def require_scopes(required: Iterable[str]):
    needed = tuple(required)
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            # assume ctx is passed (kw or positional)
            ctx: Context | None = kwargs.get("ctx") or next((a for a in args if isinstance(a, Context)), None)
            if ctx is None:
                raise TypeError("ctx: Context is required")

            auth = getattr(ctx.request_context.request.state, "auth", {}) or {}
            if not auth:
                raise PermissionError("Authentication required")

            have = set(auth.get("scopes", []))
            missing = [s for s in needed if s not in have]
            if missing:
                raise PermissionError(f"Missing required scopes: {missing}")

            return await func(*args, **kwargs)
        return wrapper
    return decorator
# tools.py (example)
from __future__ import annotations
import json
from mcp.server.fastmcp import Context, FastMCP
from authz import require_scopes

mcp = FastMCP(name="Auth0 Protected MCP Server", stateless_http=True)

@mcp.tool(name="whoami", title="Who Am I", description="Return authenticated subject", annotations={"readOnlyHint": True})
@require_scopes(["tool:whoami"])
def whoami(ctx: Context) -> str:
    info = getattr(ctx.request_context.request.state, "auth", {}) or {}
    return json.dumps({"user": info.get("extra", {}), "scopes": info.get("scopes", [])}, indent=2)

@mcp.tool(name="greet", title="Greet", description="Greets a user", annotations={"readOnlyHint": True})
@require_scopes(["tool:greet"])
def greet(name: str, ctx: Context) -> str:
    name = (name or "").strip() or "world"
    sub = (getattr(ctx.request_context.request.state, "auth", {}) or {}).get("extra", {}).get("sub")
    return f"Hello, {name}! You are authenticated as {sub}"

And then in HTTP path a Starlette exception handler for PermissionError:

async def permission_error_handler(request, exc: PermissionError):
    # You may want to differentiate missing auth vs insufficient scope.
    # If you split them into two exception classes, map them to 401 vs 403.
    status = 403
    code = "insufficient_scope"
    desc = str(exc) or "Insufficient scope"
    return JSONResponse(
        {"error": code, "error_description": desc},
        status_code=status,
        headers={"WWW-Authenticate": f'Bearer error="{code}", error_description="{desc}"'},
    )

# when building the Starlette app
starlette_app = Starlette(
    debug=...,
    routes=[...],
    lifespan=...,
    exception_handlers={PermissionError: permission_error_handler},
)

And perhaps break down:

class AuthenticationRequired(PermissionError): ...
class InsufficientScope(PermissionError): ...

This tends to be more flexible because it decouples authorization from registration, so you can reuse the same guard outside MCP tools, compose it with other decorators in whatever order you need, and unit test it without spinning up FastMCP. It also keeps metadata concerns out of the guard, which avoids hidden side effects. If you still want scopes to appear in your Protected Resource Metadata, call your existing register_scopes([...]) where you register tools.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks for this suggestion!! Decoupling definitely makes sense and it will be much cleaner and modular. I am going to update the PR to make those changes.

@btiernay
Copy link
Copy Markdown
Contributor

Some feedback from CC:


Code Review: FastMCP MCP Server with Auth0 Integration

Overview

This PR adds an example FastMCP MCP server with Auth0 authentication. The code is functional and well-structured, but would benefit from modernization to align with contemporary Python standards and best practices.

✅ Strengths

  • Clear separation of concerns with dedicated modules for auth, tools, and server
  • Proper async/await patterns throughout
  • Well-designed error hierarchy with custom exceptions
  • OAuth2 RFC compliance with Protected Resource Metadata support
  • Good use of context managers for application lifespan

🔧 Recommended Improvements

1. Type Hints & Type Safety ⭐ High Priority

Issue: Inconsistent and missing type annotations throughout the codebase.

src/auth0/init.py:69

# Current
def exception_handlers(self) -> dict[Union[int, type[Exception]], Callable]:

# Recommended
from collections.abc import Callable
from starlette.responses import Response

def exception_handlers(self) -> dict[int | type[Exception], Callable[[Request, Exception], Response]]:

src/auth0/middleware.py:28

# Current
async def dispatch(self, request: Request, call_next):

# Recommended
from starlette.responses import Response

async def dispatch(self, request: Request, call_next: Callable) -> Response:

src/tools.py:8

# Current
def register_tools(auth0Mcp):

# Recommended
def register_tools(auth0_mcp: Auth0Mcp) -> None:

2. Naming Conventions ⭐ High Priority

Issue: Inconsistent naming violates PEP 8 (mixing camelCase and snake_case).

src/tools.py:1,8

# Current
from json import dumps as jsonDumps
def register_tools(auth0Mcp):

# Recommended
import json
def register_tools(auth0_mcp: Auth0Mcp) -> None:
    # Use json.dumps() directly - it's clear and standard

src/auth0/middleware.py:45,57

# Current
clientId = decoded_and_verified_token.get('client_id') or ...
auth_data["expiresAt"] = ...

# Recommended
client_id = decoded_and_verified_token.get('client_id') or ...
auth_data["expires_at"] = ...

3. Configuration Management ⭐ High Priority

Issue: Environment variables accessed ad-hoc throughout the code.

src/server.py

# Current
load_dotenv()
audience=os.getenv("AUTH0_AUDIENCE")
port=int(os.getenv("PORT", "3001"))

# Recommended
from dataclasses import dataclass
from functools import lru_cache

@dataclass(frozen=True)
class Config:
    """Application configuration loaded from environment variables."""
    auth0_domain: str
    auth0_audience: str
    server_url: str
    port: int = 3001
    debug: bool = True
    cors_origins: list[str] = None

    @classmethod
    def from_env(cls) -> "Config":
        """Load and validate configuration from environment."""
        return cls(
            auth0_domain=os.environ["AUTH0_DOMAIN"],
            auth0_audience=os.environ["AUTH0_AUDIENCE"],
            server_url=os.getenv("MCP_SERVER_URL", "http://localhost:3001"),
            port=int(os.getenv("PORT", "3001")),
            debug=os.getenv("DEBUG", "true").lower() == "true",
            cors_origins=os.getenv("CORS_ORIGINS", "*").split(","),
        )

@lru_cache
def get_config() -> Config:
    load_dotenv()
    return Config.from_env()

4. Security Hardening ⭐ High Priority

src/server.py:54

# Current
allow_origins=["*"], # Adjust as needed for production

# Recommended
allow_origins=config.cors_origins,  # Use explicit configuration

src/auth0/middleware.py:37

# Current
token = auth_header[7:]

# Recommended
if not auth_header.lower().startswith("bearer "):
    raise MalformedAuthorizationRequest("Invalid Authorization header format")
token = auth_header[7:].strip()  # Strip any whitespace

5. Code Organization 🔸 Medium Priority

Issue: Complex inline logic in middleware makes it harder to test and maintain.

src/auth0/middleware.py:50-66

# Current
auth_data = {
    "client_id": clientId,
    "scopes": decoded_and_verified_token.get("scope", "").split()
             if decoded_and_verified_token.get("scope") else []
}
# ... more inline building

# Recommended - extract to helper method
def _build_auth_data(self, token: dict[str, Any]) -> dict[str, Any]:
    """Extract authentication data from verified token."""
    client_id = token.get('client_id') or token.get('azp')
    if not client_id:
        raise VerifyAccessTokenError("Token missing 'client_id' or 'azp' claim")
    
    scopes = token.get("scope", "").split() if token.get("scope") else []
    
    auth_data = {
        "client_id": client_id,
        "scopes": scopes,
    }
    
    if expires_at := token.get('exp'):
        auth_data["expires_at"] = expires_at
    
    # Extract extra claims with dict comprehension
    extra_fields = {'sub', 'azp', 'name', 'email', 'client_id'}
    auth_data["extra"] = {
        field: token[field]
        for field in extra_fields
        if field in token
    }
    
    return auth_data

# Then in dispatch:
request.state.auth = self._build_auth_data(decoded_and_verified_token)

6. Modern Python Idioms 🔸 Medium Priority

src/auth0/middleware.py:61-63

# Current
for field in ['azp', 'name', 'email']:
    if decoded_and_verified_token.get(field):
        extra[field] = decoded_and_verified_token.get(field)

# Recommended - dict comprehension with walrus operator
extra |= {
    field: value
    for field in ['azp', 'name', 'email']
    if (value := decoded_and_verified_token.get(field))
}

src/tools.py:31

# Current
name = (name or "").strip() or "world"

# Recommended - more explicit
name = name.strip() if name else "world"

7. Error Handling 🔸 Medium Priority

src/server.py:20-24

# Current - validation happens inside constructor
auth0_mcp = Auth0Mcp(
    name="Example FastMCP Server",
    audience=os.getenv("AUTH0_AUDIENCE"),
    domain=os.getenv("AUTH0_DOMAIN")
)

# Recommended - fail fast with clear message
if __name__ == "__main__":
    config = get_config()  # Raises KeyError if missing required vars
    auth0_mcp = Auth0Mcp(
        name="Example FastMCP Server",
        audience=config.auth0_audience,
        domain=config.auth0_domain
    )

src/auth0/middleware.py:72-75

# Current - redundant catch and re-raise
except Exception as e:
    logger.error(f"Unexpected error in middleware: {str(e)}")
    raise

# Recommended - remove redundant exception handler or be more specific
except VerifyAccessTokenError as e:
    logger.error("Token verification failed", exc_info=True)
    raise AuthenticationRequired("Invalid token") from e
# Let other exceptions propagate to global handler

8. Documentation 🔹 Low Priority

Add comprehensive docstrings with examples:

class Auth0Mcp:
    """Auth0 integration for FastMCP servers.
    
    Provides authentication middleware, authorization decorators,
    and OAuth2 Protected Resource metadata endpoints.
    
    Args:
        name: Human-readable name for the MCP server
        audience: Auth0 API identifier (OAuth2 audience)
        domain: Auth0 tenant domain (e.g., 'tenant.us.auth0.com')
    
    Raises:
        RuntimeError: If audience or domain are not provided
    
    Example:
        >>> auth0_mcp = Auth0Mcp(
        ...     name="My MCP Server",
        ...     audience="https://api.example.com",
        ...     domain="tenant.us.auth0.com"
        ... )
    """

9. Logging Improvements 🔹 Low Priority

src/auth0/init.py:98

# Current
logger.error(f"Unexpected error in: {exc}", exc_info=exc)

# Recommended - use structured logging
logger.error(
    "Unexpected error occurred",
    exc_info=True,  # Automatically includes exception info
    extra={"request_path": request.url.path}
)

📋 Priority Summary

Must Fix (High Priority)

  1. ✅ Add type hints throughout all modules
  2. ✅ Fix naming conventions (camelCase → snake_case)
  3. ✅ Centralize configuration management with validation
  4. ✅ Make CORS origins explicitly configurable

Should Fix (Medium Priority)

  1. Extract complex logic to testable helper methods
  2. Use modern Python idioms (walrus operator, dict merge with |)
  3. Improve error handling specificity

Nice to Have (Low Priority)

  1. Add comprehensive docstrings with examples
  2. Enhance structured logging
  3. Consider adding mypy type checking to CI/CD

📊 Code Quality Metrics

  • Files changed: 7 new files
  • Type coverage: ~40% (should be 90%+)
  • PEP 8 compliance: ~75% (naming issues)
  • Documentation: Minimal (needs improvement)

Conclusion

This is a solid, functional implementation that demonstrates good architectural decisions. The recommended improvements will enhance maintainability, security, and developer experience while bringing the code up to modern Python standards.

@btiernay
Copy link
Copy Markdown
Contributor

Another round of feedback, but looking very good!

Summary

The code has improved significantly with better naming conventions, modern Python features, and cleaner error handling. The remaining issues are primarily around type annotations and some minor refactoring opportunities.


High Priority Issues

1. Add Complete Type Hints

src/tools.py:8

Issue: Missing type annotations for function parameter and return type.

# Current
def register_tools(auth0Mcp):

# Recommended
from .auth0 import Auth0Mcp

def register_tools(auth0_mcp: Auth0Mcp) -> None:

Also fix: Rename parameter from auth0Mcp (camelCase) to auth0_mcp (snake_case) throughout the function.


src/auth0/__init__.py:71

Issue: Using Union[int, type[Exception]] instead of modern union syntax, and incomplete Callable type.

# Current (line 71)
def exception_handlers(self) -> dict[Union[int, type[Exception]], Callable]:

# Recommended
from starlette.responses import Response

def exception_handlers(self) -> dict[int | type[Exception], Callable[[Request, Exception], Response]]:

src/auth0/__init__.py:80

Issue: Missing return type annotation.

# Current (line 80)
def _auth_error_handler(self, request: Request, exc: Exception):

# Recommended
def _auth_error_handler(self, request: Request, exc: Exception) -> JSONResponse:

src/auth0/__init__.py:96

Issue: Missing space after colon in parameter, missing return type.

# Current (line 96)
def _generic_exception_handler(self, request:Request, exc: Exception):

# Recommended
def _generic_exception_handler(self, request: Request, exc: Exception) -> JSONResponse:

src/auth0/middleware.py:28

Issue: Missing type annotations for call_next parameter and return type.

# Current (line 28)
async def dispatch(self, request: Request, call_next):

# Recommended
from collections.abc import Callable
from starlette.responses import Response

async def dispatch(self, request: Request, call_next: Callable) -> Response:

2. Simplify Code Logic

src/tools.py:31

Issue: Overly complex expression for default value.

# Current (line 31)
name = (name or "").strip() or "world"

# Recommended
name = name.strip() if name else "world"

Rationale: More explicit and easier to understand.


Medium Priority Issues

3. Extract Complex Logic to Helper Method

src/auth0/middleware.py:44-66

Issue: Complex inline logic for building auth data makes the dispatch method harder to read and test.

Recommended: Extract to a helper method.

def _build_auth_data(self, token: dict[str, Any]) -> dict[str, Any]:
    """Build authentication data from verified token.

    Args:
        token: Decoded and verified JWT token

    Returns:
        Dictionary containing client_id, scopes, expires_at, and extra claims

    Raises:
        VerifyAccessTokenError: If token is missing required claims
    """
    client_id = token.get('client_id') or token.get('azp')
    if not client_id:
        raise VerifyAccessTokenError("Token missing 'client_id' or 'azp' claim")

    auth_data = {
        "client_id": client_id,
        "scopes": token.get("scope", "").split() if token.get("scope") else [],
    }

    if expires_at := token.get('exp'):
        auth_data["expires_at"] = expires_at

    # Extract extra claims using dict comprehension
    extra_fields = {'sub', 'client_id', 'azp', 'name', 'email'}
    auth_data["extra"] = {
        field: token[field]
        for field in extra_fields
        if field in token
    }

    return auth_data

Then in dispatch method (line ~66):

# Replace lines 44-66 with:
auth_data = self._build_auth_data(decoded_and_verified_token)
request.state.auth = auth_data

Benefits:

  • Easier to test in isolation
  • Better documentation with docstring
  • Cleaner dispatch method
  • More maintainable

4. Use Modern Dict Comprehension

src/auth0/middleware.py:61-63

Issue: Using a loop when dict comprehension would be cleaner.

# Current (lines 61-63)
for field in ['azp', 'name', 'email']:
    if decoded_and_verified_token.get(field):
        extra[field] = decoded_and_verified_token.get(field)

# Recommended
extra |= {
    field: value
    for field in ['azp', 'name', 'email']
    if (value := decoded_and_verified_token.get(field))
}

Note: This is already covered if you implement the _build_auth_data() helper method above.


Low Priority / Nice to Have

5. Add Missing Import

src/auth0/middleware.py

If implementing the _build_auth_data() helper method, add:

from typing import Any

6. Enhance Docstrings

Consider adding more comprehensive docstrings to key classes:

src/auth0/__init__.py:27

class Auth0Mcp:
    """Auth0 integration for FastMCP servers.

    Provides authentication middleware, authorization decorators,
    and OAuth2 Protected Resource metadata endpoints for MCP servers.

    Args:
        name: Human-readable name for the MCP server
        audience: Auth0 API identifier (OAuth2 audience claim)
        domain: Auth0 tenant domain (e.g., 'tenant.us.auth0.com')

    Raises:
        RuntimeError: If audience or domain are not provided

    Example:
        >>> auth0_mcp = Auth0Mcp(
        ...     name="My MCP Server",
        ...     audience="https://api.example.com",
        ...     domain="tenant.us.auth0.com"
        ... )
        >>> auth0_mcp.register_scopes(["tool:read", "tool:write"])
    """

Summary Checklist

Must Fix (High Priority)

  • Add type hints to register_tools() function (tools.py:8)
  • Fix parameter naming: auth0Mcpauth0_mcp (tools.py:8,12,14)
  • Update exception_handlers() return type (auth0/init.py:71)
  • Add return type to _auth_error_handler() (auth0/init.py:80)
  • Add return type and fix spacing in _generic_exception_handler() (auth0/init.py:96)
  • Add type hints to dispatch() method (middleware.py:28)
  • Simplify name default logic (tools.py:31)

Should Fix (Medium Priority)

  • Extract _build_auth_data() helper method (middleware.py:44-66)
  • Use dict comprehension for extra fields (middleware.py:61-63)

Nice to Have (Low Priority)

  • Add comprehensive class docstrings
  • Add missing typing.Any import if needed

Estimated Effort

  • High Priority Items: ~10 minutes
  • Medium Priority Items: ~10 minutes
  • Total: ~20 minutes to complete all recommendations

What's Already Great ✅

  • Modern Python with from __future__ import annotations
  • PEP 8 naming conventions (mostly fixed)
  • Async/await patterns used correctly
  • Good error handling hierarchy
  • Clean separation of concerns
  • OAuth2 RFC compliance
  • Improved logging practices

@patrickkang
Copy link
Copy Markdown
Contributor Author

Must Fix (High Priority)

  • Add type hints to register_tools() function
  • Fix parameter naming: auth0Mcp → auth0_mcp
  • Update exception_handlers() return type
  • Add return type to _auth_error_handler()
  • Add return type and fix spacing in _generic_exception_handler()
  • Add type hints to dispatch() method
  • Simplify name default logic

Should Fix (Medium Priority)

  • Extract _build_auth_data() helper method
  • Use dict comprehension for extra fields

Nice to Have (Low Priority)

  • Add comprehensive class docstrings
  • Add missing typing.Any import if needed

@patrickkang patrickkang requested a review from btiernay October 2, 2025 14:26

## 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).
Copy link
Copy Markdown
Contributor

@kishore7snehil kishore7snehil Oct 8, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We should add a detailed Auth0 setup section to this README or create a separate docs/AUTH0_SETUP.md file with step-by-step instructions rather than asking the user to visit any other repo.

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I agree to this, redirecting users to a different repo is a sub-optimal developer experience.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is temporary and we are working on a separate doc for setting up tenants for mcp

poetry run python -m src.server
```

## Testing
Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The testing instructions are incomplete. After starting the MCP Inspector, users need to know:

  1. How to configure the Authorization header in the Inspector
  2. How to obtain a test access token
  3. What tools are available and how to call them

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think documenting how to use a specific MCP client like MCP inspector is outside the scope of this example. This example is meant to be client agnostic

@@ -0,0 +1,7 @@
# Auth0 Configuration
Copy link
Copy Markdown
Contributor

@kishore7snehil kishore7snehil Oct 8, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We should add comments to .env.example explaining each variable. Something like this:

# Auth0 tenant domain (e.g., your-tenant.us.auth0.com)
AUTH0_DOMAIN=your-tenant.auth0.com

# Auth0 API Identifier - must match the audience in your Auth0 API configuration
AUTH0_AUDIENCE=https://api.example.com

# URL where this MCP server is accessible (used for OAuth metadata)
MCP_SERVER_URL=http://localhost:3001

# Port the server will listen on
PORT=3001

# Enable debug mode for detailed logging
DEBUG=false

# CORS origins - comma-separated list of allowed origins (* for all)
CORS_ORIGINS=*

Comment on lines +19 to +21
def from_env(cls) -> Config:
return cls(
auth0_domain=os.environ["AUTH0_DOMAIN"],
Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We can add some validations here. This will prevent cryptic errors later when the server tries to validate tokens.

@patrickkang
Copy link
Copy Markdown
Contributor Author

Closing in favour of #45

@patrickkang patrickkang closed this Oct 9, 2025
@kishore7snehil kishore7snehil mentioned this pull request Oct 10, 2025
6 tasks
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

4 participants