mcp-nixos

Instructions

Implementation Reference

• mcp_nixos/server.py:605-648 (handler)

  The main handler function for the 'nixos_stats' tool. It queries the NixOS search API for package and option counts in the specified channel and formats the results as plain text. Registered via the @mcp.tool() decorator.
  @mcp.tool() async def nixos_stats(channel: str = "unstable") -> str: """Get NixOS statistics for a channel. Args: channel: NixOS channel to get stats for (e.g., "unstable", "stable", "25.05") Returns: Plain text statistics including package/option counts """ channels = get_channels() if channel not in channels: suggestions = get_channel_suggestions(channel) return error(f"Invalid channel '{channel}'. {suggestions}") try: index = channels[channel] url = f"{NIXOS_API}/{index}/_count" # Get counts with error handling try: pkg_resp = requests.post(url, json={"query": {"term": {"type": "package"}}}, auth=NIXOS_AUTH, timeout=10) pkg_resp.raise_for_status() pkg_count = pkg_resp.json().get("count", 0) except Exception: pkg_count = 0 try: opt_resp = requests.post(url, json={"query": {"term": {"type": "option"}}}, auth=NIXOS_AUTH, timeout=10) opt_resp.raise_for_status() opt_count = opt_resp.json().get("count", 0) except Exception: opt_count = 0 if pkg_count == 0 and opt_count == 0: return error("Failed to retrieve statistics") return f"""NixOS Statistics for {channel} channel: • Packages: {pkg_count:,} • Options: {opt_count:,}""" except Exception as e: return error(str(e))
• mcp_nixos/server.py:170-170 (helper)

  Global instance of ChannelCache used by nixos_stats to resolve channel names to API indices.
  channel_cache = ChannelCache()
• mcp_nixos/server.py:180-183 (helper)

  Helper function called by nixos_stats to get the channel mappings.
  def get_channels() -> dict[str, str]: """Get current channel mappings (cached and resolved).""" return channel_cache.get_resolved()
• mcp_nixos/server.py:186-198 (helper)

  Helper for channel validation, though not directly called in nixos_stats.
  """Validate if a channel exists and is accessible.""" channels = get_channels() if channel in channels: index = channels[channel] try: resp = requests.post( f"{NIXOS_API}/{index}/_count", json={"query": {"match_all": {}}}, auth=NIXOS_AUTH, timeout=5 ) return resp.status_code == 200 and resp.json().get("count", 0) > 0 except Exception: return False return False
• mcp_nixos/server.py:55-167 (helper)

  The ChannelCache class provides dynamic discovery and caching of NixOS channel indices, essential for nixos_stats to function correctly.
  class ChannelCache: """Cache for discovered channels and resolved mappings.""" def __init__(self) -> None: """Initialize empty cache.""" self.available_channels: dict[str, str] | None = None self.resolved_channels: dict[str, str] | None = None self.using_fallback: bool = False def get_available(self) -> dict[str, str]: """Get available channels, discovering if needed.""" if self.available_channels is None: self.available_channels = self._discover_available_channels() return self.available_channels if self.available_channels is not None else {} def get_resolved(self) -> dict[str, str]: """Get resolved channel mappings, resolving if needed.""" if self.resolved_channels is None: self.resolved_channels = self._resolve_channels() return self.resolved_channels if self.resolved_channels is not None else {} def _discover_available_channels(self) -> dict[str, str]: """Discover available NixOS channels by testing API patterns.""" # Test multiple generation patterns (43, 44, 45) and versions generations = [43, 44, 45, 46] # Future-proof # Removed deprecated versions (20.09, 24.11 - EOL June 2025) versions = ["unstable", "25.05", "25.11", "26.05", "30.05"] # Current and future available = {} for gen in generations: for version in versions: pattern = f"latest-{gen}-nixos-{version}" try: resp = requests.post( f"{NIXOS_API}/{pattern}/_count", json={"query": {"match_all": {}}}, auth=NIXOS_AUTH, timeout=10, # Increased from 5s to 10s for slow connections ) if resp.status_code == 200: count = resp.json().get("count", 0) if count > 0: available[pattern] = f"{count:,} documents" except Exception: continue return available def _resolve_channels(self) -> dict[str, str]: """Resolve user-friendly channel names to actual indices.""" available = self.get_available() # If no channels were discovered, use fallback channels if not available: self.using_fallback = True return FALLBACK_CHANNELS.copy() resolved = {} # Find unstable (should be consistent) unstable_pattern = None for pattern in available: if "unstable" in pattern: unstable_pattern = pattern break if unstable_pattern: resolved["unstable"] = unstable_pattern # Find stable release (highest version number with most documents) stable_candidates = [] for pattern, count_str in available.items(): if "unstable" not in pattern: # Extract version (e.g., "25.05" from "latest-43-nixos-25.05") parts = pattern.split("-") if len(parts) >= 4: version = parts[3] # "25.05" try: # Parse version for comparison (25.05 -> 25.05) major, minor = map(int, version.split(".")) count = int(count_str.replace(",", "").replace(" documents", "")) stable_candidates.append((major, minor, version, pattern, count)) except (ValueError, IndexError): continue if stable_candidates: # Sort by version (descending), then by document count (descending) as tiebreaker stable_candidates.sort(key=lambda x: (x[0], x[1], x[4]), reverse=True) current_stable = stable_candidates[0] resolved["stable"] = current_stable[3] # pattern resolved[current_stable[2]] = current_stable[3] # version -> pattern # Add other version mappings (prefer higher generation/count for same version) version_patterns: dict[str, tuple[str, int]] = {} for _major, _minor, version, pattern, count in stable_candidates: if version not in version_patterns or count > version_patterns[version][1]: version_patterns[version] = (pattern, count) for version, (pattern, _count) in version_patterns.items(): resolved[version] = pattern # Add beta (alias for stable) if "stable" in resolved: resolved["beta"] = resolved["stable"] # If we still have no channels after all that, use fallback if not resolved: self.using_fallback = True return FALLBACK_CHANNELS.copy() return resolved