#!/usr/bin/env python3
"""MCP Server Performance Monitor"""
import time
from collections import deque
from dataclasses import dataclass
from typing import Any, Dict, Deque, List, Optional
@dataclass
class ToolMetrics:
name: str
total_calls: int = 0
successful_calls: int = 0
failed_calls: int = 0
total_time: float = 0.0
avg_time: float = 0.0
min_time: float = float('inf')
max_time: float = 0.0
cache_hits: int = 0
cache_misses: int = 0
class PerformanceMonitor:
def __init__(self, max_history: int = 1000):
self.max_history = max_history
self.tool_metrics: Dict[str, ToolMetrics] = {}
self.request_history: Deque[float] = deque(maxlen=max_history)
self.response_times: Deque[float] = deque(maxlen=max_history)
self.start_time = time.time()
self.total_requests = 0
self.successful_requests = 0
self.failed_requests = 0
def record_tool_call(self, tool_name: str, duration: float, success: bool, cache_hit: bool = False) -> None:
if tool_name not in self.tool_metrics:
self.tool_metrics[tool_name] = ToolMetrics(name=tool_name)
metrics = self.tool_metrics[tool_name]
metrics.total_calls += 1
if success:
metrics.successful_calls += 1
else:
metrics.failed_calls += 1
if cache_hit:
metrics.cache_hits += 1
else:
metrics.cache_misses += 1
metrics.total_time += duration
metrics.avg_time = metrics.total_time / metrics.total_calls
metrics.min_time = min(metrics.min_time, duration)
metrics.max_time = max(metrics.max_time, duration)
self.total_requests += 1
if success:
self.successful_requests += 1
else:
self.failed_requests += 1
self.request_history.append(time.time())
self.response_times.append(duration)
def get_overall_stats(self) -> Dict[str, Any]:
uptime = time.time() - self.start_time
avg_response = sum(self.response_times) / len(self.response_times) if self.response_times else 0
total_cache_hits = sum(m.cache_hits for m in self.tool_metrics.values())
total_attempts = sum(m.total_calls for m in self.tool_metrics.values())
cache_rate = (total_cache_hits / total_attempts * 100) if total_attempts > 0 else 0
success_rate = (self.successful_requests / self.total_requests * 100) if self.total_requests > 0 else 0
return {
'uptime_seconds': round(uptime, 2),
'total_requests': self.total_requests,
'successful_requests': self.successful_requests,
'failed_requests': self.failed_requests,
'success_rate': round(success_rate, 2),
'avg_response_time_ms': round(avg_response * 1000, 2),
'cache_hit_rate': round(cache_rate, 2),
'active_tools': len(self.tool_metrics)
}
def get_performance_report(self) -> str:
overall = self.get_overall_stats()
lines = [
"=" * 70,
"PERFORMANCE REPORT",
"=" * 70,
"",
f"Uptime: {overall['uptime_seconds']}s",
f"Total Requests: {overall['total_requests']}",
f"Success Rate: {overall['success_rate']}%",
f"Avg Response Time: {overall['avg_response_time_ms']}ms",
f"Cache Hit Rate: {overall['cache_hit_rate']}%",
"",
"=" * 70
]
return "\n".join(lines)
_monitor: Optional[PerformanceMonitor] = None
def get_monitor() -> PerformanceMonitor:
global _monitor
if _monitor is None:
_monitor = PerformanceMonitor()
return _monitor
if __name__ == '__main__':
monitor = PerformanceMonitor()
monitor.record_tool_call("read_file", 0.025, True, False)
monitor.record_tool_call("analyze_code", 0.15, True, True)
print(monitor.get_performance_report())
