diff --git a/mcp_server.py b/mcp_server.py index 10d04f9..067718d 100644 --- a/mcp_server.py +++ b/mcp_server.py @@ -8,6 +8,7 @@ from rollcalc_mcp_tools import ( analyze_transport_capacity_result, calculate_roll_diameter_result, get_article_result, + search_articles_result, ) from transport_calculation import TransportPresetKey @@ -26,6 +27,11 @@ def create_server() -> Any: """Resolve RollCalc article master data. article_number is text (including leading zeroes); optional article_name_hint preserves existing conflict and ambiguity checks.""" return get_article_result(article_number, article_name_hint) + @server.tool() + def search_articles(query: str) -> dict[str, Any]: + """Use search_articles when the user identifies a RollCalc product by name, family, designation, or descriptive article text instead of an exact article number. Do not guess or invent an article number: results are deterministic candidates from the RollCalc article master data. For an exact article number, use get_article -> calculate_roll_diameter -> optionally analyze_transport_capacity. For a product name, use search_articles first. Only when total_matches is exactly 1 may you use that candidate's exact article_number and continue through get_article and the requested workflow. When multiple candidates are returned, do NOT select the first or highest-ranked candidate and do not calculate yet: present relevant candidates with their article numbers, distinguishing names, and production_site when supplied; ask the user which article number is intended, then stop until they select it. Ranking/order expresses relevance only and never authorizes automatic selection. If no candidate is returned, say that no matching RollCalc article was found, ask for a more specific designation or article number, and do not invent an article. production_site=Malaysia is descriptive metadata only: it does not imply Bentofix, Bento 2, or any production machine.""" + return search_articles_result(query) + @server.tool() def calculate_roll_diameter( roll_length_m: float | None = None, diff --git a/roll_calculation.py b/roll_calculation.py index 71e1194..a3291f7 100644 --- a/roll_calculation.py +++ b/roll_calculation.py @@ -251,6 +251,56 @@ class ArticleRepository: return {"status": "not_requested", "article": None} + def search(self, query: str, *, limit: int = 20) -> dict[str, Any]: + """Return deterministic article-discovery candidates without resolving one.""" + if not isinstance(query, str): + return { + "status": "invalid_query", + "query": None, + "total_matches": 0, + "candidates": [], + } + if isinstance(limit, bool) or not isinstance(limit, int) or limit <= 0: + raise ValueError("limit must be a positive integer") + + query = query.strip() + query_key = _canonical_name(query) + if not query_key: + return { + "status": "search_not_found", + "query": query, + "total_matches": 0, + "candidates": [], + } + + query_tokens = _token_counts(query_key) + matches: list[tuple[bool, int, dict[str, Any]]] = [] + for article in self.articles: + name_key = _canonical_name(str(article.get("name", ""))) + name_tokens = _token_counts(name_key) + if not _tokens_contained(query_tokens, name_tokens): + continue + unmatched_tokens = _unmatched_token_count(name_tokens, query_tokens) + matches.append((name_key == query_key, unmatched_tokens, article)) + + matches.sort( + key=lambda item: ( + not item[0], + item[1], + str(item[2].get("nr", "")), + str(item[2].get("name", "")), + ) + ) + return { + "status": "search_results" if matches else "search_not_found", + "query": query, + "total_matches": len(matches), + "candidates": [ + _article_search_candidate(article) + for _, _, article in matches[:limit] + ], + } + def _matching_names(self, article_name_hint: str) -> list[dict[str, Any]]: hint_keys = _article_name_keys(article_name_hint) if not hint_keys: @@ -351,6 +401,31 @@ def _canonical_name(value: str) -> str: return " ".join(compacted) +def _token_counts(value: str) -> dict[str, int]: + counts: dict[str, int] = {} + for token in value.split(): + counts[token] = counts.get(token, 0) + 1 + return counts + + +def _tokens_contained( + query_tokens: dict[str, int], candidate_tokens: dict[str, int] +) -> bool: + return all( + candidate_tokens.get(token, 0) >= count + for token, count in query_tokens.items() + ) + + +def _unmatched_token_count( + candidate_tokens: dict[str, int], query_tokens: dict[str, int] +) -> int: + return sum( + max(0, count - query_tokens.get(token, 0)) + for token, count in candidate_tokens.items() + ) + + def _article_name_keys(value: str) -> set[str]: keys = {_canonical_name(value)} without_dimensions = re.sub( @@ -391,6 +466,22 @@ def _article_candidate(article: dict[str, Any]) -> dict[str, str]: } +def _article_search_candidate(article: dict[str, Any]) -> dict[str, Any]: + name = str(article.get("name", "")) + number = str(article.get("nr", "")) + return { + "article_number": number, + "name": " ".join(name.split()), + "width_m": _article_width(name), + "production_site": article_production_site(number), + } + + +def article_production_site(article_number: str) -> str | None: + """Return production-site metadata established by article-number rules.""" + return "Malaysia" if article_number.startswith("8") else None + + def _article_width(name: str) -> float | None: patterns = ( r"(? dict[str, Any]: + """Delegate article discovery to the shared article repository.""" + return ArticleRepository.load().search(query) + + def calculate_roll_diameter_result( *, article_number: str | None = None, diff --git a/tests/test_mcp_adapter.py b/tests/test_mcp_adapter.py index ef27d55..342db15 100644 --- a/tests/test_mcp_adapter.py +++ b/tests/test_mcp_adapter.py @@ -5,11 +5,12 @@ import unittest from unittest.mock import patch import mcp_server -from roll_calculation import calculate_roll, get_article +from roll_calculation import ArticleRepository, calculate_roll, get_article from rollcalc_mcp_tools import ( analyze_transport_capacity_result, calculate_roll_diameter_result, get_article_result, + search_articles_result, ) from transport_calculation import analyze_transport, transport_presets @@ -39,6 +40,34 @@ class McpAdapterTests(unittest.TestCase): get_article(**arguments), ) + def test_search_articles_delegates_to_domain_and_preserves_candidates(self): + query = "Bfix NSP 4900, 5,00 x 40 m" + + result = search_articles_result(query) + + self.assertEqual(result, ArticleRepository.load().search(query)) + self.assertEqual(result["status"], "search_results") + self.assertEqual( + [candidate["article_number"] for candidate in result["candidates"]], + ["180205", "8180205", "182815", "206900"], + ) + self.assertEqual( + result["candidates"][1]["production_site"], "Malaysia" + ) + + def test_search_articles_preserves_ambiguous_and_not_found_domain_results(self): + ambiguous = search_articles_result("Bentofix NSP 4900") + missing = search_articles_result("not a RollCalc product") + + self.assertGreater(ambiguous["total_matches"], 1) + self.assertGreater(len(ambiguous["candidates"]), 1) + self.assertEqual(missing, { + "status": "search_not_found", + "query": "not a RollCalc product", + "total_matches": 0, + "candidates": [], + }) + def test_calculate_roll_diameter_delegates_for_normal_weighted_and_invalid_cases(self): for arguments in ( { @@ -111,7 +140,7 @@ class McpAdapterTests(unittest.TestCase): self.assertEqual(result["analysis"]["final_rolls"], 24) self.assertEqual(result["analysis"]["limiting"]["name"], "Weight") - def test_server_registers_exactly_the_three_stdio_poc_tools(self): + def test_server_registers_the_existing_tools_and_article_search(self): mcp = types.ModuleType("mcp") server = types.ModuleType("mcp.server") fastmcp = types.ModuleType("mcp.server.fastmcp") @@ -128,6 +157,7 @@ class McpAdapterTests(unittest.TestCase): [tool.__name__ for tool in instance.tools], [ "get_article", + "search_articles", "calculate_roll_diameter", "analyze_transport_capacity", ], @@ -136,10 +166,16 @@ class McpAdapterTests(unittest.TestCase): def test_server_schema_exposes_weight_and_transport_chaining_contract(self): instance = mcp_server.create_server() tools = instance._tool_manager._tools + search_tool = tools["search_articles"] roll_schema = tools["calculate_roll_diameter"].parameters transport_tool = tools["analyze_transport_capacity"] transport_schema = transport_tool.parameters + self.assertEqual(search_tool.parameters["required"], ["query"]) + self.assertEqual( + search_tool.parameters["properties"]["query"]["type"], "string" + ) + self.assertNotIn("include_roll_weight", roll_schema["properties"]) preset_schema = transport_schema["properties"]["transport_preset"] self.assertEqual( @@ -195,6 +231,26 @@ class McpAdapterTests(unittest.TestCase): ): self.assertIn(text, description) + search_description = search_tool.description + for text in ( + "name, family, designation, or descriptive article text", + "Do not guess or invent an article number", + "deterministic candidates from the RollCalc article master data", + "total_matches is exactly 1", + "use that candidate's exact article_number", + "do NOT select the first or highest-ranked candidate", + "ask the user which article number is intended", + "do not calculate yet", + "no matching RollCalc article was found", + "more specific designation or article number", + "do not invent an article", + "production_site=Malaysia is descriptive metadata only", + "does not imply Bentofix, Bento 2, or any production machine", + "Ranking/order expresses relevance only", + "get_article -> calculate_roll_diameter -> optionally analyze_transport_capacity", + ): + self.assertIn(text, search_description) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_roll_calculation.py b/tests/test_roll_calculation.py index 5a570e2..a41cf42 100644 --- a/tests/test_roll_calculation.py +++ b/tests/test_roll_calculation.py @@ -7,6 +7,7 @@ from roll_calculation import ( CalculationRequest, CalculationState, _canonical_name, + article_production_site, calculate_roll, get_article, modify_calculation, @@ -446,6 +447,128 @@ class RollCalculationTests(unittest.TestCase): ) self.assertEqual(result["status"], "success") + def test_search_exact_visible_name_keeps_duplicate_article_numbers(self): + result = ArticleRepository.load().search("Bfix NSP 4900, 5,00 x 40 m") + + self.assertEqual(result["status"], "search_results") + self.assertEqual(result["total_matches"], 4) + self.assertEqual(result["candidates"][:2], [ + { + "article_number": "180205", + "name": "Bfix NSP 4900, 5,00 x 40 m", + "width_m": 5.0, + "production_site": None, + }, + { + "article_number": "8180205", + "name": "Bfix NSP 4900, 5,00 x 40 m", + "width_m": 5.0, + "production_site": "Malaysia", + }, + ]) + + def test_article_number_production_site_rule_only_identifies_malaysia(self): + self.assertEqual(article_production_site("8180205"), "Malaysia") + self.assertIsNone(article_production_site("180205")) + + def test_search_candidates_expose_site_without_changing_order_or_resolve(self): + repository = ArticleRepository.load() + candidates = repository.search( + "Bfix NSP 4900, 5,00 x 40 m" + )["candidates"] + + self.assertEqual( + [candidate["article_number"] for candidate in candidates], + ["180205", "8180205", "182815", "206900"], + ) + self.assertEqual( + [candidate["production_site"] for candidate in candidates], + [None, "Malaysia", None, None], + ) + resolved = repository.resolve("8180205", None) + self.assertEqual(resolved["status"], "resolved") + self.assertNotIn("production_site", resolved["article"]) + + def test_search_bentofix_alias_matches_bfix_without_resolving_one_article(self): + repository = ArticleRepository.load() + + full_name = repository.search("Bentofix NSP 4900") + abbreviated = repository.search("Bfix NSP 4900") + + self.assertEqual(full_name["candidates"], abbreviated["candidates"]) + self.assertGreater(full_name["total_matches"], 2) + self.assertIn("180205", [ + candidate["article_number"] for candidate in full_name["candidates"] + ]) + + def test_search_partial_name_returns_related_products_without_resolving_one(self): + result = ArticleRepository.load().search("NSP 4900") + + self.assertEqual(result["status"], "search_results") + self.assertGreater(result["total_matches"], 4) + self.assertIn("180205", [ + candidate["article_number"] for candidate in result["candidates"] + ]) + self.assertIn("206900", [ + candidate["article_number"] for candidate in result["candidates"] + ]) + + def test_search_qualifiers_and_dimensions_narrow_real_data_results(self): + repository = ArticleRepository.load() + base = repository.search("Bfix NSP 4900") + dimensions = repository.search("Bfix NSP 4900, 5,00 x 40 m") + qualified = repository.search("Bfix NSP 4900 GRI GCL3") + + self.assertLess(dimensions["total_matches"], base["total_matches"]) + self.assertEqual( + [candidate["article_number"] for candidate in qualified["candidates"]], + ["206900"], + ) + + def test_search_existing_family_aliases_are_equivalent(self): + repository = ArticleRepository.load() + + self.assertEqual( + repository.search("Secutex R 1801")["candidates"], + repository.search("StEx R 1801")["candidates"], + ) + self.assertEqual( + repository.search("Secugrid 40/40 Q6")["candidates"], + repository.search("Sgrid 40/40 Q6")["candidates"], + ) + + def test_search_is_stable_and_handles_empty_and_unknown_queries(self): + repository = ArticleRepository.load() + + self.assertEqual( + repository.search("NSP 4900"), repository.search("NSP 4900") + ) + self.assertEqual(repository.search(" "), { + "status": "search_not_found", + "query": "", + "total_matches": 0, + "candidates": [], + }) + self.assertEqual( + repository.search("not a RollCalc product")["status"], + "search_not_found", + ) + + def test_search_reports_total_when_the_stable_result_list_is_capped(self): + result = ArticleRepository.load().search("Secugrid") + + self.assertEqual(result["status"], "search_results") + self.assertGreater(result["total_matches"], len(result["candidates"])) + self.assertEqual(len(result["candidates"]), 20) + + def test_search_does_not_change_existing_resolve_ambiguity(self): + repository = ArticleRepository.load() + + self.assertEqual( + repository.resolve(None, "Bentofix NSP 4900")["status"], + "article_ambiguous", + ) + def test_name_only_matching_uses_deterministic_bfix_alias(self): result = self.calculate_fixture({ "article_name_hint": "Bentofix Alias Product",