mcp-internet-speed-test

Instructions

Input Schema

Input Schema (JSON Schema)

Implementation Reference

• mcp_internet_speed_test/main.py:703-759 (handler)

  The handler function for the 'get_server_info' tool. It performs HEAD requests to the provided download, upload, and latency URLs, extracts server information from the response headers using the 'extract_server_info' helper, and returns a dictionary with status codes, server info, and full headers for each URL.
  @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)

  Helper function that parses HTTP response headers to extract server details such as CDN provider (Fastly, Cloudflare, CloudFront), POP code and location, cache status, and other metadata. Used by get_server_info and other speed test functions.
  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
• mcp_internet_speed_test/main.py:703-703 (registration)

  The @mcp.tool() decorator registers the get_server_info function as an MCP tool, using its signature for input schema.
  @mcp.tool()
• mcp_internet_speed_test/main.py:704-708 (schema)

  Input schema derived from function parameters: three optional string parameters for URLs with defaults, returning dict.
  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: