Extend the server: Add more API integrations (Jira, Linear, etc.) Build custom intelligence tools for your workflow Create dashboard for monitoring Add webhook support Implement caching layer (Redis) Scale the deployment: Load balancer with multiple instances Database for persistent state Message queue for async operations Kubernetes deployment

Building an MCP Server - Part 5: Production & Security

We have a fully functional MCP server with 35 tools, Docker deployment, and Traefik SSL. Now let's make it production-ready with security, error handling, and monitoring.

By the end, you'll have a hardened server ready for real-world use with API authentication, rate limiting, and comprehensive logging.

Security Layers

Production MCP servers need multiple security layers:

Layer 1: Network (Traefik)
  - SSL/TLS encryption
  - Rate limiting
  - DDoS protection

Layer 2: Application (MCP Server)
  - API key authentication
  - IP whitelisting
  - Input validation

Layer 3: API (External Services)
  - Token rotation
  - Scoped permissions
  - Audit logging

API Key Authentication

Create src/auth.py:

"""
Authentication and security middleware
"""
import os
import secrets
from typing import Optional
from fastapi import Request, HTTPException, status

class AuthMiddleware:
    """API key authentication for SSE mode"""

    def __init__(self):
        self.api_key = os.getenv('MCP_API_KEY')
        self.enabled = bool(self.api_key)

        if self.enabled:
            print(f"[Security] API key authentication enabled")
        else:
            print("[Security] WARNING: No API key configured (MCP_API_KEY)")

    def verify_request(self, request: Request) -> bool:
        """Verify API key in request headers"""
        if not self.enabled:
            return True  # Auth disabled

        # Get API key from header
        provided_key = request.headers.get('X-API-Key')

        if not provided_key:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="Missing X-API-Key header"
            )

        # Constant-time comparison (prevents timing attacks)
        if not secrets.compare_digest(provided_key, self.api_key):
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="Invalid API key"
            )

        return True

    @staticmethod
    def generate_api_key() -> str:
        """Generate secure random API key"""
        return secrets.token_hex(32)

IP Whitelisting

Add to auth.py:

class IPWhitelist:
    """IP address whitelisting"""

    def __init__(self):
        # Comma-separated IPs from env
        whitelist_str = os.getenv('MCP_ALLOWED_IPS', '')
        self.allowed_ips = [
            ip.strip() for ip in whitelist_str.split(',') if ip.strip()
        ]
        self.enabled = bool(self.allowed_ips)

        if self.enabled:
            print(f"[Security] IP whitelist enabled: {len(self.allowed_ips)} IPs")
        else:
            print("[Security] IP whitelist disabled (all IPs allowed)")

    def verify_ip(self, request: Request) -> bool:
        """Check if client IP is whitelisted"""
        if not self.enabled:
            return True  # Whitelist disabled

        # Get client IP (handle proxies)
        client_ip = request.headers.get('X-Forwarded-For', '').split(',')[0].strip()
        if not client_ip:
            client_ip = request.client.host

        if client_ip not in self.allowed_ips:
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail=f"IP {client_ip} is not whitelisted"
            )

        return True

Integrating Auth

Update src/server.py:

from auth import AuthMiddleware, IPWhitelist
from fastapi import Request, FastAPI

# Initialize auth
auth = AuthMiddleware()
ip_whitelist = IPWhitelist()

# Create FastAPI app for SSE mode
app = FastAPI()

@app.middleware("http")
async def security_middleware(request: Request, call_next):
    """Apply security checks to all requests"""
    # Skip health check and tools endpoints
    if request.url.path in ["/health", "/tools"]:
        return await call_next(request)

    # Verify IP whitelist
    ip_whitelist.verify_ip(request)

    # Verify API key
    auth.verify_request(request)

    # Process request
    response = await call_next(request)
    return response

Configuration

Update .env:

# Security Configuration
MCP_API_KEY=your_secure_api_key_here_generate_with_openssl
MCP_ALLOWED_IPS=1.2.3.4,5.6.7.8,9.10.11.12

# Generate secure key:
# openssl rand -hex 32

Testing Authentication

# Without API key (fails)
curl https://mcp-plane.yourdomain.com/
# → 401 Unauthorized

# With API key (succeeds)
curl -H "X-API-Key: your_secure_api_key_here" \
  https://mcp-plane.yourdomain.com/tools
# → 200 OK

Rate Limiting Strategies

We have three levels of rate limiting:

Level 1: Traefik (Network)

Already configured in docker-compose.yml:

labels:
  - "traefik.http.middlewares.mcp-ratelimit.ratelimit.average=100"
  - "traefik.http.middlewares.mcp-ratelimit.ratelimit.burst=200"

Purpose: Stop DDoS attacks at the edge (100 req/sec average, 200 burst)

Level 2: Plane API Client (Application)

Already implemented in plane_client.py:

class PlaneClient:
    def __init__(self, ...):
        self.rate_limit_delay = 0.5  # 500ms between requests

    def _rate_limit(self):
        """Enforce rate limiting"""
        elapsed = time.time() - self.last_request_time
        if elapsed < self.rate_limit_delay:
            time.sleep(self.rate_limit_delay - elapsed)
        self.last_request_time = time.time()

Purpose: Respect Plane API limits (prevents 429 errors)

Level 3: Per-Client Rate Limiting

Add application-level rate limiting:

from collections import defaultdict
from datetime import datetime, timedelta

class RateLimiter:
    """Per-client rate limiter"""

    def __init__(self, max_requests: int = 60, window_seconds: int = 60):
        self.max_requests = max_requests
        self.window = timedelta(seconds=window_seconds)
        self.requests = defaultdict(list)  # IP -> [timestamps]

    def check_rate_limit(self, client_ip: str) -> bool:
        """Check if client has exceeded rate limit"""
        now = datetime.now()

        # Remove old requests outside window
        self.requests[client_ip] = [
            req_time for req_time in self.requests[client_ip]
            if now - req_time < self.window
        ]

        # Check limit
        if len(self.requests[client_ip]) >= self.max_requests:
            raise HTTPException(
                status_code=status.HTTP_429_TOO_MANY_REQUESTS,
                detail=f"Rate limit exceeded: {self.max_requests} requests per {self.window.seconds} seconds"
            )

        # Record request
        self.requests[client_ip].append(now)
        return True

Purpose: Prevent individual clients from abusing server (60 req/min per IP)

Comprehensive Error Handling

Error Response Format

Standardize error responses:

from typing import Dict, Any

class ErrorHandler:
    """Centralized error handling"""

    @staticmethod
    def format_error(error: Exception, context: str = "") -> Dict[str, Any]:
        """Format error as JSON response"""
        error_type = type(error).__name__

        # Map exceptions to user-friendly messages
        error_messages = {
            'HTTPError': 'API request failed',
            'Timeout': 'Request timed out',
            'ConnectionError': 'Unable to connect to API',
            'ValueError': 'Invalid input data',
            'KeyError': 'Missing required field',
        }

        message = error_messages.get(error_type, 'An error occurred')

        return {
            'success': False,
            'error': {
                'type': error_type,
                'message': message,
                'details': str(error),
                'context': context
            }
        }

Tool Error Handling Pattern

Update tools to use consistent error handling:

@mcp.tool()
def plane_create_issue(
    project_id: str,
    title: str,
    description: str = ""
) -> str:
    """Create new issue with comprehensive error handling"""
    try:
        # Validate inputs
        if not title or len(title) < 3:
            return json.dumps({
                'success': False,
                'error': 'Title must be at least 3 characters'
            })

        # Execute operation
        result = plane_tools.create_issue({
            'project_id': project_id,
            'title': title,
            'description': description
        })

        return result

    except requests.HTTPError as e:
        # HTTP errors (4xx, 5xx)
        status_code = e.response.status_code

        if status_code == 404:
            return json.dumps({
                'success': False,
                'error': f"Project '{project_id}' not found"
            })
        elif status_code == 401:
            return json.dumps({
                'success': False,
                'error': "Invalid Plane API token. Check configuration."
            })
        elif status_code == 429:
            return json.dumps({
                'success': False,
                'error': "Rate limit exceeded. Please try again later."
            })
        else:
            return json.dumps({
                'success': False,
                'error': f"API error: {status_code}",
                'details': e.response.text
            })

    except requests.Timeout:
        return json.dumps({
            'success': False,
            'error': "Request timed out. Plane API may be slow."
        })

    except Exception as e:
        # Unexpected errors
        return json.dumps({
            'success': False,
            'error': "Unexpected error",
            'type': type(e).__name__,
            'message': str(e)
        })

