Enterprise MCP Template

企业MCP模板

A. 生产就绪模板 用于基于Luxsant NetSuite MCP项目的实战测试模式构建具有OAuth 2.0身份验证的企业级MCP（模型上下文协议）服务器。

什么是MCP？ MCP是一种标准协议，允许AI助手（Claude、Copilot等）调用远程服务器上的“工具”（功能）。将其视为一个标准化的API，人工智能模型知道如何使用。

______________________________________________________________________

目录

• 快速开始
• 架构概述
• 项目结构
• 如何创建新的MCP服务器
• OAuth 2.0身份验证深度学习
• 库和依赖项
• 配置系统
• MCP工具模式
• API客户端模式
• 许可证管理
• 异常层次结构
• 部署指导
• 测试
• 最佳实践与准则
• 故障排除

______________________________________________________________________

快速开始

1.克隆和重命名

git clone https://github.com/YOUR_USER/enterprise-mcp-template.git my-cool-mcp
cd my-cool-mcp

2.重命名包

# 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.配置环境

cp .env.example .env
# Edit .env with your upstream API credentials

4.安装并运行

# 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.部署

# Docker build
docker compose up --build

# Or deploy to Azure Web App
az webapp up --name my-cool-mcp --runtime PYTHON:3.11

______________________________________________________________________

架构概述

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.)   |
+--------------------------------------------------+

模块依赖流

__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)

______________________________________________________________________

项目结构

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

______________________________________________________________________

如何创建新的MCP服务器

步骤1：全局查找和替换

步骤2：更新OAuth端点（server.py）

在 _build_auth_provider()，更新：

# 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"]

步骤3：更新API URL模式（config.py，API_client.py）

在 config.py UpstreamAPIConfig.build_api_base_url():

# 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"

步骤4：定义您的MCP工具（server.py）

用特定于域的CRUD工具替换通用CRUD工具：

@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)

步骤5：更新模型（Models.py）

用域实体替换示例模型：

class CustomerPayload(BaseModel):
    name: str = Field(..., description="Customer name")
    email: Optional[str] = Field(default=None)
    phone: Optional[str] = Field(default=None)
    # ... your fields

步骤6：测试和部署

# Run tests
pytest

# Local HTTP test
python -m your_package http
# Visit http://localhost:8000/health

# Docker
docker compose up --build

______________________________________________________________________

OAuth 2.0身份验证深度学习

OAuth在此模板中的工作原理

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

关键OAuth配置

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（经验教训）

1. forward_pkce=False：如果您的上游API在其自身和浏览器之间处理PKCE，请不要转发您自己的PKCE参数。您的服务器 code_verifier 与浏览器的不匹配 code_challenge，导致 invalid_grant 错误。

1. required_scopes DebugTokenVerifier：如果没有这个，通过DCR注册的客户将获得 scope="" 所有范围请求都被拒绝 invalid_scope 在到达上游之前。

1. MCP_JWT_SIGNING_KEY：如果没有稳定的密钥，OAuthProxy会在每次启动时生成一个随机密钥。容器重新启动会使所有代理令牌无效。始终投入生产。

1. MemoryStore 用于客户端存储：重新启动时重置。这实际上是好的，可以防止以前部署的过时客户端注册。

1. token_endpoint_auth_method：使用以下命令测试“client_secret_basic”和“client_sect_post” /debug/token-test 终点。错误的方法给出 invalid_client 而不是 invalid_grant.

______________________________________________________________________

库和依赖项

为什么选择FastMCP 3.0？

FastMCP 3.0是 仅 生产级MCP框架可用。主要特点：

• 本土的 host/port 支持 .run()
• 内建 OAuthProxy 用于OAuth 2.0身份验证
• DebugTokenVerifier 用于开发/测试
• get_access_token() 依赖注入
• 支持三种传输方式：stdio、SSE、HTTP
• @mcp.tool() 用于注册工具的装饰器
• @mcp.custom_route() 对于HTTP端点
• 云负载均衡器的无状态HTTP模式

为什么httpx超过了请求？

• 异步支持: httpx.AsyncClient 本机使用 async/await
• 连接池：自动重用TCP连接
• 超时控制：每个请求的粒度超时设置
• HTTP/2支持：可选HTTP/2以获得更好的性能
• 请求兼容API：易于从请求迁移

______________________________________________________________________

配置系统

所有配置使用 环境变量 遵循12因素应用程序方法。

配置层次结构

AppConfig
├── UpstreamAPIConfig   (API connection: URL, credentials, timeouts)
├── TokenStoreConfig    (Token caching: LRU size, expiry buffer)
└── ServerConfig        (Server: name, transport, host, port)

关键环境变量

\*OAuth身份验证所必需的。如果缺少身份验证，服务器将在没有身份验证的情况下运行。

