feat(example): adds fastmcp mcp example#35
Conversation
| def create_scoped_tool_decorator(auth0Mcp): | ||
| """Factory function to create a scoped_tool decorator bound to a MCP server instance.""" |
There was a problem hiding this comment.
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.
There was a problem hiding this comment.
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.
|
Some feedback from CC: Code Review: FastMCP MCP Server with Auth0 IntegrationOverviewThis 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
🔧 Recommended Improvements1. Type Hints & Type Safety ⭐ High PriorityIssue: 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 PriorityIssue: 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 standardsrc/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 PriorityIssue: 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 Prioritysrc/server.py:54 # Current
allow_origins=["*"], # Adjust as needed for production
# Recommended
allow_origins=config.cors_origins, # Use explicit configurationsrc/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 whitespace5. Code Organization 🔸 Medium PriorityIssue: 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 Prioritysrc/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 Prioritysrc/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 handler8. Documentation 🔹 Low PriorityAdd 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 Prioritysrc/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 SummaryMust Fix (High Priority)
Should Fix (Medium Priority)
Nice to Have (Low Priority)
📊 Code Quality Metrics
ConclusionThis 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. |
|
Another round of feedback, but looking very good! SummaryThe 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 Issues1. Add Complete Type Hints
|
Must Fix (High Priority)
Should Fix (Medium Priority)
Nice to Have (Low Priority)
|
|
|
||
| ## 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). |
There was a problem hiding this comment.
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.
There was a problem hiding this comment.
I agree to this, redirecting users to a different repo is a sub-optimal developer experience.
There was a problem hiding this comment.
This is temporary and we are working on a separate doc for setting up tenants for mcp
| poetry run python -m src.server | ||
| ``` | ||
|
|
||
| ## Testing |
There was a problem hiding this comment.
The testing instructions are incomplete. After starting the MCP Inspector, users need to know:
- How to configure the Authorization header in the Inspector
- How to obtain a test access token
- What tools are available and how to call them
There was a problem hiding this comment.
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 | |||
There was a problem hiding this comment.
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=*| def from_env(cls) -> Config: | ||
| return cls( | ||
| auth0_domain=os.environ["AUTH0_DOMAIN"], |
There was a problem hiding this comment.
We can add some validations here. This will prevent cryptic errors later when the server tries to validate tokens.
|
Closing in favour of #45 |
Changes
fastmcpmcp example.References
auth0-auth-js: https://github.com/auth0/auth0-auth-js/tree/main/examplesTesting
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.
Checklist