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()