fastapi-fastmcp
FastAPI + FastMCP OAuth Authentication with Scalekit
Overview
This skill documents the pattern for building production-ready MCP (Model Context Protocol) servers using FastAPI and FastMCP with OAuth 2.1 Bearer token authentication via Scalekit. This approach provides fine-grained control over authentication middleware, token validation, and server behavior compared to using FastMCP's built-in OAuth provider.
When to Use This Pattern
Use this FastAPI + FastMCP integration when you need:
- Custom middleware requirements: Implement rate limiting, request logging, or complex authorization logic beyond basic token validation
- Existing FastAPI applications: Integrate MCP tools into established FastAPI codebases without rewriting
- Advanced authorization: Enforce scope-based access control, multi-tenancy, or custom claims validation
- Full HTTP control: Manage CORS policies, health checks, and multiple endpoints alongside MCP tools
- Production requirements: Add monitoring, metrics, and deployment-specific configurations
Don't use this pattern if FastMCP's built-in OAuth provider meets your needs—the additional FastAPI layer adds complexity.
Core Architecture
Token Validation Flow
MCP Client → FastAPI Server (401 + WWW-Authenticate)
MCP Client → Scalekit (Exchange code for token)
Scalekit → MCP Client (Bearer token)
MCP Client → FastAPI Server (Request + Bearer token)
FastAPI Middleware → Scalekit SDK (Validate token)
FastAPI → MCP Tool Handler → Response
Key Components
- FastAPI Middleware: Custom HTTP middleware that intercepts all requests and validates Bearer tokens before routing to MCP handlers
- Scalekit SDK Integration: Python SDK validates JWT signatures, expiration, issuer, and audience claims
- OAuth Resource Metadata Endpoint: Well-known endpoint (
/.well-known/oauth-protected-resource) for client discovery - FastMCP Tool Registration: Standard MCP tool definitions with
@mcp.tooldecorator - WWW-Authenticate Headers: Standard OAuth 2.1 challenge response directing clients to resource metadata
Implementation Patterns
1. Environment Configuration
Required variables:
SK_ENV_URL: Scalekit environment URL (issuer)SK_CLIENT_ID+SK_CLIENT_SECRET: SDK authentication credentialsEXPECTED_AUDIENCE: The resource identifier that tokens must targetPROTECTED_RESOURCE_METADATA: Complete OAuth discovery metadata JSONPORT: Server listening port (must match registered server URL)
Security:
- Never commit
.envfiles to version control - Use secret managers in production (AWS Secrets Manager, HashiCorp Vault)
- Rotate
SK_CLIENT_SECRETregularly - Validate
EXPECTED_AUDIENCEmatches your server's public URL exactly
2. Middleware Authentication Pattern
@app.middleware("http")
async def auth_middleware(request: Request, call_next):
# Exempt public endpoints
if request.url.path in {"/health", "/.well-known/oauth-protected-resource"}:
return await call_next(request)
# Extract Bearer token
auth_header = request.headers.get("authorization")
if not auth_header or not auth_header.startswith("Bearer "):
return Response(
'{"error": "Missing Bearer token"}',
status_code=401,
headers={"WWW-Authenticate": f'Bearer realm="OAuth", resource_metadata="{RESOURCE_METADATA_URL}"'},
media_type="application/json"
)
token = auth_header.split("Bearer ", 1)[1].strip()
# Validate with Scalekit SDK
options = TokenValidationOptions(
issuer=SK_ENV_URL,
audience=[EXPECTED_AUDIENCE]
)
try:
is_valid = scalekit_client.validate_access_token(token, options=options)
if not is_valid:
raise ValueError("Invalid token")
except Exception:
return Response(
'{"error": "Token validation failed"}',
status_code=401,
headers=WWW_HEADER,
media_type="application/json"
)
return await call_next(request)
Key principles:
- Validate on every request except explicitly public endpoints
- Return 401 with WWW-Authenticate header for missing/invalid tokens
- Use Scalekit SDK for cryptographic validation—never implement custom JWT validation
- Verify issuer and audience claims to prevent token substitution attacks
3. Resource Metadata Endpoint
@app.get("/.well-known/oauth-protected-resource")
async def oauth_metadata():
if not PROTECTED_RESOURCE_METADATA:
return Response(
'{"error": "PROTECTED_RESOURCE_METADATA config missing"}',
status_code=500,
media_type="application/json"
)
metadata = json.loads(PROTECTED_RESOURCE_METADATA)
return Response(
json.dumps(metadata, indent=2),
media_type="application/json"
)
Purpose:
- Enables MCP client discovery of authorization requirements
- Clients fetch this metadata when they receive a 401 response
- Contains authorization server endpoints and supported token types
- Must be publicly accessible (no authentication required)
4. FastMCP Tool Registration
@mcp.tool(
name="greet_user",
description="Greets the user with a personalized message."
)
async def greet_user(name: str, ctx: Context | None = None) -> dict:
return {
"content": [
{
"type": "text",
"text": f"Hi {name}, welcome to Scalekit!"
}
]
}
Tool design:
- Use descriptive names and clear descriptions for AI discoverability
- Accept
ctx: Context | Noneto access authenticated user context in future - Return MCP-compliant response format with
contentarray - Tools automatically inherit authentication from middleware—no per-tool auth needed
5. Application Mounting
mcp_app = mcp.http_app(path="/")
app = FastAPI(lifespan=mcp_app.lifespan)
# Add middleware (CORS, auth, etc.)
app.add_middleware(CORSMiddleware, ...)
# Add custom endpoints
@app.get("/health")
async def health_check():
return {"status": "healthy"}
# Mount MCP at root
app.mount("/", mcp_app)
Layering order:
- Create FastMCP HTTP app
- Create FastAPI app with shared lifespan
- Add FastAPI middleware (CORS, auth)
- Register custom endpoints (health, metadata)
- Mount FastMCP app last
Security Considerations
Token Validation
- Always validate issuer: Prevents tokens from other OAuth servers being accepted
- Always validate audience: Ensures token was issued for your specific resource
- Check expiration: Scalekit SDK automatically validates
expclaim - Verify signature: SDK checks JWT signature against Scalekit's public keys
Common Vulnerabilities to Avoid
- Skipping audience validation: Tokens from other Scalekit resources could be used
- Custom JWT parsing: Use SDK validation—don't implement your own
- Logging tokens: Never log Bearer tokens in middleware or error handlers
- Missing CORS configuration: Can enable cross-origin attacks
- Hardcoded secrets: Use environment variables and secret managers
Production Hardening
- Run behind reverse proxy (Nginx, Caddy) with HTTPS termination
- Use multiple Uvicorn workers:
uvicorn main:app --workers 4 - Implement rate limiting per client/token
- Add request/response logging middleware (without tokens)
- Monitor token validation failures for attack patterns
- Set up health check endpoints for load balancers
Testing Strategy
Local Testing with MCP Inspector
npx @modelcontextprotocol/inspector@latest
- Connect to
http://localhost:3002/ - Inspector handles OAuth flow automatically
- Test each tool with various inputs
- Verify middleware logs show successful validation
Manual Token Testing
# Get token from Scalekit (via OAuth flow or test endpoint)
export TOKEN="<your-access-token>"
# Test authenticated request
curl -H "Authorization: Bearer $TOKEN" \
-H "Content-Type: application/json" \
-d '{"method":"tools/call","params":{"name":"greet_user","arguments":{"name":"Saif"}}}' \
http://localhost:3002/
# Test missing token (should return 401)
curl -v http://localhost:3002/
Integration Tests
import pytest
from fastapi.testclient import TestClient
from main import app
client = TestClient(app)
def test_missing_token():
response = client.post("/")
assert response.status_code == 401
assert "WWW-Authenticate" in response.headers
def test_invalid_token():
response = client.post(
"/",
headers={"Authorization": "Bearer invalid-token"}
)
assert response.status_code == 401
def test_health_check():
response = client.get("/health")
assert response.status_code == 200
assert response.json() == {"status": "healthy"}
Common Pitfalls
1. Mismatched Audience
Symptom: Tokens fail validation with "invalid audience" error
Cause: EXPECTED_AUDIENCE doesn't match the Server URL registered in Scalekit
Fix: Ensure both have identical values including trailing slashes
2. Middleware Order Issues
Symptom: CORS errors or authentication bypassed Cause: Mounting order affects execution sequence Fix: Add middleware before mounting MCP app; mount MCP last
3. Missing Resource Metadata
Symptom: Clients can't discover how to authenticate
Cause: PROTECTED_RESOURCE_METADATA not set or endpoint not working
Fix: Verify metadata JSON is copied correctly from Scalekit dashboard
4. Development vs Production URLs
Symptom: Works locally but fails in production
Cause: Hardcoded localhost URLs in configuration
Fix: Use environment-specific values for EXPECTED_AUDIENCE and RESOURCE_METADATA_URL
5. Token Expiration During Testing
Symptom: Tests pass initially then fail after 1 hour Cause: Access tokens expire (default 3600 seconds) Fix: Refresh tokens before each test run or implement automatic refresh
Extension Patterns
Adding Scope-Based Authorization
@app.middleware("http")
async def auth_middleware(request: Request, call_next):
# ... existing token validation ...
# Decode token to access claims (after validation)
import jwt
decoded = jwt.decode(token, options={"verify_signature": False})
scopes = decoded.get("scope", "").split()
# Attach to request state
request.state.scopes = scopes
request.state.user_id = decoded.get("sub")
return await call_next(request)
@mcp.tool()
async def admin_tool(ctx: Context) -> dict:
# Access request state in tool
if "admin" not in ctx.request_context.state.scopes:
raise PermissionError("Requires admin scope")
# ... tool logic ...
Multi-Tenancy Support
@app.middleware("http")
async def auth_middleware(request: Request, call_next):
# ... validate token ...
decoded = jwt.decode(token, options={"verify_signature": False})
org_id = decoded.get("org_id")
request.state.org_id = org_id
return await call_next(request)
@mcp.tool()
async def get_org_data(ctx: Context) -> dict:
org_id = ctx.request_context.state.org_id
# Fetch data scoped to org_id
Rate Limiting
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded
limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)
@app.post("/")
@limiter.limit("10/minute")
async def mcp_endpoint(request: Request):
# Rate-limited MCP calls
pass
Dependencies
Required Python Packages
mcp>=1.0.0 # MCP protocol implementation
fastapi>=0.104.0 # Web framework
fastmcp>=0.8.0 # FastMCP integration
uvicorn>=0.24.0 # ASGI server
pydantic>=2.5.0 # Data validation
python-dotenv>=1.0.0 # Environment variables
httpx>=0.25.0 # HTTP client
python-jose[cryptography]>=3.3.0 # JWT handling
cryptography>=41.0.0 # Cryptographic operations
scalekit-sdk-python>=2.4.0 # Scalekit authentication
starlette>=0.27.0 # ASGI toolkit
Version Pinning Strategy
- Pin exact versions in production
requirements.txt - Use
>=for development flexibility - Test version upgrades in staging before production
- Monitor security advisories for all dependencies
Complete Working Example
A full production-ready FastAPI + FastMCP server is available in the Scalekit MCP Auth Demos repository:
GitHub Repository: scalekit-inc/mcp-auth-demos/tree/main/greeting-mcp-python
This example includes:
- Complete server implementation with modular architecture
- OAuth 2.1 authentication middleware
- FastMCP tool registration
- CORS configuration and health checks
- Production-ready logging and error handling
Key Files
main.py- Main server entry point with FastAPI appsrc/config/config.py- Environment configurationsrc/lib/auth.py- OAuth discovery endpoint handlersrc/lib/middleware.py- Token validation middlewaresrc/lib/transport.py- MCP transport layer setup
Getting Started
cd greeting-mcp-python
python3 -m venv venv
source venv/bin/activate
pip install -r requirements.txt
python main.py
See README.md for complete setup instructions.
Related Resources
- FastMCP Documentation
- FastAPI Middleware Guide
- OAuth 2.1 Specification
- Scalekit MCP Authentication Docs
- MCP Protocol Specification
- Scalekit MCP Auth Demos
Changelog
- 2026-02-13: Initial skill documentation based on FastAPI + FastMCP quickstart guide
More from scalekit-inc/skills
setup-scalekit
Use when a developer is new to Scalekit and needs guidance on where to start, doesn't know which auth plugin or skill to choose, wants to connect an AI agent or agentic workflow to third-party services (Gmail, Slack, Notion, Google Calendar), needs OAuth or tool-calling auth for agents, wants to add authentication to a project but hasn't chosen an approach yet, or needs to install the Scalekit plugin for their AI coding tool (Claude Code, Codex, Copilot CLI, Cursor, or other agents).
11implementing-scalekit-fsa
Implements Scalekit full-stack authentication (FSA) including sign-up, login, logout, and secure session management using JWT tokens. Use when building or integrating user authentication with the Scalekit SDK across Node.js, Python, Go, or Java — or when the user asks about auth flows, OAuth callbacks, token refresh, or session handling with Scalekit.
4integrating-agent-auth
Integrates Scalekit Agent Auth into a project to handle OAuth flows, token storage, and automatic refresh for third-party services (Gmail, Slack, Notion, Calendar). Use when a user needs to connect to an external service, authorize OAuth access, fetch access or refresh tokens, or execute API calls on behalf of a user.
4sk-actions-custom-provider
Create or review Scalekit custom providers/connectors for proxy-only usage. Use this skill when the task is to gather API docs, infer whether a connector is OAuth, Basic, Bearer, or API Key, determine required tracked fields like domain or version, generate provider JSON, check for existing custom providers, show update diffs, run approved create or update curls, and print resolved delete curls.
3implementing-fsa-logout
Implements a complete logout flow for Scalekit FSA integrations by clearing application session cookies and redirecting the browser to Scalekit’s /oidc/logout endpoint to invalidate the Scalekit session. Use when adding or fixing logout in Node.js, Python, Go, or Java web apps that use Scalekit OIDC.
2implementing-scim-provisioning
Implements SCIM user provisioning using Scalekit's Directory API and webhooks. Use when the user asks to add SCIM, directory sync, user provisioning, deprovisioning, or lifecycle management to their existing application.
2