MiniMCP - A minimal, stateless, and lightweight framework for building MCP servers

benchmarks/minimcp/README.md

@@ -0,0 +1,78 @@
+# MiniMCP vs FastMCP · Benchmarks
+
+Latest report: [Comprehensive Benchmark Report](./reports/COMPREHENSIVE_BENCHMARK_REPORT.md)
+
+Once you've set up a development environment as described in [CONTRIBUTING.md](../../CONTRIBUTING.md), you can run the benchmark scripts.
+
+## Running Benchmarks
+
+Each transport has a separate benchmark script that can be run with the following commands. Only tool calling is used for benchmarking as other primitives aren't much different functionally. Each script produces two result files: one for sync tool calls and another for async tool calls.
+
+```bash
+# Stdio
+uv run python -m benchmarks.minimcp.macro.stdio_mcp_server_benchmark
+
+# HTTP
+uv run python -m benchmarks.minimcp.macro.http_mcp_server_benchmark
+
+# Streamable HTTP
+uv run python -m benchmarks.minimcp.macro.streamable_http_mcp_server_benchmark
+```
+
+### System Preparation - Best practice in Ubuntu
+
+The following steps can help you get consistent benchmark results. They are specifically for Ubuntu, but similar steps may exist for other operating systems.
+
+```bash
+# Stop unnecessary services
+sudo systemctl stop snapd
+sudo systemctl stop unattended-upgrades
+
+# Disable CPU frequency scaling (use performance governor)
+echo performance | sudo tee /sys/devices/system/cpu/cpu*/cpufreq/scaling_governor
+
+# Disable turbo boost for consistency (optional but recommended)
+echo 1 | sudo tee /sys/devices/system/cpu/intel_pstate/no_turbo  # Intel
+# OR for AMD:
+echo 0 | sudo tee /sys/devices/system/cpu/cpufreq/boost
+```
+
+After the above steps, you can run the benchmark scripts with `taskset` to pin to specific CPU cores. This ensures the benchmark always runs on the same CPU cores, avoiding cache misses and CPU migration overhead.
+
+```bash
+taskset -c 0-3 uv run python -m <benchmark.module>
+```
+
+### Load Profiles
+
+The benchmark uses four load profiles to test performance under different concurrency levels:
+
+| Load       | Concurrency | Iterations | Rounds | Total Messages |
+|------------|-------------|------------|--------|----------------|
+| Sequential | 1           | 30         | 40     | 1,200          |
+| Light      | 20          | 30         | 40     | 24,000         |
+| Medium     | 100         | 15         | 40     | 60,000         |
+| Heavy      | 300         | 15         | 40     | 180,000        |
+
+## Analyze Results
+
+The `analyze_results.py` script provides a visual comparison of benchmark results between MiniMCP and FastMCP. It displays response time comparisons across all load profiles with visual bar charts, performance improvements as percentages, memory usage comparisons, key findings, and metadata.
+
+You can run it for each result JSON file with:
+
+```bash
+# Stdio
+uv run python benchmarks/minimcp/analyze_results.py benchmarks/minimcp/reports/stdio_mcp_server_sync_benchmark_results.json
+
+uv run python benchmarks/minimcp/analyze_results.py benchmarks/minimcp/reports/stdio_mcp_server_async_benchmark_results.json
+
+# HTTP
+uv run python benchmarks/minimcp/analyze_results.py benchmarks/minimcp/reports/http_mcp_server_sync_benchmark_results.json
+
+uv run python benchmarks/minimcp/analyze_results.py benchmarks/minimcp/reports/http_mcp_server_async_benchmark_results.json
+
+# Streamable HTTP
+uv run python benchmarks/minimcp/analyze_results.py benchmarks/minimcp/reports/streamable_http_mcp_server_sync_benchmark_results.json
+
+uv run python benchmarks/minimcp/analyze_results.py benchmarks/minimcp/reports/streamable_http_mcp_server_async_benchmark_results.json
+```

benchmarks/minimcp/analyze_results.py

