mcp-nixos

Instructions

Implementation Reference

• mcp_nixos/server.py:1583-1583 (registration)

  Registration of the 'nixos_flakes_search' tool using @mcp.tool() decorator. This is the entry point registered with the MCP server.

  async def nixos_flakes_search(query: str, limit: int = 20, channel: str = "unstable") -> str:

• mcp_nixos/server.py:1584-1597 (handler)

  The handler function for the 'nixos_flakes_search' tool. It validates inputs minimally and delegates to the internal implementation.

  """Search NixOS flakes by name, description, owner, or repository.
  
  Searches the flake index for community-contributed packages and configurations.
  Flakes are indexed separately from official packages.
  
  Args:
      query: The search query (flake name, description, owner, or repository)
      limit: Maximum number of results to return (default: 20, max: 100)
      channel: Ignored - flakes use a separate indexing system
  
  Returns:
      Plain text list of unique flakes with their packages and metadata
  """
  return await _nixos_flakes_search_impl(query, limit, channel)

• mcp_nixos/server.py:1268-1464 (helper)

  Core implementation logic for nixos_flakes_search. Performs Elasticsearch search on flake index 'latest-43-group-manual', groups results by unique flake repository, extracts metadata like owner/repo/url/packages, and formats as plain text.

  async def _nixos_flakes_search_impl(query: str, limit: int = 20, channel: str = "unstable") -> str:
      """Internal implementation for flakes search."""
      if not 1 <= limit <= 100:
          return error("Limit must be 1-100")
  
      try:
          # Use the same alias as the web UI to get only flake packages
          flake_index = "latest-43-group-manual"
  
          # Build query for flakes
          if query.strip() == "" or query == "*":
              # Empty or wildcard query - get all flakes
              q: dict[str, Any] = {"match_all": {}}
          else:
              # Search query with multiple fields, including nested queries for flake_resolved
              q = {
                  "bool": {
                      "should": [
                          {"match": {"flake_name": {"query": query, "boost": 3}}},
                          {"match": {"flake_description": {"query": query, "boost": 2}}},
                          {"match": {"package_pname": {"query": query, "boost": 1.5}}},
                          {"match": {"package_description": query}},
                          {"wildcard": {"flake_name": {"value": f"*{query}*", "boost": 2.5}}},
                          {"wildcard": {"package_pname": {"value": f"*{query}*", "boost": 1}}},
                          {"prefix": {"flake_name": {"value": query, "boost": 2}}},
                          # Nested queries for flake_resolved fields
                          {
                              "nested": {
                                  "path": "flake_resolved",
                                  "query": {"term": {"flake_resolved.owner": query.lower()}},
                                  "boost": 2,
                              }
                          },
                          {
                              "nested": {
                                  "path": "flake_resolved",
                                  "query": {"term": {"flake_resolved.repo": query.lower()}},
                                  "boost": 2,
                              }
                          },
                      ],
                      "minimum_should_match": 1,
                  }
              }
  
          # Execute search with package filter to match web UI
          search_query = {"bool": {"filter": [{"term": {"type": "package"}}], "must": [q]}}
  
          try:
              resp = requests.post(
                  f"{NIXOS_API}/{flake_index}/_search",
                  json={"query": search_query, "size": limit * 5, "track_total_hits": True},  # Get more results
                  auth=NIXOS_AUTH,
                  timeout=10,
              )
              resp.raise_for_status()
              data = resp.json()
              hits = data.get("hits", {}).get("hits", [])
              total = data.get("hits", {}).get("total", {}).get("value", 0)
          except requests.HTTPError as e:
              if e.response and e.response.status_code == 404:
                  # No flake indices found
                  return error("Flake indices not found. Flake search may be temporarily unavailable.")
              raise
  
          # Format results as plain text
          if not hits:
              return f"""No flakes found matching '{query}'.
  
  Try searching for:
  • Popular flakes: nixpkgs, home-manager, flake-utils, devenv
  • By owner: nix-community, numtide, cachix
  • By topic: python, rust, nodejs, devops
  
  Browse flakes at:
  • GitHub: https://github.com/topics/nix-flakes
  • FlakeHub: https://flakehub.com/"""
  
          # Group hits by flake to avoid duplicates
          flakes = {}
          packages_only = []  # For entries without flake metadata
  
          for hit in hits:
              src = hit.get("_source", {})
  
              # Get flake information
              flake_name = src.get("flake_name", "").strip()
              package_pname = src.get("package_pname", "")
              resolved = src.get("flake_resolved", {})
  
              # Skip entries without any useful name
              if not flake_name and not package_pname:
                  continue
  
              # If we have flake metadata (resolved), use it to create unique key
              if isinstance(resolved, dict) and (resolved.get("owner") or resolved.get("repo") or resolved.get("url")):
                  owner = resolved.get("owner", "")
                  repo = resolved.get("repo", "")
                  url = resolved.get("url", "")
  
                  # Create a unique key based on available info
                  if owner and repo:
                      flake_key = f"{owner}/{repo}"
                      display_name = flake_name or repo or package_pname
                  elif url:
                      # Extract name from URL for git repos
                      flake_key = url
                      if "/" in url:
                          display_name = flake_name or url.rstrip("/").split("/")[-1].replace(".git", "") or package_pname
                      else:
                          display_name = flake_name or package_pname
                  else:
                      flake_key = flake_name or package_pname
                      display_name = flake_key
  
                  # Initialize flake entry if not seen
                  if flake_key not in flakes:
                      flakes[flake_key] = {
                          "name": display_name,
                          "description": src.get("flake_description") or src.get("package_description", ""),
                          "owner": owner,
                          "repo": repo,
                          "url": url,
                          "type": resolved.get("type", ""),
                          "packages": set(),  # Use set to avoid duplicates
                      }
  
                  # Add package if available
                  attr_name = src.get("package_attr_name", "")
                  if attr_name:
                      flakes[flake_key]["packages"].add(attr_name)
  
              elif flake_name:
                  # Has flake_name but no resolved metadata
                  flake_key = flake_name
  
                  if flake_key not in flakes:
                      flakes[flake_key] = {
                          "name": flake_name,
                          "description": src.get("flake_description") or src.get("package_description", ""),
                          "owner": "",
                          "repo": "",
                          "type": "",
                          "packages": set(),
                      }
  
                  # Add package if available
                  attr_name = src.get("package_attr_name", "")
                  if attr_name:
                      flakes[flake_key]["packages"].add(attr_name)
  
              else:
                  # Package without flake metadata - might still be relevant
                  packages_only.append(
                      {
                          "name": package_pname,
                          "description": src.get("package_description", ""),
                          "attr_name": src.get("package_attr_name", ""),
                      }
                  )
  
          # Build results
          results = []
          # Show both total hits and unique flakes
          if total > len(flakes):
              results.append(f"Found {total:,} total matches ({len(flakes)} unique flakes) matching '{query}':\n")
          else:
              results.append(f"Found {len(flakes)} unique flakes matching '{query}':\n")
  
          for flake in flakes.values():
              results.append(f"• {flake['name']}")
              if flake.get("owner") and flake.get("repo"):
                  results.append(
                      f"  Repository: {flake['owner']}/{flake['repo']}"
                      + (f" ({flake['type']})" if flake.get("type") else "")
                  )
              elif flake.get("url"):
                  results.append(f"  URL: {flake['url']}")
              if flake.get("description"):
                  desc = flake["description"]
                  if len(desc) > 200:
                      desc = desc[:200] + "..."
                  results.append(f"  {desc}")
              if flake["packages"]:
                  # Show max 5 packages, sorted
                  packages = sorted(flake["packages"])[:5]
                  if len(flake["packages"]) > 5:
                      results.append(f"  Packages: {', '.join(packages)}, ... ({len(flake['packages'])} total)")
                  else:
                      results.append(f"  Packages: {', '.join(packages)}")
              results.append("")
  
          return "\n".join(results).strip()
  
      except Exception as e:
          return error(str(e))

• mcp_nixos/server.py:360-362 (handler)

  Secondary usage in 'nixos_search' tool: redirects 'flakes' search_type to the same implementation.

  # Redirect flakes to dedicated function
  if search_type == "flakes":
      return await _nixos_flakes_search_impl(query, limit)