mcp-internet-speed-test

Instructions

Measure download speed using incremental file sizes.

Args:
    size_limit: Maximum file size to test (default: 100MB)

Returns:
    Dictionary with download speed results

Implementation Reference

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

  The handler function that implements the 'measure_download_speed' tool. It performs incremental downloads of binary files from GitHub raw URLs, streaming chunks, calculating real-time speed, and stops when test duration threshold (8s or 12s) is exceeded. Returns speed in Mbps with server info.

  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:464-464 (registration)

  Registers the measure_download_speed function as an MCP tool using the FastMCP decorator.

  @mcp.tool()

• mcp_internet_speed_test/main.py:465-474 (schema)

  Type hints and docstring defining the input parameter 'size_limit' (str, default '100MB') and output as dict for the measure_download_speed tool.

  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
      """

• mcp_internet_speed_test/main.py:368-460 (helper)

  Helper function called within the handler to parse HTTP response headers and extract CDN provider, POP code, location, and other server details from Fastly, Cloudflare, AWS CloudFront headers.

  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