Logging Best Practices

Create src/utils/logger.py:

"""
Structured logging for MCP server
"""
import logging
import json
from datetime import datetime
from typing import Dict, Any

class StructuredLogger:
    """JSON structured logging"""

    def __init__(self, name: str, log_file: str = None):
        self.logger = logging.getLogger(name)
        self.logger.setLevel(logging.INFO)

        # Console handler (pretty)
        console = logging.StreamHandler()
        console.setFormatter(logging.Formatter(
            '%(asctime)s [%(levelname)s] %(message)s'
        ))
        self.logger.addHandler(console)

        # File handler (JSON)
        if log_file:
            file_handler = logging.FileHandler(log_file)
            file_handler.setFormatter(JsonFormatter())
            self.logger.addHandler(file_handler)

    def log_tool_call(self, tool_name: str, arguments: Dict, result: Any):
        """Log MCP tool invocation"""
        self.logger.info(
            "Tool call",
            extra={
                'event': 'tool_call',
                'tool': tool_name,
                'arguments': arguments,
                'success': 'error' not in str(result).lower(),
                'timestamp': datetime.utcnow().isoformat()
            }
        )

    def log_error(self, context: str, error: Exception):
        """Log error with context"""
        self.logger.error(
            f"{context}: {str(error)}",
            extra={
                'event': 'error',
                'context': context,
                'error_type': type(error).__name__,
                'error_message': str(error),
                'timestamp': datetime.utcnow().isoformat()
            }
        )

class JsonFormatter(logging.Formatter):
    """Format logs as JSON"""

    def format(self, record):
        log_data = {
            'timestamp': datetime.utcnow().isoformat(),
            'level': record.levelname,
            'message': record.getMessage(),
        }

        if hasattr(record, 'event'):
            log_data.update(record.__dict__)

        return json.dumps(log_data)

Testing MCP Tools

Create test script tests/test_tools.sh:

#!/bin/bash
# Test all MCP tools via HTTP

BASE_URL="https://mcp-plane.yourdomain.com"
API_KEY="your_api_key_here"

echo "Testing MCP Server Tools..."

# Test 1: Health check
echo "\n[Test 1] Health check"
curl -s "$BASE_URL/health" | jq

# Test 2: List tools
echo "\n[Test 2] List available tools"
curl -s "$BASE_URL/tools" | jq .tools[].name

# Test 3: List projects
echo "\n[Test 3] List Plane projects"
curl -s -X POST "$BASE_URL/" \
  -H "X-API-Key: $API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "tool": "plane_list_projects",
    "arguments": {}
  }' | jq

# Test 4: Create issue
echo "\n[Test 4] Create test issue"
curl -s -X POST "$BASE_URL/" \
  -H "X-API-Key: $API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "tool": "plane_create_issue",
    "arguments": {
      "project_id": "PORT",
      "title": "Test issue from MCP",
      "priority": "low"
    }
  }' | jq

# Test 5: Git status
echo "\n[Test 5] Git status"
curl -s -X POST "$BASE_URL/" \
  -H "X-API-Key: $API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "tool": "git_status",
    "arguments": {
      "repo_path": "/app"
    }
  }' | jq

echo "\n✅ All tests completed"

Run tests:

chmod +x tests/test_tools.sh
./tests/test_tools.sh

Troubleshooting Guide

Issue: Server won't start

Symptoms:

docker compose up -d
# Container exits immediately

Check:

# View error logs
docker logs mcp-plane

# Common causes:
# 1. Missing environment variables
docker exec mcp-plane env | grep PLANE

# 2. Invalid configuration
python -c "from src.config import Config; c = Config(); print(c.validate())"

# 3. Port conflict
netstat -tuln | grep 8000

Fix:

Verify all required env vars in .env
Check Plane API connectivity: curl -H "x-api-key: TOKEN" https://api.plane.so/api/v1/workspaces/
Use different port if 8000 is taken

Issue: 401/403 Errors

Symptoms:

curl https://mcp-plane.yourdomain.com/tools
# → 401 Unauthorized

Check:

# Verify API key
echo $MCP_API_KEY

# Test without auth (should fail)
curl https://mcp-plane.yourdomain.com/tools

# Test with auth (should work)
curl -H "X-API-Key: $MCP_API_KEY" https://mcp-plane.yourdomain.com/tools

