Content
# Enterprise MCP Template
A **production-ready template** for building enterprise-grade MCP (Model Context Protocol) servers with OAuth 2.0 authentication, based on battle-tested patterns from the Luxsant NetSuite MCP project.
> **What is MCP?** MCP is a standard protocol that lets AI assistants (Claude, Copilot, etc.) call "tools" (functions) on remote servers. Think of it as a standardized API that AI models know how to use.
---
## Table of Contents
- [Quick Start](#quick-start)
- [Architecture Overview](#architecture-overview)
- [Project Structure](#project-structure)
- [How to Create a New MCP Server](#how-to-create-a-new-mcp-server)
- [OAuth 2.0 Authentication Deep Dive](#oauth-20-authentication-deep-dive)
- [Libraries & Dependencies](#libraries--dependencies)
- [Configuration System](#configuration-system)
- [MCP Tools Pattern](#mcp-tools-pattern)
- [API Client Pattern](#api-client-pattern)
- [Token Management](#token-management)
- [Exception Hierarchy](#exception-hierarchy)
- [Deployment Guide](#deployment-guide)
- [Testing](#testing)
- [Best Practices & Gotchas](#best-practices--gotchas)
- [Troubleshooting](#troubleshooting)
---
## Quick Start
### 1. Clone and rename
```bash
git clone https://github.com/YOUR_USER/enterprise-mcp-template.git my-cool-mcp
cd my-cool-mcp
```
### 2. Rename the package
```bash
# Rename the source directory
mv src/my_mcp_server src/my_cool_mcp
# Find and replace all occurrences:
# "my_mcp_server" -> "my_cool_mcp"
# "my-mcp-server" -> "my-cool-mcp"
# "{{PROJECT_NAME}}" -> "My Cool MCP"
# "{{AUTHOR}}" -> "Your Name"
```
### 3. Configure environment
```bash
cp .env.example .env
# Edit .env with your upstream API credentials
```
### 4. Install and run
```bash
# Create virtual environment
python -m venv venv
source venv/bin/activate # Linux/Mac
# or: venv\Scripts\activate # Windows
# Install dependencies
pip install -e ".[dev]"
# Run locally (stdio mode for Claude Desktop)
python -m my_cool_mcp
# Run as HTTP server
python -m my_cool_mcp http
# Run tests
pytest
```
### 5. Deploy
```bash
# Docker build
docker compose up --build
# Or deploy to Azure Web App
az webapp up --name my-cool-mcp --runtime PYTHON:3.11
```
---
## Architecture Overview
```
AI Client (Claude Desktop / VS Code / Custom)
|
| MCP Protocol (stdio / SSE / HTTP)
|
+---v----------------------------------------------+
| MCP Server (server.py) |
| +--------------------------------------------+ |
| | OAuth 2.0 Proxy (OAuthProxy) | |
| | - Handles user authentication | |
| | - Manages proxy tokens | |
| | - Token exchange with upstream | |
| +--------------------------------------------+ |
| +--------------------------------------------+ |
| | MCP Tools (@mcp.tool() functions) | |
| | - create_record() | |
| | - get_record() | |
| | - update_record() | |
| | - delete_record() | |
| | - execute_query() | |
| +--------------------------------------------+ |
| +--------------------------------------------+ |
| | HTTP Routes (/health, /debug/*) | |
| +--------------------------------------------+ |
+--------------------------------------------------+
|
| HTTPS + Bearer Token
|
+---v----------------------------------------------+
| API Client (api_client.py) |
| - HTTP requests with retry logic |
| - Response parsing |
| - Error handling |
+--------------------------------------------------+
|
| REST API calls
|
+---v----------------------------------------------+
| Upstream Service (NetSuite, Salesforce, etc.) |
+--------------------------------------------------+
```
### Module Dependency Flow
```
__main__.py / wsgi.py
-> server.py (main server, tools, OAuth, routes)
-> api_client.py (HTTP client for upstream API)
-> config.py (environment configuration)
-> models.py (Pydantic data models)
-> exceptions.py (error hierarchy)
-> auth.py (token caching & refresh)
-> config.py
-> exceptions.py
-> utils.py (logging, sanitization, helpers)
```
---
## Project Structure
```
enterprise-mcp-template/
|-- .env.example # Environment variable template
|-- .gitignore # Git ignore rules
|-- docker-compose.yml # Docker Compose for local dev
|-- Dockerfile # Multi-stage production Docker build
|-- LICENSE # MIT License
|-- main.py # Root smoke test (not the entry point)
|-- pyproject.toml # Python project configuration
|-- README.md # This file
|-- CLAUDE.md # AI agent instructions
|-- requirements.txt # Production dependencies
|-- startup.sh # Azure Web App startup script
|
|-- docs/ # Documentation
| |-- guide.pdf # PDF version of this guide
|
|-- samples/ # Example payloads
| |-- example_payload.json # Sample API request payload
|
|-- src/
| |-- my_mcp_server/ # Main package (RENAME THIS)
| |-- __init__.py # Package init with lazy imports
| |-- __main__.py # CLI entry point (python -m my_mcp_server)
| |-- server.py # *** MAIN FILE *** MCP server + tools + OAuth
| |-- api_client.py # HTTP client for upstream API
| |-- auth.py # Token management (LRU cache + refresh)
| |-- config.py # Environment-based configuration
| |-- models.py # Pydantic data models
| |-- exceptions.py # Exception hierarchy
| |-- utils.py # Utility functions
| |-- wsgi.py # ASGI entry point for production
| |-- static/
| |-- index.html # Browser-friendly status page
|
|-- tests/ # Test suite
|-- __init__.py
|-- test_config.py # Config tests
|-- test_models.py # Model tests
|-- test_auth.py # Auth/token tests
```
---
## How to Create a New MCP Server
### Step 1: Global Find & Replace
| Find | Replace With | Example |
|------|-------------|---------|
| `my_mcp_server` | Your package name (snake_case) | `salesforce_mcp` |
| `my-mcp-server` | Your package name (kebab-case) | `salesforce-mcp` |
| `{{PROJECT_NAME}}` | Display name | `Salesforce MCP Enterprise` |
| `{{AUTHOR}}` | Your name/org | `El Paso Labs` |
| `UPSTREAM_` | Your service prefix | `SALESFORCE_` |
| `example.com` | Your API domain | `salesforce.com` |
### Step 2: Update OAuth Endpoints (server.py)
In `_build_auth_provider()`, update:
```python
# BEFORE (template):
auth_endpoint = f"https://{account_id}.app.example.com/oauth2/authorize"
token_endpoint = f"https://{account_id}.api.example.com/oauth2/token"
api_scopes = ["api_access"]
# AFTER (example for NetSuite):
auth_endpoint = f"https://{account_id}.app.netsuite.com/app/login/oauth2/authorize.nl"
token_endpoint = f"https://{account_id}.suitetalk.api.netsuite.com/services/rest/auth/oauth2/v1/token"
api_scopes = ["rest_webservices"]
```
### Step 3: Update API URL Patterns (config.py, api_client.py)
In `config.py` `UpstreamAPIConfig.build_api_base_url()`:
```python
# BEFORE:
return f"https://{self.account_id}.api.example.com/v1"
# AFTER (NetSuite):
return f"https://{self.account_id}.suitetalk.api.netsuite.com/services/rest/record/v1"
```
### Step 4: Define Your MCP Tools (server.py)
Replace the generic CRUD tools with domain-specific ones:
```python
@mcp.tool()
async def create_customer(
customer_data: Dict[str, Any],
account_id: Optional[str] = None,
) -> Dict[str, Any]:
"""
Create a new customer in Salesforce.
Args:
customer_data: Customer fields (Name, Email, Phone, etc.)
account_id: Salesforce org ID
Returns:
Structured response with the created customer's ID.
"""
token = _get_oauth_token()
async with _get_client(account_id=account_id) as client:
response = await client.create_record(
access_token=token,
record_type="customer",
payload=customer_data,
)
return _serialize_response(response)
```
### Step 5: Update Models (models.py)
Replace example models with your domain entities:
```python
class CustomerPayload(BaseModel):
name: str = Field(..., description="Customer name")
email: Optional[str] = Field(default=None)
phone: Optional[str] = Field(default=None)
# ... your fields
```
### Step 6: Test and Deploy
```bash
# Run tests
pytest
# Local HTTP test
python -m your_package http
# Visit http://localhost:8000/health
# Docker
docker compose up --build
```
---
## OAuth 2.0 Authentication Deep Dive
### How OAuth Works in This Template
```
1. AI Client connects to MCP server
|
2. MCP server redirects user to upstream login page
| (via OAuthProxy)
|
3. User logs in at upstream service (NetSuite, Salesforce, etc.)
|
4. Upstream redirects back with authorization code
| -> https://your-server.com/auth/callback?code=ABC123
|
5. OAuthProxy exchanges code for access token (server-to-server)
| POST to token endpoint with client_id + client_secret
|
6. OAuthProxy stores the real token, gives client a proxy token
|
7. Client sends proxy token with each MCP tool call
|
8. OAuthProxy looks up real token, passes to tool function
|
9. Tool function uses real token to call upstream API
```
### Critical OAuth Configuration
```python
auth = OAuthProxy(
# WHERE users log in
upstream_authorization_endpoint=auth_endpoint,
# WHERE we exchange codes for tokens
upstream_token_endpoint=token_endpoint,
# OUR app's credentials
upstream_client_id=client_id,
upstream_client_secret=client_secret,
# HOW we verify proxy tokens
token_verifier=token_verifier,
# PUBLIC URL for callbacks
base_url=base_url,
# HOW we send credentials to token endpoint
# "client_secret_basic" = Authorization header (most APIs)
# "client_secret_post" = POST body parameters
token_endpoint_auth_method="client_secret_basic",
# PKCE handling - CRITICAL!
# Set to False if upstream handles PKCE with browser directly
# Set to True if you need to forward PKCE params
forward_pkce=False,
# OAuth scopes
valid_scopes=api_scopes,
# Accept any MCP client redirect URI
allowed_client_redirect_uris=None,
# Sign proxy JWTs with a stable key (set MCP_JWT_SIGNING_KEY in prod!)
jwt_signing_key=jwt_signing_key,
# Skip our consent screen (upstream has its own)
require_authorization_consent=False,
# In-memory client storage (resets on restart - intentional)
client_storage=client_storage,
)
```
### OAuth Gotchas (Lessons Learned)
1. **`forward_pkce=False`**: If your upstream API handles PKCE between itself and the browser, do NOT forward your own PKCE parameters. Your server's `code_verifier` won't match the browser's `code_challenge`, causing `invalid_grant` errors.
2. **`required_scopes` on DebugTokenVerifier**: Without this, clients registered via DCR get `scope=""` and ALL scope requests are rejected with `invalid_scope` before reaching the upstream.
3. **`MCP_JWT_SIGNING_KEY`**: Without a stable key, the OAuthProxy generates a random key on each startup. Container restarts invalidate ALL proxy tokens. Always set in production.
4. **`MemoryStore` for client storage**: Resets on restart. This is actually GOOD - prevents stale client registrations from previous deployments.
5. **`token_endpoint_auth_method`**: Test both "client_secret_basic" and "client_secret_post" using the `/debug/token-test` endpoint. The wrong method gives `invalid_client` instead of `invalid_grant`.
---
## Libraries & Dependencies
| Library | Version | Purpose | Why This Library |
|---------|---------|---------|-----------------|
| **fastmcp** | >=3.0.0b2 | MCP framework | Only production-grade MCP framework. Handles protocol, OAuth, transport. |
| **httpx** | >=0.27.0 | HTTP client | Async HTTP client with connection pooling. Superior to requests for async. |
| **pydantic** | >=2.0.0 | Data validation | Industry standard. Auto-validation, serialization, IDE support. |
| **pydantic-settings** | >=2.1.0 | Settings management | Pydantic extension for env var parsing. |
| **python-dotenv** | >=1.0.0 | .env file loading | Loads .env files for local development. |
| **loguru** | >=0.7.2 | Logging | Enhanced logging (optional, can use stdlib). |
| **gunicorn** | >=21.2.0 | Process manager | Production WSGI/ASGI server. Multi-worker, graceful restarts. |
| **uvicorn** | >=0.27.0 | ASGI server | High-performance async HTTP server. Used as gunicorn worker class. |
### Why FastMCP 3.0?
FastMCP 3.0 is the **only** production-grade MCP framework available. Key features:
- Native `host`/`port` support in `.run()`
- Built-in `OAuthProxy` for OAuth 2.0 authentication
- `DebugTokenVerifier` for development/testing
- `get_access_token()` dependency injection
- Support for three transports: stdio, SSE, HTTP
- `@mcp.tool()` decorator for registering tools
- `@mcp.custom_route()` for HTTP endpoints
- Stateless HTTP mode for cloud load balancers
### Why httpx over requests?
- **Async support**: `httpx.AsyncClient` works natively with `async`/`await`
- **Connection pooling**: Reuses TCP connections automatically
- **Timeout control**: Granular timeout settings per request
- **HTTP/2 support**: Optional HTTP/2 for better performance
- **requests-compatible API**: Easy to migrate from requests
---
## Configuration System
All configuration uses **environment variables** following the 12-Factor App methodology.
### Configuration Hierarchy
```
AppConfig
├── UpstreamAPIConfig (API connection: URL, credentials, timeouts)
├── TokenStoreConfig (Token caching: LRU size, expiry buffer)
└── ServerConfig (Server: name, transport, host, port)
```
### Key Environment Variables
| Variable | Required | Default | Description |
|----------|----------|---------|-------------|
| `UPSTREAM_ACCOUNT_ID` | Yes* | - | Account/tenant identifier |
| `UPSTREAM_OAUTH_CLIENT_ID` | Yes* | - | OAuth client ID |
| `UPSTREAM_OAUTH_CLIENT_SECRET` | Yes* | - | OAuth client secret |
| `MCP_SERVER_BASE_URL` | Yes* | - | Public URL for OAuth callbacks |
| `MCP_TRANSPORT` | No | `stdio` | Transport: stdio/sse/http |
| `MCP_PORT` | No | `8000` | Server port |
| `MCP_HOST` | No | `0.0.0.0` | Server host binding |
| `TOKEN_CACHE_ENABLED` | No | `true` | Enable token LRU cache |
| `TOKEN_EXPIRY_BUFFER_SECS` | No | `300` | Refresh buffer (seconds) |
| `MCP_JWT_SIGNING_KEY` | No | random | Stable JWT key for production |
| `LOG_LEVEL` | No | `INFO` | DEBUG/INFO/WARNING/ERROR |
| `DEBUG` | No | `false` | Enable debug mode |
*Required for OAuth authentication. Server runs without auth if missing.
### Singleton Pattern
```python
from config import get_config, set_config, reset_config
# Normal usage (reads env vars once, caches globally)
config = get_config()
base_url = config.upstream.build_api_base_url()
# Testing (override with custom config)
set_config(AppConfig(server=ServerConfig(port=9999)))
# Reset (force re-read from env)
reset_config()
```
---
## MCP Tools Pattern
Every MCP tool follows this exact pattern:
```python
@mcp.tool()
async def my_tool(
required_param: str,
optional_param: Optional[str] = None,
account_id: Optional[str] = None,
base_url: Optional[str] = None,
) -> Dict[str, Any]:
"""
Tool description (AI reads this to decide when to use the tool).
Args:
required_param: Description for AI
optional_param: Description for AI
account_id: Account ID (if not preconfigured)
base_url: Override API URL
Returns:
Structured response dict with ok, status_code, data, errors.
"""
# 1. Get OAuth token from MCP session
token = _get_oauth_token()
# 2. Create API client (async context manager for cleanup)
async with _get_client(base_url, account_id) as client:
# 3. Call the appropriate client method
response = await client.some_method(
access_token=token,
...
)
# 4. Serialize and return
return _serialize_response(response)
```
### Rules for MCP Tools
1. **Return simple Python objects** (dict, list, str, number). They're serialized to JSON.
2. **Docstrings matter**: AI reads them to decide when/how to use the tool.
3. **Parameter types matter**: FastMCP generates JSON Schema from type hints.
4. **Always use `_serialize_response()`**: Provides consistent response format.
5. **Always use `async with`**: Ensures HTTP client cleanup on error.
6. **Add `account_id` and `base_url` params**: Lets AI clients specify targets dynamically.
---
## API Client Pattern
The API client (`api_client.py`) handles all HTTP communication:
```python
async with APIClient(base_url="https://api.example.com/v1") as client:
# Generic CRUD
response = await client.create_record(token, "customer", payload)
response = await client.get_record(token, "customer", "123")
response = await client.update_record(token, "customer", "123", updates)
response = await client.delete_record(token, "customer", "123")
# Query (if your API supports it)
response = await client.execute_query(token, "SELECT * FROM Customer")
```
### Retry Logic
```
Attempt 1: Immediate
Attempt 2: Wait 0.5s (backoff_factor * 2^0)
Attempt 3: Wait 1.0s (backoff_factor * 2^1)
Attempt 4: Wait 2.0s (backoff_factor * 2^2)
```
Retries on: `429`, `500`, `502`, `503`, `504`, timeouts, connection errors.
Does NOT retry: `400`, `401`, `403`, `404`.
---
## Token Management
### LRU Token Cache
```
Token Cache (max 100 entries)
+---------+------------------+-----------+
| Key | Token | Expires |
+---------+------------------+-----------+
| sha256 | eyJhbG... | 1hr | <- Most recently used
| sha256 | eyJxyz... | 45min |
| sha256 | eyJabc... | 30min |
| ... | ... | ... |
| sha256 | eyJold... | 10min | <- Least recently used (evicted first)
+---------+------------------+-----------+
```
### Token Lifecycle
```
1. User authenticates -> access_token + refresh_token
2. Token cached with SHA-256 key
3. On each API call: check if cached token is still valid
4. If expired (with 5-min buffer): attempt refresh
5. If refresh succeeds: cache new token
6. If refresh fails: user must re-authenticate
```
---
## Exception Hierarchy
```
MCPServerError (catch-all)
├── ConfigurationError
│ ├── MissingConfigurationError
│ └── InvalidConfigurationError
├── AuthenticationError
│ ├── TokenError
│ │ ├── TokenExpiredError
│ │ ├── TokenRefreshError
│ │ └── TokenValidationError
│ └── InvalidCredentialsError
├── APIError
│ ├── ConnectionError
│ ├── TimeoutError
│ ├── RateLimitError
│ ├── NotFoundError
│ ├── ValidationError
│ ├── PermissionError
│ └── ServerError
└── RecordError
├── RecordNotFoundError
├── RecordValidationError
└── DuplicateRecordError
```
Every exception has `to_dict()` for JSON serialization and a machine-readable `code` field.
---
## Deployment Guide
### Local Development (stdio)
```bash
python -m my_mcp_server
# Communicates via stdin/stdout - used by Claude Desktop
```
### Local HTTP Server
```bash
python -m my_mcp_server http
# Available at http://localhost:8000
# Health: http://localhost:8000/health
# MCP: http://localhost:8000/mcp
```
### Docker
```bash
# Build and run
docker compose up --build
# Or standalone
docker build -t my-mcp .
docker run -p 8000:8000 --env-file .env my-mcp
```
### Azure Web App
```bash
# Option 1: Container deployment
az webapp create --name my-mcp --plan my-plan --deployment-container-image-name my-mcp:latest
# Option 2: Source deployment
az webapp up --name my-mcp --runtime PYTHON:3.11
# Set environment variables in Azure Portal:
# Settings -> Configuration -> Application settings
```
Required Azure settings:
- All `UPSTREAM_*` env vars
- `MCP_SERVER_BASE_URL=https://my-mcp.azurewebsites.net`
- `MCP_TRANSPORT=http`
- `MCP_JWT_SIGNING_KEY=<generate with: python -c "import secrets; print(secrets.token_hex(32))">`
### Claude Desktop Configuration
Add to `claude_desktop_config.json`:
```json
{
"mcpServers": {
"my-mcp": {
"url": "https://my-mcp.azurewebsites.net/mcp"
}
}
}
```
---
## Testing
```bash
# Run all tests
pytest
# With coverage
pytest --cov=my_mcp_server --cov-report=html
# Specific test file
pytest tests/test_config.py -v
# Run with verbose output
pytest -v -s
```
### Test Structure
- `test_config.py` - Environment parsing, config validation, singleton
- `test_models.py` - Pydantic model validation, serialization, factories
- `test_auth.py` - Token caching, expiry checking, LRU eviction
---
## Best Practices & Gotchas
### DO
- **Always use `async with` for API clients** - ensures HTTP connection cleanup
- **Always sanitize sensitive data before logging** - use `sanitize_for_logging()`
- **Always return `APIResponse` from tools** - consistent interface for AI clients
- **Set `MCP_JWT_SIGNING_KEY` in production** - prevents token invalidation on restart
- **Log to stderr, not stdout** - stdout is reserved for MCP protocol in stdio mode
- **Use UTC for all timestamps** - `datetime.now(timezone.utc)`
- **Add `account_id` parameter to tools** - lets AI specify targets dynamically
- **Write descriptive docstrings** - AI reads them to decide tool usage
- **Use environment variables for ALL config** - never hardcode credentials
### DON'T
- **Don't log raw tokens** - use `mask_token()` helper
- **Don't hardcode API URLs** - use config.py and env vars
- **Don't catch bare `Exception`** - use the exception hierarchy
- **Don't use `requests` library** - use `httpx` for async support
- **Don't run on stdout in stdio mode** - it corrupts MCP protocol
- **Don't skip the token expiry buffer** - tokens can expire mid-request
- **Don't use `functools.lru_cache` for tokens** - need expiry-aware eviction
- **Don't forward PKCE if upstream handles it** - causes `invalid_grant`
---
## Troubleshooting
### OAuth Issues
1. **Visit `/health`** - shows if OAuth is configured and which env vars are set
2. **Visit `/debug/logs?filter=oauth`** - shows OAuth flow logs
3. **Visit `/debug/token-test`** - tests both auth methods against upstream
4. **Visit `/debug/server-info`** - shows if container restarted (lost OAuth state)
### Common Errors
| Error | Cause | Fix |
|-------|-------|-----|
| `invalid_grant` | PKCE mismatch or expired code | Set `forward_pkce=False` |
| `invalid_client` | Wrong auth method or credentials | Try both auth methods via `/debug/token-test` |
| `invalid_scope` | Missing `required_scopes` on verifier | Add `required_scopes` to `DebugTokenVerifier` |
| `No authenticated session` | User not logged in | Connect via MCP client with OAuth support |
| Token invalidated on restart | No stable JWT key | Set `MCP_JWT_SIGNING_KEY` env var |
### Debug Endpoints
| Endpoint | Purpose |
|----------|---------|
| `GET /health` | Server status, config, OAuth info |
| `GET /debug/logs` | Recent server logs (in-memory buffer) |
| `GET /debug/logs?filter=oauth` | OAuth-specific logs |
| `GET /debug/server-info` | Instance ID, uptime, OAuth state counts |
| `GET /debug/token-test` | Test token exchange with upstream |
---
## License
MIT License - See [LICENSE](LICENSE) for details.
