#!/usr/bin/env python3 """ MCP Metrics Collector - Tool usage patterns and performance metrics Provides insights into how MIRA tools are being used """ import time import json from datetime import datetime, timedelta from typing import Dict, Any, List, Optional, Tuple from collections import defaultdict, Counter from pathlib import Path import threading import numpy as np import logging logger = logging.getLogger(__name__) class ToolUsagePattern: """Analyze tool usage patterns""" def __init__(self): self.sequences = [] # List of tool sequences self.transitions = defaultdict(Counter) # Tool A -> Tool B counts self.time_patterns = defaultdict(list) # Usage by hour of day def add_sequence(self, tools: List[str], timestamps: List[float]): """Add a sequence of tool calls""" self.sequences.append(tools) # Track transitions for i in range(len(tools) - 1): self.transitions[tools[i]][tools[i + 1]] += 1 # Track time patterns for tool, ts in zip(tools, timestamps): hour = datetime.fromtimestamp(ts).hour self.time_patterns[tool].append(hour) def get_common_sequences(self, min_length: int = 2, top_n: int = 10) -> List[Tuple[List[str], int]]: """Get most common tool sequences""" sequence_counts = Counter() for seq in self.sequences: for i in range(len(seq) - min_length + 1): subseq = tuple(seq[i:i + min_length]) sequence_counts[subseq] += 1 return [(list(seq), count) for seq, count in sequence_counts.most_common(top_n)] def get_transition_matrix(self) -> Dict[str, Dict[str, float]]: """Get probability matrix of tool transitions""" matrix = {} for from_tool, to_counts in self.transitions.items(): total = sum(to_counts.values()) matrix[from_tool] = { to_tool: count / total for to_tool, count in to_counts.items() } return matrix def get_hourly_usage(self) -> Dict[str, List[int]]: """Get usage patterns by hour""" hourly = {} for tool, hours in self.time_patterns.items(): hour_counts = [0] * 24 for h in hours: hour_counts[h] += 1 hourly[tool] = hour_counts return hourly class MetricsCollector: """ Comprehensive metrics collection for MCP server Tracks: - Tool usage frequency - Response times - Error rates - Session patterns - User behavior insights """ def __init__(self, metrics_dir: Optional[Path] = None): self.metrics_dir = metrics_dir or Path.home() / ".mira" / "metrics" self.metrics_dir.mkdir(parents=True, exist_ok=True) # Metrics storage self.tool_metrics = defaultdict(lambda: { "call_count": 0, "success_count": 0, "error_count": 0, "total_duration_ms": 0, "response_times": [], "error_types": Counter(), "parameter_patterns": Counter() }) self.session_metrics = defaultdict(lambda: { "start_time": None, "tool_sequence": [], "tool_timestamps": [], "total_calls": 0, "unique_tools": set(), "memories_created": 0, "searches_performed": 0 }) # Global metrics self.global_metrics = { "total_requests": 0, "total_errors": 0, "uptime_start": time.time(), "daily_active_sessions": set(), "peak_concurrent_sessions": 0, "current_active_sessions": 0 } # Pattern analysis self.usage_patterns = ToolUsagePattern() # Thread safety self._lock = threading.RLock() # Persistence self._load_metrics() self._start_persistence_thread() def record_tool_call(self, session_id: str, tool_name: str, params: Dict[str, Any], duration_ms: float, success: bool, error: Optional[str] = None): """Record a tool call""" with self._lock: # Update tool metrics metrics = self.tool_metrics[tool_name] metrics["call_count"] += 1 if success: metrics["success_count"] += 1 else: metrics["error_count"] += 1 if error: metrics["error_types"][error] += 1 metrics["total_duration_ms"] += duration_ms metrics["response_times"].append(duration_ms) # Keep only recent response times if len(metrics["response_times"]) > 1000: metrics["response_times"] = metrics["response_times"][-1000:] # Track parameter patterns param_pattern = self._extract_param_pattern(tool_name, params) if param_pattern: metrics["parameter_patterns"][param_pattern] += 1 # Update session metrics session = self.session_metrics[session_id] if session["start_time"] is None: session["start_time"] = time.time() session["tool_sequence"].append(tool_name) session["tool_timestamps"].append(time.time()) session["total_calls"] += 1 session["unique_tools"].add(tool_name) # Track specific behaviors if tool_name == "mira_store_memory" and success: session["memories_created"] += 1 elif "search" in tool_name and success: session["searches_performed"] += 1 # Update global metrics self.global_metrics["total_requests"] += 1 if not success: self.global_metrics["total_errors"] += 1 # Track daily active sessions today = datetime.now().strftime("%Y-%m-%d") self.global_metrics["daily_active_sessions"].add(f"{today}:{session_id}") def _extract_param_pattern(self, tool_name: str, params: Dict[str, Any]) -> Optional[str]: """Extract parameter usage pattern""" if tool_name == "mira_smart_search": # Track search query length query = params.get("query", "") length = len(query.split()) return f"words_{min(length, 10)}+" elif tool_name == "mira_store_memory": # Track memory characteristics tags = len(params.get("tags", [])) private = params.get("private", False) return f"tags_{tags}_private_{private}" elif tool_name == "mira_insights": # Track insight parameters depth = params.get("depth", "quick") has_topic = "topic" in params return f"depth_{depth}_topic_{has_topic}" return None def record_session_end(self, session_id: str): """Record end of session""" with self._lock: session = self.session_metrics.get(session_id) if session and session["tool_sequence"]: # Add to pattern analysis self.usage_patterns.add_sequence( session["tool_sequence"], session["tool_timestamps"] ) # Update active sessions self.global_metrics["current_active_sessions"] = max( 0, self.global_metrics["current_active_sessions"] - 1 ) def get_tool_stats(self, tool_name: Optional[str] = None) -> Dict[str, Any]: """Get statistics for a specific tool or all tools""" with self._lock: if tool_name: metrics = self.tool_metrics.get(tool_name, {}) if not metrics.get("call_count"): return {"error": "No data for tool"} return self._calculate_tool_stats(tool_name, metrics) else: # All tools stats = {} for name, metrics in self.tool_metrics.items(): if metrics.get("call_count"): stats[name] = self._calculate_tool_stats(name, metrics) return stats def _calculate_tool_stats(self, tool_name: str, metrics: Dict[str, Any]) -> Dict[str, Any]: """Calculate statistics for a tool""" response_times = metrics["response_times"] stats = { "tool": tool_name, "total_calls": metrics["call_count"], "success_rate": metrics["success_count"] / metrics["call_count"] if metrics["call_count"] > 0 else 0, "error_rate": metrics["error_count"] / metrics["call_count"] if metrics["call_count"] > 0 else 0, "avg_response_time_ms": metrics["total_duration_ms"] / metrics["call_count"] if metrics["call_count"] > 0 else 0 } # Response time percentiles if response_times: stats["response_time_percentiles"] = { "p50": np.percentile(response_times, 50), "p90": np.percentile(response_times, 90), "p95": np.percentile(response_times, 95), "p99": np.percentile(response_times, 99) } # Top errors if metrics["error_types"]: stats["top_errors"] = metrics["error_types"].most_common(5) # Parameter patterns if metrics["parameter_patterns"]: stats["common_patterns"] = metrics["parameter_patterns"].most_common(5) return stats def get_usage_insights(self) -> Dict[str, Any]: """Get usage pattern insights""" with self._lock: insights = { "common_sequences": self.usage_patterns.get_common_sequences(), "tool_transitions": self.usage_patterns.get_transition_matrix(), "hourly_usage": self.usage_patterns.get_hourly_usage(), "session_insights": self._get_session_insights(), "tool_relationships": self._analyze_tool_relationships() } return insights def _get_session_insights(self) -> Dict[str, Any]: """Analyze session patterns""" if not self.session_metrics: return {} session_lengths = [] tools_per_session = [] memory_creation_rate = [] for session in self.session_metrics.values(): if session["total_calls"] > 0: session_lengths.append(session["total_calls"]) tools_per_session.append(len(session["unique_tools"])) if session["memories_created"] > 0: memory_creation_rate.append( session["memories_created"] / session["total_calls"] ) insights = {} if session_lengths: insights["avg_session_length"] = np.mean(session_lengths) insights["avg_unique_tools"] = np.mean(tools_per_session) if memory_creation_rate: insights["memory_creation_rate"] = np.mean(memory_creation_rate) return insights def _analyze_tool_relationships(self) -> Dict[str, Any]: """Analyze relationships between tools""" relationships = {} # Find tools often used together tool_pairs = defaultdict(int) for session in self.session_metrics.values(): tools = list(session["unique_tools"]) for i in range(len(tools)): for j in range(i + 1, len(tools)): pair = tuple(sorted([tools[i], tools[j]])) tool_pairs[pair] += 1 if tool_pairs: relationships["frequently_paired"] = [ {"tools": list(pair), "count": count} for pair, count in sorted( tool_pairs.items(), key=lambda x: x[1], reverse=True )[:5] ] return relationships def get_performance_report(self) -> Dict[str, Any]: """Get comprehensive performance report""" with self._lock: uptime = time.time() - self.global_metrics["uptime_start"] report = { "uptime_hours": uptime / 3600, "total_requests": self.global_metrics["total_requests"], "total_errors": self.global_metrics["total_errors"], "overall_error_rate": self.global_metrics["total_errors"] / self.global_metrics["total_requests"] if self.global_metrics["total_requests"] > 0 else 0, "requests_per_hour": self.global_metrics["total_requests"] / (uptime / 3600) if uptime > 0 else 0, "unique_sessions_today": len([s for s in self.global_metrics["daily_active_sessions"] if s.startswith(datetime.now().strftime("%Y-%m-%d"))]), "tool_performance": self.get_tool_stats(), "usage_insights": self.get_usage_insights() } return report def export_prometheus_metrics(self) -> str: """Export metrics in Prometheus format""" lines = [] # Tool metrics for tool_name, metrics in self.tool_metrics.items(): safe_name = tool_name.replace("-", "_") lines.append(f'# HELP mcp_tool_calls_total Total calls to {tool_name}') lines.append(f'# TYPE mcp_tool_calls_total counter') lines.append(f'mcp_tool_calls_total{{tool="{tool_name}"}} {metrics["call_count"]}') lines.append(f'# HELP mcp_tool_errors_total Total errors for {tool_name}') lines.append(f'# TYPE mcp_tool_errors_total counter') lines.append(f'mcp_tool_errors_total{{tool="{tool_name}"}} {metrics["error_count"]}') if metrics["call_count"] > 0: avg_time = metrics["total_duration_ms"] / metrics["call_count"] lines.append(f'# HELP mcp_tool_response_time_ms Average response time for {tool_name}') lines.append(f'# TYPE mcp_tool_response_time_ms gauge') lines.append(f'mcp_tool_response_time_ms{{tool="{tool_name}"}} {avg_time:.2f}') # Global metrics lines.append('# HELP mcp_requests_total Total MCP requests') lines.append('# TYPE mcp_requests_total counter') lines.append(f'mcp_requests_total {self.global_metrics["total_requests"]}') uptime = time.time() - self.global_metrics["uptime_start"] lines.append('# HELP mcp_uptime_seconds MCP server uptime') lines.append('# TYPE mcp_uptime_seconds gauge') lines.append(f'mcp_uptime_seconds {uptime:.2f}') return '\n'.join(lines) def _save_metrics(self): """Save metrics to disk""" try: metrics_file = self.metrics_dir / f"metrics_{datetime.now().strftime('%Y%m%d')}.json" data = { "timestamp": datetime.now().isoformat(), "tool_metrics": { tool: { "call_count": m["call_count"], "success_count": m["success_count"], "error_count": m["error_count"], "total_duration_ms": m["total_duration_ms"], "error_types": dict(m["error_types"]), "parameter_patterns": dict(m["parameter_patterns"]) } for tool, m in self.tool_metrics.items() }, "global_metrics": { **self.global_metrics, "daily_active_sessions": len(self.global_metrics["daily_active_sessions"]) } } with open(metrics_file, 'w') as f: json.dump(data, f, indent=2) except Exception as e: logger.error(f"Failed to save metrics: {e}") def _load_metrics(self): """Load today's metrics if they exist""" try: metrics_file = self.metrics_dir / f"metrics_{datetime.now().strftime('%Y%m%d')}.json" if metrics_file.exists(): with open(metrics_file, 'r') as f: data = json.load(f) # Restore tool metrics for tool, metrics in data.get("tool_metrics", {}).items(): self.tool_metrics[tool].update(metrics) self.tool_metrics[tool]["error_types"] = Counter(metrics.get("error_types", {})) self.tool_metrics[tool]["parameter_patterns"] = Counter(metrics.get("parameter_patterns", {})) self.tool_metrics[tool]["response_times"] = [] logger.info(f"Loaded metrics for {len(self.tool_metrics)} tools") except Exception as e: logger.error(f"Failed to load metrics: {e}") def _start_persistence_thread(self): """Start thread to periodically save metrics""" def save_loop(): while True: time.sleep(300) # Save every 5 minutes with self._lock: self._save_metrics() thread = threading.Thread(target=save_loop, daemon=True) thread.start() # Singleton instance _metrics_collector = None def get_metrics_collector() -> MetricsCollector: """Get or create metrics collector singleton""" global _metrics_collector if _metrics_collector is None: _metrics_collector = MetricsCollector() return _metrics_collector