单例模式

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工具模式

每个MCP工具都遵循这个确切的模式：

@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)

MCP工具规则

1. 返回简单的Python对象 （字典、列表、字符串、数字）。它们被序列化为JSON。
2. 文档字符串很重要：AI读取它们以决定何时/如何使用该工具。
3. 参数类型很重要：FastMCP根据类型提示生成JSON模式。
4. 始终使用 _serialize_response()：提供一致的响应格式。
5. 始终使用 async with：确保HTTP客户端在出现错误时进行清理。
6. 添加 account_id 和 base_url 参数：允许AI客户端动态指定目标。

______________________________________________________________________

API客户端模式

API客户端(api_client.py)处理所有HTTP通信：

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")

重试逻辑

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)

检索时间： 429, 500, 502, 503, 504、超时、连接错误。 不重试： 400, 401, 403, 404.

______________________________________________________________________

许可证管理

LRU令牌缓存

Token Cache (max 100 entries)
+---------+------------------+-----------+
| Key     | Token            | Expires   |
+---------+------------------+-----------+
| sha256  | eyJhbG...        | 1hr       |  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

______________________________________________________________________

异常层次结构

MCPServerError (catch-all)
├── ConfigurationError
│   ├── MissingConfigurationError
│   └── InvalidConfigurationError
├── AuthenticationError
│   ├── TokenError
│   │   ├── TokenExpiredError
│   │   ├── TokenRefreshError
│   │   └── TokenValidationError
│   └── InvalidCredentialsError
├── APIError
│   ├── ConnectionError
│   ├── TimeoutError
│   ├── RateLimitError
│   ├── NotFoundError
│   ├── ValidationError
│   ├── PermissionError
│   └── ServerError
└── RecordError
    ├── RecordNotFoundError
    ├── RecordValidationError
    └── DuplicateRecordError

每个例外都有 to_dict() 用于JSON序列化和机器可读 code 现场。

______________________________________________________________________

部署指导

地方发展（stdio）

python -m my_mcp_server
# Communicates via stdin/stdout - used by Claude Desktop

本地HTTP服务器

python -m my_mcp_server http
# Available at http://localhost:8000
# Health: http://localhost:8000/health
# MCP: http://localhost:8000/mcp

码头工人

# 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应用程序

# 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

所需的Azure设置：

• 全部 UPSTREAM_* 环境变量
• MCP_SERVER_BASE_URL=https://my-mcp.azurewebsites.net
• MCP_TRANSPORT=http
• MCP_JWT_SIGNING_KEY=

Claude桌面配置

添加 claude_desktop_config.json:

{
  "mcpServers": {
    "my-mcp": {
      "url": "https://my-mcp.azurewebsites.net/mcp"
    }
  }
}

______________________________________________________________________

测试

# 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_config.py -环境解析、配置验证、单例
• test_models.py -Pydantic模型验证、序列化、工厂
• test_auth.py -令牌缓存、到期检查、LRU驱逐

______________________________________________________________________

最佳实践与准则

做

• 始终使用 async with 用于API客户端 -确保HTTP连接清理
• 在记录之前，始终对敏感数据进行消毒 -使用 sanitize_for_logging()
• 始终返回 APIResponse 从工具 -AI客户端的一致界面
• 集 MCP_JWT_SIGNING_KEY 生产中 -防止重新启动时令牌失效
• 记录到stderr，而不是stdout -stdout在stdio模式下保留给MCP协议
• 所有时间戳均使用UTC - datetime.now(timezone.utc)
• 添加 account_id 工具参数 -让AI动态指定目标
• 编写描述性文档字符串 -AI读取它们以决定工具的使用
• 对ALL配置使用环境变量 -从不硬编码凭据

不要

• 不记录原始令牌 -使用 mask_token() 助手
• 不硬编码API URL -使用config.py和env变量
• 不要光着身子抓 Exception -使用异常层次结构
• 不使用 requests 图书馆 -使用 httpx 用于异步支持
• 不要在stdio模式下在stdout上运行 -它破坏了MCP协议
• 不要跳过令牌到期缓冲区 -令牌可能会在请求过程中过期
• 不使用 functools.lru_cache 对于代币 -需要知道到期的驱逐
• 如果上游处理，则不转发PKCE -原因 invalid_grant

______________________________________________________________________

故障排除

OAuth问题

1. 访问 /health -显示是否配置了OAuth以及设置了哪些环境变量
2. 访问 /debug/logs?filter=oauth -显示OAuth流日志
3. 访问 /debug/token-test -针对上游测试两种身份验证方法
4. 访问 /debug/server-info -显示容器是否重新启动（丢失OAuth状态）

常见错误

调试端点

______________________________________________________________________

许可证

MIT许可证-请参阅 许可证 了解详情。