mcp-internet-speed-test

Instructions

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