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)