mcp-nixos

Instructions

Implementation Reference

• mcp_nixos/server.py:538-603 (handler)

  The nixos_channels tool handler: lists NixOS channels by discovering available indices via API, resolving friendly names like 'stable'/'unstable', with fallback mappings and status checks.
  @mcp.tool() async def nixos_channels() -> str: """List available NixOS channels with their status. Returns: Plain text list showing channel names, versions, and availability """ try: # Get resolved channels and available raw data configured = get_channels() available = channel_cache.get_available() results = [] # Show warning if using fallback channels if channel_cache.using_fallback: results.append("⚠️ WARNING: Using fallback channels (API discovery failed)") results.append(" Check network connectivity to search.nixos.org") results.append("") results.append("NixOS Channels (fallback mode):\n") else: results.append("NixOS Channels (auto-discovered):\n") # Show user-friendly channel names for name, index in sorted(configured.items()): status = "✓ Available" if index in available else "✗ Unavailable" doc_count = available.get(index, "Unknown") # Mark stable channel clearly label = f"• {name}" if name == "stable": # Extract version from index parts = index.split("-") if len(parts) >= 4: version = parts[3] label = f"• {name} (current: {version})" results.append(f"{label} → {index}") if index in available: results.append(f" Status: {status} ({doc_count})") else: if channel_cache.using_fallback: results.append(" Status: Fallback (may not be current)") else: results.append(f" Status: {status}") results.append("") # Show additional discovered channels not in our mapping if not channel_cache.using_fallback: discovered_only = set(available.keys()) - set(configured.values()) if discovered_only: results.append("Additional available channels:") for index in sorted(discovered_only): results.append(f"• {index} ({available[index]})") # Add deprecation warnings results.append("\nNote: Channels are dynamically discovered.") results.append("'stable' always points to the current stable release.") if channel_cache.using_fallback: results.append("\n⚠️ Fallback channels may not reflect the latest available versions.") results.append(" Please check your network connection to search.nixos.org.") return "\n".join(results).strip() except Exception as e: return error(str(e))
• mcp_nixos/server.py:55-167 (helper)

  ChannelCache class: discovers available channels by testing API endpoints, resolves aliases like 'stable' to concrete indices, caches results, and provides fallback static mappings.
  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
• mcp_nixos/server.py:538-538 (registration)

  The @mcp.tool() decorator registers the nixos_channels function as an MCP tool.
  @mcp.tool()
• mcp_nixos/server.py:180-183 (helper)

  get_channels() helper: retrieves the resolved channel mappings from the cache instance.
  def get_channels() -> dict[str, str]: """Get current channel mappings (cached and resolved).""" return channel_cache.get_resolved()
• mcp_nixos/server.py:170-170 (helper)

  Global channel_cache instance used by nixos_channels and related functions.
  channel_cache = ChannelCache()