suggest_fastest_mirrors
Find optimal Arch Linux mirrors by analyzing official status data and filtering by country to improve download speeds.
Instructions
[MIRRORS] Suggest optimal mirrors based on official mirror status from archlinux.org. Filters by country if specified.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| country | No | Optional country code to filter mirrors (e.g., 'US', 'DE') | |
| limit | No | Number of mirrors to suggest (default 10) |
Implementation Reference
- src/arch_ops_server/mirrors.py:198-303 (handler)The core handler function implementing the suggest_fastest_mirrors tool. It fetches mirror status data from Arch Linux's official API, filters mirrors by country (optional), activity, sync recency, and completion percentage, computes a score based on delay and average duration, sorts by score, and returns the top N mirrors with detailed metrics.async def suggest_fastest_mirrors( country: Optional[str] = None, limit: int = 10 ) -> Dict[str, Any]: """ Suggest optimal mirrors based on official mirror status. Args: country: Optional country code to filter mirrors (e.g., 'US', 'DE') limit: Number of mirrors to suggest (default 10) Returns: Dict with recommended mirrors """ logger.info(f"Fetching mirror suggestions (country={country}, limit={limit})") try: async with httpx.AsyncClient(timeout=15.0) as client: response = await client.get(MIRROR_STATUS_URL) response.raise_for_status() data = response.json() mirrors = data.get("urls", []) if not mirrors: return create_error_response( "NoData", "No mirror data available from archlinux.org" ) # Filter mirrors filtered_mirrors = [] for mirror in mirrors: # Skip if country specified and doesn't match if country and mirror.get("country_code") != country.upper(): continue # Skip if not active or has issues if not mirror.get("active", False): continue # Skip if last sync is too old (more than 24 hours) last_sync = mirror.get("last_sync") if last_sync is None: continue # Calculate score (lower is better) # Factors: completion percentage, delay, duration completion = mirror.get("completion_pct", 0) delay = mirror.get("delay", 0) or 0 # Handle None duration_avg = mirror.get("duration_avg", 0) or 0 # Skip incomplete mirrors if completion < 100: continue # Score: delay (hours) + duration (seconds converted to hours equivalent) score = delay + (duration_avg / 3600) filtered_mirrors.append({ "url": mirror.get("url"), "country": mirror.get("country"), "country_code": mirror.get("country_code"), "protocol": mirror.get("protocol"), "completion_pct": completion, "delay_hours": delay, "duration_avg": duration_avg, "duration_stddev": mirror.get("duration_stddev"), "score": round(score, 2), "last_sync": last_sync }) # Sort by score (lower is better) filtered_mirrors.sort(key=lambda x: x["score"]) # Limit results suggested_mirrors = filtered_mirrors[:limit] logger.info(f"Suggesting {len(suggested_mirrors)} mirrors") return { "suggested_count": len(suggested_mirrors), "total_available": len(filtered_mirrors), "country_filter": country, "mirrors": suggested_mirrors } except httpx.HTTPStatusError as e: logger.error(f"HTTP error fetching mirror status: {e}") return create_error_response( "HTTPError", f"Failed to fetch mirror status: HTTP {e.response.status_code}" ) except httpx.TimeoutException: logger.error("Timeout fetching mirror status") return create_error_response( "Timeout", "Request to mirror status API timed out" ) except Exception as e: logger.error(f"Failed to suggest mirrors: {e}") return create_error_response( "MirrorSuggestionError", f"Failed to suggest mirrors: {str(e)}" )
- src/arch_ops_server/__init__.py:57-62 (registration)Import statement registering the suggest_fastest_mirrors tool by exposing it from the mirrors module into the main package namespace, making it available for MCP server registration.from .mirrors import ( list_active_mirrors, test_mirror_speed, suggest_fastest_mirrors, check_mirrorlist_health )
- ToolMetadata schema defining the tool's category ('mirrors'), platform ('any'), permission ('read'), workflow ('optimize'), and relationships to other tools."suggest_fastest_mirrors": ToolMetadata( name="suggest_fastest_mirrors", category="mirrors", platform="any", permission="read", workflow="optimize", related_tools=["test_mirror_speed"], prerequisite_tools=[]
- src/arch_ops_server/__init__.py:162-162 (registration)Inclusion in the package's __all__ list, explicitly registering the tool for import via 'from arch_ops_server import suggest_fastest_mirrors'."suggest_fastest_mirrors",