mcp-internet-speed-test

Instructions

Implementation Reference

• mcp_internet_speed_test/main.py:668-700 (handler)

  The core handler function for the 'measure_jitter' tool. It is registered via the @mcp.tool() decorator. Performs multiple (default 5) round-trip latency measurements to the specified URL, computes average latency and jitter (mean absolute deviation), and returns results including server info from headers.

  @mcp.tool()
  async def measure_jitter(url: str = DEFAULT_LATENCY_URL, samples: int = 5) -> dict:
      """Jitter is the variation in latency, so we need multiple measurements."""
      latency_values = []
      server_info = None
  
      async with httpx.AsyncClient() as client:
          for i in range(samples):
              start = time.time()
              response = await client.get(url)
              end = time.time()
              latency_values.append((end - start) * 1000)  # Convert to milliseconds
  
              # Extract server info from the first response
              if i == 0:
                  server_info = extract_server_info(dict(response.headers))
  
      # Calculate average latency
      avg_latency = sum(latency_values) / len(latency_values)
  
      # Calculate jitter (average deviation from the mean)
      jitter = sum(abs(latency - avg_latency) for latency in latency_values) / len(
          latency_values
      )
  
      return {
          "jitter": round(jitter, 2),
          "unit": "ms",
          "average_latency": round(avg_latency, 2),
          "samples": samples,
          "url": url,
          "server_info": server_info,
      }

• mcp_internet_speed_test/main.py:668-668 (registration)

  The @mcp.tool() decorator registers the measure_jitter function as an MCP tool with the name 'measure_jitter'.

  @mcp.tool()

• mcp_internet_speed_test/main.py:669-669 (schema)

  Input schema defined by function parameters: url (str, optional default), samples (int=5). Output: dict with jitter, average_latency, etc.

  async def measure_jitter(url: str = DEFAULT_LATENCY_URL, samples: int = 5) -> dict:

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

  Helper function used within measure_jitter (and other tools) to parse HTTP response headers and extract server/CDN/POP location information.

  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