mcp-internet-speed-test

Instructions

Measure upload speed using incremental file sizes.

Args:
    url_upload: URL to upload data to
    size_limit: Maximum file size to test (default: 100MB)

Returns:
    Dictionary with upload speed results

Implementation Reference

• mcp_internet_speed_test/main.py:546-639 (handler)

  The core handler function for the 'measure_upload_speed' tool. It is decorated with @mcp.tool() for registration and implements incremental upload testing by sending data payloads of increasing sizes to the specified URL, calculating speed based on transfer time until the duration threshold is exceeded.

  @mcp.tool()
  async def measure_upload_speed(
      url_upload: str = DEFAULT_UPLOAD_URL, size_limit: str = "100MB"
  ) -> dict:
      """
      Measure upload speed using incremental file sizes.
  
      Args:
          url_upload: URL to upload data to
          size_limit: Maximum file size to test (default: 100MB)
  
      Returns:
          Dictionary with upload speed results
      """
      results = []
      final_result = None
  
      # Find the index of the size limit in our progression
      max_index = (
          SIZE_PROGRESSION.index(size_limit)
          if size_limit in SIZE_PROGRESSION
          else len(SIZE_PROGRESSION) - 1
      )
  
      # Only test up to the specified size limit
      async with httpx.AsyncClient() as client:
          for size_key in SIZE_PROGRESSION[: max_index + 1]:
              if size_key in ["100MB", "200MB", "500MB", "1GB"]:
                  test_duration = BASE_TEST_DURATION + ADDITIONAL_TEST_DURATION
              else:
                  test_duration = BASE_TEST_DURATION
  
              data_size = UPLOAD_SIZES[size_key]
              data = b"x" * data_size
              start = time.time()
  
              try:
                  response = await client.post(url_upload, data=data, timeout=30.0)
                  end = time.time()
                  elapsed_time = end - start
  
                  # Extract server information from headers
                  server_info = extract_server_info(dict(response.headers))
  
                  # Calculate upload speed in Mbps
                  speed_mbps = (data_size * 8) / (1024 * 1024) / elapsed_time
                  result = {
                      "size": size_key,
                      "upload_speed": round(speed_mbps, 2),
                      "elapsed_time": round(elapsed_time, 2),
                      "data_size": data_size,
                      "url": url_upload,
                      "server_info": server_info,
                  }
  
                  results.append(result)
  
                  # Set the final result to the last result
                  final_result = result
  
                  # If this test took longer than our threshold, we're done
                  if elapsed_time > test_duration:
                      break
  
              except (httpx.RequestError, httpx.HTTPStatusError, httpx.TimeoutException) as e:
                  results.append(
                      {
                          "size": size_key,
                          "error": True,
                          "message": f"HTTP Error: {str(e)}",
                          "url": url_upload,
                      }
                  )
                  # If we encounter an error, use the last successful result or continue
                  if final_result:
                      break
  
      # Return the final result or an error if all tests failed
      if final_result:
          return {
              "upload_speed": final_result["upload_speed"],
              "unit": "Mbps",
              "elapsed_time": final_result["elapsed_time"],
              "data_size": final_result["data_size"],
              "size_used": final_result["size"],
              "server_info": final_result["server_info"],
              "all_tests": results,
          }
      return {
          "error": True,
          "message": "All upload tests failed",
          "details": results,
      }