mcp-internet-speed-test

Instructions

Implementation Reference

• mcp_internet_speed_test/main.py:669-701 (handler)

  The main handler function for the 'measure_jitter' tool. It performs multiple latency pings to the specified URL, calculates the average latency and jitter (average deviation from the mean), and returns the results including server information.
  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-701 (registration)

  The @mcp.tool() decorator registers the measure_jitter function as an MCP tool, with parameters url (str, default DEFAULT_LATENCY_URL) and samples (int, default 5).
  @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:368-460 (helper)

  Helper function used by measure_jitter to extract server location and CDN information from HTTP response 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