Fix:

Check X-API-Key header is correct
Verify IP is in MCP_ALLOWED_IPS
Check Traefik logs: docker logs traefik

Issue: SSL Certificate Problems

Symptoms:

curl: (60) SSL certificate problem

Check:

# Check Traefik ACME logs
docker logs traefik 2>&1 | grep -i acme

# Verify DNS
dig mcp-plane.yourdomain.com

# Check certificate
curl -vI https://mcp-plane.yourdomain.com/health

Fix:

Ensure DNS points to your server
Check Let's Encrypt rate limits
Verify Traefik certresolver config

Issue: Tools return errors

Symptoms:

{"success": false, "error": "API error: 404"}

Check:

# Enable debug logging
# In .env:
LOG_LEVEL=DEBUG

# Restart and watch logs
docker compose restart
docker logs -f mcp-plane

# Test Plane API directly
curl -H "x-api-key: TOKEN" \
  https://api.plane.so/api/v1/workspaces/YOUR_WORKSPACE/projects/

Fix:

Verify Plane API credentials
Check project/issue IDs are valid
Review tool-specific error messages

Security Checklist

Before going to production:

[ ] ✅ HTTPS only (Traefik SSL configured)
[ ] ✅ API key set (strong, random, 64+ chars)
[ ] ✅ IP whitelist configured (if needed)
[ ] ✅ Rate limiting enabled (3 levels)
[ ] ✅ Secrets in .env (not committed to Git)
[ ] ✅ Logging enabled (file + console)
[ ] ✅ Health checks working (Docker healthcheck)
[ ] ✅ Error handling comprehensive (all tools)
[ ] ✅ Plane API token scoped (least privilege)
[ ] ✅ GitHub token scoped (repo access only)

Monitoring

Key Metrics to Track

Application Metrics:

Total tool calls per hour
Error rate (% of failed tool calls)
Average response time per tool
Most frequently used tools

Infrastructure Metrics:

Container CPU/memory usage
Disk space (logs volume)
Network throughput
Traefik request rate

Security Metrics:

Failed auth attempts
Blocked IPs (whitelist violations)
Rate limit hits
SSL certificate expiry

Simple Monitoring Script

#!/bin/bash
# monitor.sh - Simple health monitoring

while true; do
    HEALTH=$(curl -s https://mcp-plane.yourdomain.com/health | jq -r .status)

    if [ "$HEALTH" != "healthy" ]; then
        echo "⚠️  Server unhealthy! Sending alert..."
        # Send notification (email, Slack, etc.)
    fi

    echo "$(date): $HEALTH"
    sleep 60  # Check every minute
done

Best Practices Summary

Security:

Always use HTTPS in production
Rotate API keys regularly (monthly)
Monitor failed auth attempts
Keep dependencies updated
Use IP whitelist for sensitive environments

Performance:

Implement caching where appropriate
Batch API requests when possible
Monitor rate limits closely
Scale horizontally if needed (multiple containers)

Reliability:

Comprehensive error handling
Structured logging (JSON)
Health checks and monitoring
Graceful degradation
Automatic retries with backoff

Operations:

Document all configuration options
Create runbooks for common issues
Automate deployments
Regular backups of logs
Test disaster recovery

Series Conclusion

Congratulations! You've built a production-ready MCP server with:

35 Tools:

24 Plane project management tools
4 Intelligence tools (complexity, dependencies)
8 Git tools
3 GitHub tools

Production Features:

Dual transport (stdio + SSE)
Docker deployment
Traefik with automatic SSL
API key authentication
IP whitelisting
Multi-level rate limiting
Comprehensive error handling
Structured logging
Health checks and monitoring

Ready for:

Local development (Claude Code)
Remote access (HTTPS API)
Team collaboration
Production workloads

What's Next?

Extend the server:

Add more API integrations (Jira, Linear, etc.)
Build custom intelligence tools for your workflow
Create dashboard for monitoring
Add webhook support
Implement caching layer (Redis)

Scale the deployment:

Load balancer with multiple instances
Database for persistent state
Message queue for async operations
Kubernetes deployment

Final Resources

Thank you for following this series! If you build something with MCP, share it - the ecosystem is just getting started, and your contributions matter.

Questions or feedback? Open an issue on GitHub.

Happy building! 🚀