mcp-internet-speed-test

Instructions

Implementation Reference

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

  The @mcp.tool()-decorated handler function implementing the 'measure_upload_speed' MCP tool. It performs incremental upload speed tests by sending HTTP POST requests with data payloads of increasing sizes (from SIZE_PROGRESSION up to size_limit), measures elapsed time, calculates speed in Mbps, extracts server info, and returns the best result or error details.
  @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, }