arch-mcp

Instructions

Implementation Reference

• src/arch_ops_server/mirrors.py:100-195 (handler)

  The core handler function that implements the test_mirror_speed tool. It tests mirror speeds by performing HEAD requests to core.db on specified or all active mirrors, measures latency, handles timeouts/errors, sorts results by performance, and returns structured results including 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:1068-1081 (registration)

  MCP tool registration in list_tools(), including name, description, and input schema definition.
  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": [] } ),
• src/arch_ops_server/server.py:1068-1081 (schema)

  Input schema for the test_mirror_speed tool: optional mirror_url string.
  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": [] } ),
• src/arch_ops_server/server.py:1420-1426 (registration)

  Tool dispatcher in call_tool() that invokes the test_mirror_speed handler with parsed arguments and formats the response.
  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))]