mcp-internet-speed-test

Instructions

Args:
    url (str): The URL to measure latency to

Returns:
    Dictionary with latency result

Implementation Reference

• mcp_internet_speed_test/main.py:641-666 (handler)

  The core handler function for the 'measure_latency' tool, registered via @mcp.tool() decorator. Measures round-trip latency to the given URL using httpx, extracts server info, and returns latency in ms.

  @mcp.tool()
  async def measure_latency(url: str = DEFAULT_LATENCY_URL) -> dict:
      """Measure the latency
  
      Args:
          url (str): The URL to measure latency to
  
      Returns:
          Dictionary with latency result
      """
      start = time.time()
      async with httpx.AsyncClient() as client:
          response = await client.get(url)
      end = time.time()
      elapsed_time = end - start
  
      # Extract server information from headers
      server_info = extract_server_info(dict(response.headers))
  
      return {
          "latency": round(elapsed_time * 1000, 2),  # Convert to milliseconds
          "unit": "ms",
          "url": url,
          "server_info": server_info,
      }

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

  Helper function extract_server_info that parses HTTP response headers to identify CDN provider (Fastly, Cloudflare, AWS CloudFront), POP codes, locations, and other server details. Used by measure_latency to enrich the response.

  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