test_mirror_speed
Test Arch Linux mirror response times to identify optimal download sources. Measure specific mirrors or evaluate all active mirrors for performance comparison.
Instructions
[MIRRORS] Test mirror response time. Can test a specific mirror or all active mirrors. Only works on Arch Linux.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| mirror_url | No | Specific mirror URL to test, or omit to test all active mirrors |
Implementation Reference
- src/arch_ops_server/mirrors.py:100-195 (handler)The core handler function that implements the test_mirror_speed tool. It tests the response time (latency) of specified or all active Arch Linux mirrors by performing HTTP HEAD requests to core.db, handles timeouts and errors, sorts results by speed, and returns structured results with the fastest mirror.async def test_mirror_speed(mirror_url: Optional[str] = None) -> Dict[str, Any]: """ Test mirror response time. Args: mirror_url: Specific mirror URL to test, or None to test all active mirrors Returns: Dict with mirror latency results """ if not IS_ARCH: return create_error_response( "NotSupported", "This feature is only available on Arch Linux" ) logger.info(f"Testing mirror speed: {mirror_url or 'all active'}") try: mirrors_to_test = [] if mirror_url: mirrors_to_test = [mirror_url] else: # Get active mirrors result = await list_active_mirrors() if "error" in result: return result mirrors_to_test = [m["url"] for m in result.get("active_mirrors", [])] if not mirrors_to_test: return create_error_response( "NoMirrors", "No mirrors to test" ) results = [] async with httpx.AsyncClient(timeout=10.0, follow_redirects=True) as client: for mirror in mirrors_to_test: # Replace $repo and $arch with actual values for testing test_url = mirror.replace("$repo", "core").replace("$arch", "x86_64") # Add a test file path (core.db is small and always present) if not test_url.endswith('/'): test_url += '/' test_url += "core.db" try: start_time = time.time() response = await client.head(test_url) latency = (time.time() - start_time) * 1000 # Convert to ms results.append({ "mirror": mirror, "latency_ms": round(latency, 2), "status_code": response.status_code, "success": response.status_code == 200 }) except httpx.TimeoutException: results.append({ "mirror": mirror, "latency_ms": -1, "status_code": 0, "success": False, "error": "timeout" }) except Exception as e: results.append({ "mirror": mirror, "latency_ms": -1, "status_code": 0, "success": False, "error": str(e) }) # Sort by latency (successful tests first) results.sort(key=lambda x: (not x["success"], x["latency_ms"] if x["latency_ms"] > 0 else float('inf'))) logger.info(f"Tested {len(results)} mirrors") return { "tested_count": len(results), "results": results, "fastest": results[0] if results and results[0]["success"] else None } except Exception as e: logger.error(f"Failed to test mirrors: {e}") return create_error_response( "MirrorTestError", f"Failed to test mirror speed: {str(e)}" )
- src/arch_ops_server/server.py:1067-1080 (registration)MCP tool registration in @server.list_tools(). Defines the tool name, description, and input schema (optional mirror_url). This makes the tool discoverable to MCP clients.Tool( name="test_mirror_speed", description="[MIRRORS] Test mirror response time. Can test a specific mirror or all active mirrors. Only works on Arch Linux.", inputSchema={ "type": "object", "properties": { "mirror_url": { "type": "string", "description": "Specific mirror URL to test, or omit to test all active mirrors" } }, "required": [] } ),
- Dispatch handler in @server.call_tool() that validates platform (Arch only), extracts arguments, calls the core test_mirror_speed function, and formats the JSON response for MCP.elif name == "test_mirror_speed": if not IS_ARCH: return [TextContent(type="text", text="Error: test_mirror_speed only available on Arch Linux systems")] mirror_url = arguments.get("mirror_url") result = await test_mirror_speed(mirror_url=mirror_url) return [TextContent(type="text", text=json.dumps(result, indent=2))]
- Tool metadata defining category (mirrors), platform requirements, permissions, workflow, and related/prerequisite tools for better tool discovery and chaining."test_mirror_speed": ToolMetadata( name="test_mirror_speed", category="mirrors", platform="arch", permission="read", workflow="optimize", related_tools=["suggest_fastest_mirrors", "list_active_mirrors"], prerequisite_tools=["list_active_mirrors"] ),
- src/arch_ops_server/__init__.py:57-62 (registration)Import registration in package __init__.py that exposes test_mirror_speed for use in server.py and other modules.from .mirrors import ( list_active_mirrors, test_mirror_speed, suggest_fastest_mirrors, check_mirrorlist_health )