martin/RollCalcPython
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add deterministic article search to MCP

Browse Source
This commit is contained in:
admin
2026-09-25 15:13:02 +02:00
parent 6f224a1831
commit 1169df96a8
5 changed files with 284 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
mcp_server.py
+6
View File
@@ -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,
roll_calculation.py
+91
View File
@@ -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"(?<!\d)(\d{1,2}[,.]\d{1,3})\s*[x×]\s*<?\s*\d+(?:[,.]\d+)?\s*m\b",
rollcalc_mcp_tools.py
+6
View File
@@ -5,6 +5,7 @@ from __future__ import annotations
from typing import Any
from roll_calculation import (
ArticleRepository,
calculate_roll,
get_article,
transport_roll_inputs_from_roll_calculation,
@@ -20,6 +21,11 @@ def get_article_result(
return get_article(article_number, article_name_hint)
def search_articles_result(query: str) -> 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,
tests/test_mcp_adapter.py
+58 -2
View File
@@ -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()
tests/test_roll_calculation.py
+123
View File
@@ -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",