mcp-internet-speed-test

Instructions

Input Schema

Input Schema (JSON Schema)

Implementation Reference

• mcp_internet_speed_test/main.py:465-543 (handler)

  The core handler function for the 'measure_download_speed' tool. Decorated with @mcp.tool() for automatic registration in the FastMCP server. Implements incremental download testing by streaming files of progressively larger sizes from GitHub raw URLs until the download exceeds a time threshold (8-12 seconds), continuously updating speed calculations. Includes server info extraction and returns Mbps speed with details.
  async def measure_download_speed(size_limit: str = "100MB") -> dict: """ Measure download speed using incremental file sizes. Args: size_limit: Maximum file size to test (default: 100MB) Returns: Dictionary with download 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 ) # Test each file size in order, up to the specified 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 url = DEFAULT_DOWNLOAD_URLS[size_key] start = time.time() total_size = 0 async with client.stream( "GET", url, ) as response: # Extract server information from headers server_info = extract_server_info(dict(response.headers)) async for chunk in response.aiter_bytes(chunk_size=1024): if chunk: chunk_size = len(chunk) total_size += chunk_size # Check elapsed time during download current_time = time.time() elapsed_time = current_time - start # Update our final result continuously speed_mbps = ((total_size * 8) / (1024 * 1024)) / elapsed_time final_result = { "download_speed": round(speed_mbps, 2), "elapsed_time": round(elapsed_time, 2), "data_size": total_size, "size": size_key, "url": url, "server_info": server_info, } # If test duration exceeded, stop the test if elapsed_time >= test_duration: break # Return the final result or an error if all tests failed if final_result: return { "download_speed": final_result["download_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 download tests failed", "details": results, }
• mcp_internet_speed_test/main.py:368-460 (helper)

  Supporting helper function called within measure_download_speed to parse HTTP response headers and extract detailed server/CDN information, including POP codes and locations for Fastly, Cloudflare, and AWS CloudFront.
  def extract_server_info(headers: Dict[str, str]) -> Dict[str, Optional[str]]: """ Extract server information from HTTP headers. Args: headers: HTTP response headers Returns: Dictionary with server information including POP location, CDN info, etc. """ server_info = { "cdn_provider": None, "pop_code": None, "pop_location": None, "served_by": None, "via_header": None, "cache_status": None, "server_ip_info": None, "x_cache": None, } # Extract x-served-by header (Fastly specific) served_by = headers.get("x-served-by", "") if served_by: server_info["served_by"] = served_by # Extract POP code from served-by header # Format examples: cache-mex4329-MEX, cache-qro4141-QRO, cache-dfw-kdfw8210052-DFW pop_match = re.search(r"-([A-Z]{3})$", served_by) if pop_match: server_info["pop_code"] = pop_match.group(1) server_info["pop_location"] = FASTLY_POP_LOCATIONS.get( pop_match.group(1), f"Unknown location ({pop_match.group(1)})" ) server_info["cdn_provider"] = "Fastly" # Extract via header via = headers.get("via", "") if via: server_info["via_header"] = via # Extract cache status cache_status = headers.get("x-cache", "") if cache_status: server_info["x_cache"] = cache_status server_info["cache_status"] = "HIT" if "HIT" in cache_status.upper() else "MISS" # Extract Cloudflare CF-Ray header cf_ray = headers.get("cf-ray", "") if cf_ray: server_info["cf_ray"] = cf_ray # Extract data center code from CF-Ray (format: request_id-datacenter_code) cf_match = re.search(r"-([A-Z]{3})$", cf_ray) if cf_match: server_info["pop_code"] = cf_match.group(1) server_info["pop_location"] = CLOUDFLARE_POP_LOCATIONS.get( cf_match.group(1), f"Unknown location ({cf_match.group(1)})" ) server_info["cdn_provider"] = "Cloudflare" # Extract AWS CloudFront headers cf_pop = headers.get("x-amz-cf-pop", "") cf_id = headers.get("x-amz-cf-id", "") if cf_pop: server_info["cf_pop"] = cf_pop server_info["cdn_provider"] = "Amazon CloudFront" # Extract POP code from x-amz-cf-pop header (format: DFW56-P1, SIN5-C1) cf_pop_match = re.search(r"^([A-Z]{3})", cf_pop) if cf_pop_match: server_info["pop_code"] = cf_pop_match.group(1) server_info["pop_location"] = AWS_POP_LOCATIONS.get( cf_pop_match.group(1), f"Unknown location ({cf_pop_match.group(1)})" ) if cf_id: server_info["cf_id"] = cf_id if not server_info["cdn_provider"]: server_info["cdn_provider"] = "Amazon CloudFront" # Check for other CDN indicators if not server_info["cdn_provider"]: if "fastly" in headers.get("server", "").lower(): server_info["cdn_provider"] = "Fastly" elif "cloudflare" in headers.get("server", "").lower(): server_info["cdn_provider"] = "Cloudflare" elif ( "amazon" in headers.get("server", "").lower() or "aws" in headers.get("server", "").lower() ): server_info["cdn_provider"] = "Amazon CloudFront" return server_info
• mcp_internet_speed_test/main.py:465-465 (registration)

  The @mcp.tool() decorator registers the measure_download_speed function as an MCP tool in the FastMCP server instance 'mcp'.
  async def measure_download_speed(size_limit: str = "100MB") -> dict: