mcp-internet-speed-test

Instructions

Get server information for any URL without performing speed tests.

Args:
    url_download: URL to download data from
    url_upload: URL to upload data to
    url_latency: URL to measure latency to

Returns:
    Dictionary with servers information including POP location, CDN info, etc.

Implementation Reference

• mcp_internet_speed_test/main.py:703-759 (handler)

  The handler function for the get_server_info tool. It sends HEAD requests to the download, upload, and latency URLs, extracts server information from the response headers using the extract_server_info helper, and returns detailed server info including CDN provider, POP location, and headers for each endpoint. Registered via @mcp.tool() decorator.

  @mcp.tool()
  async def get_server_info(
      url_download: str = DEFAULT_DOWNLOAD_URLS["128KB"],
      url_upload: str = DEFAULT_UPLOAD_URL,
      url_latency: str = DEFAULT_LATENCY_URL,
  ) -> dict:
      """
      Get server information for any URL without performing speed tests.
  
      Args:
          url_download: URL to download data from
          url_upload: URL to upload data to
          url_latency: URL to measure latency to
  
      Returns:
          Dictionary with servers information including POP location, CDN info, etc.
      """
      async with httpx.AsyncClient() as client:
          try:
              response_url_download = await client.head(url_download, timeout=12.0)
              server_info_url_download = extract_server_info(
                  dict(response_url_download.headers)
              )
  
              response_url_upload = await client.head(url_upload, timeout=12.0)
              server_info_url_upload = extract_server_info(
                  dict(response_url_upload.headers)
              )
  
              response_url_latency = await client.head(url_latency, timeout=12.0)
              server_info_url_latency = extract_server_info(
                  dict(response_url_latency.headers)
              )
  
              return {
                  "url_download": url_download,
                  "status_code_url_download": response_url_download.status_code,
                  "server_info_url_download": server_info_url_download,
                  "headers_url_download": dict(response_url_download.headers),
                  "url_upload": url_upload,
                  "status_code_url_upload": response_url_upload.status_code,
                  "server_info_url_upload": server_info_url_upload,
                  "headers_url_upload": dict(response_url_upload.headers),
                  "url_latency": url_latency,
                  "status_code_url_latency": response_url_latency.status_code,
                  "server_info_url_latency": server_info_url_latency,
                  "headers_url_latency": dict(response_url_latency.headers),
              }
          except (httpx.RequestError, httpx.HTTPStatusError, httpx.TimeoutException) as e:
              return {
                  "error": True,
                  "message": f"Failed to get servers info: {str(e)}",
                  "url_download": url_download,
                  "url_upload": url_upload,
                  "url_latency": url_latency,
              }

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

  Supporting helper function that parses HTTP response headers to extract CDN provider (Fastly, Cloudflare, AWS), POP code, location from mappings, cache status, and other server details. Called by get_server_info for each endpoint.

  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