arch-mcp

Instructions

Implementation Reference

• src/arch_ops_server/mirrors.py:198-303 (handler)

  Core implementation of the suggest_fastest_mirrors tool: fetches mirror status from https://archlinux.org/mirrors/status/json/, filters active/complete mirrors optionally by country, scores by sync delay and avg duration, sorts ascending by score, and returns top N mirrors with metadata.
  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/server.py:1082-1101 (registration)

  MCP tool registration in list_tools(): defines the tool name, description, and input schema (country optional str, limit int default 10).
  Tool( name="suggest_fastest_mirrors", description="[MIRRORS] Suggest optimal mirrors based on official mirror status from archlinux.org. Filters by country if specified.", inputSchema={ "type": "object", "properties": { "country": { "type": "string", "description": "Optional country code to filter mirrors (e.g., 'US', 'DE')" }, "limit": { "type": "integer", "description": "Number of mirrors to suggest (default 10)", "default": 10 } }, "required": [] } ),
• src/arch_ops_server/server.py:1428-1432 (handler)

  MCP server dispatch handler in call_tool(): extracts arguments, calls the mirrors.suggest_fastest_mirrors function, and formats result as JSON TextContent.
  elif name == "suggest_fastest_mirrors": country = arguments.get("country") limit = arguments.get("limit", 10) result = await suggest_fastest_mirrors(country=country, limit=limit) return [TextContent(type="text", text=json.dumps(result, indent=2))]
• src/arch_ops_server/tool_metadata.py:402-409 (schema)

  Tool metadata definition: categorizes as 'mirrors', platform 'any', read permission, optimize workflow, related to test_mirror_speed.
  "suggest_fastest_mirrors": ToolMetadata( name="suggest_fastest_mirrors", category="mirrors", platform="any", permission="read", workflow="optimize", related_tools=["test_mirror_speed"], prerequisite_tools=[]