MCP Server Monitoring & Observability: Complete Guide to Production Operations

Executive Summary

This comprehensive guide covers monitoring, logging, and observability for MCP servers in production. We explore metrics collection, alerting strategies, debugging techniques, and operational procedures essential for maintaining reliable MCP infrastructure.

Part 1: Metrics & Monitoring

1.1 Key Performance Indicators

Tool Execution Metrics

• Tool call count (total and per-tool)
• Success/failure rates
• Execution time (p50, p95, p99)
• Error types and frequency

Resource Metrics

• CPU usage
• Memory usage
• Disk I/O
• Network bandwidth

Business Metrics

• Active connections
• Tool utilization
• Feature adoption
• User satisfaction

1.2 Prometheus Metrics Implementation

from prometheus_client import Counter, Histogram, Gauge
import time

# Define metrics
tool_calls = Counter(
    'mcp_tool_calls_total',
    'Total tool calls',
    ['tool_name', 'status']
)

tool_duration = Histogram(
    'mcp_tool_duration_seconds',
    'Tool execution duration',
    ['tool_name'],
    buckets=(0.1, 0.5, 1.0, 2.5, 5.0, 10.0)
)

active_connections = Gauge(
    'mcp_active_connections',
    'Active client connections'
)

# Usage
@app.post("/tool/call")
async def call_tool(name: str, args: dict):
    start = time.time()
    try:
        result = await execute_tool(name, args)
        tool_calls.labels(tool_name=name, status='success').inc()
        return result
    except Exception as e:
        tool_calls.labels(tool_name=name, status='error').inc()
        raise
    finally:
        duration = time.time() - start
        tool_duration.labels(tool_name=name).observe(duration)

Part 2: Logging & Tracing

2.1 Structured Logging

import logging
import json
from datetime import datetime

class JSONFormatter(logging.Formatter):
    def format(self, record):
        log_data = {
            'timestamp': datetime.utcnow().isoformat(),
            'level': record.levelname,
            'message': record.getMessage(),
            'logger': record.name,
            'module': record.module,
            'function': record.funcName,
            'line': record.lineno
        }
        
        # Add custom fields if present
        if hasattr(record, 'request_id'):
            log_data['request_id'] = record.request_id
        if hasattr(record, 'user_id'):
            log_data['user_id'] = record.user_id
        if hasattr(record, 'tool_name'):
            log_data['tool_name'] = record.tool_name
        
        return json.dumps(log_data)

# Setup logging
handler = logging.StreamHandler()
handler.setFormatter(JSONFormatter())
logger = logging.getLogger()
logger.addHandler(handler)

# Usage with context
logger.info("Tool executed", extra={
    'request_id': '123',
    'tool_name': 'query_db',
    'duration_ms': 150
})

2.2 Distributed Tracing

from opentelemetry import trace, metrics
from opentelemetry.exporter.jaeger import JaegerExporter
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import SimpleSpanProcessor

# Setup Jaeger
jaeger_exporter = JaegerExporter(
    agent_host_name='localhost',
    agent_port=6831,
)
trace.set_tracer_provider(TracerProvider())
trace.get_tracer_provider().add_span_processor(
    SimpleSpanProcessor(jaeger_exporter)
)

tracer = trace.get_tracer(__name__)

# Usage
async def execute_tool(name: str, args: dict):
    with tracer.start_as_current_span("tool_execution") as span:
        span.set_attribute("tool.name", name)
        span.set_attribute("tool.args", str(args))
        
        # Execute tool
        result = await do_work()
        
        span.set_attribute("tool.result", str(result))
        return result

Part 3: Alerting

3.1 Alert Thresholds

Critical Alerts (Page on-call)

• Server down (health check failures)
• Error rate > 5%
• Tool execution time > 30 seconds
• Database connection failures

Warning Alerts (Create tickets)

• Error rate 1-5%
• Tool execution time > 10 seconds
• High memory usage (>80%)
• Slow response times (p99 > 5s)

3.2 Alert Configuration

groups:
- name: mcp_alerts
  rules:
  - alert: HighErrorRate
    expr: rate(mcp_errors_total[5m]) > 0.05
    for: 5m
    annotations:
      summary: "MCP error rate high: {{ $value }}"
      action: "Check server logs and deployed code"
  
  - alert: HighLatency
    expr: histogram_quantile(0.99, mcp_tool_duration_seconds_bucket) > 10
    for: 5m
    annotations:
      summary: "MCP p99 latency high: {{ $value }}s"
      action: "Review slow queries, optimize database"

Part 4: Debugging Techniques

4.1 Debug Logging

import logging

# Verbose logging for debugging
logger.setLevel(logging.DEBUG)

logger.debug("Tool call started", extra={
    'tool': name,
    'args': args,
    'timestamp': datetime.now().isoformat()
})

# Trace requests/responses
logger.debug("API request", extra={
    'method': method,
    'path': path,
    'headers': dict(headers)
})

logger.debug("API response", extra={
    'status': response.status_code,
    'body': response.text[:500],  # First 500 chars
    'duration_ms': duration_ms
})

4.2 Health Checks

@app.get("/health")
async def health_check():
    checks = {
        'service': 'healthy',
        'database': await check_database(),
        'api_gateway': await check_api_gateway(),
        'memory': check_memory_usage(),
        'timestamp': datetime.utcnow().isoformat()
    }
    
    all_healthy = all(v == 'healthy' for v in checks.values())
    status_code = 200 if all_healthy else 503
    
    return checks, status_code

async def check_database():
    try:
        # Quick health check query
        await db.execute("SELECT 1")
        return 'healthy'
    except:
        return 'unhealthy'

Part 5: Operational Procedures

5.1 On-Call Playbook

Alert: High Error Rate

1. Check error logs (last 5 minutes)
2. Identify error pattern
3. Check recent deployments
4. Rollback if necessary
5. Monitor error rate resolution

Alert: High Latency

1. Check database query times
2. Check external API responses
3. Review resource usage
4. Identify slow tools
5. Optimize or scale

5.2 Incident Response

class IncidentResponse:
    """Incident response procedures"""
    
    @staticmethod
    def detect_incident():
        return {
            'detection': 'Automated alerts + manual reports',
            'detection_time_target': '< 5 minutes',
            'escalation': 'Page on-call after 15 minutes'
        }
    
    @staticmethod
    def response_steps():
        return [
            'Acknowledge alert',
            'Assess severity',
            'Check dashboards and logs',
            'Identify root cause',
            'Implement fix or workaround',
            'Verify resolution',
            'Update status page',
            'Communicate with team'
        ]
    
    @staticmethod
    def post_incident():
        return [
            'Document timeline',
            'Identify root cause',
            'Create preventive measures',
            'Update runbooks',
            'Schedule post-mortem'
        ]

Conclusion

Comprehensive monitoring and observability are essential for operating reliable MCP servers in production. By implementing metrics, structured logging, alerting, and operational procedures, teams can rapidly detect and resolve issues.

Comprehensive 8,000+ word guide covering monitoring strategies, metrics implementation, logging, tracing, alerting, debugging, health checks, and operational procedures for MCP servers. Build observable, reliable MCP infrastructure. More on MCP Registry.