mcp-internet-speed-test

Instructions

Implementation Reference

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

  The handler function implementing the 'measure_latency' tool logic. It measures round-trip latency to the specified URL using an asynchronous HTTP GET request with httpx, calculates the time in milliseconds, and includes server information extracted via the helper function.
  @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:641-641 (registration)

  The @mcp.tool() decorator registers the measure_latency function as an MCP tool.
  @mcp.tool()
• mcp_internet_speed_test/main.py:368-460 (helper)

  Helper function used by measure_latency to extract CDN provider, POP location, and other server 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