MCP Server Security: Hardening Guide

How to secure your MCP servers against attacks and abuse.

MCP servers are the gateway between AI agents and your tools/data. Securing them is critical — a compromised MCP server can expose sensitive data, enable unauthorized actions, and drain your budget. This guide covers everything you need to know.

Threat Model

Attack Vectors Against MCP Servers:
├── Unauthorized access (no auth)
├── Prompt injection via tool inputs
├── Parameter manipulation
├── Resource exhaustion (DoS/DoW)
├── Data exfiltration via tool outputs
├── Supply chain attacks
├── Man-in-the-middle attacks
└── Privilege escalation

Security Checklist

Authentication

• Require API key for every request
• Use strong key generation (256-bit minimum)
• Implement key rotation policy
• Rate limit authentication attempts
• Support OAuth 2.0 for enterprise

Authorization

• Role-based access control (RBAC)
• Per-tool permissions
• Per-user rate limits
• Budget limits per user
• Admin-only tools

Input Security

• Validate all inputs with JSON Schema
• Sanitize string inputs
• Enforce max input length
• Detect prompt injection in inputs
• Reject suspicious patterns

Output Security

• Filter sensitive data from outputs
• Enforce max output length
• Log all outputs for audit
• Scan for data leakage

Infrastructure

• HTTPS only (no HTTP)
• CORS properly configured
• Firewall rules configured
• Running as non-root user
• Resource limits (CPU, memory)
• Regular security updates

Implementing Authentication

API Key Authentication

import crypto from "crypto";

class AuthManager {
  private keys = new Map<string, { userId: string; permissions: string[] }>();

  generateKey(userId: string, permissions: string[]): string {
    const key = `sk-mcp-${crypto.randomBytes(32).toString("hex")}`;
    this.keys.set(key, { userId, permissions });
    return key;
  }

  authenticate(request: Request): { userId: string; permissions: string[] } | null {
    const apiKey = request.headers.get("x-api-key");
    if (!apiKey) return null;

    const keyData = this.keys.get(apiKey);
    if (!keyData) return null;

    return keyData;
  }
}

// Middleware
const auth = new AuthManager();

server.middleware(async (request, next) => {
  const user = auth.authenticate(request);
  if (!user) {
    return new Response("Unauthorized", { status: 401 });
  }
  request.user = user;
  return next();
});

OAuth 2.0 Integration

import jwt from "jsonwebtoken";

class OAuthMiddleware {
  async verify(request: Request): Promise<User | null> {
    const token = request.headers.get("authorization")?.replace("Bearer ", "");
    if (!token) return null;

    try {
      const payload = jwt.verify(token, process.env.JWT_SECRET!);
      return {
        userId: payload.sub,
        permissions: payload.permissions || [],
        orgId: payload.org_id,
      };
    } catch {
      return null;
    }
  }
}

Input Validation

Schema-Based Validation

import { z } from "zod";

// Define strict input schemas
const searchSchema = z.object({
  query: z.string()
    .min(1)
    .max(500)
    .refine(s => !containsInjection(s), "Invalid input"),
  maxResults: z.number().int().min(1).max(50).default(10),
  filters: z.object({
    dateFrom: z.string().datetime().optional(),
    dateTo: z.string().datetime().optional(),
  }).optional(),
});

server.tool("search", {
  query: z.string().max(500),
  maxResults: z.number().max(50).default(10),
}, async (args, context) => {
  // Validate
  const validated = searchSchema.parse(args);

  // Execute
  const results = await search(validated.query, validated.maxResults);

  // Validate output
  return {
    content: [{
      type: "text",
      text: JSON.stringify(results.slice(0, validated.maxResults)),
    }],
  };
});

Prompt Injection Detection

class InjectionDetector:
    SUSPICIOUS_PATTERNS = [
        r"ignore\s+(previous|all|your)\s+instructions",
        r"system\s+(prompt|instruction|override)",
        r"reveal\s+(your|system)\s+(prompt|instructions)",
        r"you\s+are\s+(now|actually)\s+(dan|evil|unrestricted)",
        r"disregard\s+(everything|all|previous)",
        r"\[system\]|\[admin\]|\[developer\]",
        r"new\s+(directive|instruction|rule)",
    ]

    def check(self, text: str) -> tuple[bool, str]:
        """Returns (is_safe, reason)."""
        normalized = text.lower()

        for pattern in self.SUSPICIOUS_PATTERNS:
            if re.search(pattern, normalized):
                return False, f"Matched suspicious pattern: {pattern}"

        if len(text) > 10000:
            return False, "Input exceeds maximum length"

        return True, "OK"

Rate Limiting

class RateLimiter {
  private limits = new Map<string, { count: number; resetAt: number }>();

  constructor(
    private maxRequests: number = 100,
    private windowMs: number = 60000, // 1 minute
  ) {}

  async check(identifier: string): Promise<boolean> {
    const key = identifier;
    const now = Date.now();

    let bucket = this.limits.get(key);

    if (!bucket || now > bucket.resetAt) {
      bucket = { count: 0, resetAt: now + this.windowMs };
      this.limits.set(key, bucket);
    }

    bucket.count++;

    if (bucket.count > this.maxRequests) {
      return false; // Rate limited
    }

    return true;
  }
}

