JMeter MCP Server

Instructions

Implementation Reference

• jmeter_server.py:474-581 (handler)

  The main handler function for the 'generate_visualization' MCP tool. It is decorated with @mcp.tool() for registration and implements the logic to generate different types of visualizations (time_series, distribution, comparison, html_report) from JMeter JTL files using TestResultsAnalyzer and VisualizationEngine.
  async def generate_visualization(jtl_file: str, visualization_type: str, output_file: str) -> str: """Generate visualizations of JMeter test results. Args: jtl_file: Path to the JTL file containing test results visualization_type: Type of visualization to generate (time_series, distribution, comparison, html_report) output_file: Path to save the visualization Returns: str: Path to the generated visualization file """ 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 with detailed analysis analysis_results = analyzer.analyze_file(file_path, detailed=True) # Create visualization engine output_dir = os.path.dirname(output_file) if output_file else None engine = VisualizationEngine(output_dir=output_dir) # Generate visualization based on type if visualization_type == "time_series": # Extract time series metrics time_series = analysis_results.get("detailed", {}).get("time_series", []) if not time_series: return "No time series data available for visualization." # Convert to TimeSeriesMetrics objects metrics = [] for ts_data in time_series: metrics.append(TimeSeriesMetrics( timestamp=datetime.datetime.fromisoformat(ts_data["timestamp"]), active_threads=ts_data["active_threads"], throughput=ts_data["throughput"], average_response_time=ts_data["average_response_time"], error_rate=ts_data["error_rate"] )) # Create visualization output_path = engine.create_time_series_graph( metrics, metric_name="average_response_time", output_file=output_file) return f"Time series graph generated: {output_path}" elif visualization_type == "distribution": # Extract response times samples = [] for endpoint, metrics in analysis_results.get("detailed", {}).get("endpoints", {}).items(): samples.extend([metrics["average_response_time"]] * metrics["total_samples"]) if not samples: return "No response time data available for visualization." # Create visualization output_path = engine.create_distribution_graph(samples, output_file=output_file) return f"Distribution graph generated: {output_path}" elif visualization_type == "comparison": # Extract endpoint metrics endpoints = analysis_results.get("detailed", {}).get("endpoints", {}) if not endpoints: return "No endpoint data available for visualization." # Convert to EndpointMetrics objects endpoint_metrics = {} for endpoint, metrics_data in endpoints.items(): endpoint_metrics[endpoint] = EndpointMetrics( endpoint=endpoint, total_samples=metrics_data["total_samples"], error_count=metrics_data["error_count"], error_rate=metrics_data["error_rate"], average_response_time=metrics_data["average_response_time"], median_response_time=metrics_data["median_response_time"], percentile_90=metrics_data["percentile_90"], percentile_95=metrics_data["percentile_95"], percentile_99=metrics_data["percentile_99"], min_response_time=metrics_data["min_response_time"], max_response_time=metrics_data["max_response_time"], throughput=metrics_data["throughput"], test_duration=analysis_results["summary"]["duration"] ) # Create visualization output_path = engine.create_endpoint_comparison_chart( endpoint_metrics, metric_name="average_response_time", output_file=output_file) return f"Endpoint comparison chart generated: {output_path}" elif visualization_type == "html_report": # Create HTML report output_path = engine.create_html_report(analysis_results, output_file) return f"HTML report generated: {output_path}" else: return f"Unknown visualization type: {visualization_type}. " \ f"Supported types: time_series, distribution, comparison, html_report" except ValueError as e: return f"Error generating visualization: {str(e)}" except Exception as e: return f"Error generating visualization: {str(e)}"