Model Context Protocol (MCP) - Comprehensive Guide
The Model Context Protocol (MCP) is an open standard introduced by Anthropic in November 2024 that provides a universal way to connect AI models to external data sources and tools. Think of MCP as the “USB-C port for AI applications” - a standardized interface that enables seamless integration between AI systems and various services.
Table of Contents
- Overview
- Architecture and Components
- Implementation Guide
- Security Considerations
- Performance Optimization
- Real-World Applications
- Best Practices
- Resources and References
Overview
MCP addresses a fundamental challenge in AI development: the fragmentation of data integrations. Before MCP, every new data source required custom implementation, making truly connected AI systems difficult to scale. MCP transforms this “M×N problem” (where M applications need to integrate with N tools) into an “M+N problem” by providing a common protocol.
Key Benefits
- Standardization: One protocol for all AI-to-tool connections
- Modularity: Develop and maintain specialized functionalities in isolated components
- Interoperability: Different LLM applications can share the same servers
- Extensibility: Add new capabilities without modifying client code
- Security: Centralized authentication and authorization patterns
- Simplified Development: Reduces development time and maintenance burden
- Cross-Platform Compatibility: Works across different operating systems and environments
Wide Adoption
Following its announcement, MCP was rapidly adopted by major AI providers including:
- OpenAI and Google DeepMind: Added support for MCP in their platforms
- Replit and Codeium: Integrated MCP for enhanced code assistance
- Block and Apollo: Implemented MCP for enterprise applications
- Zed, Sourcegraph, and JetBrains: Building MCP support into their developer tools
Architecture and Components
MCP follows a client-server architecture with three main components:
1. MCP Hosts
Applications that users interact with, such as:
- Claude Desktop (primary implementation)
- AI-enhanced IDEs (Cursor, VS Code with Copilot)
- Web-based LLM interfaces
- Custom agent applications
Hosts are responsible for:
- Initializing and managing multiple clients
- Providing the user interface layer
- Controlling the lifecycle of MCP connections
- Managing user authentication and permissions
2. MCP Clients
Protocol clients that maintain connections and handle communication:
- Maintain one-to-one stateful connections with servers
- Handle protocol negotiation and capability discovery
- Manage request/response communication
- Exchange information about capabilities via handshake
- Built into host applications
3. MCP Servers
Lightweight programs that bridge between MCP and external systems:
- Expose specific capabilities (tools, resources, prompts)
- Handle authentication and authorization
- Translate between MCP protocol and external APIs
- Can be local (stdio) or remote (HTTP/SSE)
- Examples: GitHub, PostgreSQL, Slack, Google Drive servers
Core Protocol Features
Resources (Application-controlled)
File-like data that can be read by clients:
// Static resource example
{
uri: "file://path/to/document",
name: "Project Documentation",
description: "Main project documentation file",
mimeType: "text/markdown"
}
// Dynamic resource with template
{
uri: "weather://current/{city}",
name: "Current Weather",
description: "Get current weather for any city",
mimeType: "application/json"
}Tools (Model-controlled)
Functions that LLMs can invoke with user approval:
{
name: "create_task",
description: "Create a new task in the project management system",
inputSchema: {
type: "object",
properties: {
title: {
type: "string",
description: "Task title"
},
description: {
type: "string",
description: "Detailed task description"
},
priority: {
type: "string",
enum: ["low", "medium", "high"],
description: "Task priority level"
},
assignee: {
type: "string",
description: "Email of person to assign task to"
}
},
required: ["title", "description"]
}
}Prompts (User-controlled)
Pre-defined templates for common workflows:
{
name: "code_review",
description: "Perform a comprehensive code review",
arguments: [
{
name: "pr_url",
description: "Pull request URL to review",
required: true
},
{
name: "focus_areas",
description: "Specific areas to focus on (security, performance, etc.)",
required: false
}
]
}Transport Mechanisms
MCP supports multiple transport methods for different deployment scenarios:
1. STDIO (Standard Input/Output)
- Use Case: Local integrations, Claude Desktop
- Advantages: Simple, efficient, no network overhead
- Implementation: JSON-RPC over stdin/stdout
- Security: Inherits process security model
2. HTTP with SSE (Server-Sent Events)
- Use Case: Remote server connections
- Advantages: Firewall-friendly, supports streaming
- Implementation: HTTP for requests, SSE for responses
- Security: Supports OAuth 2.1 authentication
3. Streamable HTTP (Recommended for Production)
- Use Case: Production deployments, multi-node architectures
- Features:
- Supports both stateful and stateless modes
- Built-in resumability for connection failures
- Choice of JSON or SSE response formats
- Better scalability for distributed systems
- Security: Full OAuth 2.1 support with token validation
4. WebSockets and UNIX Sockets
- Use Case: Specialized scenarios requiring bidirectional streaming
- Status: Supported but less commonly used
Protocol Foundation
MCP is built on JSON-RPC 2.0 with extensions for:
- Stateful Sessions: Coordinate context exchange between client and server
- Capability-Based Negotiation: Discover supported features during handshake
- Lifecycle Management: Clean initialization and shutdown procedures
- Feature Discovery: Extensible protocol with versioning support
Implementation Guide
Python Implementation with FastMCP
FastMCP is the recommended framework for Python development, providing a high-level API:
Basic Server Setup
from fastmcp import FastMCP
from fastmcp.server import Context
import httpx
from typing import List, Dict, Any
# Create an MCP server
mcp = FastMCP("weather-service")
@mcp.tool()
async def get_weather(
latitude: float,
longitude: float,
ctx: Context
) -> Dict[str, Any]:
"""Get weather forecast for a specific location.
Args:
latitude: The latitude coordinate (-90 to 90)
longitude: The longitude coordinate (-180 to 180)
ctx: MCP context for logging and capabilities
Returns:
Weather forecast data including temperature, conditions, and forecast
"""
# Validate coordinates
if not -90 <= latitude <= 90:
raise ValueError("Latitude must be between -90 and 90")
if not -180 <= longitude <= 180:
raise ValueError("Longitude must be between -180 and 180")
# Use context for logging
ctx.log.info(f"Fetching weather for {latitude}, {longitude}")
# Make API call with proper error handling
async with httpx.AsyncClient() as client:
try:
response = await client.get(
"https://api.weather.com/v1/forecast",
params={"lat": latitude, "lon": longitude},
headers={"API-Key": os.environ["WEATHER_API_KEY"]}
)
response.raise_for_status()
return response.json()
except httpx.RequestError as e:
ctx.log.error(f"Weather API request failed: {e}")
raise RuntimeError(f"Failed to fetch weather data: {e}")
@mcp.resource("weather://alerts/{region}")
async def get_weather_alerts(region: str, ctx: Context) -> str:
"""Provide weather alerts for a specific region."""
ctx.log.info(f"Fetching alerts for region: {region}")
# Validate region format
if not region.isalnum():
raise ValueError("Invalid region format")
# Fetch and return alerts
alerts = await fetch_regional_alerts(region)
return json.dumps(alerts, indent=2)
# Lifespan management for resource cleanup
@mcp.lifespan()
async def lifespan(ctx: Context):
"""Manage server lifecycle - setup and teardown."""
# Startup
ctx.log.info("Weather service starting up...")
await initialize_cache()
await verify_api_keys()
yield # Server runs
# Shutdown
ctx.log.info("Weather service shutting down...")
await cleanup_cache()
await close_connections()
# Run the server
if __name__ == "__main__":
# Set logging level via environment variable
os.environ["FASTMCP_LOG_LEVEL"] = "INFO"
mcp.run()Production Setup with Authentication and OAuth 2.1
from fastmcp import FastMCP
from fastmcp.security import TokenVerifier
from typing import Optional, Dict
import jwt
import aioredis
from datetime import datetime, timedelta
class OAuth21TokenVerifier(TokenVerifier):
"""Production-grade OAuth 2.1 token verifier with caching."""
def __init__(self,
jwks_url: str,
audience: str,
issuer: str,
cache_ttl: int = 300):
self.jwks_url = jwks_url
self.audience = audience
self.issuer = issuer
self.cache_ttl = cache_ttl
self.redis = None
self.jwks_client = None
async def initialize(self):
"""Initialize Redis cache and JWKS client."""
self.redis = await aioredis.create_redis_pool('redis://localhost')
self.jwks_client = jwt.PyJWKClient(self.jwks_url)
async def verify(self, token: str) -> Optional[Dict]:
"""Verify OAuth 2.1 access token."""
# Check cache first
cached = await self.redis.get(f"token:{token}")
if cached:
return json.loads(cached)
try:
# Get signing key from JWKS
signing_key = self.jwks_client.get_signing_key_from_jwt(token)
# Verify and decode token
claims = jwt.decode(
token,
signing_key.key,
algorithms=["RS256"],
audience=self.audience,
issuer=self.issuer,
options={"verify_exp": True}
)
# Additional validation
if not self._validate_claims(claims):
return None
# Cache valid token claims
await self.redis.setex(
f"token:{token}",
self.cache_ttl,
json.dumps(claims)
)
return claims
except jwt.InvalidTokenError as e:
logger.warning(f"Token validation failed: {e}")
return None
def _validate_claims(self, claims: Dict) -> bool:
"""Validate additional token claims."""
# Check required scopes
required_scopes = {"mcp:read", "mcp:write"}
token_scopes = set(claims.get("scope", "").split())
if not required_scopes.issubset(token_scopes):
return False
# Check token binding (DPoP)
if "cnf" in claims:
# Validate proof-of-possession
pass
return True
# Initialize secure MCP server
mcp = FastMCP("secure-data-service")
# Set up token verification
verifier = OAuth21TokenVerifier(
jwks_url="https://auth.example.com/.well-known/jwks.json",
audience="mcp-server",
issuer="https://auth.example.com"
)
@mcp.lifespan()
async def lifespan(ctx: Context):
"""Initialize security components."""
await verifier.initialize()
yield
await verifier.close()
mcp.set_token_verifier(verifier)
@mcp.tool(requires_auth=True)
async def query_sensitive_data(
query: str,
limit: int = 100,
ctx: Context
) -> List[Dict]:
"""Query sensitive data with proper authorization."""
# Access user claims from verified token
user_claims = ctx.auth_claims
user_id = user_claims.get("sub")
scopes = user_claims.get("scope", "").split()
# Log access for audit trail
await audit_log.record({
"action": "query_sensitive_data",
"user_id": user_id,
"query": query,
"timestamp": datetime.utcnow(),
"ip": ctx.request_metadata.get("remote_addr")
})
# Check fine-grained permissions
if "data:sensitive:read" not in scopes:
raise PermissionError("Insufficient permissions for sensitive data")
# Apply row-level security based on user
results = await db.query_with_rls(query, user_id, limit)
return resultsTypeScript Implementation
TypeScript provides the most mature MCP implementation with full type safety:
Basic Server Setup
import { McpServer } from "@modelcontextprotocol/sdk";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/transport/stdio";
import { z } from "zod";
// Initialize server with metadata
const server = new McpServer({
name: "weather-service",
version: "1.0.0",
description: "MCP server for weather data and forecasts"
});
// Define input validation schemas
const WeatherQuerySchema = z.object({
latitude: z.number().min(-90).max(90),
longitude: z.number().min(-180).max(180),
units: z.enum(["metric", "imperial"]).optional().default("metric")
});
// Register a tool with comprehensive error handling
server.registerTool({
name: "get_weather",
description: "Get current weather and forecast for a location",
inputSchema: WeatherQuerySchema,
handler: async (params) => {
try {
// Validate input (Zod handles this automatically)
const { latitude, longitude, units } = params;
// Call weather service
const weather = await weatherService.getWeather({
lat: latitude,
lon: longitude,
units: units
});
// Return structured response
return {
content: [
{
type: "text",
text: formatWeatherReport(weather)
},
{
type: "resource",
resource: {
uri: `weather://forecast/${latitude},${longitude}`,
mimeType: "application/json",
text: JSON.stringify(weather.forecast)
}
}
]
};
} catch (error) {
// Proper error handling with context
server.logger.error("Weather fetch failed", { error, params });
throw new Error(`Failed to fetch weather: ${error.message}`);
}
}
});
// Register a dynamic resource with template
server.registerResource({
uri: "weather://city/{cityName}/current",
name: "City Weather",
description: "Current weather for any city",
handler: async ({ uri }) => {
// Extract city from URI template
const cityName = extractCityFromUri(uri);
// Validate city name
if (!isValidCityName(cityName)) {
throw new Error("Invalid city name");
}
const weatherData = await weatherService.getWeatherByCity(cityName);
return {
contents: [
{
uri: uri,
mimeType: "application/json",
text: JSON.stringify({
city: cityName,
...weatherData,
timestamp: new Date().toISOString()
})
}
]
};
}
});
// Register a prompt for weather alerts
server.registerPrompt({
name: "weather_alert_check",
description: "Check for severe weather alerts in an area",
arguments: [
{
name: "location",
description: "City, state, or coordinates",
required: true
},
{
name: "severity",
description: "Minimum severity level (low, medium, high, extreme)",
required: false,
default: "medium"
}
],
handler: async (args) => {
const alerts = await weatherService.getAlerts(
args.location,
args.severity
);
return {
messages: [
{
role: "user",
content: {
type: "text",
text: `Check weather alerts for ${args.location}`
}
},
{
role: "assistant",
content: {
type: "text",
text: formatAlertReport(alerts)
}
}
]
};
}
});
// Start server with stdio transport
async function main() {
const transport = new StdioServerTransport();
await server.connect(transport);
// Server is now running and listening for connections
console.error("Weather MCP server started");
}
main().catch(console.error);Production Setup with Streamable HTTP Transport
import { McpServer, StreamableHTTPServerTransport } from "@modelcontextprotocol/sdk";
import express from "express";
import helmet from "helmet";
import rateLimit from "express-rate-limit";
import { validateToken } from "./auth";
import { Logger } from "winston";
import prometheus from "prom-client";
// Create Express app with security middleware
const app = express();
app.use(helmet());
app.use(express.json());
// Prometheus metrics
const httpRequestDuration = new prometheus.Histogram({
name: "mcp_http_request_duration_seconds",
help: "Duration of HTTP requests in seconds",
labelNames: ["method", "route", "status"]
});
// Rate limiting
const limiter = rateLimit({
windowMs: 60 * 1000, // 1 minute
max: 100, // 100 requests per minute
standardHeaders: true,
legacyHeaders: false,
});
// Initialize MCP server
const server = new McpServer({
name: "production-service",
version: "2.0.0"
});
// Configure transport with middleware
const transport = new StreamableHTTPServerTransport({
endpoint: "/mcp",
middleware: [
// Authentication middleware
async (req, res, next) => {
const authHeader = req.headers.authorization;
if (!authHeader?.startsWith("Bearer ")) {
return res.status(401).json({
error: "Missing or invalid authorization header"
});
}
const token = authHeader.substring(7);
const claims = await validateToken(token);
if (!claims) {
return res.status(401).json({
error: "Invalid or expired token"
});
}
// Attach claims to request for use in handlers
req.userClaims = claims;
next();
},
// Request logging and metrics
(req, res, next) => {
const start = Date.now();
res.on("finish", () => {
const duration = (Date.now() - start) / 1000;
httpRequestDuration
.labels(req.method, req.route?.path || req.path, res.statusCode)
.observe(duration);
logger.info("MCP request", {
method: req.method,
path: req.path,
status: res.statusCode,
duration,
user: req.userClaims?.sub
});
});
next();
}
],
// Configure streaming options
streamOptions: {
maxChunkSize: 1024 * 64, // 64KB chunks
compressionLevel: 6
}
});
// Advanced tool with context access
server.registerTool({
name: "execute_privileged_operation",
description: "Execute an operation requiring elevated privileges",
inputSchema: z.object({
operation: z.string(),
parameters: z.record(z.any())
}),
handler: async (params, context) => {
// Access user claims from context
const userClaims = context.meta?.userClaims;
if (!userClaims?.roles?.includes("admin")) {
throw new Error("Insufficient privileges");
}
// Log privileged operation
await auditLog.logPrivilegedOperation({
user: userClaims.sub,
operation: params.operation,
parameters: params.parameters,
timestamp: new Date(),
requestId: context.requestId
});
// Execute operation with transaction
const result = await db.transaction(async (trx) => {
return await executeOperation(params.operation, params.parameters, trx);
});
return {
content: [{
type: "text",
text: `Operation completed: ${JSON.stringify(result)}`
}]
};
}
});
// Health check endpoint
app.get("/health", (req, res) => {
const health = {
status: "healthy",
version: "2.0.0",
uptime: process.uptime(),
timestamp: new Date().toISOString()
};
res.json(health);
});
// Metrics endpoint
app.get("/metrics", async (req, res) => {
res.set("Content-Type", prometheus.register.contentType);
res.end(await prometheus.register.metrics());
});
// Mount MCP transport
app.use("/mcp", limiter, transport.handler);
// Error handling
app.use((err, req, res, next) => {
logger.error("Unhandled error", { error: err, path: req.path });
res.status(500).json({ error: "Internal server error" });
});
// Start server
const PORT = process.env.PORT || 3000;
app.listen(PORT, () => {
logger.info(`MCP server running on port ${PORT}`);
});
// Graceful shutdown
process.on("SIGTERM", async () => {
logger.info("SIGTERM received, shutting down gracefully");
await server.close();
process.exit(0);
});Security Considerations
Critical Vulnerabilities and Mitigations
1. Authentication and Authorization Issues
Vulnerability: MCP specification treats authentication as optional, leading to insecure implementations.
Mitigation:
# Always enforce authentication for production
@mcp.tool(requires_auth=True)
async def sensitive_operation(data: str, ctx: Context) -> str:
# Verify user has required permissions
if "write:sensitive" not in ctx.auth_claims.get("scope", "").split():
raise PermissionError("Insufficient permissions")
# Implement operation-specific authorization
resource_owner = await get_resource_owner(data)
if resource_owner != ctx.auth_claims["sub"]:
raise PermissionError("Not authorized for this resource")2. Session Management Vulnerabilities
Vulnerability: Session IDs in URLs can be exposed through logs, referrer headers, or browser history.
Mitigation:
// Use secure session management
class SecureSessionManager {
private sessions = new Map<string, SessionData>();
createSession(userId: string): string {
// Generate cryptographically secure session ID
const sessionId = crypto.randomBytes(32).toString("hex");
// Store session with metadata
this.sessions.set(sessionId, {
userId,
createdAt: Date.now(),
lastAccessed: Date.now(),
ipAddress: getClientIp()
});
// Set expiration
setTimeout(() => {
this.sessions.delete(sessionId);
}, SESSION_TIMEOUT);
return sessionId;
}
validateSession(sessionId: string): SessionData | null {
const session = this.sessions.get(sessionId);
if (!session) return null;
// Check session age
if (Date.now() - session.createdAt > MAX_SESSION_AGE) {
this.sessions.delete(sessionId);
return null;
}
// Update last accessed
session.lastAccessed = Date.now();
return session;
}
}3. Indirect Prompt Injection
Vulnerability: Malicious content in resources can inject commands into AI responses.
Mitigation:
import re
from typing import Any
class ContentSanitizer:
"""Sanitize content to prevent prompt injection attacks."""
# Patterns that might indicate injection attempts
SUSPICIOUS_PATTERNS = [
r"ignore previous instructions",
r"disregard all prior",
r"system prompt",
r"<\|im_start\|>", # Model control tokens
r"\[INST\]", # Instruction markers
r"```system", # Code block abuse
]
@classmethod
def sanitize(cls, content: str) -> str:
"""Remove potential injection attempts from content."""
# Check for suspicious patterns
for pattern in cls.SUSPICIOUS_PATTERNS:
if re.search(pattern, content, re.IGNORECASE):
# Log potential attack
logger.warning(f"Potential injection attempt detected: {pattern}")
# Remove the suspicious content
content = re.sub(pattern, "[REDACTED]", content, flags=re.IGNORECASE)
# Remove zero-width characters often used to hide instructions
content = cls._remove_invisible_chars(content)
# Escape special formatting that might be interpreted as instructions
content = cls._escape_special_formatting(content)
return content
@staticmethod
def _remove_invisible_chars(text: str) -> str:
"""Remove zero-width and other invisible Unicode characters."""
invisible_chars = [
'\u200b', # Zero-width space
'\u200c', # Zero-width non-joiner
'\u200d', # Zero-width joiner
'\ufeff', # Zero-width no-break space
'\u2060', # Word joiner
]
for char in invisible_chars:
text = text.replace(char, '')
return text
@mcp.resource("content://sanitized/{id}")
async def get_sanitized_content(id: str, ctx: Context) -> str:
"""Retrieve content with sanitization applied."""
raw_content = await fetch_content(id)
sanitized = ContentSanitizer.sanitize(raw_content)
return sanitized4. Command Injection Risks
Vulnerability: Improper handling of user input in system commands.
Mitigation:
import subprocess
import shlex
from pathlib import Path
class SecureCommandExecutor:
"""Execute system commands safely."""
ALLOWED_COMMANDS = {
"convert": ["/usr/bin/convert"],
"ffmpeg": ["/usr/bin/ffmpeg"],
"git": ["/usr/bin/git"]
}
ALLOWED_EXTENSIONS = {".jpg", ".png", ".mp4", ".git"}
@classmethod
async def execute(cls, command: str, args: List[str]) -> str:
"""Execute command with strict validation."""
# Validate command is allowed
if command not in cls.ALLOWED_COMMANDS:
raise ValueError(f"Command '{command}' not allowed")
# Get full command path
cmd_path = cls.ALLOWED_COMMANDS[command][0]
# Validate all arguments
safe_args = []
for arg in args:
if cls._is_safe_argument(arg):
safe_args.append(arg)
else:
raise ValueError(f"Unsafe argument: {arg}")
# Build command with subprocess (never use shell=True)
cmd = [cmd_path] + safe_args
try:
result = subprocess.run(
cmd,
capture_output=True,
text=True,
timeout=30, # Prevent hanging
check=True
)
return result.stdout
except subprocess.CalledProcessError as e:
raise RuntimeError(f"Command failed: {e.stderr}")
@staticmethod
def _is_safe_argument(arg: str) -> bool:
"""Validate argument is safe."""
# Check for shell metacharacters
dangerous_chars = set(';|&$`<>(){}[]!\\')
if any(char in arg for char in dangerous_chars):
return False
# Validate file paths
if arg.startswith('/') or arg.startswith('./'):
path = Path(arg)
# Ensure path doesn't escape allowed directories
try:
path.resolve()
# Check extension if it's a file
if path.is_file() and path.suffix not in cls.ALLOWED_EXTENSIONS:
return False
except:
return False
return True
@mcp.tool()
async def convert_image(
input_path: str,
output_format: str,
ctx: Context
) -> str:
"""Safely convert image format."""
# Validate format
if output_format not in ["jpg", "png", "webp"]:
raise ValueError("Invalid output format")
# Build safe output path
input_p = Path(input_path)
output_path = input_p.with_suffix(f".{output_format}")
# Execute conversion safely
result = await SecureCommandExecutor.execute(
"convert",
[str(input_path), str(output_path)]
)
return f"Image converted successfully: {output_path}"Security Best Practices
1. Zero-Trust Security Architecture
from typing import Dict, List, Optional
import asyncio
class ZeroTrustMCPServer:
"""MCP server implementing zero-trust principles."""
def __init__(self):
self.mcp = FastMCP("zero-trust-server")
self.policy_engine = PolicyEngine()
self.risk_analyzer = RiskAnalyzer()
async def authorize_request(
self,
user_claims: Dict,
resource: str,
action: str,
context: Dict
) -> bool:
"""Authorize every request with continuous verification."""
# 1. Verify identity continuously
if not await self.verify_identity(user_claims):
return False
# 2. Assess risk score
risk_score = await self.risk_analyzer.calculate_risk(
user_claims=user_claims,
resource=resource,
action=action,
context=context
)
if risk_score > 0.7: # High risk
# Require additional authentication
if not await self.require_mfa(user_claims["sub"]):
return False
# 3. Apply fine-grained policies
policy_result = await self.policy_engine.evaluate(
subject=user_claims,
resource=resource,
action=action,
environment=context
)
# 4. Log decision for audit
await self.audit_log.record_authorization(
decision=policy_result,
user=user_claims["sub"],
resource=resource,
action=action,
risk_score=risk_score
)
return policy_result.allowed
async def verify_identity(self, claims: Dict) -> bool:
"""Continuously verify user identity."""
# Check token is still valid
if not await self.token_verifier.is_valid(claims["jti"]):
return False
# Verify device binding if present
if "device_id" in claims:
if not await self.verify_device(claims["device_id"]):
return False
# Check for anomalous behavior
if await self.is_anomalous_activity(claims["sub"]):
return False
return True2. Human-in-the-Loop Controls
interface ApprovalRequest {
id: string;
tool: string;
parameters: any;
risk_level: "low" | "medium" | "high" | "critical";
user: string;
timestamp: Date;
}
class HumanInTheLoopManager {
private pendingApprovals = new Map<string, ApprovalRequest>();
async requireApproval(
tool: string,
parameters: any,
context: any
): Promise<boolean> {
// Assess risk level
const riskLevel = this.assessRisk(tool, parameters);
if (riskLevel === "low") {
// Auto-approve low risk operations
return true;
}
// Create approval request
const request: ApprovalRequest = {
id: generateId(),
tool,
parameters,
risk_level: riskLevel,
user: context.user,
timestamp: new Date()
};
// Store pending request
this.pendingApprovals.set(request.id, request);
// Notify approvers based on risk level
if (riskLevel === "critical") {
await this.notifyApprovers(request, ["security-team", "admin"]);
} else {
await this.notifyApprovers(request, [context.user]);
}
// Wait for approval with timeout
const approved = await this.waitForApproval(request.id, 300000); // 5 min timeout
// Log decision
await auditLog.record({
type: "approval_decision",
request,
approved,
approver: approved ? this.getApprover(request.id) : null
});
return approved;
}
private assessRisk(tool: string, parameters: any): string {
// Critical operations
if (tool.includes("delete") || tool.includes("drop")) {
return "critical";
}
// High risk based on scope
if (parameters.affect_all || parameters.force) {
return "high";
}
// Medium risk for modifications
if (tool.includes("update") || tool.includes("modify")) {
return "medium";
}
return "low";
}
}
// Integration with MCP tool
server.registerTool({
name: "delete_records",
handler: async (params, context) => {
// Require human approval for deletions
const approved = await humanInTheLoop.requireApproval(
"delete_records",
params,
context
);
if (!approved) {
throw new Error("Operation cancelled: approval denied");
}
// Proceed with deletion
return await deleteRecords(params);
}
});3. Comprehensive Monitoring and Auditing
import structlog
from prometheus_client import Counter, Histogram, Gauge
import json
from datetime import datetime
class MCPSecurityMonitor:
"""Comprehensive security monitoring for MCP servers."""
def __init__(self):
# Structured logging
self.logger = structlog.get_logger()
# Metrics
self.request_counter = Counter(
'mcp_requests_total',
'Total MCP requests',
['method', 'tool', 'user', 'status']
)
self.auth_failures = Counter(
'mcp_auth_failures_total',
'Authentication failures',
['reason', 'user']
)
self.risk_score_histogram = Histogram(
'mcp_request_risk_score',
'Risk score distribution',
['tool']
)
self.active_sessions = Gauge(
'mcp_active_sessions',
'Number of active sessions'
)
async def log_request(
self,
tool: str,
params: Dict,
user_claims: Dict,
result: Any,
error: Optional[Exception] = None
):
"""Log request with full context for security analysis."""
request_id = generate_request_id()
# Structure log entry
log_entry = {
"request_id": request_id,
"timestamp": datetime.utcnow().isoformat(),
"tool": tool,
"user": {
"id": user_claims.get("sub"),
"roles": user_claims.get("roles", []),
"ip": get_client_ip(),
"user_agent": get_user_agent()
},
"parameters": self._sanitize_params(params),
"result": {
"success": error is None,
"error": str(error) if error else None
},
"security": {
"risk_score": await self.calculate_risk_score(tool, params),
"anomaly_detected": await self.detect_anomaly(user_claims, tool),
"rate_limit_remaining": get_rate_limit_remaining(user_claims["sub"])
}
}
# Log based on outcome
if error:
self.logger.error("mcp_request_failed", **log_entry)
else:
self.logger.info("mcp_request_success", **log_entry)
# Update metrics
status = "error" if error else "success"
self.request_counter.labels(
method="tool_call",
tool=tool,
user=user_claims.get("sub", "anonymous"),
status=status
).inc()
# Alert on suspicious activity
if log_entry["security"]["anomaly_detected"]:
await self.alert_security_team(log_entry)
def _sanitize_params(self, params: Dict) -> Dict:
"""Remove sensitive data from parameters for logging."""
sensitive_keys = {"password", "token", "secret", "key", "credential"}
def sanitize(obj):
if isinstance(obj, dict):
return {
k: "[REDACTED]" if any(s in k.lower() for s in sensitive_keys) else sanitize(v)
for k, v in obj.items()
}
elif isinstance(obj, list):
return [sanitize(item) for item in obj]
else:
return obj
return sanitize(params)
async def detect_anomaly(self, user_claims: Dict, tool: str) -> bool:
"""Detect anomalous behavior using ML/statistical methods."""
user_id = user_claims.get("sub")
# Check unusual access patterns
recent_tools = await self.get_recent_tools(user_id)
if tool not in recent_tools and len(recent_tools) > 10:
return True
# Check access frequency
access_rate = await self.get_access_rate(user_id, tool)
if access_rate > self.get_normal_rate(user_id, tool) * 3:
return True
# Check time-based anomalies
current_hour = datetime.utcnow().hour
usual_hours = await self.get_usual_access_hours(user_id)
if current_hour not in usual_hours:
return True
return FalsePerformance Optimization
Architecture for Scale
1. Connection Pooling and Resource Management
from asyncio import Queue, Semaphore
from contextlib import asynccontextmanager
import aioboto3
import asyncpg
class ResourcePoolManager:
"""Manage connection pools for various resources."""
def __init__(self):
# Database connection pool
self.db_pool = None
# AWS service pools
self.s3_semaphore = Semaphore(50) # Limit concurrent S3 operations
self.boto_session = None
# HTTP client pool
self.http_client = None
# Cache connection pool
self.redis_pool = None
async def initialize(self):
"""Initialize all connection pools."""
# PostgreSQL pool
self.db_pool = await asyncpg.create_pool(
host=os.environ["DB_HOST"],
database=os.environ["DB_NAME"],
user=os.environ["DB_USER"],
password=os.environ["DB_PASSWORD"],
min_size=10,
max_size=50,
max_inactive_connection_lifetime=300,
command_timeout=60
)
# Redis pool for caching
self.redis_pool = await aioredis.create_redis_pool(
'redis://localhost',
minsize=5,
maxsize=20
)
# HTTP client with connection pooling
self.http_client = httpx.AsyncClient(
limits=httpx.Limits(
max_keepalive_connections=100,
max_connections=200,
keepalive_expiry=30
),
timeout=httpx.Timeout(30.0),
http2=True # Enable HTTP/2 for better performance
)
# AWS session
self.boto_session = aioboto3.Session()
@asynccontextmanager
async def get_db_connection(self):
"""Get a database connection from pool."""
async with self.db_pool.acquire() as connection:
yield connection
@asynccontextmanager
async def get_s3_client(self):
"""Get S3 client with concurrency control."""
async with self.s3_semaphore:
async with self.boto_session.client('s3') as s3:
yield s3
# Global resource manager
resource_manager = ResourcePoolManager()
@mcp.lifespan()
async def lifespan(ctx: Context):
"""Initialize and cleanup resources."""
await resource_manager.initialize()
yield
await resource_manager.cleanup()
@mcp.tool()
async def query_database(
query: str,
params: List[Any],
ctx: Context
) -> List[Dict]:
"""Execute database query with connection pooling."""
async with resource_manager.get_db_connection() as conn:
# Use prepared statements for better performance
stmt = await conn.prepare(query)
rows = await stmt.fetch(*params)
return [dict(row) for row in rows]2. Advanced Caching Strategies
import { LRUCache } from "lru-cache";
import Redis from "ioredis";
import { createHash } from "crypto";
class MultiLevelCache {
private l1Cache: LRUCache<string, any>;
private l2Cache: Redis;
private cacheTTL: Map<string, number>;
constructor() {
// L1: In-memory LRU cache
this.l1Cache = new LRUCache({
max: 1000,
ttl: 1000 * 60 * 5, // 5 minutes default
updateAgeOnGet: true,
updateAgeOnHas: true
});
// L2: Redis cache
this.l2Cache = new Redis({
host: process.env.REDIS_HOST,
port: 6379,
maxRetriesPerRequest: 3,
enableOfflineQueue: false
});
// TTL configuration per cache key pattern
this.cacheTTL = new Map([
["user:*", 300], // 5 minutes
["weather:*", 1800], // 30 minutes
["static:*", 86400], // 24 hours
]);
}
async get<T>(key: string): Promise<T | null> {
// Check L1 cache first
const l1Result = this.l1Cache.get(key);
if (l1Result !== undefined) {
return l1Result;
}
// Check L2 cache
const l2Result = await this.l2Cache.get(key);
if (l2Result) {
const parsed = JSON.parse(l2Result);
// Promote to L1 cache
this.l1Cache.set(key, parsed);
return parsed;
}
return null;
}
async set<T>(key: string, value: T, ttl?: number): Promise<void> {
// Determine TTL
const finalTTL = ttl || this.getTTLForKey(key);
// Set in both caches
this.l1Cache.set(key, value, { ttl: finalTTL * 1000 });
await this.l2Cache.setex(key, finalTTL, JSON.stringify(value));
}
async invalidate(pattern: string): Promise<void> {
// Invalidate L1 cache
for (const key of this.l1Cache.keys()) {
if (this.matchesPattern(key, pattern)) {
this.l1Cache.delete(key);
}
}
// Invalidate L2 cache
const keys = await this.l2Cache.keys(pattern);
if (keys.length > 0) {
await this.l2Cache.del(...keys);
}
}
private getTTLForKey(key: string): number {
for (const [pattern, ttl] of this.cacheTTL) {
if (this.matchesPattern(key, pattern)) {
return ttl;
}
}
return 300; // Default 5 minutes
}
private matchesPattern(key: string, pattern: string): boolean {
const regex = new RegExp("^" + pattern.replace("*", ".*") + "$");
return regex.test(key);
}
}
// Cache decorator for MCP tools
function cached(options?: { ttl?: number; keyGenerator?: Function }) {
return function (target: any, propertyKey: string, descriptor: PropertyDescriptor) {
const originalMethod = descriptor.value;
descriptor.value = async function (...args: any[]) {
// Generate cache key
const cacheKey = options?.keyGenerator
? options.keyGenerator(...args)
: `${propertyKey}:${createHash("sha256").update(JSON.stringify(args)).digest("hex")}`;
// Check cache
const cached = await cache.get(cacheKey);
if (cached !== null) {
return cached;
}
// Execute method
const result = await originalMethod.apply(this, args);
// Cache result
await cache.set(cacheKey, result, options?.ttl);
return result;
};
return descriptor;
};
}
// Usage example
class WeatherService {
@cached({ ttl: 1800 }) // Cache for 30 minutes
async getWeather(lat: number, lon: number): Promise<WeatherData> {
return await fetchWeatherAPI(lat, lon);
}
@cached({
ttl: 3600,
keyGenerator: (city: string) => `weather:city:${city.toLowerCase()}`
})
async getWeatherByCity(city: string): Promise<WeatherData> {
return await fetchWeatherByCityAPI(city);
}
}3. Batch Processing and Request Coalescing
from typing import List, Dict, Any, Callable
import asyncio
from collections import defaultdict
from dataclasses import dataclass
@dataclass
class BatchRequest:
key: str
args: tuple
future: asyncio.Future
class BatchProcessor:
"""Batch multiple requests to reduce overhead."""
def __init__(
self,
batch_fn: Callable,
batch_size: int = 100,
batch_timeout: float = 0.1
):
self.batch_fn = batch_fn
self.batch_size = batch_size
self.batch_timeout = batch_timeout
self.pending_requests: List[BatchRequest] = []
self.processing = False
self._lock = asyncio.Lock()
async def process(self, key: str, *args) -> Any:
"""Add request to batch and wait for result."""
future = asyncio.Future()
request = BatchRequest(key=key, args=args, future=future)
async with self._lock:
self.pending_requests.append(request)
# Start batch processing if not already running
if not self.processing:
asyncio.create_task(self._process_batch())
return await future
async def _process_batch(self):
"""Process accumulated requests in batch."""
self.processing = True
# Wait for more requests or timeout
await asyncio.sleep(self.batch_timeout)
async with self._lock:
if not self.pending_requests:
self.processing = False
return
# Take up to batch_size requests
batch = self.pending_requests[:self.batch_size]
self.pending_requests = self.pending_requests[self.batch_size:]
# Continue processing if more requests
if self.pending_requests:
asyncio.create_task(self._process_batch())
else:
self.processing = False
# Group requests by key for efficient processing
grouped = defaultdict(list)
for req in batch:
grouped[req.key].append(req)
# Process batch
try:
results = await self.batch_fn(grouped)
# Distribute results
for key, requests in grouped.items():
if key in results:
for i, req in enumerate(requests):
if isinstance(results[key], list) and i < len(results[key]):
req.future.set_result(results[key][i])
else:
req.future.set_result(results[key])
else:
for req in requests:
req.future.set_exception(
KeyError(f"No result for key: {key}")
)
except Exception as e:
# Set exception for all requests
for requests in grouped.values():
for req in requests:
req.future.set_exception(e)
# Example usage for database queries
async def batch_database_query(grouped_requests: Dict[str, List[BatchRequest]]) -> Dict:
"""Execute multiple database queries in a single round trip."""
results = {}
async with resource_manager.get_db_connection() as conn:
# Build combined query
combined_query = []
params = []
param_count = 0
for table, requests in grouped_requests.items():
ids = [req.args[0] for req in requests]
placeholders = ", ".join([f"${i}" for i in range(param_count + 1, param_count + len(ids) + 1)])
combined_query.append(
f"SELECT * FROM {table} WHERE id IN ({placeholders})"
)
params.extend(ids)
param_count += len(ids)
# Execute combined query
full_query = " UNION ALL ".join(combined_query)
rows = await conn.fetch(full_query, *params)
# Group results by table
for row in rows:
table = row["table_name"] # Assuming table_name column exists
if table not in results:
results[table] = []
results[table].append(dict(row))
return results
# Create batch processor
db_batch_processor = BatchProcessor(
batch_fn=batch_database_query,
batch_size=100,
batch_timeout=0.05 # 50ms
)
@mcp.tool()
async def get_user_data(user_id: str, ctx: Context) -> Dict:
"""Get user data with automatic batching."""
# This will be automatically batched with other concurrent requests
user_data = await db_batch_processor.process("users", user_id)
return user_data4. Streaming and Memory-Efficient Processing
import { Transform, pipeline } from "stream";
import { promisify } from "util";
const pipelineAsync = promisify(pipeline);
class StreamingMCPServer {
// Handle large file processing without loading into memory
async processLargeFile(
filePath: string,
processFunc: (chunk: any) => any
): Promise<void> {
const readStream = fs.createReadStream(filePath, {
highWaterMark: 64 * 1024 // 64KB chunks
});
const processStream = new Transform({
objectMode: true,
transform(chunk, encoding, callback) {
try {
const processed = processFunc(chunk);
callback(null, processed);
} catch (error) {
callback(error);
}
}
});
const writeStream = fs.createWriteStream(filePath + ".processed");
await pipelineAsync(readStream, processStream, writeStream);
}
// Stream results to client
async *streamDatabaseResults(query: string, params: any[]): AsyncGenerator<any> {
const client = await this.pool.connect();
try {
// Use cursor for memory-efficient streaming
await client.query("BEGIN");
await client.query(`DECLARE mycursor CURSOR FOR ${query}`, params);
const batchSize = 1000;
while (true) {
const result = await client.query(`FETCH ${batchSize} FROM mycursor`);
if (result.rows.length === 0) break;
for (const row of result.rows) {
yield row;
}
}
await client.query("CLOSE mycursor");
await client.query("COMMIT");
} finally {
client.release();
}
}
}
// Register streaming tool
server.registerTool({
name: "export_large_dataset",
handler: async (params) => {
const { query, format } = params;
// Create streaming response
return {
content: [{
type: "stream",
stream: async function* () {
let rowCount = 0;
// Stream header
if (format === "csv") {
yield "id,name,value\n";
}
// Stream data
for await (const row of streamingServer.streamDatabaseResults(query, [])) {
rowCount++;
if (format === "csv") {
yield `${row.id},${row.name},${row.value}\n`;
} else {
yield JSON.stringify(row) + "\n";
}
// Progress update every 10k rows
if (rowCount % 10000 === 0) {
yield `[PROGRESS] Processed ${rowCount} rows\n`;
}
}
yield `[COMPLETE] Exported ${rowCount} rows\n`;
}
}]
};
}
});Monitoring and Observability
# Comprehensive monitoring setup with Prometheus and OpenTelemetry
from prometheus_client import Counter, Histogram, Gauge, Info
from opentelemetry import trace
from opentelemetry.exporter.jaeger import JaegerExporter
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
import psutil
import asyncio
class MCPObservability:
"""Complete observability for MCP servers."""
def __init__(self, service_name: str):
self.service_name = service_name
# Metrics
self.request_counter = Counter(
'mcp_requests_total',
'Total number of MCP requests',
['method', 'tool', 'status']
)
self.request_duration = Histogram(
'mcp_request_duration_seconds',
'MCP request duration',
['method', 'tool'],
buckets=[0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0]
)
self.active_requests = Gauge(
'mcp_active_requests',
'Number of active MCP requests',
['method']
)
self.cache_hits = Counter(
'mcp_cache_hits_total',
'Cache hit count',
['cache_type']
)
self.cache_misses = Counter(
'mcp_cache_misses_total',
'Cache miss count',
['cache_type']
)
self.error_counter = Counter(
'mcp_errors_total',
'Total number of errors',
['error_type', 'tool']
)
# System metrics
self.cpu_usage = Gauge('mcp_cpu_usage_percent', 'CPU usage percentage')
self.memory_usage = Gauge('mcp_memory_usage_bytes', 'Memory usage in bytes')
self.connection_pool_size = Gauge(
'mcp_connection_pool_size',
'Database connection pool size',
['pool_name']
)
# Service info
self.service_info = Info('mcp_service', 'MCP service information')
self.service_info.info({
'version': '2.0.0',
'service': service_name,
'environment': os.environ.get('ENVIRONMENT', 'development')
})
# Tracing setup
self._setup_tracing()
# Start background metrics collection
asyncio.create_task(self._collect_system_metrics())
def _setup_tracing(self):
"""Setup distributed tracing with Jaeger."""
tracer_provider = TracerProvider()
trace.set_tracer_provider(tracer_provider)
jaeger_exporter = JaegerExporter(
agent_host_name=os.environ.get('JAEGER_HOST', 'localhost'),
agent_port=int(os.environ.get('JAEGER_PORT', 6831)),
service_name=self.service_name
)
span_processor = BatchSpanProcessor(jaeger_exporter)
tracer_provider.add_span_processor(span_processor)
self.tracer = trace.get_tracer(__name__)
async def _collect_system_metrics(self):
"""Collect system metrics periodically."""
while True:
try:
# CPU usage
self.cpu_usage.set(psutil.cpu_percent(interval=1))
# Memory usage
memory = psutil.virtual_memory()
self.memory_usage.set(memory.used)
# Connection pool metrics
if hasattr(resource_manager, 'db_pool'):
self.connection_pool_size.labels('postgres').set(
resource_manager.db_pool.size
)
await asyncio.sleep(10) # Collect every 10 seconds
except Exception as e:
logger.error(f"Failed to collect system metrics: {e}")
await asyncio.sleep(60)
def track_request(self, method: str, tool: str):
"""Decorator to track MCP requests."""
def decorator(func):
async def wrapper(*args, **kwargs):
# Start timer
start_time = asyncio.get_event_loop().time()
# Create trace span
with self.tracer.start_as_current_span(
f"mcp.{method}.{tool}",
attributes={
"mcp.method": method,
"mcp.tool": tool,
"mcp.service": self.service_name
}
) as span:
# Track active requests
self.active_requests.labels(method).inc()
try:
# Execute function
result = await func(*args, **kwargs)
# Record success
self.request_counter.labels(method, tool, 'success').inc()
span.set_status(trace.Status(trace.StatusCode.OK))
return result
except Exception as e:
# Record error
self.request_counter.labels(method, tool, 'error').inc()
self.error_counter.labels(type(e).__name__, tool).inc()
# Add error to span
span.record_exception(e)
span.set_status(
trace.Status(trace.StatusCode.ERROR, str(e))
)
raise
finally:
# Record duration
duration = asyncio.get_event_loop().time() - start_time
self.request_duration.labels(method, tool).observe(duration)
# Update active requests
self.active_requests.labels(method).dec()
# Add duration to span
span.set_attribute("mcp.duration_seconds", duration)
return wrapper
return decorator
# Global observability instance
observability = MCPObservability("my-mcp-service")
# Usage example
@mcp.tool()
@observability.track_request("tool", "get_weather")
async def get_weather(lat: float, lon: float, ctx: Context) -> Dict:
"""Get weather with full observability."""
# Check cache
cache_key = f"weather:{lat}:{lon}"
cached = await cache.get(cache_key)
if cached:
observability.cache_hits.labels("weather").inc()
return cached
observability.cache_misses.labels("weather").inc()
# Fetch weather
weather = await fetch_weather_api(lat, lon)
# Cache result
await cache.set(cache_key, weather, ttl=1800)
return weatherReal-World Applications
Financial Services Integration
# Production-ready trading system MCP server
from decimal import Decimal
import uuid
from typing import Optional, List, Dict
from enum import Enum
class OrderType(Enum):
MARKET = "market"
LIMIT = "limit"
STOP = "stop"
STOP_LIMIT = "stop_limit"
class TradingMCPServer:
"""MCP server for financial trading operations."""
def __init__(self):
self.mcp = FastMCP("trading-system")
self.risk_engine = RiskEngine()
self.compliance = ComplianceEngine()
self.market_data = MarketDataService()
@mcp.tool(requires_auth=True)
async def execute_trade(
self,
symbol: str,
quantity: int,
order_type: OrderType,
price: Optional[Decimal],
stop_price: Optional[Decimal],
ctx: Context
) -> Dict:
"""Execute a trading order with comprehensive risk and compliance checks."""
order_id = str(uuid.uuid4())
user_id = ctx.auth_claims["sub"]
# Start distributed trace
with ctx.tracer.start_span("execute_trade") as span:
span.set_attribute("order.id", order_id)
span.set_attribute("order.symbol", symbol)
span.set_attribute("order.quantity", quantity)
try:
# 1. Pre-trade compliance checks
compliance_result = await self.compliance.check_pre_trade(
user_id=user_id,
symbol=symbol,
quantity=quantity,
order_type=order_type
)
if not compliance_result.approved:
raise ComplianceError(f"Trade rejected: {compliance_result.reason}")
# 2. Get current market data
market_data = await self.market_data.get_quote(symbol)
# 3. Risk assessment
risk_assessment = await self.risk_engine.assess_order(
user_id=user_id,
symbol=symbol,
quantity=quantity,
order_type=order_type,
price=price or market_data.last_price,
account_value=await self.get_account_value(user_id)
)
if risk_assessment.risk_score > 0.8:
# High risk - require additional confirmation
if not await self.require_risk_acknowledgment(
user_id, risk_assessment
):
raise RiskError("Trade cancelled due to high risk")
# 4. Calculate order value and fees
order_value = self.calculate_order_value(
quantity, price or market_data.last_price
)
fees = self.calculate_fees(order_value, order_type)
# 5. Check account balance
if not await self.check_buying_power(
user_id, order_value + fees
):
raise InsufficientFundsError("Insufficient buying power")
# 6. Place order with exchange
exchange_order = await self.place_exchange_order(
order_id=order_id,
symbol=symbol,
quantity=quantity,
order_type=order_type,
price=price,
stop_price=stop_price
)
# 7. Record order in database
await self.record_order(
order_id=order_id,
user_id=user_id,
symbol=symbol,
quantity=quantity,
order_type=order_type,
price=price,
stop_price=stop_price,
exchange_order_id=exchange_order.id,
fees=fees
)
# 8. Send real-time notification
await self.send_order_notification(user_id, order_id, "placed")
# 9. Audit trail
await self.audit_log.record_trade(
order_id=order_id,
user_id=user_id,
action="place_order",
details={
"symbol": symbol,
"quantity": quantity,
"order_type": order_type.value,
"price": str(price) if price else None,
"risk_score": risk_assessment.risk_score,
"compliance_check": compliance_result.check_id
}
)
return {
"order_id": order_id,
"status": "placed",
"symbol": symbol,
"quantity": quantity,
"order_type": order_type.value,
"price": str(price) if price else "market",
"estimated_value": str(order_value),
"fees": str(fees),
"timestamp": datetime.utcnow().isoformat()
}
except Exception as e:
span.record_exception(e)
span.set_status(trace.Status(trace.StatusCode.ERROR))
# Log failure
await self.audit_log.record_trade_failure(
order_id=order_id,
user_id=user_id,
error=str(e)
)
raise
@mcp.tool(requires_auth=True)
async def get_portfolio_analysis(
self,
include_recommendations: bool = False,
ctx: Context
) -> Dict:
"""Provide comprehensive portfolio analysis with AI insights."""
user_id = ctx.auth_claims["sub"]
# Get portfolio data
positions = await self.get_user_positions(user_id)
account_value = await self.get_account_value(user_id)
# Calculate metrics
analysis = {
"account_value": str(account_value),
"positions": len(positions),
"asset_allocation": await self.calculate_asset_allocation(positions),
"risk_metrics": await self.calculate_risk_metrics(positions),
"performance": await self.calculate_performance(user_id),
"diversification_score": await self.calculate_diversification(positions)
}
if include_recommendations:
# Generate AI-powered recommendations
recommendations = await self.generate_recommendations(
positions=positions,
risk_profile=await self.get_risk_profile(user_id),
market_conditions=await self.market_data.get_market_conditions()
)
analysis["recommendations"] = recommendations
return analysisHealthcare Data Platform
# HIPAA-compliant healthcare MCP server
from cryptography.fernet import Fernet
import hashlib
from typing import List, Dict, Optional
class HealthcareMCPServer:
"""MCP server for healthcare data with HIPAA compliance."""
def __init__(self):
self.mcp = FastMCP("healthcare-platform")
self.encryption_key = Fernet(os.environ["ENCRYPTION_KEY"])
self.consent_manager = ConsentManager()
self.phi_auditor = PHIAuditor()
@mcp.resource("patient://{patient_id}/records/{record_type}")
async def get_patient_records(
self,
patient_id: str,
record_type: str,
ctx: Context
) -> Dict:
"""Retrieve patient records with full HIPAA compliance."""
provider_id = ctx.auth_claims.get("provider_id")
provider_type = ctx.auth_claims.get("provider_type")
# Validate provider credentials
if not await self.validate_provider(provider_id, provider_type):
raise AuthorizationError("Invalid healthcare provider credentials")
# Check patient consent
consent = await self.consent_manager.check_consent(
patient_id=patient_id,
provider_id=provider_id,
record_type=record_type,
purpose="treatment" # Purpose of use
)
if not consent.is_granted:
# Check for emergency override
if not ctx.auth_claims.get("emergency_override"):
raise AuthorizationError(
f"Patient consent required for {record_type} access"
)
# Log emergency access
await self.phi_auditor.log_emergency_access(
patient_id=patient_id,
provider_id=provider_id,
record_type=record_type,
justification=ctx.auth_claims.get("emergency_justification")
)
# Retrieve encrypted records
encrypted_records = await self.db.get_patient_records(
patient_id=patient_id,
record_type=record_type
)
# Decrypt records
records = []
for encrypted_record in encrypted_records:
decrypted = self.decrypt_phi(encrypted_record)
# Apply minimum necessary principle
filtered = self.apply_minimum_necessary(
decrypted,
provider_type,
consent.allowed_fields
)
# De-identify if required
if consent.requires_deidentification:
filtered = self.deidentify_record(filtered)
records.append(filtered)
# Comprehensive audit log
await self.phi_auditor.log_access(
patient_id=patient_id,
provider_id=provider_id,
record_type=record_type,
records_accessed=len(records),
fields_accessed=self.get_accessed_fields(records),
purpose="treatment",
consent_id=consent.id,
ip_address=ctx.request_metadata.get("ip"),
timestamp=datetime.utcnow()
)
return {
"patient_id": self.hash_patient_id(patient_id), # Never expose raw IDs
"record_type": record_type,
"records": records,
"consent_id": consent.id,
"access_timestamp": datetime.utcnow().isoformat()
}
@mcp.tool(requires_auth=True)
async def create_clinical_note(
self,
patient_id: str,
note_type: str,
content: str,
diagnoses: List[str],
procedures: List[str],
ctx: Context
) -> Dict:
"""Create clinical note with NLP enhancement and compliance."""
provider_id = ctx.auth_claims["provider_id"]
# Validate clinical privileges
if not await self.check_clinical_privileges(
provider_id, "create_notes"
):
raise AuthorizationError("Insufficient clinical privileges")
# NLP processing for clinical insights
nlp_results = await self.process_clinical_text(content)
# Check for potential errors or inconsistencies
if nlp_results.has_warnings:
# Require confirmation for potential issues
if not await self.confirm_warnings(
nlp_results.warnings,
ctx
):
return {
"status": "cancelled",
"warnings": nlp_results.warnings
}
# Encrypt PHI
encrypted_content = self.encrypt_phi({
"content": content,
"diagnoses": diagnoses,
"procedures": procedures,
"nlp_insights": nlp_results.insights
})
# Create note with transaction
note_id = await self.db.transaction(async (tx) => {
# Create note
note_id = await tx.create_clinical_note(
patient_id=patient_id,
provider_id=provider_id,
note_type=note_type,
encrypted_content=encrypted_content,
metadata={
"created_at": datetime.utcnow(),
"version": "1.0",
"nlp_version": nlp_results.version
}
)
# Update patient summary
await tx.update_patient_summary(
patient_id=patient_id,
diagnoses=diagnoses,
procedures=procedures
)
# Create audit entry
await tx.create_audit_entry({
"action": "create_clinical_note",
"patient_id": patient_id,
"provider_id": provider_id,
"note_id": note_id,
"note_type": note_type
})
return note_id
})
# Send HL7 message to EHR
await self.send_hl7_message(
message_type="MDM^T02", # Medical document management
note_id=note_id,
patient_id=patient_id,
provider_id=provider_id
)
return {
"note_id": note_id,
"status": "created",
"nlp_insights": nlp_results.insights,
"timestamp": datetime.utcnow().isoformat()
}
def decrypt_phi(self, encrypted_data: bytes) -> Dict:
"""Decrypt PHI with audit trail."""
try:
decrypted = self.encryption_key.decrypt(encrypted_data)
return json.loads(decrypted)
except Exception as e:
self.phi_auditor.log_decryption_failure(e)
raise
def encrypt_phi(self, data: Dict) -> bytes:
"""Encrypt PHI for storage."""
json_data = json.dumps(data)
return self.encryption_key.encrypt(json_data.encode())
def hash_patient_id(self, patient_id: str) -> str:
"""Create consistent hash of patient ID for logging."""
return hashlib.sha256(
f"{patient_id}{os.environ['PATIENT_SALT']}".encode()
).hexdigest()[:16]Manufacturing IoT Platform
# Industrial IoT MCP server with predictive maintenance
from scipy import signal
import numpy as np
from sklearn.ensemble import IsolationForest
import pandas as pd
class ManufacturingMCPServer:
"""MCP server for industrial IoT and predictive maintenance."""
def __init__(self):
self.mcp = FastMCP("manufacturing-platform")
self.telemetry_buffer = TelemetryBuffer()
self.ml_models = MLModelManager()
self.alert_manager = AlertManager()
self.digital_twin = DigitalTwinEngine()
@mcp.tool()
async def monitor_equipment_health(
self,
equipment_id: str,
window_minutes: int = 5,
ctx: Context
) -> Dict:
"""Real-time equipment health monitoring with ML-based anomaly detection."""
# Retrieve recent telemetry data
telemetry = await self.telemetry_buffer.get_recent_data(
equipment_id=equipment_id,
window=timedelta(minutes=window_minutes)
)
if not telemetry:
return {
"status": "no_data",
"equipment_id": equipment_id,
"message": "No telemetry data available"
}
# Convert to dataframe for analysis
df = pd.DataFrame(telemetry)
# Extract key metrics
metrics = {
"temperature": df["temperature"].mean(),
"vibration": df["vibration_rms"].mean(),
"pressure": df["pressure"].mean(),
"rpm": df["rpm"].mean(),
"power_consumption": df["power_kw"].mean()
}
# Perform spectral analysis on vibration data
vibration_spectrum = await self.analyze_vibration_spectrum(
df["vibration_raw"].values
)
# Run anomaly detection
anomaly_scores = await self.detect_anomalies(df)
# Check against threshold rules
alerts = await self.check_thresholds(equipment_id, metrics)
# Predict remaining useful life (RUL)
rul_prediction = await self.ml_models.predict_rul(
equipment_id=equipment_id,
features=self.extract_features(df),
historical_data=await self.get_historical_features(equipment_id)
)
# Update digital twin
await self.digital_twin.update_state(
equipment_id=equipment_id,
metrics=metrics,
anomaly_scores=anomaly_scores
)
# Generate health score
health_score = self.calculate_health_score(
metrics=metrics,
anomaly_scores=anomaly_scores,
rul_days=rul_prediction["days_remaining"]
)
# Prepare response
response = {
"equipment_id": equipment_id,
"timestamp": datetime.utcnow().isoformat(),
"health_score": health_score,
"status": self.get_health_status(health_score),
"metrics": metrics,
"anomalies": {
"detected": any(score > 0.7 for score in anomaly_scores.values()),
"scores": anomaly_scores,
"components": self.identify_anomalous_components(anomaly_scores)
},
"vibration_analysis": {
"dominant_frequency": vibration_spectrum["dominant_freq"],
"harmonics": vibration_spectrum["harmonics"],
"bearing_fault_probability": vibration_spectrum["bearing_fault_prob"]
},
"predictive_maintenance": {
"rul_days": rul_prediction["days_remaining"],
"confidence": rul_prediction["confidence"],
"recommended_action": rul_prediction["recommended_action"],
"next_maintenance": rul_prediction["next_maintenance_date"]
},
"alerts": alerts
}
# Trigger alerts if necessary
if health_score < 0.3 or alerts:
await self.trigger_maintenance_workflow(
equipment_id=equipment_id,
severity=self.calculate_severity(health_score, alerts),
details=response
)
return response
@mcp.tool()
async def optimize_production_line(
self,
line_id: str,
optimization_goal: str,
constraints: Dict,
ctx: Context
) -> Dict:
"""AI-driven production line optimization."""
# Get current line configuration
current_config = await self.get_line_configuration(line_id)
# Collect performance data
performance_data = await self.collect_line_performance(
line_id=line_id,
duration=timedelta(hours=24)
)
# Run optimization based on goal
if optimization_goal == "throughput":
optimization = await self.optimize_for_throughput(
current_config=current_config,
performance_data=performance_data,
constraints=constraints
)
elif optimization_goal == "quality":
optimization = await self.optimize_for_quality(
current_config=current_config,
performance_data=performance_data,
constraints=constraints
)
elif optimization_goal == "energy_efficiency":
optimization = await self.optimize_for_energy(
current_config=current_config,
performance_data=performance_data,
constraints=constraints
)
else:
raise ValueError(f"Unknown optimization goal: {optimization_goal}")
# Simulate optimization results
simulation = await self.digital_twin.simulate_configuration(
line_id=line_id,
new_config=optimization["recommended_config"],
duration=timedelta(hours=8)
)
# Calculate expected improvements
improvements = {
"throughput_increase": simulation["throughput_delta"],
"quality_improvement": simulation["quality_delta"],
"energy_reduction": simulation["energy_delta"],
"roi_days": simulation["payback_period"]
}
return {
"line_id": line_id,
"optimization_goal": optimization_goal,
"current_performance": performance_data["summary"],
"recommended_changes": optimization["changes"],
"expected_improvements": improvements,
"implementation_plan": optimization["implementation_plan"],
"risks": optimization["identified_risks"],
"simulation_confidence": simulation["confidence"]
}
async def analyze_vibration_spectrum(
self,
vibration_data: np.ndarray
) -> Dict:
"""Perform FFT analysis for bearing fault detection."""
# Apply windowing
windowed = vibration_data * signal.windows.hann(len(vibration_data))
# Compute FFT
fft = np.fft.rfft(windowed)
freqs = np.fft.rfftfreq(len(windowed), d=1/self.sampling_rate)
# Find peaks
peaks, properties = signal.find_peaks(
np.abs(fft),
height=np.max(np.abs(fft)) * 0.1
)
# Identify bearing fault frequencies
bearing_faults = self.detect_bearing_faults(freqs[peaks], np.abs(fft[peaks]))
return {
"dominant_freq": freqs[peaks[0]] if peaks.size > 0 else 0,
"harmonics": freqs[peaks].tolist(),
"bearing_fault_prob": bearing_faults["probability"],
"fault_type": bearing_faults["type"] if bearing_faults["probability"] > 0.7 else None
}
async def detect_anomalies(self, df: pd.DataFrame) -> Dict[str, float]:
"""Multi-variate anomaly detection using Isolation Forest."""
# Select features for anomaly detection
features = ["temperature", "vibration_rms", "pressure", "rpm", "power_kw"]
X = df[features].values
# Load pre-trained model
model = await self.ml_models.get_anomaly_model(df.iloc[0]["equipment_type"])
# Predict anomaly scores
scores = model.decision_function(X)
# Normalize to 0-1 range
normalized_scores = (scores - scores.min()) / (scores.max() - scores.min())
# Calculate per-feature anomaly contribution
feature_scores = {}
for i, feature in enumerate(features):
# Create dataset with feature permuted
X_permuted = X.copy()
np.random.shuffle(X_permuted[:, i])
# Calculate importance based on score change
permuted_scores = model.decision_function(X_permuted)
feature_scores[feature] = np.mean(np.abs(scores - permuted_scores))
# Normalize feature scores
total = sum(feature_scores.values())
if total > 0:
feature_scores = {k: v/total for k, v in feature_scores.items()}
return {
"overall": float(np.mean(normalized_scores)),
**feature_scores
}Best Practices
Development Guidelines
1. Error Handling and Recovery
from typing import TypeVar, Callable, Any
import functools
import asyncio
T = TypeVar('T')
class MCPErrorHandler:
"""Comprehensive error handling for MCP servers."""
@staticmethod
def with_retry(
max_attempts: int = 3,
backoff_factor: float = 2.0,
max_delay: float = 60.0,
retryable_exceptions: tuple = (ConnectionError, TimeoutError)
):
"""Decorator for automatic retry with exponential backoff."""
def decorator(func: Callable[..., T]) -> Callable[..., T]:
@functools.wraps(func)
async def wrapper(*args, **kwargs) -> T:
last_exception = None
delay = 1.0
for attempt in range(max_attempts):
try:
return await func(*args, **kwargs)
except retryable_exceptions as e:
last_exception = e
if attempt < max_attempts - 1:
# Log retry attempt
logger.warning(
f"Attempt {attempt + 1} failed: {e}. "
f"Retrying in {delay}s..."
)
await asyncio.sleep(delay)
delay = min(delay * backoff_factor, max_delay)
else:
# Final attempt failed
logger.error(
f"All {max_attempts} attempts failed: {e}"
)
except Exception as e:
# Non-retryable exception
logger.error(f"Non-retryable error: {e}")
raise
# All attempts exhausted
raise last_exception
return wrapper
return decorator
@staticmethod
def with_fallback(fallback_value: Any = None):
"""Decorator to provide fallback value on error."""
def decorator(func: Callable[..., T]) -> Callable[..., T]:
@functools.wraps(func)
async def wrapper(*args, **kwargs) -> T:
try:
return await func(*args, **kwargs)
except Exception as e:
logger.warning(
f"Function {func.__name__} failed: {e}. "
f"Using fallback value: {fallback_value}"
)
return fallback_value
return wrapper
return decorator
@staticmethod
def with_circuit_breaker(
failure_threshold: int = 5,
recovery_timeout: float = 60.0
):
"""Circuit breaker pattern for failing services."""
class CircuitBreaker:
def __init__(self):
self.failure_count = 0
self.last_failure_time = None
self.is_open = False
async def call(self, func, *args, **kwargs):
# Check if circuit is open
if self.is_open:
if (datetime.now() - self.last_failure_time).seconds < recovery_timeout:
raise CircuitBreakerOpenError(
"Circuit breaker is open"
)
else:
# Try to reset
self.is_open = False
self.failure_count = 0
try:
result = await func(*args, **kwargs)
# Success - reset failure count
self.failure_count = 0
return result
except Exception as e:
self.failure_count += 1
self.last_failure_time = datetime.now()
if self.failure_count >= failure_threshold:
self.is_open = True
logger.error(
f"Circuit breaker opened after "
f"{self.failure_count} failures"
)
raise
breaker = CircuitBreaker()
def decorator(func: Callable[..., T]) -> Callable[..., T]:
@functools.wraps(func)
async def wrapper(*args, **kwargs) -> T:
return await breaker.call(func, *args, **kwargs)
return wrapper
return decorator
# Usage examples
@mcp.tool()
@MCPErrorHandler.with_retry(max_attempts=3)
@MCPErrorHandler.with_circuit_breaker(failure_threshold=5)
async def fetch_external_data(url: str, ctx: Context) -> Dict:
"""Fetch data with retry and circuit breaker."""
async with httpx.AsyncClient() as client:
response = await client.get(url, timeout=30.0)
response.raise_for_status()
return response.json()
@mcp.tool()
@MCPErrorHandler.with_fallback(fallback_value={"status": "unavailable"})
async def get_service_status(service_name: str, ctx: Context) -> Dict:
"""Get service status with fallback."""
return await check_service_health(service_name)2. Testing Strategies
# Comprehensive testing for MCP servers
import pytest
from fastmcp.testing import MCPTestClient
from unittest.mock import AsyncMock, patch
import asyncio
class TestMCPServer:
"""Test suite for MCP server functionality."""
@pytest.fixture
async def test_client(self):
"""Create test client with mocked dependencies."""
# Mock external services
with patch('app.weather_api') as mock_weather:
mock_weather.get_forecast.return_value = {
"temperature": 72,
"conditions": "sunny"
}
# Create test client
client = MCPTestClient(mcp_server)
# Set test authentication
client.set_auth_claims({
"sub": "test-user",
"scope": "read write",
"roles": ["user"]
})
yield client
@pytest.mark.asyncio
async def test_tool_execution(self, test_client):
"""Test basic tool execution."""
result = await test_client.call_tool(
"get_weather",
{"latitude": 37.7749, "longitude": -122.4194}
)
assert result.success
assert "temperature" in result.data
assert result.data["temperature"] == 72
@pytest.mark.asyncio
async def test_authentication_required(self, test_client):
"""Test authentication enforcement."""
# Remove auth claims
test_client.set_auth_claims(None)
with pytest.raises(AuthenticationError):
await test_client.call_tool("secure_operation", {})
@pytest.mark.asyncio
async def test_input_validation(self, test_client):
"""Test input validation."""
with pytest.raises(ValidationError) as exc_info:
await test_client.call_tool(
"get_weather",
{"latitude": 200, "longitude": -122} # Invalid latitude
)
assert "latitude" in str(exc_info.value)
@pytest.mark.asyncio
async def test_concurrent_requests(self, test_client):
"""Test handling of concurrent requests."""
# Create multiple concurrent requests
tasks = [
test_client.call_tool(
"get_weather",
{"latitude": lat, "longitude": -122}
)
for lat in range(30, 40)
]
results = await asyncio.gather(*tasks)
# All should succeed
assert all(r.success for r in results)
assert len(results) == 10
@pytest.mark.asyncio
async def test_resource_access(self, test_client):
"""Test resource retrieval."""
result = await test_client.get_resource("weather://current/seattle")
assert result.mime_type == "application/json"
assert "temperature" in result.content
@pytest.mark.asyncio
async def test_error_handling(self, test_client):
"""Test proper error handling."""
# Mock service failure
with patch('app.weather_api.get_forecast') as mock:
mock.side_effect = ConnectionError("Service unavailable")
with pytest.raises(ServiceUnavailableError):
await test_client.call_tool(
"get_weather",
{"latitude": 37, "longitude": -122}
)
@pytest.mark.asyncio
async def test_rate_limiting(self, test_client):
"""Test rate limiting enforcement."""
# Make requests up to limit
for _ in range(100):
await test_client.call_tool("simple_operation", {})
# Next request should be rate limited
with pytest.raises(RateLimitExceededError):
await test_client.call_tool("simple_operation", {})
@pytest.mark.parametrize("latitude,longitude,expected", [
(37.7749, -122.4194, "San Francisco"),
(40.7128, -74.0060, "New York"),
(51.5074, -0.1278, "London"),
])
async def test_location_detection(
self,
test_client,
latitude,
longitude,
expected
):
"""Test location detection with multiple inputs."""
result = await test_client.call_tool(
"detect_location",
{"latitude": latitude, "longitude": longitude}
)
assert result.data["city"] == expected
# Integration tests
class TestMCPIntegration:
"""Integration tests for MCP server."""
@pytest.fixture
async def live_server(self):
"""Start live MCP server for integration testing."""
server_process = await start_mcp_server(
port=0, # Random port
config="test_config.yaml"
)
yield server_process
await server_process.terminate()
@pytest.mark.integration
async def test_end_to_end_workflow(self, live_server):
"""Test complete workflow with live server."""
# Create client
client = MCPClient(f"http://localhost:{live_server.port}")
# Authenticate
token = await client.authenticate("test-user", "test-pass")
client.set_token(token)
# Execute workflow
result = await client.execute_workflow([
{"tool": "fetch_data", "params": {"source": "api"}},
{"tool": "process_data", "params": {"format": "json"}},
{"tool": "store_result", "params": {"destination": "cache"}}
])
assert result.success
assert len(result.steps) == 3
assert all(step.success for step in result.steps)
# Performance tests
class TestMCPPerformance:
"""Performance testing for MCP server."""
@pytest.mark.performance
async def test_throughput(self, test_client):
"""Test request throughput."""
start_time = asyncio.get_event_loop().time()
request_count = 1000
# Send requests concurrently
tasks = [
test_client.call_tool("simple_operation", {"value": i})
for i in range(request_count)
]
results = await asyncio.gather(*tasks)
duration = asyncio.get_event_loop().time() - start_time
throughput = request_count / duration
assert all(r.success for r in results)
assert throughput > 100 # At least 100 req/s
print(f"Throughput: {throughput:.2f} req/s")
@pytest.mark.performance
async def test_latency(self, test_client):
"""Test request latency."""
latencies = []
for _ in range(100):
start = asyncio.get_event_loop().time()
await test_client.call_tool("simple_operation", {})
latency = asyncio.get_event_loop().time() - start
latencies.append(latency)
# Calculate percentiles
p50 = np.percentile(latencies, 50)
p95 = np.percentile(latencies, 95)
p99 = np.percentile(latencies, 99)
assert p50 < 0.01 # 10ms median
assert p95 < 0.05 # 50ms 95th percentile
assert p99 < 0.1 # 100ms 99th percentile
print(f"Latency - P50: {p50*1000:.2f}ms, "
f"P95: {p95*1000:.2f}ms, P99: {p99*1000:.2f}ms")3. Documentation Standards
"""
MCP Server Documentation Example
This module implements a production-ready MCP server with comprehensive
documentation following best practices.
"""
from typing import List, Dict, Optional, Union
from pydantic import BaseModel, Field
from fastmcp import FastMCP
# API Models with detailed documentation
class WeatherRequest(BaseModel):
"""Request model for weather queries.
Attributes:
latitude: Geographic latitude (-90 to 90 degrees)
longitude: Geographic longitude (-180 to 180 degrees)
units: Temperature units (celsius or fahrenheit)
include_forecast: Whether to include 7-day forecast
"""
latitude: float = Field(
...,
ge=-90,
le=90,
description="Geographic latitude in decimal degrees"
)
longitude: float = Field(
...,
ge=-180,
le=180,
description="Geographic longitude in decimal degrees"
)
units: str = Field(
"celsius",
pattern="^(celsius|fahrenheit)$",
description="Temperature unit system"
)
include_forecast: bool = Field(
False,
description="Include 7-day forecast in response"
)
class WeatherResponse(BaseModel):
"""Response model for weather data.
Attributes:
location: Human-readable location name
current: Current weather conditions
forecast: Optional 7-day forecast
alerts: Active weather alerts for the area
"""
location: str
current: Dict[str, Union[float, str]]
forecast: Optional[List[Dict]] = None
alerts: List[Dict] = Field(default_factory=list)
# MCP Server with comprehensive documentation
mcp = FastMCP(
"weather-service",
description="""
Weather Service MCP Server
Provides real-time weather data and forecasts through a secure,
high-performance API. Features include:
- Current conditions for any global location
- 7-day weather forecasts
- Severe weather alerts
- Historical weather data access
- Batch location queries
Authentication: OAuth 2.0 bearer tokens required
Rate Limits: 1000 requests per hour per user
"""
)
@mcp.tool(
description="""
Get current weather and optional forecast for a location.
This tool retrieves real-time weather data from multiple sources
and provides consolidated, accurate weather information.
Rate limited to 100 requests per minute per user.
Requires 'weather:read' scope.
Example:
{
"latitude": 37.7749,
"longitude": -122.4194,
"units": "fahrenheit",
"include_forecast": true
}
""",
examples=[
{
"description": "Get current weather for San Francisco",
"params": {
"latitude": 37.7749,
"longitude": -122.4194
}
},
{
"description": "Get weather with forecast in Fahrenheit",
"params": {
"latitude": 40.7128,
"longitude": -74.0060,
"units": "fahrenheit",
"include_forecast": True
}
}
]
)
async def get_weather(
request: WeatherRequest,
ctx: Context
) -> WeatherResponse:
"""
Retrieve weather data for a specific location.
Args:
request: Weather query parameters
ctx: MCP context with auth claims and metadata
Returns:
WeatherResponse with current conditions and optional forecast
Raises:
RateLimitError: If rate limit exceeded
AuthorizationError: If missing required scope
WeatherAPIError: If weather service unavailable
Note:
Results are cached for 5 minutes to improve performance
and reduce API calls to upstream providers.
"""
# Implementation here
pass
# API documentation endpoint
@mcp.resource("docs://api/reference")
async def get_api_documentation(ctx: Context) -> Dict:
"""Provide comprehensive API documentation."""
return {
"openapi": "3.0.0",
"info": {
"title": "Weather Service MCP API",
"version": "1.0.0",
"description": mcp.description
},
"paths": generate_openapi_paths(mcp),
"components": {
"schemas": generate_schemas([WeatherRequest, WeatherResponse]),
"securitySchemes": {
"bearer": {
"type": "http",
"scheme": "bearer",
"bearerFormat": "JWT"
}
}
}
}Deployment Best Practices
1. Container Deployment
# Production Dockerfile for MCP server
FROM python:3.11-slim as builder
# Install build dependencies
RUN apt-get update && apt-get install -y \
gcc \
g++ \
&& rm -rf /var/lib/apt/lists/*
# Create virtual environment
RUN python -m venv /opt/venv
ENV PATH="/opt/venv/bin:$PATH"
# Install Python dependencies
COPY requirements.txt .
RUN pip install --upgrade pip && \
pip install --no-cache-dir -r requirements.txt
# Production image
FROM python:3.11-slim
# Install runtime dependencies
RUN apt-get update && apt-get install -y \
ca-certificates \
curl \
&& rm -rf /var/lib/apt/lists/*
# Copy virtual environment from builder
COPY --from=builder /opt/venv /opt/venv
ENV PATH="/opt/venv/bin:$PATH"
# Create non-root user
RUN useradd -m -u 1000 mcpuser && \
mkdir -p /app && \
chown -R mcpuser:mcpuser /app
WORKDIR /app
# Copy application code
COPY --chown=mcpuser:mcpuser . .
# Security hardening
RUN chmod -R 755 /app && \
find /app -type f -name "*.py" -exec chmod 644 {} \;
# Switch to non-root user
USER mcpuser
# Health check
HEALTHCHECK --interval=30s --timeout=3s --start-period=40s --retries=3 \
CMD python -c "import httpx; httpx.get('http://localhost:8080/health').raise_for_status()"
# Expose port
EXPOSE 8080
# Set environment variables
ENV PYTHONUNBUFFERED=1 \
PYTHONDONTWRITEBYTECODE=1 \
MCP_LOG_LEVEL=INFO
# Run the application
CMD ["python", "-m", "uvicorn", "app.main:app", \
"--host", "0.0.0.0", \
"--port", "8080", \
"--workers", "4", \
"--loop", "uvloop", \
"--access-log", \
"--log-config", "logging.yaml"]2. Kubernetes Deployment
# kubernetes/mcp-server.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
name: mcp-server
labels:
app: mcp-server
version: v1
spec:
replicas: 3
selector:
matchLabels:
app: mcp-server
template:
metadata:
labels:
app: mcp-server
version: v1
annotations:
prometheus.io/scrape: "true"
prometheus.io/port: "8080"
prometheus.io/path: "/metrics"
spec:
serviceAccountName: mcp-server
securityContext:
runAsNonRoot: true
runAsUser: 1000
fsGroup: 1000
containers:
- name: mcp-server
image: myregistry/mcp-server:latest
imagePullPolicy: Always
ports:
- containerPort: 8080
name: http
protocol: TCP
env:
- name: MCP_LOG_LEVEL
value: "INFO"
- name: DATABASE_URL
valueFrom:
secretKeyRef:
name: mcp-secrets
key: database-url
- name: JWT_SECRET
valueFrom:
secretKeyRef:
name: mcp-secrets
key: jwt-secret
resources:
requests:
memory: "256Mi"
cpu: "250m"
limits:
memory: "1Gi"
cpu: "1000m"
livenessProbe:
httpGet:
path: /health
port: 8080
initialDelaySeconds: 30
periodSeconds: 10
timeoutSeconds: 3
failureThreshold: 3
readinessProbe:
httpGet:
path: /ready
port: 8080
initialDelaySeconds: 5
periodSeconds: 5
timeoutSeconds: 2
successThreshold: 1
failureThreshold: 3
volumeMounts:
- name: config
mountPath: /app/config
readOnly: true
- name: cache
mountPath: /app/cache
securityContext:
allowPrivilegeEscalation: false
readOnlyRootFilesystem: true
capabilities:
drop:
- ALL
volumes:
- name: config
configMap:
name: mcp-config
- name: cache
emptyDir:
sizeLimit: 1Gi
---
apiVersion: v1
kind: Service
metadata:
name: mcp-server
labels:
app: mcp-server
spec:
type: ClusterIP
ports:
- port: 80
targetPort: 8080
protocol: TCP
name: http
selector:
app: mcp-server
---
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
name: mcp-server-hpa
spec:
scaleTargetRef:
apiVersion: apps/v1
kind: Deployment
name: mcp-server
minReplicas: 3
maxReplicas: 10
metrics:
- type: Resource
resource:
name: cpu
target:
type: Utilization
averageUtilization: 70
- type: Resource
resource:
name: memory
target:
type: Utilization
averageUtilization: 80
- type: Pods
pods:
metric:
name: mcp_requests_per_second
target:
type: AverageValue
averageValue: "100"
---
apiVersion: policy/v1
kind: PodDisruptionBudget
metadata:
name: mcp-server-pdb
spec:
minAvailable: 2
selector:
matchLabels:
app: mcp-serverResources and References
Official Documentation
- Model Context Protocol Specification - Official protocol documentation
- Python SDK Documentation - Python implementation guide
- TypeScript SDK Documentation - TypeScript implementation guide
- Anthropic MCP Docs - Claude-specific integration
Community Resources
- Claude MCP Community - Community hub with examples and discussions
- MCP Server Examples - Pre-built server implementations
- FastMCP Documentation - High-level Python framework
- MCP for Beginners - Microsoft’s comprehensive tutorial
Security Resources
- OWASP Top 10 for LLM Applications - Security guidelines
- MCP Security Best Practices - Official security guide
- OAuth 2.1 Specification - Authentication standard
- Zero Trust Architecture - NIST guidelines
Related Knowledge Base Topics
- Claude Code Hooks - Integration with Claude Code’s hook system
- Security Deep Dive - Comprehensive security patterns
- Deployment Guide - Production deployment strategies
- Testing Guide - Testing best practices
- Performance Optimization - Advanced optimization techniques
- Multi-Agent MCP Experiments - Cutting-edge research
Tools and Libraries
- Testing: pytest-mcp, mcp-test-client
- Monitoring: Prometheus exporters, OpenTelemetry integration
- Security: mcp-security-scanner, oauth2-proxy
- Development: mcp-cli, mcp-debugger
Conclusion
The Model Context Protocol represents a paradigm shift in how AI applications interact with external systems. By providing a standardized, secure, and scalable protocol, MCP enables developers to build sophisticated AI integrations without the complexity of custom implementations.
Key takeaways:
- Always prioritize security - Authentication should never be optional in production
- Design for scale - Use connection pooling, caching, and async patterns
- Monitor everything - Comprehensive observability is crucial for production systems
- Test thoroughly - Include unit, integration, and performance tests
- Document extensively - Clear documentation accelerates adoption
As the MCP ecosystem continues to grow, staying updated with the latest security advisories, protocol updates, and best practices is essential. The standardization that MCP brings to AI-tool communication is just the beginning - the real power lies in the innovative applications that developers will build on this foundation.
Whether you’re building a simple integration or a complex multi-agent system, MCP provides the tools and patterns needed to create robust, secure, and scalable AI-powered applications. The future of AI development is modular, interoperable, and secure - and MCP is leading the way.