@@ -0,0 +1,198 @@
+"""
+Analyze and visualize MCP server benchmark results.
+Analyzer generated by Claude 4.5 Sonnet.
+"""
+
+import json
+import sys
+from pathlib import Path
+from typing import Any
+
+LOAD_INFO = {
+    "sequential_load": ("Sequential Load", "1 concurrent request"),
+    "light_load": ("Light Load", "20 concurrent requests"),
+    "medium_load": ("Medium Load", "100 concurrent requests"),
+    "heavy_load": ("Heavy Load", "300 concurrent requests"),
+}
+
+
+def load_results(json_path: Path) -> dict[str, Any]:
+    """Load benchmark results from JSON file."""
+
+    if not json_path.exists():
+        print(f"Error: File not found: {json_path}")
+        sys.exit(1)
+
+    with open(json_path) as f:
+        return json.load(f)
+
+
+def calculate_improvement(minimcp_val: float, fastmcp_val: float, lower_is_better: bool = True) -> float:
+    """Calculate percentage improvement."""
+    if lower_is_better:
+        return ((fastmcp_val - minimcp_val) / fastmcp_val) * 100
+    else:
+        return ((minimcp_val - fastmcp_val) / fastmcp_val) * 100
+
+
+def print_title(title: str) -> None:
+    # Bold + Underline
+    print("\033[1m\033[4m" + title + "\033[0m\n")
+
+
+def organize_results(results: dict[str, Any]) -> tuple[dict[str, Any], dict[str, Any]]:
+    """Organize results by server and load."""
+    data: dict[str, dict[str, Any]] = {}
+    for result in results["results"]:
+        server = result["server_name"]
+        load = result["load_name"]
+        if server not in data:
+            data[server] = {}
+        data[server][load] = result["metrics"]
+
+    return data["minimcp"], data["fastmcp"]
+
+
+def print_metadata(results: dict[str, Any]) -> None:
+    """Print metadata."""
+    min, sec = divmod(results["metadata"]["duration_seconds"], 60)
+    print(f"Date: {results['metadata']['timestamp']}")
+    print(f"Duration: {min:.0f}m {sec:.0f}s\n")
+
+
+def print_key_findings(results: dict[str, Any]) -> None:
+    """Print key findings section."""
+    print_title("Key Findings")
+
+    minimcp, fastmcp = organize_results(results)
+
+    # Response time improvements (excluding sequential)
+    response_improvements: list[float] = []
+    for load in ["light_load", "medium_load", "heavy_load"]:
+        min_rt = minimcp[load]["response_time"]["mean"]
+        fast_rt = fastmcp[load]["response_time"]["mean"]
+        improvement = calculate_improvement(min_rt, fast_rt, lower_is_better=True)
+        response_improvements.append(improvement)
+
+    rt_min = min(response_improvements)
+    rt_max = max(response_improvements)
+
+    # Throughput improvements
+    throughput_improvements: list[float] = []
+    for load in ["light_load", "medium_load", "heavy_load"]:
+        min_tp = minimcp[load]["throughput_rps"]["mean"]
+        fast_tp = fastmcp[load]["throughput_rps"]["mean"]
+        improvement = calculate_improvement(min_tp, fast_tp, lower_is_better=False)
+        throughput_improvements.append(improvement)
+
+    tp_min = min(throughput_improvements)
+    tp_max = max(throughput_improvements)
+
+    # Memory improvements
+    memory_improvements: list[float] = []
+    for load in ["medium_load", "heavy_load"]:
+        min_mem = minimcp[load]["max_memory_usage"]["mean"]
+        fast_mem = fastmcp[load]["max_memory_usage"]["mean"]
+        improvement = calculate_improvement(min_mem, fast_mem, lower_is_better=True)
+        memory_improvements.append(improvement)
+
+    mem_min = min(memory_improvements)
+    mem_max = max(memory_improvements)
+
+    print(
+        f"- MiniMCP outperforms FastMCP by ~{rt_min:.0f}-{rt_max:.0f}% in response time across "
+        "all concurrent load scenarios"
+    )
+    print(f"- MiniMCP achieves ~{tp_min:.0f}-{tp_max:.0f}% higher throughput than FastMCP")
+
+    # Handle memory improvements (can be positive or negative)
+    if mem_min >= 0 and mem_max >= 0:
+        print(f"- MiniMCP uses ~{mem_min:.0f}-{mem_max:.0f}% less memory under medium to heavy loads")
+    elif mem_min < 0 and mem_max < 0:
+        print(f"- MiniMCP uses ~{abs(mem_max):.0f}-{abs(mem_min):.0f}% more memory under medium to heavy loads")
+    else:
+        print(
+            f"- MiniMCP memory usage varies from {mem_min:.0f}% to {mem_max:.0f}% compared to FastMCP under medium "
+            "to heavy loads"
+        )
+    print()
+
+
+def print_response_time_visualization(results: dict[str, Any]) -> None:
+    """Print response time visualization."""
+    print_title("Response Time Visualization (smaller is better)")
+
+    minimcp, fastmcp = organize_results(results)
+
+    for load_key, (title, subtitle) in LOAD_INFO.items():
+        min_rt = minimcp[load_key]["response_time"]["mean"] * 1000  # to ms
+        fast_rt = fastmcp[load_key]["response_time"]["mean"] * 1000
+        improvement = calculate_improvement(min_rt, fast_rt, lower_is_better=True)
+
+        # Scale bars (max 50 chars for fastmcp)
+        max_val = max(min_rt, fast_rt)
+        fast_bars = int((fast_rt / max_val) * 50)
+        min_bars = int((min_rt / max_val) * 50)
+
+        # Determine if minimcp is better or worse
+        if improvement > 0:
+            status = f"✓ {improvement:.1f}% faster"
+        else:
+            status = f"✗ {abs(improvement):.1f}% slower"
+
+        print(f"{title} ({subtitle})")
+        print(f"minimcp  {'▓' * min_bars} {min_rt:.2f}ms  {status}")
+        print(f"fastmcp  {'▓' * fast_bars} {fast_rt:.2f}ms")
+        print()
+    print()
+
+
+def print_memory_visualization(results: dict[str, Any]) -> None:
+    """Print maximum memory usage visualization."""
+    print_title("Maximum Memory Usage Visualization (smaller is better)")
+
+    minimcp, fastmcp = organize_results(results)
+
+    for load_key, (title, subtitle) in LOAD_INFO.items():
+        min_mem = minimcp[load_key]["max_memory_usage"]["mean"]
+        fast_mem = fastmcp[load_key]["max_memory_usage"]["mean"]
+        improvement = calculate_improvement(min_mem, fast_mem, lower_is_better=True)
+
+        # Scale bars (max 50 chars for the higher value)
+        max_val = max(min_mem, fast_mem)
+        min_bars = int((min_mem / max_val) * 50)
+        fast_bars = int((fast_mem / max_val) * 50)
+
+        # Determine if minimcp is better or worse
+        if improvement > 0:
+            status = f"✓ {improvement:.1f}% lower"
+        else:
+            status = f"✗ {abs(improvement):.1f}% higher"
+
+        print(f"{title} ({subtitle})")
+        print(f"minimcp  {'▓' * min_bars} {min_mem:,.0f} KB  {status}")
+        print(f"fastmcp  {'▓' * fast_bars} {fast_mem:,.0f} KB")
+        print()
+    print()
+
+
+def main() -> None:
+    """Main entry point."""
+    if len(sys.argv) != 2:
+        print("Usage: python analyze_results.py <results.json>")
+        sys.exit(1)
+
+    json_path = Path(sys.argv[1])
+    results = load_results(json_path)
+
+    print()
+    print_title("Benchmark Analysis")
+
+    print_metadata(results)
+    print_key_findings(results)
+    print_response_time_visualization(results)
+    print_memory_visualization(results)
+
+
+if __name__ == "__main__":
+    main()

benchmarks/minimcp/configs.py

@@ -0,0 +1,24 @@
+import os
+
+from benchmarks.minimcp.core.mcp_server_benchmark import Load
+
+# --- Server Configuration ---
+
+SERVER_HOST = os.environ.get("TEST_SERVER_HOST", "127.0.0.1")
+SERVER_PORT = int(os.environ.get("TEST_SERVER_PORT", "30789"))
+
+HTTP_MCP_PATH = "/mcp"
+
+
+# --- Paths ---
+
+REPORTS_DIR = "benchmarks/minimcp/reports"
+
+# --- Load Configuration ---
+
+LOADS = [
+    Load(name="sequential_load", concurrency=1, iterations=30, rounds=40),
+    Load(name="light_load", concurrency=20, iterations=30, rounds=40),
+    Load(name="medium_load", concurrency=100, iterations=15, rounds=40),
+    Load(name="heavy_load", concurrency=300, iterations=15, rounds=40),
+]