MCP Server Best Practices
<overview> Production-ready MCP servers require attention to security, reliability, performance, and maintainability. This guide covers essential best practices for building robust servers. </overview>
Overview
MCP Server Best Practices
<overview> Production-ready MCP servers require attention to security, reliability, performance, and maintainability. This guide covers essential best practices for building robust servers. </overview>Security
<input_validation> Always Validate Inputs:
// TypeScript - Use Zod for strict validation
const FileReadSchema = z.object({
path: z.string()
.min(1, "Path required")
.max(500, "Path too long")
.refine(
(path) => !path.includes(".."),
"Path traversal not allowed"
)
.refine(
(path) => !path.startsWith("/etc"),
"System directories not allowed"
),
});
async function readFileTool(args: z.infer<typeof FileReadSchema>) {
// Validation happens automatically via Zod
const validated = FileReadSchema.parse(args);
// Additional runtime checks
const fullPath = path.resolve(ALLOWED_DIR, validated.path);
if (!fullPath.startsWith(ALLOWED_DIR)) {
throw new Error("Access denied: Path outside allowed directory");
}
// Safe to proceed
return await fs.readFile(fullPath, "utf-8");
}
# Python - Use Pydantic with validators
from pydantic import BaseModel, Field, field_validator
from pathlib import Path
class FileReadArgs(BaseModel):
path: str = Field(min_length=1, max_length=500)
@field_validator('path')
@classmethod
def validate_path(cls, v: str) -> str:
# Prevent path traversal
if ".." in v:
raise ValueError("Path traversal not allowed")
# Prevent system directories
if v.startswith("/etc") or v.startswith("/sys"):
raise ValueError("System directories not allowed")
return v
async def read_file_tool(args: FileReadArgs) -> TextContent:
# Additional runtime checks
full_path = (Path(ALLOWED_DIR) / args.path).resolve()
if not str(full_path).startswith(ALLOWED_DIR):
raise ValueError("Access denied: Path outside allowed directory")
# Safe to proceed
async with aiofiles.open(full_path, "r") as f:
content = await f.read()
return TextContent(type="text", text=content)
Key principles:
- Validate all inputs with strict schemas
- Check for path traversal attacks (
.., absolute paths) - Whitelist allowed directories/operations
- Validate at schema level AND runtime
- Never trust user input </input_validation>
<secrets_management> Secrets Management:
// TypeScript - Environment variables, never hardcode
dotenv.config();
interface Config {
apiKey: string;
dbPassword: string;
}
function loadConfig(): Config {
const apiKey = process.env.API_KEY;
const dbPassword = process.env.DB_PASSWORD;
if (!apiKey || !dbPassword) {
throw new Error("Missing required environment variables");
}
// NEVER log secrets
console.error("Config loaded successfully");
return { apiKey, dbPassword };
}
const config = loadConfig();
// NEVER return secrets to Claude
@app.list_resources()
async def list_resources() -> list[Resource]:
return [
Resource(
uri="config://server",
name="Server Config",
description="Server configuration (secrets redacted)",
)
]
@app.read_resource()
async def read_resource(uri: str) -> str:
if uri == "config://server":
return json.dumps({
"endpoint": config.api_endpoint,
"timeout": config.timeout,
# NEVER expose secrets:
# "apiKey": config.apiKey, ❌
})
# Python - Use python-dotenv or environment variables
from dataclasses import dataclass
@dataclass
class Config:
api_key: str
db_password: str
@classmethod
def from_env(cls) -> 'Config':
api_key = os.getenv("API_KEY")
db_password = os.getenv("DB_PASSWORD")
if not api_key or not db_password:
raise ValueError("Missing required environment variables")
# NEVER log secrets
print("Config loaded successfully", file=sys.stderr)
return cls(api_key=api_key, db_password=db_password)
config = Config.from_env()
Key principles:
- Use environment variables for secrets
- Never hardcode credentials
- Never log secrets (even in debug mode)
- Never return secrets to Claude
- Use
.envfor development, proper secret management in production - Rotate secrets regularly </secrets_management>
<rate_limiting> Rate Limiting and Resource Protection:
// TypeScript - Simple rate limiter
class RateLimiter {
private requests = new Map<string, number[]>();
check(key: string, limit: number, windowMs: number): boolean {
const now = Date.now();
const requests = this.requests.get(key) || [];
// Remove old requests outside window
const recent = requests.filter((time) => now - time < windowMs);
if (recent.length >= limit) {
return false; // Rate limited
}
recent.push(now);
this.requests.set(key, recent);
return true;
}
}
const limiter = new RateLimiter();
async function callTool(name: string, args: any) {
// Rate limit: 10 requests per minute per tool
if (!limiter.check(name, 10, 60000)) {
throw new Error(`Rate limit exceeded for ${name}`);
}
// Proceed with tool execution
return await executeTool(name, args);
}
# Python - Rate limiter with asyncio
from collections import defaultdict
from datetime import datetime, timedelta
from typing import Dict, List
class RateLimiter:
def __init__(self):
self.requests: Dict[str, List[datetime]] = defaultdict(list)
def check(self, key: str, limit: int, window_seconds: int) -> bool:
now = datetime.now()
cutoff = now - timedelta(seconds=window_seconds)
# Remove old requests
self.requests[key] = [
req_time for req_time in self.requests[key]
if req_time > cutoff
]
if len(self.requests[key]) >= limit:
return False # Rate limited
self.requests[key].append(now)
return True
limiter = RateLimiter()
async def call_tool(name: str, args: dict) -> list[TextContent]:
# Rate limit: 10 requests per minute per tool
if not limiter.check(name, limit=10, window_seconds=60):
raise ValueError(f"Rate limit exceeded for {name}")
# Proceed with tool execution
return await execute_tool(name, args)
</rate_limiting>
<sql_injection> SQL Injection Prevention:
// TypeScript - ALWAYS use parameterized queries
const pool = new Pool({ connectionString: process.env.DATABASE_URL });
// ✅ CORRECT - Parameterized query
async function getUserById(id: string) {
const result = await pool.query(
"SELECT * FROM users WHERE id = $1",
[id]
);
return result.rows[0];
}
// ❌ WRONG - String concatenation (SQL injection!)
async function getUserByIdWrong(id: string) {
const result = await pool.query(
`SELECT * FROM users WHERE id = '${id}'`
);
return result.rows[0];
}
# Python - Use parameterized queries with asyncpg
async def get_user_by_id(user_id: str) -> dict:
conn = await asyncpg.connect(DATABASE_URL)
try:
# ✅ CORRECT - Parameterized query
row = await conn.fetchrow(
"SELECT * FROM users WHERE id = $1",
user_id
)
return dict(row) if row else None
finally:
await conn.close()
# ❌ WRONG - String formatting (SQL injection!)
async def get_user_by_id_wrong(user_id: str) -> dict:
conn = await asyncpg.connect(DATABASE_URL)
try:
row = await conn.fetchrow(
f"SELECT * FROM users WHERE id = '{user_id}'"
)
return dict(row) if row else None
finally:
await conn.close()
</sql_injection>
<authentication> **Authentication & Authorization**:MCP servers may need to authenticate users or protect sensitive operations. Use OAuth 2.1 for production scenarios.
// TypeScript - OAuth Resource Server with FastMCP
class JWTTokenVerifier implements TokenVerifier {
async verifyToken(token: string): Promise {
try {
// Verify JWT token (use a library like jose)
const payload = await verifyJWT(token, process.env.JWT_PUBLIC_KEY);
return {
sub: payload.sub,
scope: payload.scope || "",
exp: payload.exp,
};
} catch (error) {
return null;
}
}
}
const mcp = new FastMCP("Protected API", {
tokenVerifier: new JWTTokenVerifier(),
auth: {
issuerUrl: "https://auth.example.com",
resourceServerUrl: "http://localhost:3000",
requiredScopes: ["api:read"],
},
});
// Tools automatically protected by auth
mcp.tool("get_sensitive_data", async (args, ctx) => {
// Access token info from context
const token = ctx.auth?.accessToken;
if (!token) {
throw new Error("Unauthorized");
}
// Check scopes
if (!token.scope.includes("data:read")) {
throw new Error("Insufficient permissions");
}
return { data: "sensitive information" };
});
# Python - OAuth Resource Server
from mcp.server.fastmcp import FastMCP
from mcp.server.auth.provider import TokenVerifier, AccessToken
from pydantic import AnyHttpUrl
class JWTTokenVerifier(TokenVerifier):
async def verify_token(self, token: str) -> AccessToken | None:
try:
# Verify JWT (use PyJWT)
payload = jwt.decode(
token,
os.environ["JWT_PUBLIC_KEY"],
algorithms=["RS256"]
)
return AccessToken(
sub=payload["sub"],
scope=payload.get("scope", ""),
exp=payload["exp"],
)
except jwt.InvalidTokenError:
return None
mcp = FastMCP(
"Protected API",
token_verifier=JWTTokenVerifier(),
auth=AuthSettings(
issuer_url=AnyHttpUrl("https://auth.example.com"),
resource_server_url=AnyHttpUrl("http://localhost:3000"),
required_scopes=["api:read"],
),
)
@mcp.tool()
async def get_sensitive_data(ctx: Context) -> str:
"""Get sensitive data (requires authentication)."""
# Access token from context
token = ctx.request_context.auth.access_token
if not token:
raise ValueError("Unauthorized")
# Check scopes
if "data:read" not in token.scope:
raise ValueError("Insufficient permissions")
return "sensitive information"
API Key Authentication (simpler, less secure):
// TypeScript - Simple API key auth
const API_KEY = process.env.API_KEY;
server.setRequestHandler("tools/call", async (request) => {
// Check API key in request metadata
const apiKey = request.params._meta?.apiKey;
if (apiKey !== API_KEY) {
throw new Error("Invalid API key");
}
// Proceed with tool execution
return await handleTool(request.params.name, request.params.arguments);
});
# Python - API key in environment variables
API_KEY = os.environ.get("API_KEY")
@mcp.call_tool()
async def call_tool(name: str, arguments: dict, ctx: Context) -> list[TextContent]:
# Extract API key from request metadata
api_key = arguments.get("_api_key")
if api_key != API_KEY:
raise ValueError("Invalid API key")
# Proceed with tool execution
return await execute_tool(name, arguments)
Key principles:
- Use OAuth 2.1 for production (proper token verification, scope checking)
- API keys only for simple/internal use cases
- Never log tokens or API keys
- Verify authentication on every tool call
- Check authorization (scopes/permissions) per operation
- Return 401 for authentication failures, 403 for authorization failures
- Token verification should be fast (cache public keys) </authentication>
Dependency Isolation
<why_it_matters> The Problem: Dependency Conflicts Break Everything
Real story: A pydantic version conflict broke 6 MCP servers simultaneously. One server updated pydantic to 2.10, breaking 5 other servers that required pydantic 2.9. All MCPs failed to start because they shared the same global Python interpreter.
When MCP servers share Python interpreters or global package installations:
- One server's dependencies can break other servers (version conflicts cascade)
- Upgrades become dangerous (updating one server risks breaking others)
- Debugging is impossible (which server caused the conflict?)
- Rollbacks require reinstalling everything (no per-server isolation)
The Solution: Every MCP server needs its own isolated environment </why_it_matters>
<uv_tooling>
Primary Approach: uv (Official MCP Recommendation)
uv is the official tool for Python MCP servers. It automatically creates isolated environments per-project and manages dependencies without global installs.
Development workflow:
# Initialize new MCP server with uv
uv init my-mcp-server
cd my-mcp-server
# Add dependencies
uv add mcp aiohttp
# Development/testing
uv run mcp dev server.py
# Install for Claude Desktop
uv run mcp install server.py
Claude Desktop configuration (automatic isolation):
{
"mcpServers": {
"my-server": {
"command": "uv",
"args": [
"--directory",
"/Users/username/Developer/mcp/my-server",
"run",
"python",
"server.py"
]
}
}
}
The --directory flag tells uv to:
- Use the project's local environment (
.venv/) - Install dependencies from
pyproject.tomlautomatically - Isolate this server from all others
Published servers (for distribution):
# Users install with uvx (no global pollution)
uvx mcp-server-name
Real examples from working configuration:
{
"Workshop": {
"command": "uv",
"args": [
"--directory",
"/Users/lexchristopherson/Developer/workshops/mcp-server",
"run",
"python",
"server.py"
]
},
"finance": {
"command": "uv",
"args": [
"--directory",
"/Users/lexchristopherson/Developer/finance/mcp-server-finance",
"run",
"python",
"server.py"
]
},
"zoom": {
"command": "uv",
"args": [
"--directory",
"/Users/lexchristopherson/Developer/mcp/zoom-mcp",
"run",
"python",
"-m",
"zoom_mcp.server"
],
"env": {
"ZOOM_ACCOUNT_ID": "...",
"ZOOM_CLIENT_ID": "...",
"ZOOM_CLIENT_SECRET": "..."
}
}
}
Each server runs in complete isolation with its own dependencies. </uv_tooling>
<anti_patterns> What NOT To Do
❌ Never use bare python or python3 commands:
{
"my-server": {
"command": "python",
"args": ["/path/to/server.p