This is page 3 of 3. Use http://codebase.md/utensils/mcp-nixos?page={x} to view the full context.
# Directory Structure
```
├── .claude
│ ├── agents
│ │ ├── mcp-server-architect.md
│ │ ├── nix-expert.md
│ │ └── python-expert.md
│ ├── commands
│ │ └── release.md
│ └── settings.json
├── .dockerignore
├── .envrc
├── .github
│ └── workflows
│ ├── ci.yml
│ ├── claude-code-review.yml
│ ├── claude.yml
│ ├── deploy-flakehub.yml
│ ├── deploy-website.yml
│ └── publish.yml
├── .gitignore
├── .mcp.json
├── .pre-commit-config.yaml
├── CLAUDE.md
├── Dockerfile
├── flake.lock
├── flake.nix
├── LICENSE
├── MANIFEST.in
├── mcp_nixos
│ ├── __init__.py
│ └── server.py
├── pyproject.toml
├── pytest.ini
├── README.md
├── RELEASE_NOTES.md
├── RELEASE_WORKFLOW.md
├── smithery.yaml
├── tests
│ ├── __init__.py
│ ├── conftest.py
│ ├── test_channels.py
│ ├── test_edge_cases.py
│ ├── test_evals.py
│ ├── test_flakes.py
│ ├── test_integration.py
│ ├── test_main.py
│ ├── test_mcp_behavior.py
│ ├── test_mcp_tools.py
│ ├── test_nixhub.py
│ ├── test_nixos_stats.py
│ ├── test_options.py
│ ├── test_plain_text_output.py
│ ├── test_real_world_scenarios.py
│ ├── test_regression.py
│ └── test_server.py
├── uv.lock
└── website
├── .eslintignore
├── .eslintrc.json
├── .gitignore
├── .prettierignore
├── .prettierrc
├── .vscode
│ └── settings.json
├── app
│ ├── about
│ │ └── page.tsx
│ ├── docs
│ │ └── claude.html
│ ├── globals.css
│ ├── layout.tsx
│ ├── page.tsx
│ ├── test-code-block
│ │ └── page.tsx
│ └── usage
│ └── page.tsx
├── components
│ ├── AnchorHeading.tsx
│ ├── ClientFooter.tsx
│ ├── ClientNavbar.tsx
│ ├── CodeBlock.tsx
│ ├── CollapsibleSection.tsx
│ ├── FeatureCard.tsx
│ ├── Footer.tsx
│ └── Navbar.tsx
├── metadata-checker.html
├── netlify.toml
├── next.config.js
├── package-lock.json
├── package.json
├── postcss.config.js
├── public
│ ├── favicon
│ │ ├── android-chrome-192x192.png
│ │ ├── android-chrome-512x512.png
│ │ ├── apple-touch-icon.png
│ │ ├── browserconfig.xml
│ │ ├── favicon-16x16.png
│ │ ├── favicon-32x32.png
│ │ ├── favicon.ico
│ │ ├── mstile-150x150.png
│ │ ├── README.md
│ │ └── site.webmanifest
│ ├── images
│ │ ├── .gitkeep
│ │ ├── attribution.md
│ │ ├── claude-logo.png
│ │ ├── JamesBrink.jpeg
│ │ ├── mcp-nixos.png
│ │ ├── nixos-snowflake-colour.svg
│ │ ├── og-image.png
│ │ ├── sean-callan.png
│ │ └── utensils-logo.png
│ ├── robots.txt
│ └── sitemap.xml
├── README.md
├── tailwind.config.js
├── tsconfig.json
└── windsurf_deployment.yaml
```
# Files
--------------------------------------------------------------------------------
/tests/test_evals.py:
--------------------------------------------------------------------------------
```python
"""Basic evaluation tests for MCP-NixOS to validate AI usability."""
from dataclasses import dataclass
from unittest.mock import Mock, patch
import pytest
from mcp_nixos import server
def get_tool_function(tool_name: str):
"""Get the underlying function from a FastMCP tool."""
tool = getattr(server, tool_name)
if hasattr(tool, "fn"):
return tool.fn
return tool
# Get the underlying functions for direct use
darwin_search = get_tool_function("darwin_search")
darwin_info = get_tool_function("darwin_info")
darwin_list_options = get_tool_function("darwin_list_options")
darwin_options_by_prefix = get_tool_function("darwin_options_by_prefix")
home_manager_search = get_tool_function("home_manager_search")
home_manager_info = get_tool_function("home_manager_info")
home_manager_list_options = get_tool_function("home_manager_list_options")
home_manager_options_by_prefix = get_tool_function("home_manager_options_by_prefix")
nixos_info = get_tool_function("nixos_info")
nixos_search = get_tool_function("nixos_search")
nixos_stats = get_tool_function("nixos_stats")
# Removed duplicate classes - kept the more comprehensive versions below
class TestErrorHandlingEvals:
"""Evaluations for error scenarios."""
@pytest.fixture(autouse=True)
def mock_channel_validation(self):
"""Mock channel validation to always pass for 'unstable'."""
with patch("mcp_nixos.server.channel_cache") as mock_cache:
mock_cache.get_available.return_value = {"unstable": "latest-45-nixos-unstable"}
mock_cache.get_resolved.return_value = {"unstable": "latest-45-nixos-unstable"}
with patch("mcp_nixos.server.validate_channel") as mock_validate:
mock_validate.return_value = True
yield mock_cache
@pytest.mark.asyncio
async def test_invalid_channel_error(self):
"""User specifies invalid channel - should get clear error."""
result = await nixos_search("firefox", channel="invalid-channel")
# Should get a clear error message
assert "Error (ERROR): Invalid channel 'invalid-channel'" in result
@patch("mcp_nixos.server.requests.post")
@pytest.mark.asyncio
async def test_package_not_found(self, mock_post):
"""User searches for non-existent package."""
mock_response = Mock()
mock_response.json.return_value = {"hits": {"hits": []}}
mock_response.raise_for_status = Mock()
mock_post.return_value = mock_response
result = await nixos_info("nonexistentpackage", type="package")
# Should get informative not found error
assert "Error (NOT_FOUND): Package 'nonexistentpackage' not found" in result
class TestCompleteScenarioEval:
"""End-to-end scenario evaluation."""
@pytest.fixture(autouse=True)
def mock_channel_validation(self):
"""Mock channel validation to always pass for 'unstable'."""
with patch("mcp_nixos.server.channel_cache") as mock_cache:
mock_cache.get_available.return_value = {"unstable": "latest-45-nixos-unstable"}
mock_cache.get_resolved.return_value = {"unstable": "latest-45-nixos-unstable"}
with patch("mcp_nixos.server.validate_channel") as mock_validate:
mock_validate.return_value = True
yield mock_cache
@patch("mcp_nixos.server.requests.post")
@patch("mcp_nixos.server.requests.get")
@pytest.mark.asyncio
async def test_complete_firefox_installation_flow(self, mock_get, mock_post):
"""Complete flow: user wants Firefox with specific Home Manager config."""
# Step 1: Search for Firefox package
search_resp = Mock()
search_resp.json.return_value = {
"hits": {
"hits": [
{
"_source": {
"package_pname": "firefox",
"package_pversion": "121.0",
"package_description": "A web browser built from Firefox source tree",
}
}
]
}
}
search_resp.raise_for_status = Mock()
# Step 2: Get package details
info_resp = Mock()
info_resp.json.return_value = {
"hits": {
"hits": [
{
"_source": {
"package_pname": "firefox",
"package_pversion": "121.0",
"package_description": "A web browser built from Firefox source tree",
"package_homepage": ["https://www.mozilla.org/firefox/"],
"package_license_set": ["MPL-2.0"],
}
}
]
}
}
info_resp.raise_for_status = Mock()
# Step 3: Search Home Manager options
hm_resp = Mock()
hm_resp.content = b"""
<html>
<dt>programs.firefox.enable</dt>
<dd>
<p>Whether to enable Firefox.</p>
<span class="term">Type: boolean</span>
</dd>
</html>
"""
hm_resp.raise_for_status = Mock()
mock_post.side_effect = [search_resp, info_resp]
mock_get.return_value = hm_resp
# Execute the flow
# 1. Search for Firefox
result1 = await nixos_search("firefox")
assert "Found 1 packages matching 'firefox':" in result1
assert "• firefox (121.0)" in result1
# 2. Get detailed info
result2 = await nixos_info("firefox")
assert "Package: firefox" in result2
assert "Homepage: https://www.mozilla.org/firefox/" in result2
# 3. Check Home Manager options
result3 = await home_manager_search("firefox")
assert "• programs.firefox.enable" in result3
# AI should now have all info needed to guide user through installation
# ===== Content from test_evals_comprehensive.py =====
@dataclass
class EvalScenario:
"""Represents an evaluation scenario."""
name: str
user_query: str
expected_tool_calls: list[str]
success_criteria: list[str]
description: str = ""
@dataclass
class EvalResult:
"""Result of running an evaluation."""
scenario: EvalScenario
passed: bool
score: float # 0.0 to 1.0
tool_calls_made: list[tuple[str, dict, str]] # (tool_name, args, result)
criteria_met: dict[str, bool]
reasoning: str
class MockAIAssistant:
"""Simulates an AI assistant using the MCP tools."""
def __init__(self):
self.tool_calls = []
async def process_query(self, query: str) -> list[tuple[str, dict, str]]:
"""Process a user query and return tool calls made."""
self.tool_calls = []
# Simulate AI decision making based on query
if ("install" in query.lower() or "get" in query.lower()) and any(
pkg in query.lower() for pkg in ["vscode", "firefox", "git"]
):
await self._handle_package_installation(query)
elif ("configure" in query.lower() or "set up" in query.lower()) and "nginx" in query.lower():
await self._handle_service_configuration(query)
elif (
"home manager" in query.lower()
or "should i configure" in query.lower()
or ("manage" in query.lower() and "home manager" in query.lower())
):
await self._handle_home_manager_query(query)
elif "dock" in query.lower() and ("darwin" in query.lower() or "macos" in query.lower()):
await self._handle_darwin_query(query)
elif "difference between" in query.lower():
await self._handle_comparison_query(query)
return self.tool_calls
async def _make_tool_call(self, tool_name: str, **kwargs) -> str:
"""Make a tool call and record it."""
# Map tool names to actual functions
tools = {
"nixos_search": nixos_search,
"nixos_info": nixos_info,
"nixos_stats": nixos_stats,
"home_manager_search": home_manager_search,
"home_manager_info": home_manager_info,
"home_manager_list_options": home_manager_list_options,
"home_manager_options_by_prefix": home_manager_options_by_prefix,
"darwin_search": darwin_search,
"darwin_info": darwin_info,
"darwin_list_options": darwin_list_options,
"darwin_options_by_prefix": darwin_options_by_prefix,
}
if tool_name in tools:
result = await tools[tool_name](**kwargs)
self.tool_calls.append((tool_name, kwargs, result))
return result
return ""
async def _handle_package_installation(self, query: str):
"""Handle package installation queries."""
# Extract package name
package = None
if "vscode" in query.lower():
package = "vscode"
elif "firefox" in query.lower():
package = "firefox"
elif "git" in query.lower():
package = "git"
if package:
# Search for the package
await self._make_tool_call("nixos_search", query=package, search_type="packages")
# If it's a command, also search programs
if package == "git":
await self._make_tool_call("nixos_search", query=package, search_type="programs")
# Get detailed info
await self._make_tool_call("nixos_info", name=package, type="package")
async def _handle_service_configuration(self, query: str):
"""Handle service configuration queries."""
if "nginx" in query.lower():
# Search for nginx options
await self._make_tool_call("nixos_search", query="services.nginx", search_type="options")
# Get specific option info
await self._make_tool_call("nixos_info", name="services.nginx.enable", type="option")
await self._make_tool_call("nixos_info", name="services.nginx.virtualHosts", type="option")
async def _handle_home_manager_query(self, query: str):
"""Handle Home Manager related queries."""
if "git" in query.lower():
# Search both system and user options
await self._make_tool_call("nixos_search", query="git", search_type="packages")
await self._make_tool_call("home_manager_search", query="programs.git")
await self._make_tool_call("home_manager_info", name="programs.git.enable")
elif "shell" in query.lower():
# Handle shell configuration queries
await self._make_tool_call("home_manager_search", query="programs.zsh")
await self._make_tool_call("home_manager_info", name="programs.zsh.enable")
await self._make_tool_call("home_manager_options_by_prefix", option_prefix="programs.zsh")
async def _handle_darwin_query(self, query: str):
"""Handle Darwin/macOS queries."""
if "dock" in query.lower():
await self._make_tool_call("darwin_search", query="system.defaults.dock")
await self._make_tool_call("darwin_info", name="system.defaults.dock.autohide")
await self._make_tool_call("darwin_options_by_prefix", option_prefix="system.defaults.dock")
async def _handle_comparison_query(self, query: str):
"""Handle package comparison queries."""
if "firefox" in query.lower():
await self._make_tool_call("nixos_search", query="firefox", search_type="packages")
await self._make_tool_call("nixos_info", name="firefox", type="package")
await self._make_tool_call("nixos_info", name="firefox-esr", type="package")
class EvalFramework:
"""Framework for running and scoring evaluations."""
def __init__(self):
self.assistant = MockAIAssistant()
async def run_eval(self, scenario: EvalScenario) -> EvalResult:
"""Run a single evaluation scenario."""
# Have the assistant process the query
tool_calls = await self.assistant.process_query(scenario.user_query)
# Check which criteria were met
criteria_met = self._check_criteria(scenario, tool_calls)
# Calculate score
score = sum(1 for met in criteria_met.values() if met) / len(criteria_met)
passed = score >= 0.7 # 70% threshold
# Generate reasoning
reasoning = self._generate_reasoning(scenario, tool_calls, criteria_met)
return EvalResult(
scenario=scenario,
passed=passed,
score=score,
tool_calls_made=tool_calls,
criteria_met=criteria_met,
reasoning=reasoning,
)
def _check_criteria(self, scenario: EvalScenario, tool_calls: list[tuple[str, dict, str]]) -> dict[str, bool]:
"""Check which success criteria were met."""
criteria_met = {}
# Check expected tool calls
expected_tools = set()
for expected_call in scenario.expected_tool_calls:
# Parse expected call (handle "await" prefix)
if expected_call.startswith("await "):
tool_name = expected_call[6:].split("(")[0] # Skip "await "
else:
tool_name = expected_call.split("(")[0]
expected_tools.add(tool_name)
actual_tools = {call[0] for call in tool_calls}
criteria_met["made_expected_tool_calls"] = expected_tools.issubset(actual_tools)
# Check specific criteria based on scenario
all_results = "\n".join(call[2] for call in tool_calls)
for criterion in scenario.success_criteria:
if "finds" in criterion and "package" in criterion:
# Check if package was found
criteria_met[criterion] = any("Found" in call[2] and "packages" in call[2] for call in tool_calls)
elif "mentions" in criterion:
# Check if certain text is mentioned
key_term = criterion.split("mentions")[1].strip()
criteria_met[criterion] = key_term.lower() in all_results.lower()
elif "provides" in criterion:
# Check if examples/syntax provided
criteria_met[criterion] = bool(tool_calls) and len(all_results) > 100
elif "explains" in criterion:
# Check if explanation provided (has meaningful content)
criteria_met[criterion] = len(all_results) > 200
else:
# Default: assume met if we have results
criteria_met[criterion] = bool(tool_calls)
return criteria_met
def _generate_reasoning(
self, scenario: EvalScenario, tool_calls: list[tuple[str, dict, str]], criteria_met: dict[str, bool]
) -> str:
"""Generate reasoning about the evaluation result."""
parts = []
# Tool usage
if tool_calls:
parts.append(f"Made {len(tool_calls)} tool calls")
else:
parts.append("No tool calls made")
# Criteria summary
met_count = sum(1 for met in criteria_met.values() if met)
parts.append(f"Met {met_count}/{len(criteria_met)} criteria")
# Specific issues
for criterion, met in criteria_met.items():
if not met:
parts.append(f"Failed: {criterion}")
return "; ".join(parts)
class TestPackageDiscoveryEvals:
"""Evaluations for package discovery scenarios."""
def setup_method(self):
self.framework = EvalFramework()
@patch("mcp_nixos.server.es_query")
@pytest.mark.asyncio
async def test_eval_find_vscode_package(self, mock_query):
"""Eval: User wants to install VSCode."""
# Mock responses
mock_query.return_value = [
{
"_source": {
"package_pname": "vscode",
"package_pversion": "1.85.0",
"package_description": "Open source code editor by Microsoft",
}
}
]
scenario = EvalScenario(
name="find_vscode",
user_query="I want to install VSCode on NixOS",
expected_tool_calls=[
"await nixos_search(query='vscode', search_type='packages')",
"await nixos_info(name='vscode', type='package')",
],
success_criteria=["finds vscode package", "mentions configuration.nix", "provides installation syntax"],
)
result = await self.framework.run_eval(scenario)
# Verify evaluation
assert result.passed
assert result.score >= 0.7
assert len(result.tool_calls_made) >= 2
assert any("vscode" in str(call) for call in result.tool_calls_made)
@patch("mcp_nixos.server.es_query")
@pytest.mark.asyncio
async def test_eval_find_git_command(self, mock_query):
"""Eval: User wants git command."""
# Mock different responses for different queries
def query_side_effect(*args, **kwargs):
query = args[1]
if "program" in str(query):
return [{"_source": {"package_programs": ["git"], "package_pname": "git"}}]
return [
{
"_source": {
"package_pname": "git",
"package_pversion": "2.43.0",
"package_description": "Distributed version control system",
}
}
]
mock_query.side_effect = query_side_effect
scenario = EvalScenario(
name="find_git_command",
user_query="How do I get the 'git' command on NixOS?",
expected_tool_calls=[
"await nixos_search(query='git', search_type='programs')",
"await nixos_info(name='git', type='package')",
],
success_criteria=[
"identifies git package",
"explains system vs user installation",
"shows both environment.systemPackages and Home Manager options",
],
)
result = await self.framework.run_eval(scenario)
assert result.passed
assert any("programs" in str(call[1]) for call in result.tool_calls_made)
@patch("mcp_nixos.server.es_query")
@pytest.mark.asyncio
async def test_eval_package_comparison(self, mock_query):
"""Eval: User needs to compare packages."""
# Mock responses for firefox variants
def query_side_effect(*args, **kwargs):
return [
{
"_source": {
"package": {
"pname": "firefox",
"version": "120.0",
"description": "Mozilla Firefox web browser",
}
}
}
]
mock_query.side_effect = query_side_effect
scenario = EvalScenario(
name="compare_firefox_variants",
user_query="What's the difference between firefox and firefox-esr?",
expected_tool_calls=[
"await nixos_search(query='firefox', search_type='packages')",
"await nixos_info(name='firefox', type='package')",
"await nixos_info(name='firefox-esr', type='package')",
],
success_criteria=[
"explains ESR vs regular versions",
"mentions stability vs features trade-off",
"provides configuration examples for both",
],
)
result = await self.framework.run_eval(scenario)
# Check that comparison tools were called
assert len(result.tool_calls_made) >= 2
assert any("firefox-esr" in str(call) for call in result.tool_calls_made)
class TestServiceConfigurationEvals:
"""Evaluations for service configuration scenarios."""
def setup_method(self):
self.framework = EvalFramework()
@patch("mcp_nixos.server.es_query")
@pytest.mark.asyncio
async def test_eval_nginx_setup(self, mock_query):
"""Eval: User wants to set up nginx."""
mock_query.return_value = [
{
"_source": {
"option_name": "services.nginx.enable",
"option_type": "boolean",
"option_description": "Whether to enable nginx web server",
}
}
]
scenario = EvalScenario(
name="nginx_setup",
user_query="How do I set up nginx on NixOS to serve static files?",
expected_tool_calls=[
"await nixos_search(query='services.nginx', search_type='options')",
"await nixos_info(name='services.nginx.enable', type='option')",
"await nixos_info(name='services.nginx.virtualHosts', type='option')",
],
success_criteria=[
"enables nginx service",
"configures virtual host",
"explains directory structure",
"mentions firewall configuration",
"provides complete configuration.nix example",
],
)
result = await self.framework.run_eval(scenario)
assert len(result.tool_calls_made) >= 2
assert any("nginx" in call[2] for call in result.tool_calls_made)
@patch("mcp_nixos.server.es_query")
@pytest.mark.asyncio
async def test_eval_database_setup(self, mock_query):
"""Eval: User wants PostgreSQL setup."""
mock_query.return_value = [
{
"_source": {
"option": {
"option_name": "services.postgresql.enable",
"option_type": "boolean",
"option_description": "Whether to enable PostgreSQL",
}
}
}
]
scenario = EvalScenario(
name="postgresql_setup",
user_query="Set up PostgreSQL with a database for my app",
expected_tool_calls=[
"await nixos_search(query='services.postgresql', search_type='options')",
"await nixos_info(name='services.postgresql.enable', type='option')",
"await nixos_info(name='services.postgresql.ensureDatabases', type='option')",
"await nixos_info(name='services.postgresql.ensureUsers', type='option')",
],
success_criteria=[
"enables postgresql service",
"creates database",
"sets up user with permissions",
"explains connection details",
"mentions backup considerations",
],
)
# This scenario would need more complex mocking in real implementation
# For now, just verify the structure works
result = await self.framework.run_eval(scenario)
assert isinstance(result, EvalResult)
class TestHomeManagerIntegrationEvals:
"""Evaluations for Home Manager vs system configuration."""
def setup_method(self):
self.framework = EvalFramework()
@patch("mcp_nixos.server.es_query")
@patch("mcp_nixos.server.parse_html_options")
@pytest.mark.asyncio
async def test_eval_user_vs_system_config(self, mock_parse, mock_query):
"""Eval: User confused about where to configure git."""
# Mock system package
mock_query.return_value = [
{
"_source": {
"package": {
"pname": "git",
"version": "2.43.0",
"description": "Distributed version control system",
}
}
}
]
# Mock Home Manager options
mock_parse.return_value = [
{"name": "programs.git.enable", "type": "boolean", "description": "Enable git"},
{"name": "programs.git.userName", "type": "string", "description": "Git user name"},
]
scenario = EvalScenario(
name="git_config_location",
user_query="Should I configure git in NixOS or Home Manager?",
expected_tool_calls=[
"await nixos_search(query='git', search_type='packages')",
"await home_manager_search(query='programs.git')",
"await home_manager_info(name='programs.git.enable')",
],
success_criteria=[
"explains system vs user configuration",
"recommends Home Manager for user configs",
"shows both approaches",
"explains when to use each",
],
)
result = await self.framework.run_eval(scenario)
assert len(result.tool_calls_made) >= 3
assert any("home_manager" in call[0] for call in result.tool_calls_made)
@patch("mcp_nixos.server.parse_html_options")
@pytest.mark.asyncio
async def test_eval_dotfiles_management(self, mock_parse):
"""Eval: User wants to manage shell config."""
mock_parse.return_value = [
{"name": "programs.zsh.enable", "type": "boolean", "description": "Enable zsh"},
{"name": "programs.zsh.oh-my-zsh.enable", "type": "boolean", "description": "Enable Oh My Zsh"},
]
scenario = EvalScenario(
name="shell_config",
user_query="How do I manage my shell configuration with Home Manager?",
expected_tool_calls=[
"await home_manager_search(query='programs.zsh')",
"await home_manager_info(name='programs.zsh.enable')",
"await home_manager_options_by_prefix(option_prefix='programs.zsh')",
],
success_criteria=[
"enables shell program",
"explains configuration options",
"mentions aliases and plugins",
"provides working example",
],
)
result = await self.framework.run_eval(scenario)
assert any("zsh" in str(call) for call in result.tool_calls_made)
class TestDarwinPlatformEvals:
"""Evaluations for macOS/nix-darwin scenarios."""
def setup_method(self):
self.framework = EvalFramework()
@patch("mcp_nixos.server.parse_html_options")
@pytest.mark.asyncio
async def test_eval_macos_dock_settings(self, mock_parse):
"""Eval: User wants to configure macOS dock."""
mock_parse.return_value = [
{"name": "system.defaults.dock.autohide", "type": "boolean", "description": "Auto-hide dock"},
{"name": "system.defaults.dock.tilesize", "type": "integer", "description": "Dock icon size"},
]
scenario = EvalScenario(
name="macos_dock_config",
user_query="How do I configure dock settings with nix-darwin?",
expected_tool_calls=[
"await darwin_search(query='system.defaults.dock')",
"await darwin_info(name='system.defaults.dock.autohide')",
"await darwin_options_by_prefix(option_prefix='system.defaults.dock')",
],
success_criteria=[
"finds dock configuration options",
"explains autohide and other settings",
"provides darwin-configuration.nix example",
"mentions darwin-rebuild command",
],
)
result = await self.framework.run_eval(scenario)
assert len(result.tool_calls_made) >= 2
assert any("darwin" in call[0] for call in result.tool_calls_made)
assert any("dock" in str(call) for call in result.tool_calls_made)
class TestEvalReporting:
"""Test evaluation reporting functionality."""
@pytest.mark.asyncio
async def test_eval_result_generation(self):
"""Test that eval results are properly generated."""
scenario = EvalScenario(
name="test_scenario",
user_query="Test query",
expected_tool_calls=["await nixos_search(query='test')"],
success_criteria=["finds test package"],
)
result = EvalResult(
scenario=scenario,
passed=True,
score=1.0,
tool_calls_made=[
("nixos_search", {"query": "test"}, "Found 1 packages matching 'test':\n\n• test (1.0.0)")
],
criteria_met={"finds test package": True},
reasoning="Made 1 tool calls; Met 1/1 criteria",
)
assert result.passed
assert result.score == 1.0
assert len(result.tool_calls_made) == 1
def test_eval_scoring(self):
"""Test evaluation scoring logic."""
# Create a scenario with multiple criteria
EvalScenario(
name="multi_criteria",
user_query="Test with multiple criteria",
expected_tool_calls=[],
success_criteria=["criterion1", "criterion2", "criterion3"],
)
# Test partial success
criteria_met = {"criterion1": True, "criterion2": True, "criterion3": False}
score = sum(1 for met in criteria_met.values() if met) / len(criteria_met)
assert score == pytest.approx(0.666, rel=0.01)
assert score < 0.7 # Below passing threshold
def generate_eval_report(self, results: list[EvalResult]) -> str:
"""Generate a report from evaluation results."""
total = len(results)
passed = sum(1 for r in results if r.passed)
avg_score = sum(r.score for r in results) / total if total > 0 else 0
report = f"""# MCP-NixOS Evaluation Report
## Summary
- Total Evaluations: {total}
- Passed: {passed} ({passed / total * 100:.1f}%)
- Average Score: {avg_score:.2f}
## Detailed Results
"""
for result in results:
status = "✅ PASS" if result.passed else "❌ FAIL"
report += f"\n### {status} {result.scenario.name} (Score: {result.score:.2f})\n"
report += f"Query: {result.scenario.user_query}\n"
report += f"Reasoning: {result.reasoning}\n"
return report
class TestCompleteEvalSuite:
"""Run complete evaluation suite."""
@pytest.mark.integration
@pytest.mark.asyncio
async def test_run_all_evals(self):
"""Run all evaluation scenarios and generate report."""
# This would run all eval scenarios and generate a comprehensive report
# For brevity, just verify the structure exists
all_scenarios = [
EvalScenario(
name="basic_package_install",
user_query="How do I install Firefox?",
expected_tool_calls=["await nixos_search(query='firefox')"],
success_criteria=["finds firefox package"],
),
EvalScenario(
name="service_config",
user_query="Configure nginx web server",
expected_tool_calls=["await nixos_search(query='nginx', search_type='options')"],
success_criteria=["finds nginx options"],
),
EvalScenario(
name="home_manager_usage",
user_query="Should I use Home Manager for git config?",
expected_tool_calls=["await home_manager_search(query='git')"],
success_criteria=["recommends Home Manager"],
),
]
assert len(all_scenarios) >= 3
assert all(isinstance(s, EvalScenario) for s in all_scenarios)
if __name__ == "__main__":
pytest.main([__file__, "-v"])
```
--------------------------------------------------------------------------------
/tests/test_channels.py:
--------------------------------------------------------------------------------
```python
#!/usr/bin/env python3
"""Tests for robust channel handling functionality."""
from unittest.mock import Mock, patch
import pytest
import requests
from mcp_nixos import server
from mcp_nixos.server import (
channel_cache,
get_channel_suggestions,
get_channels,
validate_channel,
)
def get_tool_function(tool_name: str):
"""Get the underlying function from a FastMCP tool."""
tool = getattr(server, tool_name)
if hasattr(tool, "fn"):
return tool.fn
return tool
# Get the underlying functions for direct use
nixos_channels = get_tool_function("nixos_channels")
nixos_info = get_tool_function("nixos_info")
nixos_search = get_tool_function("nixos_search")
nixos_stats = get_tool_function("nixos_stats")
class TestChannelHandling:
"""Test robust channel handling functionality."""
@patch("requests.post")
def test_discover_available_channels_success(self, mock_post):
"""Test successful channel discovery."""
# Mock successful responses for some channels (note: 24.11 removed from version list)
mock_responses = {
"latest-43-nixos-unstable": {"count": 151798},
"latest-43-nixos-25.05": {"count": 151698},
}
def side_effect(url, **kwargs):
mock_resp = Mock()
for pattern, response in mock_responses.items():
if pattern in url:
mock_resp.status_code = 200
mock_resp.json.return_value = response
return mock_resp
# Default to 404 for unknown patterns
mock_resp.status_code = 404
return mock_resp
mock_post.side_effect = side_effect
# Clear cache first
channel_cache.available_channels = None
result = channel_cache.get_available()
assert "latest-43-nixos-unstable" in result
assert "latest-43-nixos-25.05" in result
assert "151,798 documents" in result["latest-43-nixos-unstable"]
@patch("requests.post")
def test_discover_available_channels_with_cache(self, mock_post):
"""Test that channel discovery uses cache."""
# Set up cache
channel_cache.available_channels = {"test": "cached"}
result = channel_cache.get_available()
# Should return cached result without making API calls
assert result == {"test": "cached"}
mock_post.assert_not_called()
@patch("mcp_nixos.server.get_channels")
@patch("requests.post")
def test_validate_channel_success(self, mock_post, mock_get_channels):
"""Test successful channel validation."""
mock_get_channels.return_value = {"stable": "latest-43-nixos-25.05"}
mock_resp = Mock()
mock_resp.status_code = 200
mock_resp.json.return_value = {"count": 100000}
mock_post.return_value = mock_resp
result = validate_channel("stable")
assert result is True
@patch("mcp_nixos.server.get_channels")
def test_validate_channel_failure(self, mock_get_channels):
"""Test channel validation failure."""
mock_get_channels.return_value = {"stable": "latest-43-nixos-25.05"}
result = validate_channel("nonexistent")
assert result is False
def test_validate_channel_invalid_name(self):
"""Test validation of channel not in CHANNELS."""
result = validate_channel("totally-invalid")
assert result is False
@patch("mcp_nixos.server.get_channels")
def test_get_channel_suggestions_similar(self, mock_get_channels):
"""Test getting suggestions for similar channel names."""
# Mock the available channels
mock_get_channels.return_value = {
"unstable": "latest-43-nixos-unstable",
"stable": "latest-43-nixos-25.05",
"25.05": "latest-43-nixos-25.05",
"24.11": "latest-43-nixos-24.11",
"beta": "latest-43-nixos-25.05",
}
result = get_channel_suggestions("unstabl")
assert "unstable" in result
result = get_channel_suggestions("24")
assert "24.11" in result
@patch("mcp_nixos.server.get_channels")
def test_get_channel_suggestions_fallback(self, mock_get_channels):
"""Test fallback suggestions for completely invalid names."""
# Mock the available channels
mock_get_channels.return_value = {
"unstable": "latest-43-nixos-unstable",
"stable": "latest-43-nixos-25.05",
"25.05": "latest-43-nixos-25.05",
"24.11": "latest-43-nixos-24.11",
"beta": "latest-43-nixos-25.05",
}
result = get_channel_suggestions("totally-random-xyz")
assert "unstable" in result
assert "stable" in result
assert "25.05" in result
@patch("mcp_nixos.server.channel_cache.get_available")
@patch("mcp_nixos.server.channel_cache.get_resolved")
@pytest.mark.asyncio
async def test_nixos_channels_tool(self, mock_resolved, mock_discover):
"""Test nixos_channels tool output."""
mock_discover.return_value = {
"latest-43-nixos-unstable": "151,798 documents",
"latest-43-nixos-25.05": "151,698 documents",
"latest-43-nixos-24.11": "142,034 documents",
}
mock_resolved.return_value = {
"unstable": "latest-43-nixos-unstable",
"stable": "latest-43-nixos-25.05",
"25.05": "latest-43-nixos-25.05",
"24.11": "latest-43-nixos-24.11",
"beta": "latest-43-nixos-25.05",
}
result = await nixos_channels()
assert "NixOS Channels" in result # Match both old and new format
assert "unstable → latest-43-nixos-unstable" in result or "unstable \u2192 latest-43-nixos-unstable" in result
assert "stable" in result and "latest-43-nixos-25.05" in result
assert "✓ Available" in result
assert "151,798 documents" in result
@patch("mcp_nixos.server.channel_cache.get_available")
@patch("mcp_nixos.server.channel_cache.get_resolved")
@pytest.mark.asyncio
async def test_nixos_channels_with_unavailable(self, mock_resolved, mock_discover):
"""Test nixos_channels tool with some unavailable channels."""
# Only return some channels as available
mock_discover.return_value = {"latest-43-nixos-unstable": "151,798 documents"}
mock_resolved.return_value = {
"unstable": "latest-43-nixos-unstable",
"stable": "latest-43-nixos-25.05", # Not available
"25.05": "latest-43-nixos-25.05",
}
# Mock that we're not using fallback (partial availability)
channel_cache.using_fallback = False
result = await nixos_channels()
assert "✓ Available" in result
assert "✗ Unavailable" in result or "Fallback" in result
@patch("mcp_nixos.server.channel_cache.get_available")
@pytest.mark.asyncio
async def test_nixos_channels_with_extra_discovered(self, mock_discover):
"""Test nixos_channels with extra discovered channels."""
mock_discover.return_value = {
"latest-43-nixos-unstable": "151,798 documents",
"latest-43-nixos-25.05": "151,698 documents",
"latest-44-nixos-unstable": "152,000 documents", # New channel
}
# Mock that we're not using fallback
channel_cache.using_fallback = False
result = await nixos_channels()
# If not using fallback, should show additional channels
if not channel_cache.using_fallback:
assert "Additional available channels:" in result or "latest-44-nixos-unstable" in result
@pytest.mark.asyncio
async def test_nixos_stats_with_invalid_channel(self):
"""Test nixos_stats with invalid channel shows suggestions."""
result = await nixos_stats("invalid-channel")
assert "Error (ERROR):" in result
assert "Invalid channel 'invalid-channel'" in result
assert "Available channels:" in result
@pytest.mark.asyncio
async def test_nixos_search_with_invalid_channel(self):
"""Test nixos_search with invalid channel shows suggestions."""
result = await nixos_search("test", channel="invalid-channel")
assert "Error (ERROR):" in result
assert "Invalid channel 'invalid-channel'" in result
assert "Available channels:" in result
@patch("mcp_nixos.server.channel_cache.get_resolved")
def test_channel_mappings_dynamic(self, mock_resolved):
"""Test that dynamic channel mappings work correctly."""
# Mock the resolved channels
mock_resolved.return_value = {
"stable": "latest-43-nixos-25.05",
"unstable": "latest-43-nixos-unstable",
"25.05": "latest-43-nixos-25.05",
"24.11": "latest-43-nixos-24.11",
"beta": "latest-43-nixos-25.05",
}
channels = get_channels()
# Should have basic channels
assert "stable" in channels
assert "unstable" in channels
# Stable should point to a valid channel index
assert channels["stable"].startswith("latest-")
assert "nixos" in channels["stable"]
# Unstable should point to unstable index
assert "unstable" in channels["unstable"]
@patch("requests.post")
def test_discover_channels_handles_exceptions(self, mock_post):
"""Test channel discovery handles network exceptions gracefully."""
mock_post.side_effect = requests.ConnectionError("Network error")
# Clear cache
channel_cache.available_channels = None
result = channel_cache.get_available()
# Should return empty dict when all requests fail
assert result == {}
@patch("requests.post")
def test_validate_channel_handles_exceptions(self, mock_post):
"""Test channel validation handles exceptions gracefully."""
mock_post.side_effect = requests.ConnectionError("Network error")
result = validate_channel("stable")
assert result is False
@patch("mcp_nixos.server.channel_cache.get_available")
@pytest.mark.asyncio
async def test_nixos_channels_handles_exceptions(self, mock_discover):
"""Test nixos_channels tool handles exceptions gracefully."""
mock_discover.side_effect = Exception("Discovery failed")
result = await nixos_channels()
assert "Error (ERROR):" in result
assert "Discovery failed" in result
@patch("mcp_nixos.server.get_channels")
def test_channel_suggestions_for_legacy_channels(self, mock_get_channels):
"""Test suggestions work for legacy channel references."""
mock_get_channels.return_value = {
"stable": "latest-43-nixos-25.05",
"unstable": "latest-43-nixos-unstable",
"25.05": "latest-43-nixos-25.05",
"24.11": "latest-43-nixos-24.11",
"beta": "latest-43-nixos-25.05",
}
# Test old stable reference
result = get_channel_suggestions("20.09")
assert "24.11" in result or "stable" in result
# Test partial version
result = get_channel_suggestions("25")
assert "25.05" in result
@patch("requests.post")
def test_discover_channels_filters_empty_indices(self, mock_post):
"""Test that discovery filters out indices with 0 documents."""
def side_effect(url, **kwargs):
mock_resp = Mock()
if "empty-index" in url:
mock_resp.status_code = 200
mock_resp.json.return_value = {"count": 0} # Empty index
else:
mock_resp.status_code = 200
mock_resp.json.return_value = {"count": 100000}
return mock_resp
mock_post.side_effect = side_effect
# Clear cache
channel_cache.available_channels = None
# This should work with the actual test patterns
result = channel_cache.get_available()
# Should not include any indices with 0 documents
for _, info in result.items():
# Check that it doesn't start with "0 documents"
assert not info.startswith("0 documents")
# ===== Content from test_dynamic_channels.py =====
class TestDynamicChannelLifecycle:
"""Test dynamic channel detection and lifecycle management."""
def setup_method(self):
"""Clear caches before each test."""
channel_cache.available_channels = None
channel_cache.resolved_channels = None
@patch("requests.post")
def test_channel_discovery_future_proof(self, mock_post):
"""Test discovery works with future NixOS releases."""
# Simulate future release state
future_responses = {
"latest-44-nixos-unstable": {"count": 160000},
"latest-44-nixos-25.11": {"count": 155000}, # New stable
"latest-44-nixos-25.05": {"count": 152000}, # Old stable
"latest-43-nixos-25.05": {"count": 151000}, # Legacy
}
def side_effect(url, **kwargs):
mock_resp = Mock()
for pattern, response in future_responses.items():
if pattern in url:
mock_resp.status_code = 200
mock_resp.json.return_value = response
return mock_resp
mock_resp.status_code = 404
return mock_resp
mock_post.side_effect = side_effect
# Test discovery
available = channel_cache.get_available()
assert "latest-44-nixos-unstable" in available
assert "latest-44-nixos-25.11" in available
# Test resolution - should pick 25.11 as new stable
channels = channel_cache.get_resolved()
assert channels["stable"] == "latest-44-nixos-25.11"
assert channels["unstable"] == "latest-44-nixos-unstable"
assert channels["25.11"] == "latest-44-nixos-25.11"
assert channels["25.05"] == "latest-44-nixos-25.05"
@patch("requests.post")
def test_stable_detection_by_version_priority(self, mock_post):
"""Test stable detection prioritizes higher version numbers."""
# Same generation, different versions
responses = {
"latest-43-nixos-24.11": {"count": 150000},
"latest-43-nixos-25.05": {"count": 140000}, # Lower count but higher version
"latest-43-nixos-unstable": {"count": 155000},
}
def side_effect(url, **kwargs):
mock_resp = Mock()
for pattern, response in responses.items():
if pattern in url:
mock_resp.status_code = 200
mock_resp.json.return_value = response
return mock_resp
mock_resp.status_code = 404
return mock_resp
mock_post.side_effect = side_effect
channels = channel_cache.get_resolved()
# Should pick 25.05 despite lower count (higher version)
assert channels["stable"] == "latest-43-nixos-25.05"
@patch("requests.post")
def test_stable_detection_by_count_when_same_version(self, mock_post):
"""Test stable detection uses count as tiebreaker."""
responses = {
"latest-43-nixos-25.05": {"count": 150000},
"latest-44-nixos-25.05": {"count": 155000}, # Higher count, same version
"latest-43-nixos-unstable": {"count": 160000},
}
def side_effect(url, **kwargs):
mock_resp = Mock()
for pattern, response in responses.items():
if pattern in url:
mock_resp.status_code = 200
mock_resp.json.return_value = response
return mock_resp
mock_resp.status_code = 404
return mock_resp
mock_post.side_effect = side_effect
channels = channel_cache.get_resolved()
# Should pick higher count for same version
assert channels["stable"] == "latest-44-nixos-25.05"
@patch("requests.post")
def test_channel_discovery_handles_no_channels(self, mock_post):
"""Test graceful handling when no channels are available."""
mock_post.return_value = Mock(status_code=404)
available = channel_cache.get_available()
assert available == {}
channels = channel_cache.get_resolved()
# Should use fallback channels when discovery fails
assert channels != {}
assert "stable" in channels
assert "unstable" in channels
assert channel_cache.using_fallback is True
@patch("requests.post")
def test_channel_discovery_partial_availability(self, mock_post):
"""Test handling when only some channels are available."""
responses = {
"latest-43-nixos-unstable": {"count": 150000},
# No stable releases available
}
def side_effect(url, **kwargs):
mock_resp = Mock()
for pattern, response in responses.items():
if pattern in url:
mock_resp.status_code = 200
mock_resp.json.return_value = response
return mock_resp
mock_resp.status_code = 404
return mock_resp
mock_post.side_effect = side_effect
channels = channel_cache.get_resolved()
assert channels["unstable"] == "latest-43-nixos-unstable"
assert "stable" not in channels # No stable release found
@patch("mcp_nixos.server.channel_cache.get_resolved")
@pytest.mark.asyncio
async def test_nixos_stats_with_dynamic_channels(self, mock_resolve):
"""Test nixos_stats works with dynamically resolved channels."""
mock_resolve.return_value = {
"stable": "latest-44-nixos-25.11",
"unstable": "latest-44-nixos-unstable",
}
with patch("requests.post") as mock_post:
# Mock successful response
mock_resp = Mock()
mock_resp.status_code = 200
mock_resp.json.return_value = {"count": 1000}
mock_resp.raise_for_status.return_value = None
mock_post.return_value = mock_resp
# Should work with new stable
result = await nixos_stats("stable")
# Should not error and should contain statistics
assert "NixOS Statistics" in result
assert "stable" in result
# Should have made API calls
assert mock_post.called
@patch("mcp_nixos.server.channel_cache.get_resolved")
@pytest.mark.asyncio
async def test_nixos_search_with_dynamic_channels(self, mock_resolve):
"""Test nixos_search works with dynamically resolved channels."""
mock_resolve.return_value = {
"stable": "latest-44-nixos-25.11",
"unstable": "latest-44-nixos-unstable",
}
with patch("mcp_nixos.server.es_query") as mock_es:
mock_es.return_value = []
result = await nixos_search("test", channel="stable")
assert "No packages found" in result
@patch("mcp_nixos.server.channel_cache.get_available")
@pytest.mark.asyncio
async def test_nixos_channels_tool_shows_current_stable(self, mock_discover):
"""Test nixos_channels tool clearly shows current stable version."""
mock_discover.return_value = {
"latest-44-nixos-25.11": "155,000 documents",
"latest-44-nixos-unstable": "160,000 documents",
}
with patch("mcp_nixos.server.channel_cache.get_resolved") as mock_resolve:
mock_resolve.return_value = {
"stable": "latest-44-nixos-25.11",
"25.11": "latest-44-nixos-25.11",
"unstable": "latest-44-nixos-unstable",
}
result = await nixos_channels()
assert "stable (current: 25.11)" in result
assert "latest-44-nixos-25.11" in result
assert "dynamically discovered" in result
@pytest.mark.asyncio
async def test_channel_suggestions_work_with_dynamic_channels(self):
"""Test channel suggestions work with dynamic resolution."""
with patch("mcp_nixos.server.get_channels") as mock_get:
mock_get.return_value = {
"stable": "latest-44-nixos-25.11",
"unstable": "latest-44-nixos-unstable",
"25.11": "latest-44-nixos-25.11",
}
result = await nixos_stats("invalid-channel")
assert "Available channels:" in result
assert any(ch in result for ch in ["stable", "unstable"])
@patch("requests.post")
def test_caching_behavior(self, mock_post):
"""Test that caching works correctly."""
responses = {
"latest-43-nixos-unstable": {"count": 150000},
"latest-43-nixos-25.05": {"count": 145000},
}
call_count = 0
def side_effect(url, **kwargs):
nonlocal call_count
call_count += 1
mock_resp = Mock()
for pattern, response in responses.items():
if pattern in url:
mock_resp.status_code = 200
mock_resp.json.return_value = response
return mock_resp
mock_resp.status_code = 404
return mock_resp
mock_post.side_effect = side_effect
# First call should hit API
channels1 = get_channels()
first_call_count = call_count
# Second call should use cache
channels2 = get_channels()
second_call_count = call_count
assert channels1 == channels2
assert second_call_count == first_call_count # No additional API calls
@patch("requests.post")
def test_malformed_version_handling(self, mock_post):
"""Test handling of malformed version numbers."""
responses = {
"latest-43-nixos-unstable": {"count": 150000},
"latest-43-nixos-badversion": {"count": 145000}, # Invalid version
"latest-43-nixos-25.05": {"count": 140000}, # Valid version
}
def side_effect(url, **kwargs):
mock_resp = Mock()
for pattern, response in responses.items():
if pattern in url:
mock_resp.status_code = 200
mock_resp.json.return_value = response
return mock_resp
mock_resp.status_code = 404
return mock_resp
mock_post.side_effect = side_effect
channels = channel_cache.get_resolved()
# Should ignore malformed version and use valid one
assert channels["stable"] == "latest-43-nixos-25.05"
assert "badversion" not in channels
@patch("requests.post")
def test_network_error_handling(self, mock_post):
"""Test handling of network errors during discovery."""
mock_post.side_effect = requests.ConnectionError("Network error")
available = channel_cache.get_available()
assert available == {}
channels = channel_cache.get_resolved()
# Should use fallback channels when network fails
assert channels != {}
assert "stable" in channels
assert "unstable" in channels
assert channel_cache.using_fallback is True
@patch("requests.post")
def test_zero_document_filtering(self, mock_post):
"""Test that channels with zero documents are filtered out."""
responses = {
"latest-43-nixos-unstable": {"count": 150000},
"latest-43-nixos-25.05": {"count": 0}, # Empty index
"latest-43-nixos-25.11": {"count": 140000},
}
def side_effect(url, **kwargs):
mock_resp = Mock()
for pattern, response in responses.items():
if pattern in url:
mock_resp.status_code = 200
mock_resp.json.return_value = response
return mock_resp
mock_resp.status_code = 404
return mock_resp
mock_post.side_effect = side_effect
available = channel_cache.get_available()
assert "latest-43-nixos-unstable" in available
assert "latest-43-nixos-25.05" not in available # Filtered out
assert "latest-43-nixos-25.11" in available
@patch("requests.post")
def test_version_comparison_edge_cases(self, mock_post):
"""Test version comparison with edge cases."""
# Note: 20.09 not in test since it's no longer in version list
responses = {
"latest-43-nixos-unstable": {"count": 150000},
"latest-43-nixos-25.05": {"count": 145000}, # Current
"latest-43-nixos-30.05": {"count": 140000}, # Future
}
def side_effect(url, **kwargs):
mock_resp = Mock()
for pattern, response in responses.items():
if pattern in url:
mock_resp.status_code = 200
mock_resp.json.return_value = response
return mock_resp
mock_resp.status_code = 404
return mock_resp
mock_post.side_effect = side_effect
channels = channel_cache.get_resolved()
# Should pick highest version (30.05)
assert channels["stable"] == "latest-43-nixos-30.05"
assert "25.05" in channels
assert "30.05" in channels
@patch("mcp_nixos.server.channel_cache.get_available")
def test_beta_alias_behavior(self, mock_discover):
"""Test that beta is always an alias for stable."""
mock_discover.return_value = {
"latest-44-nixos-25.11": "155,000 documents",
"latest-44-nixos-unstable": "160,000 documents",
}
channels = channel_cache.get_resolved()
assert "beta" in channels
assert channels["beta"] == channels["stable"]
@pytest.mark.asyncio
async def test_integration_with_all_tools(self):
"""Test that all tools work with dynamic channels."""
with patch("mcp_nixos.server.get_channels") as mock_get:
mock_get.return_value = {
"stable": "latest-44-nixos-25.11",
"unstable": "latest-44-nixos-unstable",
}
with patch("mcp_nixos.server.es_query") as mock_es:
mock_es.return_value = []
with patch("requests.post") as mock_post:
# Mock successful response for nixos_stats
mock_resp = Mock()
mock_resp.status_code = 200
mock_resp.json.return_value = {"count": 1000}
mock_resp.raise_for_status.return_value = None
mock_post.return_value = mock_resp
# Test all tools that use channels
tools_to_test = [
lambda: nixos_search("test", channel="stable"),
lambda: nixos_info("test", channel="stable"),
lambda: nixos_stats("stable"),
]
for tool in tools_to_test:
result = await tool()
# Should not error due to channel resolution
assert (
"Error" not in result
or "not found" in result
or "No packages found" in result
or "NixOS Statistics" in result
)
# ===== Tests for Fallback Channel Behavior (Issue #52 fix) =====
class TestFallbackChannels:
"""Test fallback channel behavior when API discovery fails."""
def setup_method(self):
"""Clear caches before each test."""
channel_cache.available_channels = None
channel_cache.resolved_channels = None
channel_cache.using_fallback = False
@patch("requests.post")
def test_fallback_when_all_api_calls_fail(self, mock_post):
"""Test that fallback channels are used when all API calls fail."""
# Simulate complete API failure
mock_post.side_effect = requests.Timeout("Connection timeout")
channels = channel_cache.get_resolved()
# Should use fallback channels
assert channel_cache.using_fallback is True
assert "stable" in channels
assert "unstable" in channels
assert "25.05" in channels
assert "beta" in channels
assert channels["stable"] == "latest-44-nixos-25.05"
@patch("requests.post")
def test_fallback_when_api_returns_empty(self, mock_post):
"""Test fallback when API returns empty results."""
# Mock API returning empty results
mock_resp = Mock()
mock_resp.status_code = 200
mock_resp.json.return_value = {"count": 0}
mock_post.return_value = mock_resp
channels = channel_cache.get_resolved()
# Should use fallback channels
assert channel_cache.using_fallback is True
assert "stable" in channels
@patch("requests.post")
@pytest.mark.asyncio
async def test_nixos_search_works_with_fallback(self, mock_post):
"""Test that nixos_search works when using fallback channels."""
# Simulate API failure for discovery
mock_post.side_effect = requests.Timeout("Connection timeout")
# Clear cache to force rediscovery
channel_cache.available_channels = None
channel_cache.resolved_channels = None
# Mock es_query to return empty results
with patch("mcp_nixos.server.es_query") as mock_es:
mock_es.return_value = []
# This should NOT fail with "Invalid channel 'stable'"
result = await nixos_search("test", channel="stable")
# Should work and return "No packages found" not an error about invalid channel
assert "Invalid channel" not in result
assert "No packages found" in result or "Error" not in result
@patch("requests.post")
@pytest.mark.asyncio
async def test_nixos_channels_shows_fallback_warning(self, mock_post):
"""Test that nixos_channels shows a warning when using fallback."""
# Simulate API failure
mock_post.side_effect = requests.ConnectionError("Network error")
# Clear cache
channel_cache.available_channels = None
channel_cache.resolved_channels = None
result = await nixos_channels()
# Should show fallback warning
assert "WARNING" in result or "fallback" in result.lower()
assert "stable" in result # Should still show channels
@patch("mcp_nixos.server.get_channels")
def test_get_channel_suggestions_works_with_fallback(self, mock_get):
"""Test channel suggestions work when using fallback channels."""
# Mock fallback channels
mock_get.return_value = {
"stable": "latest-44-nixos-25.05",
"unstable": "latest-44-nixos-unstable",
"25.05": "latest-44-nixos-25.05",
"beta": "latest-44-nixos-25.05",
}
result = get_channel_suggestions("invalid")
# Should provide suggestions from fallback channels
assert "stable" in result or "unstable" in result
@patch("requests.post")
def test_no_fallback_when_api_succeeds(self, mock_post):
"""Test that fallback is NOT used when API works correctly."""
# Mock successful API response
responses = {
"latest-44-nixos-unstable": {"count": 150000},
"latest-44-nixos-25.05": {"count": 145000},
}
def side_effect(url, **kwargs):
mock_resp = Mock()
for pattern, response in responses.items():
if pattern in url:
mock_resp.status_code = 200
mock_resp.json.return_value = response
return mock_resp
mock_resp.status_code = 404
return mock_resp
mock_post.side_effect = side_effect
channels = channel_cache.get_resolved()
# Should NOT use fallback
assert channel_cache.using_fallback is False
assert "stable" in channels
@patch("requests.post")
@pytest.mark.asyncio
async def test_all_tools_work_with_fallback(self, mock_post):
"""Test that all channel-based tools work with fallback channels."""
# Simulate API failure
mock_post.side_effect = requests.Timeout("Timeout")
# Clear cache
channel_cache.available_channels = None
channel_cache.resolved_channels = None
# Mock es_query
with patch("mcp_nixos.server.es_query") as mock_es:
mock_es.return_value = []
# Test various tools - none should fail with "Invalid channel"
result1 = await nixos_search("test", channel="stable")
assert "Invalid channel" not in result1
result2 = await nixos_info("vim", channel="stable")
assert "Invalid channel" not in result2
result3 = await nixos_stats("stable")
# nixos_stats might error, but not due to invalid channel
if "Error" in result3:
assert "Invalid channel" not in result3
```
--------------------------------------------------------------------------------
/tests/test_mcp_behavior.py:
--------------------------------------------------------------------------------
```python
#!/usr/bin/env python3
"""MCP behavior evaluation tests for real-world usage scenarios."""
from unittest.mock import Mock, patch
import pytest
from mcp_nixos import server
def get_tool_function(tool_name: str):
"""Get the underlying function from a FastMCP tool."""
tool = getattr(server, tool_name)
if hasattr(tool, "fn"):
return tool.fn
return tool
# Get the underlying functions for direct use
darwin_info = get_tool_function("darwin_info")
darwin_options_by_prefix = get_tool_function("darwin_options_by_prefix")
darwin_search = get_tool_function("darwin_search")
darwin_stats = get_tool_function("darwin_stats")
home_manager_info = get_tool_function("home_manager_info")
home_manager_list_options = get_tool_function("home_manager_list_options")
home_manager_options_by_prefix = get_tool_function("home_manager_options_by_prefix")
home_manager_search = get_tool_function("home_manager_search")
home_manager_stats = get_tool_function("home_manager_stats")
nixos_channels = get_tool_function("nixos_channels")
nixos_info = get_tool_function("nixos_info")
nixos_search = get_tool_function("nixos_search")
nixos_stats = get_tool_function("nixos_stats")
class MockAssistant:
"""Mock AI assistant for testing MCP tool usage patterns."""
def __init__(self):
self.tool_calls = []
self.responses = []
async def use_tool(self, tool_name: str, **kwargs) -> str:
"""Simulate using an MCP tool."""
from mcp_nixos import server
self.tool_calls.append({"tool": tool_name, "args": kwargs})
# Call the actual tool - get the underlying function from FastMCP tool
tool_func = getattr(server, tool_name)
if hasattr(tool_func, "fn"):
# FastMCP wrapped function - use the underlying function
result = await tool_func.fn(**kwargs)
else:
# Direct function call
result = await tool_func(**kwargs)
self.responses.append(result)
return result
def analyze_response(self, response: str) -> dict[str, bool | int]:
"""Analyze tool response for key information."""
analysis = {
"has_results": "Found" in response or ":" in response,
"is_error": "Error" in response,
"has_bullet_points": "•" in response,
"line_count": len(response.strip().split("\n")),
"mentions_not_found": "not found" in response.lower(),
}
return analysis
@pytest.mark.evals
class TestMCPBehaviorEvals:
"""Test MCP tool behavior in realistic scenarios."""
@pytest.mark.asyncio
async def test_scenario_install_package(self):
"""User wants to install a specific package."""
assistant = MockAssistant()
# Step 1: Search for the package
response1 = await assistant.use_tool("nixos_search", query="neovim", search_type="packages", limit=5)
analysis1 = assistant.analyze_response(response1)
assert analysis1["has_results"] or analysis1["mentions_not_found"]
assert not analysis1["is_error"]
# Step 2: Get detailed info if found
if analysis1["has_results"]:
response2 = await assistant.use_tool("nixos_info", name="neovim", type="package")
analysis2 = assistant.analyze_response(response2)
assert "Package:" in response2
assert "Version:" in response2
assert not analysis2["is_error"]
# Verify tool usage pattern
assert len(assistant.tool_calls) >= 1
assert assistant.tool_calls[0]["tool"] == "nixos_search"
@pytest.mark.asyncio
async def test_scenario_configure_service(self):
"""User wants to configure a NixOS service."""
assistant = MockAssistant()
# Step 1: Search for service options
response1 = await assistant.use_tool("nixos_search", query="nginx", search_type="options", limit=10)
# Step 2: Get specific option details
if "services.nginx.enable" in response1:
response2 = await assistant.use_tool("nixos_info", name="services.nginx.enable", type="option")
assert "Type: boolean" in response2
assert "Default:" in response2
@pytest.mark.asyncio
async def test_scenario_explore_home_manager(self):
"""User wants to explore Home Manager configuration."""
assistant = MockAssistant()
# Step 1: List categories
response1 = await assistant.use_tool("home_manager_list_options")
assert "programs" in response1
assert "services" in response1
# Step 2: Explore programs category
await assistant.use_tool("home_manager_options_by_prefix", option_prefix="programs")
# Step 3: Search for specific program
response3 = await assistant.use_tool("home_manager_search", query="firefox", limit=5)
# Step 4: Get details on specific option
if "programs.firefox.enable" in response3:
response4 = await assistant.use_tool("home_manager_info", name="programs.firefox.enable")
assert "Option:" in response4
@pytest.mark.asyncio
async def test_scenario_macos_configuration(self):
"""User wants to configure macOS with nix-darwin."""
assistant = MockAssistant()
# Step 1: Search for Homebrew integration
await assistant.use_tool("darwin_search", query="homebrew", limit=10)
# Step 2: Explore system defaults
response2 = await assistant.use_tool("darwin_options_by_prefix", option_prefix="system.defaults")
# Step 3: Get specific dock settings
if "system.defaults.dock" in response2:
response3 = await assistant.use_tool("darwin_options_by_prefix", option_prefix="system.defaults.dock")
# Check for autohide option
if "autohide" in response3:
response4 = await assistant.use_tool("darwin_info", name="system.defaults.dock.autohide")
assert "Option:" in response4
@pytest.mark.asyncio
async def test_scenario_compare_channels(self):
"""User wants to compare packages across channels."""
assistant = MockAssistant()
package = "postgresql"
channels = ["unstable", "stable"]
results = {}
for channel in channels:
response = await assistant.use_tool("nixos_info", name=package, type="package", channel=channel)
if "Version:" in response:
# Extract version
for line in response.split("\n"):
if line.startswith("Version:"):
results[channel] = line.split("Version:")[1].strip()
# User can now compare versions across channels
assert len(assistant.tool_calls) == len(channels)
@pytest.mark.asyncio
async def test_scenario_find_package_by_program(self):
"""User wants to find which package provides a specific program."""
assistant = MockAssistant()
# Search for package that provides 'gcc'
response = await assistant.use_tool("nixos_search", query="gcc", search_type="programs", limit=10)
analysis = assistant.analyze_response(response)
if analysis["has_results"]:
assert "provided by" in response
assert "gcc" in response.lower()
@pytest.mark.asyncio
async def test_scenario_complex_option_exploration(self):
"""User wants to understand complex NixOS options."""
assistant = MockAssistant()
# Look for virtualisation options
response1 = await assistant.use_tool(
"nixos_search", query="virtualisation.docker", search_type="options", limit=20
)
if "virtualisation.docker.enable" in response1:
# Get details on enable option
await assistant.use_tool("nixos_info", name="virtualisation.docker.enable", type="option")
# Search for related options
await assistant.use_tool("nixos_search", query="docker", search_type="options", limit=10)
# Verify we get comprehensive docker configuration options
assert any(r for r in assistant.responses if "docker" in r.lower())
@pytest.mark.asyncio
async def test_scenario_git_configuration(self):
"""User wants to configure git with Home Manager."""
assistant = MockAssistant()
# Explore git options
response1 = await assistant.use_tool("home_manager_options_by_prefix", option_prefix="programs.git")
# Count git-related options
git_options = response1.count("programs.git")
assert git_options > 10 # Git should have many options
# Look for specific features
features = ["delta", "lfs", "signing", "aliases"]
found_features = sum(1 for f in features if f in response1)
assert found_features >= 2 # Should find at least some features
@pytest.mark.asyncio
async def test_scenario_error_recovery(self):
"""Test how tools handle errors and guide users."""
assistant = MockAssistant()
# Try invalid channel
response1 = await assistant.use_tool("nixos_search", query="test", channel="invalid-channel")
assert "Error" in response1
assert "Invalid channel" in response1
# Try non-existent package
response2 = await assistant.use_tool("nixos_info", name="definitely-not-a-real-package-12345", type="package")
assert "not found" in response2.lower()
# Try invalid type
response3 = await assistant.use_tool("nixos_search", query="test", search_type="invalid-type")
assert "Error" in response3
assert "Invalid type" in response3
@pytest.mark.asyncio
async def test_scenario_bulk_option_discovery(self):
"""User wants to discover all options for a service."""
assistant = MockAssistant()
# Search for all nginx options
response1 = await assistant.use_tool("nixos_search", query="services.nginx", search_type="options", limit=50)
if "Found" in response1:
# Count unique option types
option_types = set()
for line in response1.split("\n"):
if "Type:" in line:
option_type = line.split("Type:")[1].strip()
option_types.add(option_type)
# nginx should have various option types
assert len(option_types) >= 2
@pytest.mark.asyncio
async def test_scenario_multi_tool_workflow(self):
"""Test realistic multi-step workflows."""
assistant = MockAssistant()
# Workflow: Set up a development environment
# 1. Check statistics
stats = await assistant.use_tool("nixos_stats")
assert "Packages:" in stats
# 2. Search for development tools
dev_tools = ["vscode", "git", "docker", "nodejs"]
for tool in dev_tools[:2]: # Test first two to save time
response = await assistant.use_tool("nixos_search", query=tool, search_type="packages", limit=3)
if "Found" in response:
# Get info on first result
package_name = None
for line in response.split("\n"):
if line.startswith("•"):
# Extract package name
package_name = line.split("•")[1].split("(")[0].strip()
break
if package_name:
info = await assistant.use_tool("nixos_info", name=package_name, type="package")
assert "Package:" in info
# 3. Configure git in Home Manager
await assistant.use_tool("home_manager_search", query="git", limit=10)
# Verify workflow completed
assert len(assistant.tool_calls) >= 4
assert not any("Error" in r for r in assistant.responses[:3]) # First 3 should succeed
@pytest.mark.asyncio
async def test_scenario_performance_monitoring(self):
"""Monitor performance characteristics of tool calls."""
import time
assistant = MockAssistant()
timings = {}
# Time different operations
operations = [
("nixos_stats", {}),
("nixos_search", {"query": "python", "limit": 20}),
("home_manager_list_options", {}),
("darwin_search", {"query": "system", "limit": 10}),
]
for op_name, op_args in operations:
start = time.time()
try:
await assistant.use_tool(op_name, **op_args)
elapsed = time.time() - start
timings[op_name] = elapsed
except Exception:
timings[op_name] = -1
# All operations should complete reasonably quickly
for op, timing in timings.items():
if timing > 0:
assert timing < 30, f"{op} took too long: {timing}s"
@pytest.mark.asyncio
async def test_scenario_option_value_types(self):
"""Test understanding different option value types."""
assistant = MockAssistant()
# Search for options with different types
type_examples = {
"boolean": "enable",
"string": "description",
"list": "allowedTCPPorts",
"attribute set": "extraConfig",
}
found_types = {}
for type_name, search_term in type_examples.items():
response = await assistant.use_tool("nixos_search", query=search_term, search_type="options", limit=5)
if "Type:" in response:
found_types[type_name] = response
# Should find at least some different types
assert len(found_types) >= 2
# ===== Content from test_mcp_behavior_comprehensive.py =====
class TestMCPBehaviorComprehensive:
"""Test real-world usage patterns based on actual tool testing results."""
@pytest.mark.asyncio
async def test_nixos_package_discovery_flow(self):
"""Test typical package discovery workflow."""
# 1. Search for packages
with patch("mcp_nixos.server.es_query") as mock_es:
mock_es.return_value = [
{
"_source": {
"type": "package",
"package_pname": "git",
"package_pversion": "2.49.0",
"package_description": "Distributed version control system",
}
},
{
"_source": {
"type": "package",
"package_pname": "gitoxide",
"package_pversion": "0.40.0",
"package_description": "Rust implementation of Git",
}
},
]
result = await nixos_search("git", limit=5)
assert "git (2.49.0)" in result
assert "Distributed version control system" in result
assert "gitoxide" in result
# 2. Get detailed info about a specific package
with patch("mcp_nixos.server.es_query") as mock_es:
mock_es.return_value = [
{
"_source": {
"type": "package",
"package_pname": "git",
"package_pversion": "2.49.0",
"package_description": "Distributed version control system",
"package_homepage": ["https://git-scm.com/"],
"package_license_set": ["GNU General Public License v2.0"],
}
}
]
result = await nixos_info("git")
assert "Package: git" in result
assert "Version: 2.49.0" in result
assert "Homepage: https://git-scm.com/" in result
assert "License: GNU General Public License v2.0" in result
@pytest.mark.asyncio
async def test_nixos_channel_awareness(self):
"""Test channel discovery and usage."""
# 1. List available channels
with patch("mcp_nixos.server.channel_cache.get_available") as mock_discover:
with patch("mcp_nixos.server.channel_cache.get_resolved") as mock_resolved:
mock_discover.return_value = {
"latest-43-nixos-unstable": "151,798 documents",
"latest-43-nixos-25.05": "151,698 documents",
"latest-43-nixos-25.11": "152,000 documents",
}
mock_resolved.return_value = {
"unstable": "latest-43-nixos-unstable",
"stable": "latest-43-nixos-25.11",
"25.05": "latest-43-nixos-25.05",
"25.11": "latest-43-nixos-25.11",
"beta": "latest-43-nixos-25.11",
}
# Mock that we're not using fallback
from mcp_nixos.server import channel_cache
channel_cache.using_fallback = False
result = await nixos_channels()
assert "NixOS Channels" in result
assert "stable (current: 25.11)" in result
assert "unstable" in result
assert "✓ Available" in result
# 2. Get stats for a channel
with patch("requests.post") as mock_post:
mock_resp = Mock()
mock_resp.status_code = 200
mock_resp.json.side_effect = [
{"count": 129865}, # packages
{"count": 21933}, # options
]
mock_resp.raise_for_status.return_value = None
mock_post.return_value = mock_resp
result = await nixos_stats()
assert "NixOS Statistics" in result
assert "129,865" in result
assert "21,933" in result
@pytest.mark.asyncio
async def test_home_manager_option_discovery_flow(self):
"""Test typical Home Manager option discovery workflow."""
# 1. Search for options
with patch("mcp_nixos.server.parse_html_options") as mock_parse:
mock_parse.return_value = [
{
"name": "programs.git.enable",
"type": "boolean",
"description": "Whether to enable Git",
},
{
"name": "programs.git.userName",
"type": "string",
"description": "Default Git username",
},
{
"name": "programs.git.userEmail",
"type": "string",
"description": "Default Git email",
},
]
result = await home_manager_search("git", limit=3)
assert "programs.git.enable" in result
assert "programs.git.userName" in result
assert "programs.git.userEmail" in result
# 2. Browse by prefix to find exact option names
with patch("mcp_nixos.server.parse_html_options") as mock_parse:
mock_parse.return_value = [
{
"name": "programs.git.enable",
"type": "boolean",
"description": "Whether to enable Git",
},
{
"name": "programs.git.aliases",
"type": "attribute set of string",
"description": "Git aliases",
},
{
"name": "programs.git.delta.enable",
"type": "boolean",
"description": "Whether to enable delta syntax highlighting",
},
]
result = await home_manager_options_by_prefix("programs.git")
assert "programs.git.enable" in result
assert "programs.git.aliases" in result
assert "programs.git.delta.enable" in result
# 3. Get specific option info (requires exact name)
with patch("mcp_nixos.server.parse_html_options") as mock_parse:
mock_parse.return_value = [
{
"name": "programs.git.enable",
"type": "boolean",
"description": "Whether to enable Git",
}
]
result = await home_manager_info("programs.git.enable")
assert "Option: programs.git.enable" in result
assert "Type: boolean" in result
assert "Whether to enable Git" in result
@pytest.mark.asyncio
async def test_home_manager_category_exploration(self):
"""Test exploring Home Manager categories."""
with patch("mcp_nixos.server.parse_html_options") as mock_parse:
# Simulate real category distribution
mock_parse.return_value = [
{"name": "programs.git.enable", "type": "", "description": ""},
{"name": "programs.vim.enable", "type": "", "description": ""},
{"name": "services.gpg-agent.enable", "type": "", "description": ""},
{"name": "home.packages", "type": "", "description": ""},
{"name": "accounts.email.accounts", "type": "", "description": ""},
]
result = await home_manager_list_options()
assert "Home Manager option categories" in result
assert "programs (2 options)" in result
assert "services (1 options)" in result
assert "home (1 options)" in result
assert "accounts (1 options)" in result
@pytest.mark.asyncio
async def test_darwin_system_configuration_flow(self):
"""Test typical Darwin configuration workflow."""
# 1. Search for system options
with patch("mcp_nixos.server.parse_html_options") as mock_parse:
mock_parse.return_value = [
{
"name": "system.defaults.dock.autohide",
"type": "boolean",
"description": "Whether to automatically hide the dock",
},
{
"name": "system.defaults.NSGlobalDomain.AppleInterfaceStyle",
"type": "string",
"description": "Set to 'Dark' to enable dark mode",
},
{
"name": "system.stateVersion",
"type": "string",
"description": "The nix-darwin state version",
},
]
result = await darwin_search("system", limit=3)
assert "system.defaults.dock.autohide" in result
assert "system.defaults.NSGlobalDomain.AppleInterfaceStyle" in result
assert "system.stateVersion" in result
# 2. Browse system options by prefix
with patch("mcp_nixos.server.parse_html_options") as mock_parse:
mock_parse.return_value = [
{
"name": "system.defaults.dock.autohide",
"type": "boolean",
"description": "Whether to automatically hide the dock",
},
{
"name": "system.defaults.dock.autohide-delay",
"type": "float",
"description": "Dock autohide delay",
},
{
"name": "system.defaults.dock.orientation",
"type": "string",
"description": "Dock position on screen",
},
]
result = await darwin_options_by_prefix("system.defaults.dock")
assert "system.defaults.dock.autohide" in result
assert "system.defaults.dock.autohide-delay" in result
assert "system.defaults.dock.orientation" in result
@pytest.mark.asyncio
async def test_error_handling_with_suggestions(self):
"""Test error handling provides helpful suggestions."""
# Invalid channel
with patch("mcp_nixos.server.get_channels") as mock_get:
mock_get.return_value = {
"stable": "latest-43-nixos-25.05",
"unstable": "latest-43-nixos-unstable",
"25.05": "latest-43-nixos-25.05",
"24.11": "latest-43-nixos-24.11",
}
result = await nixos_search("test", channel="24.05")
assert "Invalid channel" in result
assert "Available channels:" in result
assert "24.11" in result or "25.05" in result
@pytest.mark.asyncio
async def test_cross_tool_consistency(self):
"""Test that different tools provide consistent information."""
# Channel consistency
with patch("mcp_nixos.server.get_channels") as mock_get:
channels = {
"stable": "latest-43-nixos-25.05",
"unstable": "latest-43-nixos-unstable",
"25.05": "latest-43-nixos-25.05",
"beta": "latest-43-nixos-25.05",
}
mock_get.return_value = channels
# All tools should accept the same channels
for channel in ["stable", "unstable", "25.05", "beta"]:
with patch("mcp_nixos.server.es_query") as mock_es:
mock_es.return_value = []
result = await nixos_search("test", channel=channel)
assert "Error" not in result or "Invalid channel" not in result
@pytest.mark.asyncio
async def test_real_world_git_configuration_scenario(self):
"""Test a complete Git configuration discovery scenario."""
# User wants to configure Git in Home Manager
# Step 1: Search for git-related options
with patch("mcp_nixos.server.parse_html_options") as mock_parse:
mock_parse.return_value = [
{
"name": "programs.git.enable",
"type": "boolean",
"description": "Whether to enable Git",
},
{
"name": "programs.git.userName",
"type": "string",
"description": "Default Git username",
},
]
result = await home_manager_search("git user")
assert "programs.git.userName" in result
# Step 2: Browse all git options
with patch("mcp_nixos.server.parse_html_options") as mock_parse:
mock_parse.return_value = [
{"name": "programs.git.enable", "type": "boolean", "description": "Whether to enable Git"},
{"name": "programs.git.userName", "type": "string", "description": "Default Git username"},
{"name": "programs.git.userEmail", "type": "string", "description": "Default Git email"},
{"name": "programs.git.signing.key", "type": "string", "description": "GPG signing key"},
{
"name": "programs.git.signing.signByDefault",
"type": "boolean",
"description": "Sign commits by default",
},
]
result = await home_manager_options_by_prefix("programs.git")
assert "programs.git.userName" in result
assert "programs.git.userEmail" in result
assert "programs.git.signing.key" in result
# Step 3: Get details for specific options
with patch("mcp_nixos.server.parse_html_options") as mock_parse:
mock_parse.return_value = [
{
"name": "programs.git.signing.signByDefault",
"type": "boolean",
"description": "Whether to sign commits by default",
}
]
result = await home_manager_info("programs.git.signing.signByDefault")
assert "Type: boolean" in result
assert "sign commits by default" in result
@pytest.mark.asyncio
async def test_performance_with_large_result_sets(self):
"""Test handling of large result sets efficiently."""
# Home Manager has 2000+ options
with patch("mcp_nixos.server.parse_html_options") as mock_parse:
# Simulate large option set
mock_options = []
for i in range(2129): # Actual count from testing
mock_options.append(
{
"name": f"programs.option{i}",
"type": "string",
"description": f"Option {i}",
}
)
mock_parse.return_value = mock_options
result = await home_manager_list_options()
assert "2129 options" in result or "programs (" in result
@pytest.mark.asyncio
async def test_package_not_found_behavior(self):
"""Test behavior when packages/options are not found."""
# Package not found
with patch("mcp_nixos.server.es_query") as mock_es:
mock_es.return_value = []
result = await nixos_info("nonexistent-package")
assert "not found" in result.lower()
# Option not found
with patch("mcp_nixos.server.parse_html_options") as mock_parse:
mock_parse.return_value = []
result = await home_manager_info("nonexistent.option")
assert "not found" in result.lower()
@pytest.mark.asyncio
async def test_channel_migration_scenario(self):
"""Test that users can migrate from old to new channels."""
# User on 24.11 wants to upgrade to 25.05
with patch("mcp_nixos.server.get_channels") as mock_get:
mock_get.return_value = {
"stable": "latest-43-nixos-25.05",
"25.05": "latest-43-nixos-25.05",
"24.11": "latest-43-nixos-24.11",
"unstable": "latest-43-nixos-unstable",
}
# Can still query old channel
with patch("mcp_nixos.server.es_query") as mock_es:
mock_es.return_value = []
result = await nixos_search("test", channel="24.11")
assert "Error" not in result or "Invalid channel" not in result
# Can query new stable
with patch("mcp_nixos.server.es_query") as mock_es:
mock_es.return_value = []
result = await nixos_search("test", channel="stable")
assert "Error" not in result or "Invalid channel" not in result
@pytest.mark.asyncio
async def test_option_type_information(self):
"""Test that option type information is properly displayed."""
test_cases = [
("boolean option", "boolean", "programs.git.enable"),
("string option", "string", "programs.git.userName"),
("attribute set", "attribute set of string", "programs.git.aliases"),
("list option", "list of string", "home.packages"),
("complex type", "null or string or signed integer", "services.dunst.settings.global.offset"),
]
for desc, type_str, option_name in test_cases:
with patch("mcp_nixos.server.parse_html_options") as mock_parse:
mock_parse.return_value = [
{
"name": option_name,
"type": type_str,
"description": f"Test {desc}",
}
]
result = await home_manager_info(option_name)
assert f"Type: {type_str}" in result
@pytest.mark.asyncio
@patch("mcp_nixos.server.parse_html_options")
async def test_stats_functions_limitations(self, mock_parse):
"""Test that stats functions return actual statistics now."""
# Mock parsed options for Home Manager
mock_parse.return_value = [
{"name": "programs.git.enable", "type": "boolean", "description": "Enable git"},
{"name": "programs.zsh.enable", "type": "boolean", "description": "Enable zsh"},
{"name": "services.gpg-agent.enable", "type": "boolean", "description": "Enable GPG agent"},
{"name": "home.packages", "type": "list", "description": "Packages to install"},
{"name": "wayland.windowManager.sway.enable", "type": "boolean", "description": "Enable Sway"},
{"name": "xsession.enable", "type": "boolean", "description": "Enable X session"},
]
# Home Manager stats now return actual statistics
result = await home_manager_stats()
assert "Home Manager Statistics:" in result
assert "Total options:" in result
assert "Categories:" in result
assert "Top categories:" in result
# Mock parsed options for Darwin
mock_parse.return_value = [
{"name": "services.nix-daemon.enable", "type": "boolean", "description": "Enable nix-daemon"},
{"name": "system.defaults.dock.autohide", "type": "boolean", "description": "Auto-hide dock"},
{"name": "launchd.agents.test", "type": "attribute set", "description": "Launchd agents"},
{"name": "programs.zsh.enable", "type": "boolean", "description": "Enable zsh"},
{"name": "homebrew.enable", "type": "boolean", "description": "Enable Homebrew"},
]
# Darwin stats now return actual statistics
result = await darwin_stats()
assert "nix-darwin Statistics:" in result
assert "Total options:" in result
assert "Categories:" in result
assert "Top categories:" in result
```
--------------------------------------------------------------------------------
/tests/test_flakes.py:
--------------------------------------------------------------------------------
```python
"""Evaluation tests for flake search and improved stats functionality."""
from unittest.mock import MagicMock, Mock, patch
import pytest
import requests
from mcp_nixos import server
def get_tool_function(tool_name: str):
"""Get the underlying function from a FastMCP tool."""
tool = getattr(server, tool_name)
if hasattr(tool, "fn"):
return tool.fn
return tool
# Get the underlying functions for direct use
darwin_stats = get_tool_function("darwin_stats")
home_manager_stats = get_tool_function("home_manager_stats")
nixos_flakes_search = get_tool_function("nixos_flakes_search")
nixos_flakes_stats = get_tool_function("nixos_flakes_stats")
nixos_search = get_tool_function("nixos_search")
class TestFlakeSearchEvals:
"""Test flake search functionality with real-world scenarios."""
@pytest.fixture(autouse=True)
def mock_channel_validation(self):
"""Mock channel validation to always pass for 'unstable'."""
with patch("mcp_nixos.server.channel_cache") as mock_cache:
mock_cache.get_available.return_value = {"unstable": "latest-45-nixos-unstable"}
mock_cache.get_resolved.return_value = {"unstable": "latest-45-nixos-unstable"}
with patch("mcp_nixos.server.validate_channel") as mock_validate:
mock_validate.return_value = True
yield mock_cache
@pytest.fixture
def mock_flake_response(self):
"""Mock response for flake search results."""
return {
"hits": {
"total": {"value": 3},
"hits": [
{
"_source": {
"flake_attr_name": "neovim",
"flake_name": "nixpkgs",
"flake_url": "github:NixOS/nixpkgs",
"flake_description": "Vim-fork focused on extensibility and usability",
"flake_platforms": ["x86_64-linux", "aarch64-linux", "x86_64-darwin", "aarch64-darwin"],
}
},
{
"_source": {
"flake_attr_name": "packages.x86_64-linux.neovim",
"flake_name": "neovim-nightly",
"flake_url": "github:nix-community/neovim-nightly-overlay",
"flake_description": "Neovim nightly builds",
"flake_platforms": ["x86_64-linux"],
}
},
{
"_source": {
"flake_attr_name": "packages.aarch64-darwin.neovim",
"flake_name": "neovim-nightly",
"flake_url": "github:nix-community/neovim-nightly-overlay",
"flake_description": "Neovim nightly builds",
"flake_platforms": ["aarch64-darwin"],
}
},
],
}
}
@pytest.fixture
def mock_popular_flakes_response(self):
"""Mock response for popular flakes."""
return {
"hits": {
"total": {"value": 5},
"hits": [
{
"_source": {
"flake_attr_name": "homeConfigurations.example",
"flake_name": "home-manager",
"flake_url": "github:nix-community/home-manager",
"flake_description": "Manage a user environment using Nix",
"flake_platforms": ["x86_64-linux", "aarch64-linux", "x86_64-darwin", "aarch64-darwin"],
}
},
{
"_source": {
"flake_attr_name": "nixosConfigurations.example",
"flake_name": "nixos-hardware",
"flake_url": "github:NixOS/nixos-hardware",
"flake_description": "NixOS modules to support various hardware",
"flake_platforms": ["x86_64-linux", "aarch64-linux"],
}
},
{
"_source": {
"flake_attr_name": "devShells.x86_64-linux.default",
"flake_name": "devenv",
"flake_url": "github:cachix/devenv",
"flake_description": (
"Fast, Declarative, Reproducible, and Composable Developer Environments"
),
"flake_platforms": ["x86_64-linux", "x86_64-darwin"],
}
},
{
"_source": {
"flake_attr_name": "packages.x86_64-linux.agenix",
"flake_name": "agenix",
"flake_url": "github:ryantm/agenix",
"flake_description": "age-encrypted secrets for NixOS",
"flake_platforms": ["x86_64-linux", "aarch64-linux", "x86_64-darwin", "aarch64-darwin"],
}
},
{
"_source": {
"flake_attr_name": "packages.x86_64-darwin.agenix",
"flake_name": "agenix",
"flake_url": "github:ryantm/agenix",
"flake_description": "age-encrypted secrets for NixOS",
"flake_platforms": ["x86_64-darwin", "aarch64-darwin"],
}
},
],
}
}
@pytest.fixture
def mock_empty_response(self):
"""Mock empty response."""
return {"hits": {"total": {"value": 0}, "hits": []}}
@patch("requests.post")
@pytest.mark.asyncio
async def test_flake_search_basic(self, mock_post, mock_flake_response):
"""Test basic flake search functionality."""
mock_post.return_value.status_code = 200
mock_post.return_value.json.return_value = mock_flake_response
result = await nixos_search("neovim", search_type="flakes")
# Verify API call
mock_post.assert_called_once()
call_args = mock_post.call_args
assert "_search" in call_args[0][0]
# Check query structure - now using json parameter instead of data
query_data = call_args[1]["json"]
# The query now uses bool->filter->term for type filtering
assert "query" in query_data
assert "size" in query_data
# Verify output format
assert "unique flakes" in result
assert "• nixpkgs" in result or "• neovim" in result
assert "• neovim-nightly" in result
@patch("requests.post")
@pytest.mark.asyncio
async def test_flake_search_deduplication(self, mock_post, mock_flake_response):
"""Test that flake deduplication works correctly."""
mock_post.return_value.status_code = 200
mock_post.return_value.json.return_value = mock_flake_response
result = await nixos_search("neovim", search_type="flakes")
# Should deduplicate neovim-nightly entries
assert result.count("neovim-nightly") == 1
# But should show it has multiple packages
assert "Neovim nightly builds" in result
@patch("requests.post")
@pytest.mark.asyncio
async def test_flake_search_popular(self, mock_post, mock_popular_flakes_response):
"""Test searching for popular flakes."""
mock_post.return_value.status_code = 200
mock_post.return_value.json.return_value = mock_popular_flakes_response
result = await nixos_search("home-manager devenv agenix", search_type="flakes")
assert "Found 5 total matches (4 unique flakes)" in result or "Found 4 unique flakes" in result
assert "• home-manager" in result
assert "• devenv" in result
assert "• agenix" in result
assert "Manage a user environment using Nix" in result
assert "Fast, Declarative, Reproducible, and Composable Developer Environments" in result
assert "age-encrypted secrets for NixOS" in result
@patch("requests.post")
@pytest.mark.asyncio
async def test_flake_search_no_results(self, mock_post, mock_empty_response):
"""Test flake search with no results."""
mock_post.return_value.status_code = 200
mock_post.return_value.json.return_value = mock_empty_response
result = await nixos_search("nonexistentflake123", search_type="flakes")
assert "No flakes found" in result
@patch("requests.post")
@pytest.mark.asyncio
async def test_flake_search_wildcard(self, mock_post):
"""Test flake search with wildcard patterns."""
mock_response = {
"hits": {
"total": {"value": 2},
"hits": [
{
"_source": {
"flake_attr_name": "nixvim",
"flake_name": "nixvim",
"flake_url": "github:nix-community/nixvim",
"flake_description": "Configure Neovim with Nix",
"flake_platforms": ["x86_64-linux", "x86_64-darwin"],
}
},
{
"_source": {
"flake_attr_name": "vim-startify",
"flake_name": "vim-plugins",
"flake_url": "github:m15a/nixpkgs-vim-extra-plugins",
"flake_description": "Extra Vim plugins for Nix",
"flake_platforms": ["x86_64-linux"],
}
},
],
}
}
mock_post.return_value.status_code = 200
mock_post.return_value.json.return_value = mock_response
result = await nixos_search("*vim*", search_type="flakes")
assert "Found 2 unique flakes" in result
assert "• nixvim" in result
assert "• vim-plugins" in result
@patch("requests.post")
@pytest.mark.asyncio
async def test_flake_search_error_handling(self, mock_post):
"""Test flake search error handling."""
mock_response = MagicMock()
mock_response.status_code = 500
mock_response.content = b"Internal Server Error"
# Create an HTTPError with a response attribute
http_error = requests.HTTPError("500 Server Error")
http_error.response = mock_response
mock_response.raise_for_status.side_effect = http_error
mock_post.return_value = mock_response
result = await nixos_search("test", search_type="flakes")
assert "Error" in result
# The actual error message will be the exception string
assert "'NoneType' object has no attribute 'status_code'" not in result
@patch("requests.post")
@pytest.mark.asyncio
async def test_flake_search_malformed_response(self, mock_post):
"""Test handling of malformed flake responses."""
mock_response = {
"hits": {
"total": {"value": 1},
"hits": [
{
"_source": {
"flake_attr_name": "broken",
# Missing required fields
}
}
],
}
}
mock_post.return_value.status_code = 200
mock_post.return_value.json.return_value = mock_response
result = await nixos_search("broken", search_type="flakes")
# Should handle gracefully - with missing fields, no flakes will be created
assert "Found 1 total matches (0 unique flakes)" in result
class TestImprovedStatsEvals:
"""Test improved stats functionality."""
@patch("requests.get")
@pytest.mark.asyncio
async def test_home_manager_stats_with_data(self, mock_get):
"""Test home_manager_stats returns actual statistics."""
mock_html = """
<html>
<body>
<dl class="variablelist">
<dt id="opt-programs.git.enable">programs.git.enable</dt>
<dd>Enable git</dd>
<dt id="opt-programs.vim.enable">programs.vim.enable</dt>
<dd>Enable vim</dd>
<dt id="opt-services.gpg-agent.enable">services.gpg-agent.enable</dt>
<dd>Enable gpg-agent</dd>
</dl>
</body>
</html>
"""
mock_get.return_value.status_code = 200
mock_get.return_value.content = mock_html.encode("utf-8")
result = await home_manager_stats()
assert "Home Manager Statistics:" in result
assert "Total options: 3" in result
assert "Categories:" in result
assert "- programs: 2 options" in result
assert "- services: 1 options" in result
@patch("requests.get")
@pytest.mark.asyncio
async def test_home_manager_stats_error_handling(self, mock_get):
"""Test home_manager_stats error handling."""
mock_get.return_value.status_code = 404
mock_get.return_value.content = b"Not Found"
result = await home_manager_stats()
assert "Error" in result
@patch("requests.get")
@pytest.mark.asyncio
async def test_darwin_stats_with_data(self, mock_get):
"""Test darwin_stats returns actual statistics."""
mock_html = """
<html>
<body>
<div id="toc">
<dl>
<dt><a href="#opt-system.defaults.dock.autohide">system.defaults.dock.autohide</a></dt>
<dd>Auto-hide the dock</dd>
<dt><a href="#opt-system.defaults.finder.ShowPathbar">system.defaults.finder.ShowPathbar</a></dt>
<dd>Show path bar in Finder</dd>
<dt><a href="#opt-homebrew.enable">homebrew.enable</a></dt>
<dd>Enable Homebrew</dd>
<dt><a href="#opt-homebrew.casks">homebrew.casks</a></dt>
<dd>List of Homebrew casks to install</dd>
</dl>
</div>
</body>
</html>
"""
mock_get.return_value.status_code = 200
mock_get.return_value.content = mock_html.encode("utf-8")
result = await darwin_stats()
assert "nix-darwin Statistics:" in result
assert "Total options: 4" in result
assert "Categories:" in result
assert "- system: 2 options" in result
assert "- homebrew: 2 options" in result
@patch("requests.get")
@pytest.mark.asyncio
async def test_darwin_stats_error_handling(self, mock_get):
"""Test darwin_stats error handling."""
mock_get.return_value.status_code = 500
mock_get.return_value.content = b"Server Error"
result = await darwin_stats()
assert "Error" in result
@patch("requests.get")
@pytest.mark.asyncio
async def test_stats_with_complex_categories(self, mock_get):
"""Test stats functions with complex nested categories."""
mock_html = """
<html>
<body>
<dl class="variablelist">
<dt id="opt-programs.git.enable">programs.git.enable</dt>
<dd>Enable git</dd>
<dt id="opt-programs.git.signing.key">programs.git.signing.key</dt>
<dd>GPG signing key</dd>
<dt id="opt-services.xserver.displayManager.gdm.enable">services.xserver.displayManager.gdm.enable</dt>
<dd>Enable GDM</dd>
<dt id="opt-home.packages">home.packages</dt>
<dd>List of packages</dd>
</dl>
</body>
</html>
"""
mock_get.return_value.status_code = 200
mock_get.return_value.content = mock_html.encode("utf-8")
result = await home_manager_stats()
assert "Total options: 4" in result
assert "- programs: 2 options" in result
assert "- services: 1 options" in result
assert "- home: 1 options" in result
@patch("requests.get")
@pytest.mark.asyncio
async def test_stats_with_empty_html(self, mock_get):
"""Test stats functions with empty HTML."""
mock_get.return_value.status_code = 200
mock_get.return_value.content = b"<html><body></body></html>"
result = await home_manager_stats()
# When no options are found, the function returns an error
assert "Error" in result
assert "Failed to fetch Home Manager statistics" in result
class TestRealWorldScenarios:
"""Test real-world usage scenarios for flake search and stats."""
@pytest.fixture(autouse=True)
def mock_channel_validation(self):
"""Mock channel validation to always pass for 'unstable'."""
with patch("mcp_nixos.server.channel_cache") as mock_cache:
mock_cache.get_available.return_value = {"unstable": "latest-45-nixos-unstable"}
mock_cache.get_resolved.return_value = {"unstable": "latest-45-nixos-unstable"}
with patch("mcp_nixos.server.validate_channel") as mock_validate:
mock_validate.return_value = True
yield mock_cache
@patch("requests.post")
@pytest.mark.asyncio
async def test_developer_workflow_flake_search(self, mock_post):
"""Test a developer searching for development environment flakes."""
# First search for devenv
devenv_response = {
"hits": {
"total": {"value": 1},
"hits": [
{
"_source": {
"flake_attr_name": "devShells.x86_64-linux.default",
"flake_name": "devenv",
"flake_url": "github:cachix/devenv",
"flake_description": (
"Fast, Declarative, Reproducible, and Composable Developer Environments"
),
"flake_platforms": ["x86_64-linux", "x86_64-darwin"],
}
}
],
}
}
mock_post.return_value.status_code = 200
mock_post.return_value.json.return_value = devenv_response
result = await nixos_search("devenv", search_type="flakes")
assert "• devenv" in result
assert "Fast, Declarative, Reproducible, and Composable Developer Environments" in result
assert "Developer Environments" in result
@patch("requests.post")
@pytest.mark.asyncio
async def test_system_configuration_flake_search(self, mock_post):
"""Test searching for system configuration flakes."""
config_response = {
"hits": {
"total": {"value": 3},
"hits": [
{
"_source": {
"flake_attr_name": "nixosModules.default",
"flake_name": "impermanence",
"flake_url": "github:nix-community/impermanence",
"flake_description": (
"Modules to help you handle persistent state on systems with ephemeral root storage"
),
"flake_platforms": ["x86_64-linux", "aarch64-linux"],
}
},
{
"_source": {
"flake_attr_name": "nixosModules.home-manager",
"flake_name": "home-manager",
"flake_url": "github:nix-community/home-manager",
"flake_description": "Manage a user environment using Nix",
"flake_platforms": ["x86_64-linux", "aarch64-linux", "x86_64-darwin", "aarch64-darwin"],
}
},
{
"_source": {
"flake_attr_name": "nixosModules.sops",
"flake_name": "sops-nix",
"flake_url": "github:Mic92/sops-nix",
"flake_description": "Atomic secret provisioning for NixOS based on sops",
"flake_platforms": ["x86_64-linux", "aarch64-linux"],
}
},
],
}
}
mock_post.return_value.status_code = 200
mock_post.return_value.json.return_value = config_response
result = await nixos_search("nixosModules", search_type="flakes")
assert "Found 3 unique flakes" in result
assert "• impermanence" in result
assert "• home-manager" in result
assert "• sops-nix" in result
assert "ephemeral root storage" in result
assert "secret provisioning" in result
@patch("requests.get")
@patch("requests.post")
@pytest.mark.asyncio
async def test_combined_workflow_stats_and_search(self, mock_post, mock_get):
"""Test a workflow combining stats check and targeted search."""
# First, check Home Manager stats
stats_html = """
<html>
<body>
<dl class="variablelist">
<dt id="opt-programs.neovim.enable">programs.neovim.enable</dt>
<dd>Enable neovim</dd>
<dt id="opt-programs.neovim.plugins">programs.neovim.plugins</dt>
<dd>List of vim plugins</dd>
<dt id="opt-programs.vim.enable">programs.vim.enable</dt>
<dd>Enable vim</dd>
</dl>
</body>
</html>
"""
mock_get.return_value.status_code = 200
mock_get.return_value.content = stats_html.encode("utf-8")
stats_result = await home_manager_stats()
assert "Total options: 3" in stats_result
assert "- programs: 3 options" in stats_result
# Then search for related flakes
flake_response = {
"hits": {
"total": {"value": 1},
"hits": [
{
"_source": {
"flake_attr_name": "homeManagerModules.nixvim",
"flake_name": "nixvim",
"flake_url": "github:nix-community/nixvim",
"flake_description": "Configure Neovim with Nix",
"flake_platforms": ["x86_64-linux", "x86_64-darwin"],
}
}
],
}
}
mock_post.return_value.status_code = 200
mock_post.return_value.json.return_value = flake_response
search_result = await nixos_search("nixvim", search_type="flakes")
assert "• nixvim" in search_result
assert "Configure Neovim with Nix" in search_result
if __name__ == "__main__":
pytest.main([__file__, "-v"])
# ===== Content from test_flake_search_improved.py =====
class TestImprovedFlakeSearch:
"""Test improved flake search functionality."""
@pytest.fixture
def mock_empty_flake_response(self):
"""Mock response for empty query with various flake types."""
return {
"hits": {
"total": {"value": 894},
"hits": [
{
"_source": {
"flake_name": "",
"flake_description": "Home Manager for Nix",
"package_pname": "home-manager",
"package_attr_name": "docs-json",
"flake_source": {"type": "github", "owner": "nix-community", "repo": "home-manager"},
"flake_resolved": {"type": "github", "owner": "nix-community", "repo": "home-manager"},
}
},
{
"_source": {
"flake_name": "haskell.nix",
"flake_description": "Alternative Haskell Infrastructure for Nixpkgs",
"package_pname": "hix",
"package_attr_name": "hix",
"flake_source": {"type": "github", "owner": "input-output-hk", "repo": "haskell.nix"},
"flake_resolved": {"type": "github", "owner": "input-output-hk", "repo": "haskell.nix"},
}
},
{
"_source": {
"flake_name": "nix-vscode-extensions",
"flake_description": (
"VS Code Marketplace (~40K) and Open VSX (~3K) extensions as Nix expressions."
),
"package_pname": "updateExtensions",
"package_attr_name": "updateExtensions",
"flake_source": {
"type": "github",
"owner": "nix-community",
"repo": "nix-vscode-extensions",
},
"flake_resolved": {
"type": "github",
"owner": "nix-community",
"repo": "nix-vscode-extensions",
},
}
},
{
"_source": {
"flake_name": "",
"flake_description": "A Python wrapper for the Trovo API",
"package_pname": "python3.11-python-trovo-0.1.7",
"package_attr_name": "default",
"flake_source": {"type": "git", "url": "https://codeberg.org/wolfangaukang/python-trovo"},
"flake_resolved": {"type": "git", "url": "https://codeberg.org/wolfangaukang/python-trovo"},
}
},
],
}
}
@patch("requests.post")
@pytest.mark.asyncio
async def test_empty_query_returns_all_flakes(self, mock_post, mock_empty_flake_response):
"""Test that empty query returns all flakes."""
mock_post.return_value.status_code = 200
mock_post.return_value.json.return_value = mock_empty_flake_response
result = await nixos_flakes_search("", limit=50)
# Should use match_all query for empty search
call_args = mock_post.call_args
query_data = call_args[1]["json"]
# The query is wrapped in bool->filter->must structure
assert "match_all" in str(query_data["query"])
# Should show results
assert "4 unique flakes" in result
assert "home-manager" in result
assert "haskell.nix" in result
assert "nix-vscode-extensions" in result
@patch("requests.post")
@pytest.mark.asyncio
async def test_wildcard_query_returns_all_flakes(self, mock_post, mock_empty_flake_response):
"""Test that * query returns all flakes."""
mock_post.return_value.status_code = 200
mock_post.return_value.json.return_value = mock_empty_flake_response
await nixos_flakes_search("*", limit=50) # Result not used in this test
# Should use match_all query for wildcard
call_args = mock_post.call_args
query_data = call_args[1]["json"]
# The query is wrapped in bool->filter->must structure
assert "match_all" in str(query_data["query"])
@patch("requests.post")
@pytest.mark.asyncio
async def test_search_by_owner(self, mock_post):
"""Test searching by owner like nix-community."""
mock_response = {
"hits": {
"total": {"value": 2},
"hits": [
{
"_source": {
"flake_name": "home-manager",
"flake_description": "Home Manager for Nix",
"package_pname": "home-manager",
"flake_resolved": {"type": "github", "owner": "nix-community", "repo": "home-manager"},
}
}
],
}
}
mock_post.return_value.status_code = 200
mock_post.return_value.json.return_value = mock_response
await nixos_flakes_search("nix-community", limit=20) # Result tested via assertions
# Should search in owner field
call_args = mock_post.call_args
query_data = call_args[1]["json"]
# The query structure has bool->filter and bool->must
assert "nix-community" in str(query_data["query"])
@patch("requests.post")
@pytest.mark.asyncio
async def test_deduplication_by_repo(self, mock_post):
"""Test that multiple packages from same repo are deduplicated."""
mock_response = {
"hits": {
"total": {"value": 4},
"hits": [
{
"_source": {
"flake_name": "",
"package_pname": "hix",
"package_attr_name": "hix",
"flake_resolved": {"owner": "input-output-hk", "repo": "haskell.nix"},
}
},
{
"_source": {
"flake_name": "",
"package_pname": "hix-build",
"package_attr_name": "hix-build",
"flake_resolved": {"owner": "input-output-hk", "repo": "haskell.nix"},
}
},
{
"_source": {
"flake_name": "",
"package_pname": "hix-env",
"package_attr_name": "hix-env",
"flake_resolved": {"owner": "input-output-hk", "repo": "haskell.nix"},
}
},
],
}
}
mock_post.return_value.status_code = 200
mock_post.return_value.json.return_value = mock_response
result = await nixos_flakes_search("haskell", limit=20)
# Should show only one flake with multiple packages
assert "1 unique flakes" in result
assert "input-output-hk/haskell.nix" in result
assert "Packages: hix, hix-build, hix-env" in result
@patch("requests.post")
@pytest.mark.asyncio
async def test_handles_flakes_without_name(self, mock_post):
"""Test handling flakes with empty flake_name."""
mock_response = {
"hits": {
"total": {"value": 1},
"hits": [
{
"_source": {
"flake_name": "",
"flake_description": "Home Manager for Nix",
"package_pname": "home-manager",
"flake_resolved": {"owner": "nix-community", "repo": "home-manager"},
}
}
],
}
}
mock_post.return_value.status_code = 200
mock_post.return_value.json.return_value = mock_response
result = await nixos_flakes_search("home-manager", limit=20)
# Should use repo name when flake_name is empty
assert "home-manager" in result
assert "nix-community/home-manager" in result
@patch("requests.post")
@pytest.mark.asyncio
async def test_no_results_shows_suggestions(self, mock_post):
"""Test that no results shows helpful suggestions."""
mock_response = {"hits": {"total": {"value": 0}, "hits": []}}
mock_post.return_value.status_code = 200
mock_post.return_value.json.return_value = mock_response
result = await nixos_flakes_search("nonexistent", limit=20)
assert "No flakes found" in result
assert "Popular flakes: nixpkgs, home-manager, flake-utils, devenv" in result
assert "By owner: nix-community, numtide, cachix" in result
assert "GitHub: https://github.com/topics/nix-flakes" in result
assert "FlakeHub: https://flakehub.com/" in result
@patch("requests.post")
@pytest.mark.asyncio
async def test_handles_git_urls(self, mock_post):
"""Test handling of non-GitHub Git URLs."""
mock_response = {
"hits": {
"total": {"value": 1},
"hits": [
{
"_source": {
"flake_name": "",
"package_pname": "python-trovo",
"flake_resolved": {"type": "git", "url": "https://codeberg.org/wolfangaukang/python-trovo"},
}
}
],
}
}
mock_post.return_value.status_code = 200
mock_post.return_value.json.return_value = mock_response
result = await nixos_flakes_search("python", limit=20)
assert "python-trovo" in result
@patch("requests.post")
@pytest.mark.asyncio
async def test_search_tracks_total_hits(self, mock_post):
"""Test that search tracks total hits."""
mock_response = {"hits": {"total": {"value": 894}, "hits": []}}
mock_post.return_value.status_code = 200
mock_post.return_value.json.return_value = mock_response
# Make the call
await nixos_flakes_search("", limit=20)
# Check that track_total_hits was set
call_args = mock_post.call_args
query_data = call_args[1]["json"]
assert query_data.get("track_total_hits") is True
@patch("requests.post")
@pytest.mark.asyncio
async def test_increased_size_multiplier(self, mock_post):
"""Test that we request more results to account for duplicates."""
mock_response = {"hits": {"total": {"value": 0}, "hits": []}}
mock_post.return_value.status_code = 200
mock_post.return_value.json.return_value = mock_response
await nixos_flakes_search("test", limit=20)
# Should request more than limit to account for duplicates
call_args = mock_post.call_args
query_data = call_args[1]["json"]
assert query_data["size"] > 20 # Should be limit * 5 = 100
# ===== Content from test_flake_search.py =====
class TestFlakeSearch:
"""Test flake search functionality."""
@pytest.mark.asyncio
@patch("mcp_nixos.server.requests.post")
async def test_flakes_search_empty_query(self, mock_post):
"""Test flake search with empty query returns all flakes."""
# Mock response
mock_response = Mock()
mock_response.status_code = 200
mock_response.json.return_value = {
"hits": {
"total": {"value": 100},
"hits": [
{
"_source": {
"flake_name": "home-manager",
"flake_description": "Home Manager for Nix",
"flake_resolved": {
"type": "github",
"owner": "nix-community",
"repo": "home-manager",
},
"package_pname": "home-manager",
"package_attr_name": "default",
}
}
],
}
}
mock_post.return_value = mock_response
result = await nixos_flakes_search("", limit=10)
assert "Found 100 total matches" in result
assert "home-manager" in result
assert "nix-community/home-manager" in result
assert "Home Manager for Nix" in result
# Verify the query structure
call_args = mock_post.call_args
query_data = call_args[1]["json"]["query"]
# Should have a bool query with filter and must
assert "bool" in query_data
assert "filter" in query_data["bool"]
assert "must" in query_data["bool"]
@pytest.mark.asyncio
@patch("mcp_nixos.server.requests.post")
async def test_flakes_search_with_query(self, mock_post):
"""Test flake search with specific query."""
# Mock response
mock_response = Mock()
mock_response.status_code = 200
mock_response.json.return_value = {
"hits": {
"total": {"value": 5},
"hits": [
{
"_source": {
"flake_name": "devenv",
"flake_description": "Fast, Declarative, Reproducible Developer Environments",
"flake_resolved": {
"type": "github",
"owner": "cachix",
"repo": "devenv",
},
"package_pname": "devenv",
"package_attr_name": "default",
}
}
],
}
}
mock_post.return_value = mock_response
result = await nixos_flakes_search("devenv", limit=10)
assert "Found 5" in result
assert "devenv" in result
assert "cachix/devenv" in result
assert "Fast, Declarative" in result
# Verify the query structure has filter and inner bool
call_args = mock_post.call_args
query_data = call_args[1]["json"]["query"]
assert "bool" in query_data
assert "filter" in query_data["bool"]
assert "must" in query_data["bool"]
# The actual search query is inside must
inner_query = query_data["bool"]["must"][0]
assert "bool" in inner_query
assert "should" in inner_query["bool"]
@pytest.mark.asyncio
@patch("mcp_nixos.server.requests.post")
async def test_flakes_search_no_results(self, mock_post):
"""Test flake search with no results."""
# Mock response
mock_response = Mock()
mock_response.status_code = 200
mock_response.json.return_value = {"hits": {"total": {"value": 0}, "hits": []}}
mock_post.return_value = mock_response
result = await nixos_flakes_search("nonexistent", limit=10)
assert "No flakes found matching 'nonexistent'" in result
assert "Try searching for:" in result
assert "Popular flakes:" in result
@pytest.mark.asyncio
@patch("mcp_nixos.server.requests.post")
async def test_flakes_search_deduplication(self, mock_post):
"""Test flake search properly deduplicates flakes."""
# Mock response with duplicate flakes
mock_response = Mock()
mock_response.status_code = 200
mock_response.json.return_value = {
"hits": {
"total": {"value": 4},
"hits": [
{
"_source": {
"flake_name": "nixpkgs",
"flake_resolved": {"type": "github", "owner": "NixOS", "repo": "nixpkgs"},
"package_pname": "hello",
"package_attr_name": "hello",
}
},
{
"_source": {
"flake_name": "nixpkgs",
"flake_resolved": {"type": "github", "owner": "NixOS", "repo": "nixpkgs"},
"package_pname": "git",
"package_attr_name": "git",
}
},
],
}
}
mock_post.return_value = mock_response
result = await nixos_flakes_search("nixpkgs", limit=10)
# Should show 1 unique flake with 2 packages
assert "Found 4 total matches (1 unique flakes)" in result
assert "nixpkgs" in result
assert "NixOS/nixpkgs" in result
assert "Packages: git, hello" in result
@pytest.mark.asyncio
@patch("mcp_nixos.server.requests.post")
async def test_flakes_stats(self, mock_post):
"""Test flake statistics."""
# Mock responses
mock_count_response = Mock()
mock_count_response.status_code = 200
mock_count_response.json.return_value = {"count": 452176}
# Mock search response for sampling
mock_search_response = Mock()
mock_search_response.status_code = 200
mock_search_response.json.return_value = {
"hits": {
"hits": [
{
"_source": {
"flake_resolved": {
"url": "https://github.com/nix-community/home-manager",
"type": "github",
},
"package_pname": "home-manager",
}
},
{
"_source": {
"flake_resolved": {"url": "https://github.com/NixOS/nixpkgs", "type": "github"},
"package_pname": "hello",
}
},
]
}
}
mock_post.side_effect = [mock_count_response, mock_search_response]
result = await nixos_flakes_stats()
assert "Available flakes: 452,176" in result
# Stats now samples documents, not using aggregations
# So we won't see the mocked aggregation values
@pytest.mark.asyncio
@patch("mcp_nixos.server.requests.post")
async def test_flakes_search_error_handling(self, mock_post):
"""Test flake search error handling."""
# Mock 404 response with HTTPError
from requests import HTTPError
mock_response = Mock()
mock_response.status_code = 404
error = HTTPError()
error.response = mock_response
mock_response.raise_for_status.side_effect = error
mock_post.return_value = mock_response
result = await nixos_flakes_search("test", limit=10)
assert "Error" in result
assert "Flake indices not found" in result
# ===== Content from test_flakes_stats_eval.py =====
class TestFlakesStatsEval:
"""Test evaluations for flakes statistics and counting."""
@pytest.mark.asyncio
@patch("mcp_nixos.server.requests.post")
async def test_get_total_flakes_count(self, mock_post):
"""Eval: User asks 'how many flakes are there?'"""
# Mock flakes stats responses
def side_effect(*args, **kwargs):
url = args[0]
if "/_count" in url:
# Count request
mock_response = Mock()
mock_response.status_code = 200
mock_response.json.return_value = {"count": 4500}
return mock_response
# Regular search request
# Search request to get sample documents
mock_response = Mock()
mock_response.status_code = 200
mock_response.json.return_value = {
"hits": {
"total": {"value": 4500},
"hits": [
{
"_source": {
"flake_resolved": {"url": "https://github.com/NixOS/nixpkgs", "type": "github"},
"package_pname": "hello",
}
},
{
"_source": {
"flake_resolved": {
"url": "https://github.com/nix-community/home-manager",
"type": "github",
},
"package_pname": "home-manager",
}
},
]
* 10, # Simulate more hits
}
}
return mock_response
mock_post.side_effect = side_effect
# Get flakes stats
result = await nixos_flakes_stats()
# Should show available flakes count (formatted with comma)
assert "Available flakes:" in result
assert "4,500" in result # Matches our mock data
# Should show unique repositories count
assert "Unique repositories:" in result
# The actual count depends on unique URLs in mock data
# Should show breakdown by type
assert "Flake types:" in result
assert "github:" in result # Our mock data only has github type
# Should show top contributors
assert "Top contributors:" in result
assert "NixOS:" in result
assert "nix-community:" in result
@pytest.mark.asyncio
@patch("mcp_nixos.server.requests.post")
async def test_flakes_search_shows_total_count(self, mock_post):
"""Eval: Flakes search should show total matching flakes."""
# Mock search response with multiple hits
mock_response = Mock()
mock_response.status_code = 200
mock_response.json.return_value = {
"hits": {
"total": {"value": 156},
"hits": [
{
"_source": {
"flake_name": "nixpkgs",
"flake_description": "Nix Packages collection",
"flake_resolved": {
"owner": "NixOS",
"repo": "nixpkgs",
},
"package_attr_name": "packages.x86_64-linux.hello",
}
},
{
"_source": {
"flake_name": "nixpkgs",
"flake_description": "Nix Packages collection",
"flake_resolved": {
"owner": "NixOS",
"repo": "nixpkgs",
},
"package_attr_name": "packages.x86_64-linux.git",
}
},
],
}
}
mock_post.return_value = mock_response
# Search for nix
result = await nixos_flakes_search("nix", limit=2)
# Should show both total matches and unique flakes count
assert "total matches" in result
assert "unique flakes" in result
assert "nixpkgs" in result
@pytest.mark.asyncio
@patch("mcp_nixos.server.requests.post")
async def test_flakes_wildcard_search_shows_all(self, mock_post):
"""Eval: User searches with '*' to see all flakes."""
# Mock response with many flakes
mock_response = Mock()
mock_response.status_code = 200
mock_response.json.return_value = {
"hits": {
"total": {"value": 4500},
"hits": [
{
"_source": {
"flake_name": "devenv",
"flake_description": "Development environments",
"flake_resolved": {"owner": "cachix", "repo": "devenv"},
"package_attr_name": "packages.x86_64-linux.devenv",
}
},
{
"_source": {
"flake_name": "home-manager",
"flake_description": "Manage user configuration",
"flake_resolved": {"owner": "nix-community", "repo": "home-manager"},
"package_attr_name": "packages.x86_64-linux.home-manager",
}
},
{
"_source": {
"flake_name": "",
"flake_description": "Flake utilities",
"flake_resolved": {"owner": "numtide", "repo": "flake-utils"},
"package_attr_name": "lib.eachDefaultSystem",
}
},
],
}
}
mock_post.return_value = mock_response
# Wildcard search
result = await nixos_flakes_search("*", limit=10)
# Should show total count
assert "total matches" in result
# Should list some flakes
assert "devenv" in result
assert "home-manager" in result
@pytest.mark.asyncio
@patch("mcp_nixos.server.requests.post")
async def test_flakes_stats_with_no_flakes(self, mock_post):
"""Eval: Flakes stats when no flakes are indexed."""
# Mock empty response
def side_effect(*args, **kwargs):
url = args[0]
mock_response = Mock()
mock_response.status_code = 200
if "/_count" in url:
# Count request
mock_response.json.return_value = {"count": 0}
else:
# Search with aggregations
mock_response.json.return_value = {
"hits": {"total": {"value": 0}},
"aggregations": {
"unique_flakes": {"value": 0},
"flake_types": {"buckets": []},
"top_owners": {"buckets": []},
},
}
return mock_response
mock_post.side_effect = side_effect
result = await nixos_flakes_stats()
# Should handle empty case gracefully
assert "Available flakes: 0" in result
@pytest.mark.asyncio
@patch("mcp_nixos.server.requests.post")
async def test_flakes_stats_error_handling(self, mock_post):
"""Eval: Flakes stats handles API errors gracefully."""
# Mock 404 error
mock_response = Mock()
mock_response.status_code = 404
mock_response.raise_for_status.side_effect = Exception("Not found")
mock_post.return_value = mock_response
result = await nixos_flakes_stats()
# Should return error message
assert "Error" in result
assert "Flake indices not found" in result or "Not found" in result
@pytest.mark.asyncio
@patch("mcp_nixos.server.requests.post")
async def test_compare_flakes_vs_packages(self, mock_post):
"""Eval: User wants to understand flakes vs packages relationship."""
# First call: flakes stats
mock_flakes_response = Mock()
mock_flakes_response.status_code = 200
mock_flakes_response.json.return_value = {
"hits": {"total": {"value": 4500}},
"aggregations": {
"unique_flakes": {"value": 894},
"flake_types": {
"buckets": [
{"key": "github", "doc_count": 3800},
]
},
"top_contributors": {
"buckets": [
{"key": "NixOS", "doc_count": 450},
]
},
},
}
# Second call: regular packages stats (for comparison)
mock_packages_response = Mock()
mock_packages_response.json.return_value = {
"aggregations": {
"attr_count": {"value": 151798},
"option_count": {"value": 20156},
"program_count": {"value": 3421},
"license_count": {"value": 125},
"maintainer_count": {"value": 3254},
"platform_counts": {"buckets": []},
}
}
def side_effect(*args, **kwargs):
url = args[0]
if "latest-43-group-manual" in url:
if "/_count" in url:
# Count request
mock_response = Mock()
mock_response.status_code = 200
mock_response.json.return_value = {"count": 4500}
return mock_response
# Search request - return sample hits
mock_response = Mock()
mock_response.status_code = 200
mock_response.json.return_value = {
"hits": {
"hits": [
{
"_source": {
"flake_resolved": {"url": "https://github.com/NixOS/nixpkgs", "type": "github"}
}
}
]
* 5
}
}
return mock_response
return mock_packages_response
mock_post.side_effect = side_effect
# Get flakes stats
flakes_result = await nixos_flakes_stats()
assert "Available flakes:" in flakes_result
assert "4,500" in flakes_result # From our mock
# Should also show unique repositories
assert "Unique repositories:" in flakes_result
```
--------------------------------------------------------------------------------
/mcp_nixos/server.py:
--------------------------------------------------------------------------------
```python
#!/usr/bin/env python3
"""MCP-NixOS Server - Model Context Protocol tools for NixOS, Home Manager, and nix-darwin.
Provides search and query capabilities for:
- NixOS packages, options, and programs via Elasticsearch API
- Home Manager configuration options via HTML documentation parsing
- nix-darwin (macOS) configuration options via HTML documentation parsing
All responses are formatted as human-readable plain text for optimal LLM interaction.
"""
import re
from typing import Any
import requests
from bs4 import BeautifulSoup
from fastmcp import FastMCP
class APIError(Exception):
"""Custom exception for API-related errors."""
class DocumentParseError(Exception):
"""Custom exception for document parsing errors."""
mcp = FastMCP("mcp-nixos")
# API Configuration
NIXOS_API = "https://search.nixos.org/backend"
NIXOS_AUTH = ("aWVSALXpZv", "X8gPHnzL52wFEekuxsfQ9cSh")
# Base channel patterns - these are dynamic and auto-discovered
BASE_CHANNELS = {
"unstable": "nixos-unstable",
"24.11": "nixos-24.11",
"25.05": "nixos-25.05",
}
# Fallback channels when API discovery fails
# These are static mappings based on most recent known patterns
FALLBACK_CHANNELS = {
"unstable": "latest-44-nixos-unstable",
"stable": "latest-44-nixos-25.05",
"25.05": "latest-44-nixos-25.05",
"25.11": "latest-44-nixos-25.11", # For when 25.11 is released
"beta": "latest-44-nixos-25.05",
}
HOME_MANAGER_URL = "https://nix-community.github.io/home-manager/options.xhtml"
DARWIN_URL = "https://nix-darwin.github.io/nix-darwin/manual/index.html"
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
# Create a single instance of the cache
channel_cache = ChannelCache()
def error(msg: str, code: str = "ERROR") -> str:
"""Format error as plain text."""
# Ensure msg is always a string, even if empty
msg = str(msg) if msg is not None else ""
return f"Error ({code}): {msg}"
def get_channels() -> dict[str, str]:
"""Get current channel mappings (cached and resolved)."""
return channel_cache.get_resolved()
def validate_channel(channel: str) -> bool:
"""Validate if a channel exists and is accessible."""
channels = get_channels()
if channel in channels:
index = channels[channel]
try:
resp = requests.post(
f"{NIXOS_API}/{index}/_count", json={"query": {"match_all": {}}}, auth=NIXOS_AUTH, timeout=5
)
return resp.status_code == 200 and resp.json().get("count", 0) > 0
except Exception:
return False
return False
def get_channel_suggestions(invalid_channel: str) -> str:
"""Get helpful suggestions for invalid channels."""
channels = get_channels()
available = list(channels.keys())
suggestions = []
# Find similar channel names
invalid_lower = invalid_channel.lower()
for channel in available:
if invalid_lower in channel.lower() or channel.lower() in invalid_lower:
suggestions.append(channel)
if not suggestions:
# Fallback to most common channels
common = ["unstable", "stable", "beta"]
# Also include version numbers
version_channels = [ch for ch in available if "." in ch and ch.replace(".", "").isdigit()]
common.extend(version_channels[:2]) # Add up to 2 version channels
suggestions = [ch for ch in common if ch in available]
if not suggestions:
suggestions = available[:4] # First 4 available
return f"Available channels: {', '.join(suggestions)}"
def es_query(index: str, query: dict[str, Any], size: int = 20) -> list[dict[str, Any]]:
"""Execute Elasticsearch query."""
try:
resp = requests.post(
f"{NIXOS_API}/{index}/_search", json={"query": query, "size": size}, auth=NIXOS_AUTH, timeout=10
)
resp.raise_for_status()
data = resp.json()
# Handle malformed responses gracefully
if isinstance(data, dict) and "hits" in data:
hits = data.get("hits", {})
if isinstance(hits, dict) and "hits" in hits:
return list(hits.get("hits", []))
return []
except requests.Timeout as exc:
raise APIError("API error: Connection timed out") from exc
except requests.HTTPError as exc:
raise APIError(f"API error: {str(exc)}") from exc
except Exception as exc:
raise APIError(f"API error: {str(exc)}") from exc
def parse_html_options(url: str, query: str = "", prefix: str = "", limit: int = 100) -> list[dict[str, str]]:
"""Parse options from HTML documentation."""
try:
resp = requests.get(url, timeout=30) # Increase timeout for large docs
resp.raise_for_status()
# Use resp.content to let BeautifulSoup handle encoding detection
# This prevents encoding errors like "unknown encoding: windows-1252"
soup = BeautifulSoup(resp.content, "html.parser")
options = []
# Get all dt elements
dts = soup.find_all("dt")
for dt in dts:
# Get option name
name = ""
if "home-manager" in url:
# Home Manager uses anchor IDs like "opt-programs.git.enable"
anchor = dt.find("a", id=True)
if anchor:
anchor_id = anchor.get("id", "")
# Remove "opt-" prefix and convert underscores
if anchor_id.startswith("opt-"):
name = anchor_id[4:] # Remove "opt-" prefix
# Convert _name_ placeholders back to <name>
name = name.replace("_name_", "<name>")
else:
# Fallback to text content
name_elem = dt.find(string=True, recursive=False)
if name_elem:
name = name_elem.strip()
else:
name = dt.get_text(strip=True)
else:
# Darwin and fallback - use text content
name = dt.get_text(strip=True)
# Skip if it doesn't look like an option (must contain a dot)
# But allow single-word options in some cases
if "." not in name and len(name.split()) > 1:
continue
# Filter by query or prefix
if query and query.lower() not in name.lower():
continue
if prefix and not (name.startswith(prefix + ".") or name == prefix):
continue
# Find the corresponding dd element
dd = dt.find_next_sibling("dd")
if dd:
# Extract description (first p tag or direct text)
desc_elem = dd.find("p")
if desc_elem:
description = desc_elem.get_text(strip=True)
else:
# Get first text node, handle None case
text = dd.get_text(strip=True)
description = text.split("\n")[0] if text else ""
# Extract type info - look for various patterns
type_info = ""
# Pattern 1: <span class="term">Type: ...</span>
type_elem = dd.find("span", class_="term")
if type_elem and "Type:" in type_elem.get_text():
type_info = type_elem.get_text(strip=True).replace("Type:", "").strip()
# Pattern 2: Look for "Type:" in text
elif "Type:" in dd.get_text():
text = dd.get_text()
type_start = text.find("Type:") + 5
type_end = text.find("\n", type_start)
if type_end == -1:
type_end = len(text)
type_info = text[type_start:type_end].strip()
options.append(
{
"name": name,
"description": description[:200] if len(description) > 200 else description,
"type": type_info,
}
)
if len(options) >= limit:
break
return options
except Exception as exc:
raise DocumentParseError(f"Failed to fetch docs: {str(exc)}") from exc
@mcp.tool()
async def nixos_search(query: str, search_type: str = "packages", limit: int = 20, channel: str = "unstable") -> str:
"""Search NixOS packages, options, or programs.
Args:
query: Search term to look for
search_type: Type of search - "packages", "options", "programs", or "flakes"
limit: Maximum number of results to return (1-100)
channel: NixOS channel to search in (e.g., "unstable", "stable", "25.05")
Returns:
Plain text results with bullet points or error message
"""
if search_type not in ["packages", "options", "programs", "flakes"]:
return error(f"Invalid type '{search_type}'")
channels = get_channels()
if channel not in channels:
suggestions = get_channel_suggestions(channel)
return error(f"Invalid channel '{channel}'. {suggestions}")
if not 1 <= limit <= 100:
return error("Limit must be 1-100")
# Redirect flakes to dedicated function
if search_type == "flakes":
return await _nixos_flakes_search_impl(query, limit)
try:
# Build query with correct field names
if search_type == "packages":
q = {
"bool": {
"must": [{"term": {"type": "package"}}],
"should": [
{"match": {"package_pname": {"query": query, "boost": 3}}},
{"match": {"package_description": query}},
],
"minimum_should_match": 1,
}
}
elif search_type == "options":
# Use wildcard for option names to handle hierarchical names like services.nginx.enable
q = {
"bool": {
"must": [{"term": {"type": "option"}}],
"should": [
{"wildcard": {"option_name": f"*{query}*"}},
{"match": {"option_description": query}},
],
"minimum_should_match": 1,
}
}
else: # programs
q = {
"bool": {
"must": [{"term": {"type": "package"}}],
"should": [
{"match": {"package_programs": {"query": query, "boost": 2}}},
{"match": {"package_pname": query}},
],
"minimum_should_match": 1,
}
}
hits = es_query(channels[channel], q, limit)
# Format results as plain text
if not hits:
return f"No {search_type} found matching '{query}'"
results = []
results.append(f"Found {len(hits)} {search_type} matching '{query}':\n")
for hit in hits:
src = hit.get("_source", {})
if search_type == "packages":
name = src.get("package_pname", "")
version = src.get("package_pversion", "")
desc = src.get("package_description", "")
results.append(f"• {name} ({version})")
if desc:
results.append(f" {desc}")
results.append("")
elif search_type == "options":
name = src.get("option_name", "")
opt_type = src.get("option_type", "")
desc = src.get("option_description", "")
# Strip HTML tags from description
if desc and "<rendered-html>" in desc:
# Remove outer rendered-html tags
desc = desc.replace("<rendered-html>", "").replace("</rendered-html>", "")
# Remove common HTML tags
desc = re.sub(r"<[^>]+>", "", desc)
desc = desc.strip()
results.append(f"• {name}")
if opt_type:
results.append(f" Type: {opt_type}")
if desc:
results.append(f" {desc}")
results.append("")
else: # programs
programs = src.get("package_programs", [])
pkg_name = src.get("package_pname", "")
# Check if query matches any program exactly (case-insensitive)
query_lower = query.lower()
matched_programs = [p for p in programs if p.lower() == query_lower]
for prog in matched_programs:
results.append(f"• {prog} (provided by {pkg_name})")
results.append("")
return "\n".join(results).strip()
except Exception as e:
return error(str(e))
@mcp.tool()
async def nixos_info(name: str, type: str = "package", channel: str = "unstable") -> str: # pylint: disable=redefined-builtin
"""Get detailed info about a NixOS package or option.
Args:
name: Name of the package or option to look up
type: Type of lookup - "package" or "option"
channel: NixOS channel to search in (e.g., "unstable", "stable", "25.05")
Returns:
Plain text details about the package/option or error message
"""
info_type = type # Avoid shadowing built-in
if info_type not in ["package", "option"]:
return error("Type must be 'package' or 'option'")
channels = get_channels()
if channel not in channels:
suggestions = get_channel_suggestions(channel)
return error(f"Invalid channel '{channel}'. {suggestions}")
try:
# Exact match query with correct field names
field = "package_pname" if info_type == "package" else "option_name"
query = {"bool": {"must": [{"term": {"type": info_type}}, {"term": {field: name}}]}}
hits = es_query(channels[channel], query, 1)
if not hits:
return error(f"{info_type.capitalize()} '{name}' not found", "NOT_FOUND")
src = hits[0].get("_source", {})
if info_type == "package":
info = []
info.append(f"Package: {src.get('package_pname', '')}")
info.append(f"Version: {src.get('package_pversion', '')}")
desc = src.get("package_description", "")
if desc:
info.append(f"Description: {desc}")
homepage = src.get("package_homepage", [])
if homepage:
if isinstance(homepage, list):
homepage = homepage[0] if homepage else ""
info.append(f"Homepage: {homepage}")
licenses = src.get("package_license_set", [])
if licenses:
info.append(f"License: {', '.join(licenses)}")
return "\n".join(info)
# Option type
info = []
info.append(f"Option: {src.get('option_name', '')}")
opt_type = src.get("option_type", "")
if opt_type:
info.append(f"Type: {opt_type}")
desc = src.get("option_description", "")
if desc:
# Strip HTML tags from description
if "<rendered-html>" in desc:
desc = desc.replace("<rendered-html>", "").replace("</rendered-html>", "")
desc = re.sub(r"<[^>]+>", "", desc)
desc = desc.strip()
info.append(f"Description: {desc}")
default = src.get("option_default", "")
if default:
info.append(f"Default: {default}")
example = src.get("option_example", "")
if example:
info.append(f"Example: {example}")
return "\n".join(info)
except Exception as e:
return error(str(e))
@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.tool()
async def nixos_stats(channel: str = "unstable") -> str:
"""Get NixOS statistics for a channel.
Args:
channel: NixOS channel to get stats for (e.g., "unstable", "stable", "25.05")
Returns:
Plain text statistics including package/option counts
"""
channels = get_channels()
if channel not in channels:
suggestions = get_channel_suggestions(channel)
return error(f"Invalid channel '{channel}'. {suggestions}")
try:
index = channels[channel]
url = f"{NIXOS_API}/{index}/_count"
# Get counts with error handling
try:
pkg_resp = requests.post(url, json={"query": {"term": {"type": "package"}}}, auth=NIXOS_AUTH, timeout=10)
pkg_resp.raise_for_status()
pkg_count = pkg_resp.json().get("count", 0)
except Exception:
pkg_count = 0
try:
opt_resp = requests.post(url, json={"query": {"term": {"type": "option"}}}, auth=NIXOS_AUTH, timeout=10)
opt_resp.raise_for_status()
opt_count = opt_resp.json().get("count", 0)
except Exception:
opt_count = 0
if pkg_count == 0 and opt_count == 0:
return error("Failed to retrieve statistics")
return f"""NixOS Statistics for {channel} channel:
• Packages: {pkg_count:,}
• Options: {opt_count:,}"""
except Exception as e:
return error(str(e))
@mcp.tool()
async def home_manager_search(query: str, limit: int = 20) -> str:
"""Search Home Manager configuration options.
Searches through available Home Manager options by name and description.
Args:
query: The search query string to match against option names and descriptions
limit: Maximum number of results to return (default: 20, max: 100)
Returns:
Plain text list of matching options with name, type, and description
"""
if not 1 <= limit <= 100:
return error("Limit must be 1-100")
try:
options = parse_html_options(HOME_MANAGER_URL, query, "", limit)
if not options:
return f"No Home Manager options found matching '{query}'"
results = []
results.append(f"Found {len(options)} Home Manager options matching '{query}':\n")
for opt in options:
results.append(f"• {opt['name']}")
if opt["type"]:
results.append(f" Type: {opt['type']}")
if opt["description"]:
results.append(f" {opt['description']}")
results.append("")
return "\n".join(results).strip()
except Exception as e:
return error(str(e))
@mcp.tool()
async def home_manager_info(name: str) -> str:
"""Get detailed information about a specific Home Manager option.
Requires an exact option name match. If not found, suggests similar options.
Args:
name: The exact option name (e.g., 'programs.git.enable')
Returns:
Plain text with option details (name, type, description) or error with suggestions
"""
try:
# Search more broadly first
options = parse_html_options(HOME_MANAGER_URL, name, "", 100)
# Look for exact match
for opt in options:
if opt["name"] == name:
info = []
info.append(f"Option: {name}")
if opt["type"]:
info.append(f"Type: {opt['type']}")
if opt["description"]:
info.append(f"Description: {opt['description']}")
return "\n".join(info)
# If not found, check if there are similar options to suggest
if options:
suggestions = []
for opt in options[:5]: # Show up to 5 suggestions
if name in opt["name"] or opt["name"].startswith(name + "."):
suggestions.append(opt["name"])
if suggestions:
return error(
f"Option '{name}' not found. Did you mean one of these?\n"
+ "\n".join(f" • {s}" for s in suggestions)
+ f"\n\nTip: Use home_manager_options_by_prefix('{name}') to browse all options with this prefix.",
"NOT_FOUND",
)
return error(
f"Option '{name}' not found.\n"
+ f"Tip: Use home_manager_options_by_prefix('{name}') to browse available options.",
"NOT_FOUND",
)
except Exception as e:
return error(str(e))
@mcp.tool()
async def home_manager_stats() -> str:
"""Get statistics about Home Manager options.
Retrieves overall statistics including total options, categories, and top categories.
Returns:
Plain text summary with total options, category count, and top 5 categories
"""
try:
# Parse all options to get statistics
options = parse_html_options(HOME_MANAGER_URL, limit=5000)
if not options:
return error("Failed to fetch Home Manager statistics")
# Count categories
categories: dict[str, int] = {}
for opt in options:
cat = opt["name"].split(".")[0]
categories[cat] = categories.get(cat, 0) + 1
# Count types
types: dict[str, int] = {}
for opt in options:
opt_type = opt.get("type", "unknown")
if opt_type:
# Simplify complex types
if "null or" in opt_type:
opt_type = "nullable"
elif "list of" in opt_type:
opt_type = "list"
elif "attribute set" in opt_type:
opt_type = "attribute set"
types[opt_type] = types.get(opt_type, 0) + 1
# Build statistics
return f"""Home Manager Statistics:
• Total options: {len(options):,}
• Categories: {len(categories)}
• Top categories:
- programs: {categories.get("programs", 0):,} options
- services: {categories.get("services", 0):,} options
- home: {categories.get("home", 0):,} options
- wayland: {categories.get("wayland", 0):,} options
- xsession: {categories.get("xsession", 0):,} options"""
except Exception as e:
return error(str(e))
@mcp.tool()
async def home_manager_list_options() -> str:
"""List all Home Manager option categories.
Enumerates all top-level categories with their option counts.
Returns:
Plain text list of categories sorted alphabetically with option counts
"""
try:
# Get more options to see all categories (default 100 is too few)
options = parse_html_options(HOME_MANAGER_URL, limit=5000)
categories: dict[str, int] = {}
for opt in options:
name = opt["name"]
# Process option names
if name and not name.startswith("."):
if "." in name:
cat = name.split(".")[0]
else:
cat = name # Option without dot is its own category
# Valid categories should:
# - Be more than 1 character
# - Be a valid identifier (allows underscores)
# - Not be common value words
# - Match typical nix option category patterns
if (
len(cat) > 1 and cat.isidentifier() and (cat.islower() or cat.startswith("_"))
): # This ensures valid identifier
# Additional filtering for known valid categories
valid_categories = {
"accounts",
"dconf",
"editorconfig",
"fonts",
"gtk",
"home",
"i18n",
"launchd",
"lib",
"manual",
"news",
"nix",
"nixgl",
"nixpkgs",
"pam",
"programs",
"qt",
"services",
"specialisation",
"systemd",
"targets",
"wayland",
"xdg",
"xresources",
"xsession",
}
# Only include if it's in the known valid list or looks like a typical category
if cat in valid_categories or (len(cat) >= 3 and not any(char.isdigit() for char in cat)):
categories[cat] = categories.get(cat, 0) + 1
results = []
results.append(f"Home Manager option categories ({len(categories)} total):\n")
# Sort by count descending, then alphabetically
sorted_cats = sorted(categories.items(), key=lambda x: (-x[1], x[0]))
for cat, count in sorted_cats:
results.append(f"• {cat} ({count} options)")
return "\n".join(results)
except Exception as e:
return error(str(e))
@mcp.tool()
async def home_manager_options_by_prefix(option_prefix: str) -> str:
"""Get Home Manager options matching a specific prefix.
Useful for browsing options under a category or finding exact option names.
Args:
option_prefix: The prefix to match (e.g., 'programs.git' or 'services')
Returns:
Plain text list of options with the given prefix, including descriptions
"""
try:
options = parse_html_options(HOME_MANAGER_URL, "", option_prefix)
if not options:
return f"No Home Manager options found with prefix '{option_prefix}'"
results = []
results.append(f"Home Manager options with prefix '{option_prefix}' ({len(options)} found):\n")
for opt in sorted(options, key=lambda x: x["name"]):
results.append(f"• {opt['name']}")
if opt["description"]:
results.append(f" {opt['description']}")
results.append("")
return "\n".join(results).strip()
except Exception as e:
return error(str(e))
@mcp.tool()
async def darwin_search(query: str, limit: int = 20) -> str:
"""Search nix-darwin (macOS) configuration options.
Searches through available nix-darwin options by name and description.
Args:
query: The search query string to match against option names and descriptions
limit: Maximum number of results to return (default: 20, max: 100)
Returns:
Plain text list of matching options with name, type, and description
"""
if not 1 <= limit <= 100:
return error("Limit must be 1-100")
try:
options = parse_html_options(DARWIN_URL, query, "", limit)
if not options:
return f"No nix-darwin options found matching '{query}'"
results = []
results.append(f"Found {len(options)} nix-darwin options matching '{query}':\n")
for opt in options:
results.append(f"• {opt['name']}")
if opt["type"]:
results.append(f" Type: {opt['type']}")
if opt["description"]:
results.append(f" {opt['description']}")
results.append("")
return "\n".join(results).strip()
except Exception as e:
return error(str(e))
@mcp.tool()
async def darwin_info(name: str) -> str:
"""Get detailed information about a specific nix-darwin option.
Requires an exact option name match. If not found, suggests similar options.
Args:
name: The exact option name (e.g., 'system.defaults.dock.autohide')
Returns:
Plain text with option details (name, type, description) or error with suggestions
"""
try:
# Search more broadly first
options = parse_html_options(DARWIN_URL, name, "", 100)
# Look for exact match
for opt in options:
if opt["name"] == name:
info = []
info.append(f"Option: {name}")
if opt["type"]:
info.append(f"Type: {opt['type']}")
if opt["description"]:
info.append(f"Description: {opt['description']}")
return "\n".join(info)
# If not found, check if there are similar options to suggest
if options:
suggestions = []
for opt in options[:5]: # Show up to 5 suggestions
if name in opt["name"] or opt["name"].startswith(name + "."):
suggestions.append(opt["name"])
if suggestions:
return error(
f"Option '{name}' not found. Did you mean one of these?\n"
+ "\n".join(f" • {s}" for s in suggestions)
+ f"\n\nTip: Use darwin_options_by_prefix('{name}') to browse all options with this prefix.",
"NOT_FOUND",
)
return error(
f"Option '{name}' not found.\n"
+ f"Tip: Use darwin_options_by_prefix('{name}') to browse available options.",
"NOT_FOUND",
)
except Exception as e:
return error(str(e))
@mcp.tool()
async def darwin_stats() -> str:
"""Get statistics about nix-darwin options.
Retrieves overall statistics including total options, categories, and top categories.
Returns:
Plain text summary with total options, category count, and top 5 categories
"""
try:
# Parse all options to get statistics
options = parse_html_options(DARWIN_URL, limit=3000)
if not options:
return error("Failed to fetch nix-darwin statistics")
# Count categories
categories: dict[str, int] = {}
for opt in options:
cat = opt["name"].split(".")[0]
categories[cat] = categories.get(cat, 0) + 1
# Count types
types: dict[str, int] = {}
for opt in options:
opt_type = opt.get("type", "unknown")
if opt_type:
# Simplify complex types
if "null or" in opt_type:
opt_type = "nullable"
elif "list of" in opt_type:
opt_type = "list"
elif "attribute set" in opt_type:
opt_type = "attribute set"
types[opt_type] = types.get(opt_type, 0) + 1
# Build statistics
return f"""nix-darwin Statistics:
• Total options: {len(options):,}
• Categories: {len(categories)}
• Top categories:
- services: {categories.get("services", 0):,} options
- system: {categories.get("system", 0):,} options
- launchd: {categories.get("launchd", 0):,} options
- programs: {categories.get("programs", 0):,} options
- homebrew: {categories.get("homebrew", 0):,} options"""
except Exception as e:
return error(str(e))
@mcp.tool()
async def darwin_list_options() -> str:
"""List all nix-darwin option categories.
Enumerates all top-level categories with their option counts.
Returns:
Plain text list of categories sorted alphabetically with option counts
"""
try:
# Get more options to see all categories (default 100 is too few)
options = parse_html_options(DARWIN_URL, limit=2000)
categories: dict[str, int] = {}
for opt in options:
name = opt["name"]
# Process option names
if name and not name.startswith("."):
if "." in name:
cat = name.split(".")[0]
else:
cat = name # Option without dot is its own category
# Valid categories should:
# - Be more than 1 character
# - Be a valid identifier (allows underscores)
# - Not be common value words
# - Match typical nix option category patterns
if (
len(cat) > 1 and cat.isidentifier() and (cat.islower() or cat.startswith("_"))
): # This ensures valid identifier
# Additional filtering for known valid Darwin categories
valid_categories = {
"documentation",
"environment",
"fonts",
"homebrew",
"ids",
"launchd",
"networking",
"nix",
"nixpkgs",
"power",
"programs",
"security",
"services",
"system",
"targets",
"time",
"users",
}
# Only include if it's in the known valid list or looks like a typical category
if cat in valid_categories or (len(cat) >= 3 and not any(char.isdigit() for char in cat)):
categories[cat] = categories.get(cat, 0) + 1
results = []
results.append(f"nix-darwin option categories ({len(categories)} total):\n")
# Sort by count descending, then alphabetically
sorted_cats = sorted(categories.items(), key=lambda x: (-x[1], x[0]))
for cat, count in sorted_cats:
results.append(f"• {cat} ({count} options)")
return "\n".join(results)
except Exception as e:
return error(str(e))
@mcp.tool()
async def darwin_options_by_prefix(option_prefix: str) -> str:
"""Get nix-darwin options matching a specific prefix.
Useful for browsing options under a category or finding exact option names.
Args:
option_prefix: The prefix to match (e.g., 'system.defaults' or 'services')
Returns:
Plain text list of options with the given prefix, including descriptions
"""
try:
options = parse_html_options(DARWIN_URL, "", option_prefix)
if not options:
return f"No nix-darwin options found with prefix '{option_prefix}'"
results = []
results.append(f"nix-darwin options with prefix '{option_prefix}' ({len(options)} found):\n")
for opt in sorted(options, key=lambda x: x["name"]):
results.append(f"• {opt['name']}")
if opt["description"]:
results.append(f" {opt['description']}")
results.append("")
return "\n".join(results).strip()
except Exception as e:
return error(str(e))
@mcp.tool()
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))
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))
def _version_key(version_str: str) -> tuple[int, int, int]:
"""Convert version string to tuple for proper sorting."""
try:
parts = version_str.split(".")
# Handle versions like "3.9.9" or "3.10.0-rc1"
numeric_parts = []
for part in parts[:3]: # Major.Minor.Patch
# Extract numeric part
numeric = ""
for char in part:
if char.isdigit():
numeric += char
else:
break
if numeric:
numeric_parts.append(int(numeric))
else:
numeric_parts.append(0)
# Pad with zeros if needed
while len(numeric_parts) < 3:
numeric_parts.append(0)
return (numeric_parts[0], numeric_parts[1], numeric_parts[2])
except Exception:
return (0, 0, 0)
def _format_nixhub_found_version(package_name: str, version: str, found_version: dict[str, Any]) -> str:
"""Format a found version for display."""
results = []
results.append(f"✓ Found {package_name} version {version}\n")
last_updated = found_version.get("last_updated", "")
if last_updated:
try:
from datetime import datetime
dt = datetime.fromisoformat(last_updated.replace("Z", "+00:00"))
formatted_date = dt.strftime("%Y-%m-%d %H:%M UTC")
results.append(f"Last updated: {formatted_date}")
except Exception:
results.append(f"Last updated: {last_updated}")
platforms_summary = found_version.get("platforms_summary", "")
if platforms_summary:
results.append(f"Platforms: {platforms_summary}")
# Show commit hashes
platforms = found_version.get("platforms", [])
if platforms:
results.append("\nNixpkgs commits:")
seen_commits = set()
for platform in platforms:
attr_path = platform.get("attribute_path", "")
commit_hash = platform.get("commit_hash", "")
if commit_hash and commit_hash not in seen_commits:
seen_commits.add(commit_hash)
if re.match(r"^[a-fA-F0-9]{40}$", commit_hash):
results.append(f"• {commit_hash}")
if attr_path:
results.append(f" Attribute: {attr_path}")
results.append("\nTo use this version:")
results.append("1. Pin nixpkgs to one of the commit hashes above")
results.append("2. Install using the attribute path")
return "\n".join(results)
def _format_nixhub_release(release: dict[str, Any], package_name: str | None = None) -> list[str]:
"""Format a single NixHub release for display."""
results = []
version = release.get("version", "unknown")
last_updated = release.get("last_updated", "")
platforms_summary = release.get("platforms_summary", "")
platforms = release.get("platforms", [])
results.append(f"• Version {version}")
if last_updated:
# Format date nicely
try:
from datetime import datetime
dt = datetime.fromisoformat(last_updated.replace("Z", "+00:00"))
formatted_date = dt.strftime("%Y-%m-%d %H:%M UTC")
results.append(f" Last updated: {formatted_date}")
except Exception:
results.append(f" Last updated: {last_updated}")
if platforms_summary:
results.append(f" Platforms: {platforms_summary}")
# Show commit hashes and attribute paths for each platform (avoid duplicates)
if platforms:
seen_commits = set()
for platform in platforms:
commit_hash = platform.get("commit_hash", "")
attr_path = platform.get("attribute_path", "")
if commit_hash and commit_hash not in seen_commits:
seen_commits.add(commit_hash)
# Validate commit hash format (40 hex chars)
if re.match(r"^[a-fA-F0-9]{40}$", commit_hash):
results.append(f" Nixpkgs commit: {commit_hash}")
else:
results.append(f" Nixpkgs commit: {commit_hash} (warning: invalid format)")
# Show attribute path if different from package name
if attr_path and package_name and attr_path != package_name:
results.append(f" Attribute: {attr_path}")
return results
@mcp.tool()
async def nixos_flakes_search(query: str, limit: int = 20, channel: str = "unstable") -> str:
"""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.tool()
async def nixhub_package_versions(package_name: str, limit: int = 10) -> str:
"""Get version history and nixpkgs commit hashes for a specific package from NixHub.io.
Use this tool when users need specific package versions or commit hashes for reproducible builds.
Args:
package_name: Name of the package to query (e.g., "firefox", "python")
limit: Maximum number of versions to return (default: 10, max: 50)
Returns:
Plain text with package info and version history including commit hashes
"""
# Validate inputs
if not package_name or not package_name.strip():
return error("Package name is required")
# Sanitize package name - only allow alphanumeric, hyphens, underscores, dots
if not re.match(r"^[a-zA-Z0-9\-_.]+$", package_name):
return error("Invalid package name. Only letters, numbers, hyphens, underscores, and dots are allowed")
if not 1 <= limit <= 50:
return error("Limit must be between 1 and 50")
try:
# Construct NixHub API URL with the _data parameter
url = f"https://www.nixhub.io/packages/{package_name}?_data=routes%2F_nixhub.packages.%24pkg._index"
# Make request with timeout and proper headers
headers = {"Accept": "application/json", "User-Agent": "mcp-nixos/1.0.0"} # Identify ourselves
resp = requests.get(url, headers=headers, timeout=15)
# Handle different HTTP status codes
if resp.status_code == 404:
return error(f"Package '{package_name}' not found in NixHub", "NOT_FOUND")
if resp.status_code >= 500:
# NixHub returns 500 for non-existent packages with unusual names
# Check if the package name looks suspicious
if len(package_name) > 30 or package_name.count("-") > 5:
return error(f"Package '{package_name}' not found in NixHub", "NOT_FOUND")
return error("NixHub service temporarily unavailable", "SERVICE_ERROR")
resp.raise_for_status()
# Parse JSON response
data = resp.json()
# Validate response structure
if not isinstance(data, dict):
return error("Invalid response format from NixHub")
# Extract package info
# Use the requested package name, not what API returns (e.g., user asks for python3, API returns python)
name = package_name
summary = data.get("summary", "")
releases = data.get("releases", [])
if not releases:
return f"Package: {name}\nNo version history available in NixHub"
# Build results
results = []
results.append(f"Package: {name}")
if summary:
results.append(f"Description: {summary}")
results.append(f"Total versions: {len(releases)}")
results.append("")
# Limit results
shown_releases = releases[:limit]
results.append(f"Version history (showing {len(shown_releases)} of {len(releases)}):\n")
for release in shown_releases:
results.extend(_format_nixhub_release(release, name))
results.append("")
# Add usage hint
if shown_releases and any(r.get("platforms", [{}])[0].get("commit_hash") for r in shown_releases):
results.append("To use a specific version in your Nix configuration:")
results.append("1. Pin nixpkgs to the commit hash")
results.append("2. Use the attribute path to install the package")
return "\n".join(results).strip()
except requests.Timeout:
return error("Request to NixHub timed out", "TIMEOUT")
except requests.RequestException as e:
return error(f"Network error accessing NixHub: {str(e)}", "NETWORK_ERROR")
except ValueError as e:
return error(f"Failed to parse NixHub response: {str(e)}", "PARSE_ERROR")
except Exception as e:
return error(f"Unexpected error: {str(e)}")
@mcp.tool()
async def nixhub_find_version(package_name: str, version: str) -> str:
"""Find a specific version of a package in NixHub with smart search.
Automatically searches with increasing limits to find the requested version.
Args:
package_name: Name of the package to query (e.g., "ruby", "python")
version: Specific version to find (e.g., "2.6.7", "3.5.9")
Returns:
Plain text with version info and commit hash if found, or helpful message if not
"""
# Validate inputs
if not package_name or not package_name.strip():
return error("Package name is required")
if not version or not version.strip():
return error("Version is required")
# Sanitize inputs
if not re.match(r"^[a-zA-Z0-9\-_.]+$", package_name):
return error("Invalid package name. Only letters, numbers, hyphens, underscores, and dots are allowed")
# Try with incremental limits
limits_to_try = [10, 25, 50]
found_version = None
all_versions: list[dict[str, Any]] = []
for limit in limits_to_try:
try:
# Make request - handle special cases for package names
nixhub_name = package_name
# Common package name mappings
if package_name == "python":
nixhub_name = "python3"
elif package_name == "python2":
nixhub_name = "python"
url = f"https://www.nixhub.io/packages/{nixhub_name}?_data=routes%2F_nixhub.packages.%24pkg._index"
headers = {"Accept": "application/json", "User-Agent": "mcp-nixos/1.0.0"}
resp = requests.get(url, headers=headers, timeout=15)
if resp.status_code == 404:
return error(f"Package '{package_name}' not found in NixHub", "NOT_FOUND")
if resp.status_code >= 500:
return error("NixHub service temporarily unavailable", "SERVICE_ERROR")
resp.raise_for_status()
data = resp.json()
if not isinstance(data, dict):
return error("Invalid response format from NixHub")
releases = data.get("releases", [])
# Collect all versions seen
for release in releases[:limit]:
release_version = release.get("version", "")
if release_version and release_version not in [v["version"] for v in all_versions]:
all_versions.append({"version": release_version, "release": release})
# Check if this is the version we're looking for
if release_version == version:
found_version = release
break
if found_version:
break
except requests.Timeout:
return error("Request to NixHub timed out", "TIMEOUT")
except requests.RequestException as e:
return error(f"Network error accessing NixHub: {str(e)}", "NETWORK_ERROR")
except Exception as e:
return error(f"Unexpected error: {str(e)}")
# Format response
if found_version:
return _format_nixhub_found_version(package_name, version, found_version)
# Version not found - provide helpful information
results = []
results.append(f"✗ {package_name} version {version} not found in NixHub\n")
# Show available versions
if all_versions:
results.append(f"Available versions (checked {len(all_versions)} total):")
# Sort versions properly using version comparison
sorted_versions = sorted(all_versions, key=lambda x: _version_key(x["version"]), reverse=True)
# Find newest and oldest
newest = sorted_versions[0]["version"]
oldest = sorted_versions[-1]["version"]
results.append(f"• Newest: {newest}")
results.append(f"• Oldest: {oldest}")
# Show version range summary
major_versions = set()
for v in all_versions:
parts = v["version"].split(".")
if parts:
major_versions.add(parts[0])
if major_versions:
results.append(f"• Major versions available: {', '.join(sorted(major_versions, reverse=True))}")
# Check if requested version is older than available
try:
requested_parts = version.split(".")
oldest_parts = oldest.split(".")
if len(requested_parts) >= 2 and len(oldest_parts) >= 2:
req_major = int(requested_parts[0])
req_minor = int(requested_parts[1])
old_major = int(oldest_parts[0])
old_minor = int(oldest_parts[1])
if req_major < old_major or (req_major == old_major and req_minor < old_minor):
results.append(f"\nVersion {version} is older than the oldest available ({oldest})")
results.append("This version may have been removed after reaching end-of-life.")
except (ValueError, IndexError):
pass
results.append("\nAlternatives:")
results.append("• Use a newer version if possible")
results.append("• Build from source with a custom derivation")
results.append("• Use Docker/containers with the specific version")
results.append("• Find an old nixpkgs commit from before the version was removed")
return "\n".join(results)
def main() -> None:
"""Run the MCP server."""
mcp.run()
if __name__ == "__main__":
main()
```