""" Monitoring and diagnostics tools implementation """ import json from typing import Any, Dict, List from mcp.types import TextContent from ..utils.api_client import get_api_client from ..utils.cache import cleanup_cache, clear_cache, get_http_cache, get_response_cache from ..utils.health import get_system_health, quick_health_check from ..utils.logging import log_tool_end, log_tool_start from ..utils.metrics import get_performance_report async def get_system_health_tool(args: dict[str, Any]) -> list[TextContent]: """Get comprehensive system health report.""" start_time = log_tool_start("get_system_health", args) try: detailed = args.get("detailed", False) if detailed: health_report = get_system_health() else: # Quick health check health_statuses = quick_health_check() health_report = { "timestamp": health_statuses, "overall_status": "healthy" if all(s == "healthy" for s in health_statuses.values()) else "warning", "components": list(health_statuses.keys()), } log_tool_end("get_system_health", start_time, health_report, True) return [ TextContent( type="text", text=json.dumps(health_report, indent=2, ensure_ascii=False), ) ] except Exception as e: log_tool_end("get_system_health", start_time, {}, False, str(e)) return [TextContent(type="text", text=f"Health check failed: {e}")] async def get_performance_metrics_tool(args: dict[str, Any]) -> list[TextContent]: """Get detailed performance metrics.""" start_time = log_tool_start("get_performance_metrics", args) try: time_range = args.get("time_range", "1h") include_histogram = args.get("include_histogram", False) # Get metrics report report = get_performance_report() # Filter by time range if needed (simplified implementation) if time_range != "1h": # In a real implementation, you would filter the time series data # For now, just include the time range in the response report["time_range_requested"] = time_range if not include_histogram: # Remove histogram data if not requested report.pop("histograms", None) log_tool_end("get_performance_metrics", start_time, report, True) return [TextContent(type="text", text=json.dumps(report, indent=2, ensure_ascii=False))] except Exception as e: log_tool_end("get_performance_metrics", start_time, {}, False, str(e)) return [TextContent(type="text", text=f"Performance metrics retrieval failed: {e}")] async def get_cache_stats_tool(args: dict[str, Any]) -> list[TextContent]: """Get detailed cache statistics.""" start_time = log_tool_start("get_cache_stats", args) try: include_keys = args.get("include_keys", False) # Get cache statistics response_cache = get_response_cache() cache_stats = response_cache.get_stats() # Get HTTP cache info http_cache = get_http_cache() http_cache_lru = http_cache.cache report = { "response_cache": cache_stats, "http_cache": { "size": http_cache_lru.size(), "max_size": http_cache_lru.max_size, "default_ttl": http_cache_lru.default_ttl, }, "total_cached_items": cache_stats.get("total_items", 0), } # Clean up expired entries expired_cleaned = cleanup_cache() if expired_cleaned > 0: report["expired_entries_cleaned"] = expired_cleaned if include_keys and cache_stats.get("total_items", 0) > 0: # Include sample cache keys (first 5) sample_keys = list(http_cache_lru.cache.keys())[:5] report["sample_keys"] = sample_keys log_tool_end("get_cache_stats", start_time, report, True) return [TextContent(type="text", text=json.dumps(report, indent=2, ensure_ascii=False))] except Exception as e: log_tool_end("get_cache_stats", start_time, {}, False, str(e)) return [TextContent(type="text", text=f"Cache stats retrieval failed: {e}")] async def clear_cache_tool(args: dict[str, Any]) -> list[TextContent]: """Clear cache entries.""" start_time = log_tool_start("clear_cache", args) try: cache_type = args.get("cache_type", "all") if cache_type in ["all", "http"]: http_cache_size = get_http_cache().cache.size() else: http_cache_size = 0 if cache_type in ["all", "response"]: response_cache_size = get_response_cache().size() else: response_cache_size = 0 # Clear caches if cache_type in ["all", "http"]: clear_cache() elif cache_type == "response": get_response_cache().clear() # Get API client and clear its cache api_client = get_api_client() api_client.clear_cache() report = { "cache_type_cleared": cache_type, "http_cache_entries_removed": http_cache_size, "response_cache_entries_removed": response_cache_size, "api_client_cache_cleared": True, "success": True, } log_tool_end("clear_cache", start_time, report, True) return [TextContent(type="text", text=json.dumps(report, indent=2, ensure_ascii=False))] except Exception as e: log_tool_end("clear_cache", start_time, {}, False, str(e)) return [TextContent(type="text", text=f"Cache clearing failed: {e}")] async def get_connection_stats_tool(args: dict[str, Any]) -> list[TextContent]: """Get connection pool statistics.""" start_time = log_tool_start("get_connection_stats", args) try: api_client = get_api_client() connection_stats = api_client.get_connection_pool_stats() # Add API client performance info performance_info = api_client.get_performance_info() report = { "connection_pool": connection_stats, "api_client_performance": performance_info, "connection_health": "good" if connection_stats.get("stats_available", False) else "unknown", } log_tool_end("get_connection_stats", start_time, report, True) return [TextContent(type="text", text=json.dumps(report, indent=2, ensure_ascii=False))] except Exception as e: log_tool_end("get_connection_stats", start_time, {}, False, str(e)) return [TextContent(type="text", text=f"Connection stats retrieval failed: {e}")] async def run_diagnostic_tool(args: dict[str, Any]) -> list[TextContent]: """Run comprehensive diagnostic check.""" start_time = log_tool_start("run_diagnostic", args) try: component = args.get("component", "all") verbose = args.get("verbose", False) # Get system health health_report = get_system_health() # Get performance metrics performance_report = get_performance_report() # Get cache stats cache_stats = get_response_cache().get_stats() http_cache_stats = get_http_cache().cache.get_stats() # Get API client info api_client = get_api_client() api_performance = api_client.get_performance_info() diagnostic_report = { "diagnostic_timestamp": health_report.get("timestamp"), "overall_health": health_report.get("overall_status"), "component_statuses": health_report.get("results", {}), "performance_summary": { "uptime_seconds": performance_report.get("uptime_seconds", 0), "total_http_requests": performance_report.get("counters", {}).get( "http_requests_total", 0 ), "cache_hit_rate": performance_report.get("derived_metrics", {}).get( "cache_hit_rate", 0 ), }, "cache_status": { "response_cache": cache_stats, "http_cache": http_cache_stats, }, "api_client_status": api_performance, "recommendations": [], } # Generate recommendations if health_report.get("overall_status") != "healthy": diagnostic_report["recommendations"].append("Check unhealthy components") if cache_stats.get("total_items", 0) == 0: diagnostic_report["recommendations"].append( "Cache is empty - consider checking cache configuration" ) if api_performance.get("cache_enabled") == False: diagnostic_report["recommendations"].append( "HTTP cache is disabled - consider enabling for better performance" ) if len(diagnostic_report["recommendations"]) == 0: diagnostic_report["recommendations"].append("All systems operational") if not verbose: # Remove detailed data for non-verbose mode diagnostic_report.pop("component_statuses", None) diagnostic_report.pop("performance_summary", None) log_tool_end("run_diagnostic", start_time, diagnostic_report, True) return [ TextContent( type="text", text=json.dumps(diagnostic_report, indent=2, ensure_ascii=False), ) ] except Exception as e: log_tool_end("run_diagnostic", start_time, {}, False, str(e)) return [TextContent(type="text", text=f"Diagnostic failed: {e}")]