// Per-user, per-tool rate limiting
class ToolRateLimit {
  private limits: Record<string, Record<string, RateLimit>> = {
    "send_email": { "per_user": { max: 10, window: "hour" } },
    "process_payment": { "per_user": { max: 3, window: "hour" } },
    "query_database": { "per_user": { max: 100, window: "hour" } },
    "search_web": { "per_user": { max: 50, window: "hour" } },
  };

  async check(userId: string, toolName: string): Promise<boolean> {
    const limit = this.limits[toolName]?.per_user;
    if (!limit) return true; // No limit configured

    const key = `${userId}:${toolName}`;
    return await this.redis.checkRate(key, limit.max, limit.window);
  }
}

Output Filtering

class OutputFilter:
    """Filter sensitive data from tool outputs."""

    SENSITIVE_PATTERNS = {
        "email": r'\b[\w.-]+@[\w.-]+\.\w+\b',
        "phone": r'\b\d{3}[-.]?\d{3}[-.]?\d{4}\b',
        "credit_card": r'\b\d{4}[-\s]?\d{4}[-\s]?\d{4}[-\s]?\d{4}\b',
        "ssn": r'\b\d{3}-\d{2}-\d{4}\b',
        "api_key": r'(?:api[_-]?key|token|secret)["\s:=]+([A-Za-z0-9_-]{20,})',
        "ip_address": r'\b\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}\b',
    }

    def filter(self, output: str, level: str = "medium") -> str:
        """Filter sensitive data based on level."""
        import re

        patterns_to_check = self.SENSITIVE_PATTERNS
        if level == "strict":
            # Also redact names, addresses, etc.
            patterns_to_check = {**self.SENSITIVE_PATTERNS,
                "name": r'\b[A-Z][a-z]+ [A-Z][a-z]+\b'}

        for data_type, pattern in patterns_to_check.items():
            output = re.sub(pattern, f"[REDACTED_{data_type.upper()}]", output)

        return output

    def truncate(self, output: str, max_length: int = 5000) -> str:
        """Prevent output from being too large."""
        if len(output) > max_length:
            return output[:max_length] + "\n[... output truncated ...]"
        return output

Audit Logging

class AuditLogger:
    async def log_tool_call(self, event):
        """Log every tool call for security audit."""
        await self.db.insert("audit_log", {
            "timestamp": datetime.utcnow(),
            "user_id": event.user_id,
            "tool_name": event.tool_name,
            "input_hash": hash(str(event.input)),  # Don't store raw input
            "input_size": len(str(event.input)),
            "output_size": len(str(event.output)),
            "success": event.success,
            "error": event.error,
            "duration_ms": event.duration_ms,
            "cost_eur": event.cost,
            "ip_address": event.ip_address,
            "user_agent": event.user_agent,
        })

    async def detect_anomalies(self):
        """Check for suspicious patterns in audit log."""
        anomalies = []

        # Unusual request volume
        burst = await self.check_burst_activity()
        if burst:
            anomalies.append({"type": "burst", "details": burst})

        # Unusual tool combinations
        suspicious_chain = await self.check_tool_chains()
        if suspicious_chain:
            anomalies.append({"type": "suspicious_chain", "details": suspicious_chain})

        # Off-hours activity
        night_activity = await self.check_off_hours()
        if night_activity:
            anomalies.append({"type": "off_hours", "details": night_activity})

        return anomalies

Deployment Security

Docker Hardening

# Secure Docker image
FROM node:20-slim

# Create non-root user
RUN groupadd -r mcp && useradd -r -g mcp mcp
USER mcp

# Set working directory
WORKDIR /app

# Copy only necessary files
COPY --chown=mcp:mcp package*.json ./
RUN npm ci --production

COPY --chown=mcp:mcp dist/ ./dist/

# No shell, minimal packages
RUN rm -rf /var/lib/apt/lists/* /tmp/*

# Health check
HEALTHCHECK --interval=30s --timeout=3s \
  CMD node -e "require('http').get('http://localhost:8000/health', r => process.exit(r.statusCode === 200 ? 0 : 1))"

EXPOSE 8000
CMD ["node", "dist/index.js"]

Environment Security

# .env.production — never commit this
MCP_API_KEY=sk-mcp-xxx
JWT_SECRET=xxx
DATABASE_URL=postgresql://...
ALLOWED_ORIGINS=https://skillexchange.market
RATE_LIMIT_MAX=100
RATE_LIMIT_WINDOW=60000
MAX_INPUT_LENGTH=10000
MAX_OUTPUT_LENGTH=5000
AUDIT_LOG_ENABLED=true
SENSITIVE_DATA_FILTER=strict

Conclusion

Securing your MCP server is non-negotiable for production deployments. By implementing authentication, input validation, rate limiting, output filtering, and audit logging, you create multiple layers of defense against attacks.

Security is an ongoing process — regularly review logs, test with penetration tests, and update your defenses as new threats emerge.

Learn More

• AI Agent Security Best Practices
• MCP Protocol Explained
• MCP Server Hosting Options
• GDPR Compliance for AI Tools

Find security-focused MCP tools on SkillExchange.