JMeter MCP Server

Instructions

Implementation Reference

• jmeter_server.py:150-302 (handler)

  The core handler function for the 'analyze_jmeter_results' tool. Decorated with @mcp.tool() for MCP registration. Analyzes JMeter JTL file using TestResultsAnalyzer, computes summary metrics, detailed endpoint/bottleneck analysis, insights, and formats into a comprehensive string report.
  @mcp.tool() async def analyze_jmeter_results(jtl_file: str, detailed: bool = False) -> str: """Analyze JMeter test results and provide a summary of key metrics and insights. Args: jtl_file: Path to the JTL file containing test results detailed: Whether to include detailed analysis (default: False) Returns: str: Analysis results in a formatted string """ try: analyzer = TestResultsAnalyzer() # Validate file exists file_path = Path(jtl_file) if not file_path.exists(): return f"Error: JTL file not found: {jtl_file}" try: # Analyze the file analysis_results = analyzer.analyze_file(file_path, detailed=detailed) # Format the results as a string result_str = f"Analysis of {jtl_file}:\n\n" # Add summary information summary = analysis_results.get("summary", {}) result_str += "Summary:\n" result_str += f"- Total samples: {summary.get('total_samples', 'N/A')}\n" result_str += f"- Error count: {summary.get('error_count', 'N/A')} ({summary.get('error_rate', 'N/A'):.2f}%)\n" result_str += f"- Response times (ms):\n" result_str += f" - Average: {summary.get('average_response_time', 'N/A'):.2f}\n" result_str += f" - Median: {summary.get('median_response_time', 'N/A'):.2f}\n" result_str += f" - 90th percentile: {summary.get('percentile_90', 'N/A'):.2f}\n" result_str += f" - 95th percentile: {summary.get('percentile_95', 'N/A'):.2f}\n" result_str += f" - 99th percentile: {summary.get('percentile_99', 'N/A'):.2f}\n" result_str += f" - Min: {summary.get('min_response_time', 'N/A'):.2f}\n" result_str += f" - Max: {summary.get('max_response_time', 'N/A'):.2f}\n" result_str += f"- Throughput: {summary.get('throughput', 'N/A'):.2f} requests/second\n" result_str += f"- Start time: {summary.get('start_time', 'N/A')}\n" result_str += f"- End time: {summary.get('end_time', 'N/A')}\n" result_str += f"- Duration: {summary.get('duration', 'N/A'):.2f} seconds\n\n" # Add detailed information if requested if detailed and "detailed" in analysis_results: detailed_info = analysis_results["detailed"] # Add endpoint information endpoints = detailed_info.get("endpoints", {}) if endpoints: result_str += "Endpoint Analysis:\n" for endpoint, metrics in endpoints.items(): result_str += f"- {endpoint}:\n" result_str += f" - Samples: {metrics.get('total_samples', 'N/A')}\n" result_str += f" - Errors: {metrics.get('error_count', 'N/A')} ({metrics.get('error_rate', 'N/A'):.2f}%)\n" result_str += f" - Average response time: {metrics.get('average_response_time', 'N/A'):.2f} ms\n" result_str += f" - 95th percentile: {metrics.get('percentile_95', 'N/A'):.2f} ms\n" result_str += f" - Throughput: {metrics.get('throughput', 'N/A'):.2f} requests/second\n" result_str += "\n" # Add bottleneck information bottlenecks = detailed_info.get("bottlenecks", {}) if bottlenecks: result_str += "Bottleneck Analysis:\n" # Slow endpoints slow_endpoints = bottlenecks.get("slow_endpoints", []) if slow_endpoints: result_str += "- Slow Endpoints:\n" for endpoint in slow_endpoints: result_str += f" - {endpoint.get('endpoint')}: {endpoint.get('response_time'):.2f} ms " result_str += f"(Severity: {endpoint.get('severity')})\n" result_str += "\n" # Error-prone endpoints error_endpoints = bottlenecks.get("error_prone_endpoints", []) if error_endpoints: result_str += "- Error-Prone Endpoints:\n" for endpoint in error_endpoints: result_str += f" - {endpoint.get('endpoint')}: {endpoint.get('error_rate'):.2f}% " result_str += f"(Severity: {endpoint.get('severity')})\n" result_str += "\n" # Anomalies anomalies = bottlenecks.get("anomalies", []) if anomalies: result_str += "- Response Time Anomalies:\n" for anomaly in anomalies[:3]: # Show only top 3 anomalies result_str += f" - At {anomaly.get('timestamp')}: " result_str += f"Expected {anomaly.get('expected_value'):.2f} ms, " result_str += f"Got {anomaly.get('actual_value'):.2f} ms " result_str += f"({anomaly.get('deviation_percentage'):.2f}% deviation)\n" result_str += "\n" # Concurrency impact concurrency = bottlenecks.get("concurrency_impact", {}) if concurrency: result_str += "- Concurrency Impact:\n" correlation = concurrency.get("correlation", 0) result_str += f" - Correlation between threads and response time: {correlation:.2f}\n" if concurrency.get("has_degradation", False): result_str += f" - Performance degradation detected at {concurrency.get('degradation_threshold')} threads\n" else: result_str += " - No significant performance degradation detected with increasing threads\n" result_str += "\n" # Add insights and recommendations insights = detailed_info.get("insights", {}) if insights: result_str += "Insights and Recommendations:\n" # Recommendations recommendations = insights.get("recommendations", []) if recommendations: result_str += "- Top Recommendations:\n" for rec in recommendations[:3]: # Show only top 3 recommendations result_str += f" - [{rec.get('priority_level', 'medium').upper()}] {rec.get('issue')}\n" result_str += f" Recommendation: {rec.get('recommendation')}\n" result_str += f" Expected Impact: {rec.get('expected_impact')}\n" result_str += "\n" # Scaling insights scaling_insights = insights.get("scaling_insights", []) if scaling_insights: result_str += "- Scaling Insights:\n" for insight in scaling_insights[:2]: # Show only top 2 insights result_str += f" - {insight.get('topic')}: {insight.get('description')}\n" result_str += "\n" # Add time series information (just a summary) time_series = detailed_info.get("time_series", []) if time_series: result_str += "Time Series Analysis:\n" result_str += f"- Intervals: {len(time_series)}\n" result_str += f"- Interval duration: 5 seconds\n" # Calculate average throughput and response time over intervals avg_throughput = sum(ts.get('throughput', 0) for ts in time_series) / len(time_series) avg_response_time = sum(ts.get('average_response_time', 0) for ts in time_series) / len(time_series) result_str += f"- Average throughput over intervals: {avg_throughput:.2f} requests/second\n" result_str += f"- Average response time over intervals: {avg_response_time:.2f} ms\n\n" return result_str except ValueError as e: return f"Error analyzing JTL file: {str(e)}" except Exception as e: return f"Error analyzing JMeter results: {str(e)}"