mcp-nixos

Instructions

Implementation Reference

• mcp_nixos/server.py:1145-1266 (handler)

  The handler function for the 'nixos_flakes_stats' tool. It queries the NixOS Elasticsearch API's flake index ('latest-43-group-manual') to retrieve statistics including total flake packages, unique repositories (by sampling 10k documents), flake types, and top 5 contributors by extracting from flake URLs (GitHub, Codeberg, sr.ht). Formats results as plain text.
  async def nixos_flakes_stats() -> str: """Get statistics about available NixOS flakes. Retrieves statistics from the flake search index including total packages, unique repositories, flake types, and top contributors. Returns: Plain text summary with flake statistics and top contributors """ try: # Use the same alias as the web UI for accurate counts flake_index = "latest-43-group-manual" # Get total count of flake packages (not options or apps) try: resp = requests.post( f"{NIXOS_API}/{flake_index}/_count", json={"query": {"term": {"type": "package"}}}, auth=NIXOS_AUTH, timeout=10, ) resp.raise_for_status() total_packages = resp.json().get("count", 0) except requests.HTTPError as e: if e.response.status_code == 404: return error("Flake indices not found. Flake search may be temporarily unavailable.") raise # Get unique flakes by sampling documents # Since aggregations on text fields don't work, we'll sample and count manually unique_urls = set() type_counts: dict[str, int] = {} contributor_counts: dict[str, int] = {} try: # Get a large sample of documents to count unique flakes resp = requests.post( f"{NIXOS_API}/{flake_index}/_search", json={ "size": 10000, # Get a large sample "query": {"term": {"type": "package"}}, # Only packages "_source": ["flake_resolved", "flake_name", "package_pname"], }, auth=NIXOS_AUTH, timeout=10, ) resp.raise_for_status() data = resp.json() hits = data.get("hits", {}).get("hits", []) # Process hits to extract unique URLs for hit in hits: src = hit.get("_source", {}) resolved = src.get("flake_resolved", {}) if isinstance(resolved, dict) and "url" in resolved: url = resolved["url"] unique_urls.add(url) # Count types flake_type = resolved.get("type", "unknown") type_counts[flake_type] = type_counts.get(flake_type, 0) + 1 # Extract contributor from URL contributor = None if "github.com/" in url: parts = url.split("github.com/")[1].split("/") if parts: contributor = parts[0] elif "codeberg.org/" in url: parts = url.split("codeberg.org/")[1].split("/") if parts: contributor = parts[0] elif "sr.ht/~" in url: parts = url.split("sr.ht/~")[1].split("/") if parts: contributor = parts[0] if contributor: contributor_counts[contributor] = contributor_counts.get(contributor, 0) + 1 unique_count = len(unique_urls) # Format type info type_info = [] for type_name, count in sorted(type_counts.items(), key=lambda x: x[1], reverse=True)[:5]: if type_name: type_info.append(f" - {type_name}: {count:,}") # Format contributor info owner_info = [] for contributor, count in sorted(contributor_counts.items(), key=lambda x: x[1], reverse=True)[:5]: owner_info.append(f" - {contributor}: {count:,} packages") except Exception: # Fallback if query fails unique_count = 0 type_info = [] owner_info = [] # Build statistics results = [] results.append("NixOS Flakes Statistics:") results.append(f"• Available flakes: {total_packages:,}") if unique_count > 0: results.append(f"• Unique repositories: {unique_count:,}") if type_info: results.append("• Flake types:") results.extend(type_info) if owner_info: results.append("• Top contributors:") results.extend(owner_info) results.append("\nNote: Flakes are community-contributed and indexed separately from official packages.") return "\n".join(results) except Exception as e: return error(str(e))
• mcp_nixos/server.py:1145-1145 (registration)

  Registration of the 'nixos_flakes_stats' tool using the FastMCP @mcp.tool() decorator.
  async def nixos_flakes_stats() -> str: