measure_jitter
Measure network jitter by analyzing latency variation across multiple samples to assess connection stability for internet speed testing.
Instructions
Jitter is the variation in latency, so we need multiple measurements.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| url | No | https://httpi.dev/get | |
| samples | No |
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()
- 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:
- 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