Harden MCP article and weight workflows

This commit is contained in:
2026-09-25 16:34:49 +02:00
parent 1169df96a8
commit 580e9caa5e
6 changed files with 181 additions and 12 deletions
+15 -5
View File
@@ -118,8 +118,10 @@ Authentication is implemented with `Flask-HTTPAuth`. The current code checks `BE
`mcp_server.py` provides a local stdio-only Model Context Protocol server. It is `mcp_server.py` provides a local stdio-only Model Context Protocol server. It is
a thin adapter over the same domain services used by Flask: article lookup, a thin adapter over the same domain services used by Flask: article lookup,
direct roll calculation, and transport analysis. The available PoC tools are direct roll calculation, material-weight calculation, and transport analysis.
`get_article`, `calculate_roll_diameter`, and `analyze_transport_capacity`. The available PoC tools are `get_article`, `search_articles`,
`calculate_material_weight`, `calculate_roll_diameter`, and
`analyze_transport_capacity`.
Install the pinned dependencies, including `mcp==1.26.0`, in the existing Install the pinned dependencies, including `mcp==1.26.0`, in the existing
environment, then start it with: environment, then start it with:
@@ -136,11 +138,19 @@ Tool calling semantics:
- `get_article` accepts textual `article_number` values (including leading - `get_article` accepts textual `article_number` values (including leading
zeroes) and an optional `article_name_hint`. zeroes) and an optional `article_name_hint`.
- `search_articles` deterministically returns master-data candidates for a
product name or designation. Its ordering is retrieval relevance, not a
product recommendation: when multiple candidates remain, callers must ask
the user to select an article number rather than choosing one.
- `calculate_material_weight` calculates material-only weight from roll length,
width, and area weight. It requires no core diameter, does not infer one, and
cannot return a roll diameter or a core-inclusive total weight.
- `calculate_roll_diameter` uses `roll_length_m` for the material length on one - `calculate_roll_diameter` uses `roll_length_m` for the material length on one
roll. Diameter and thickness inputs are in mm; `width_m` is in m; roll. Diameter and thickness inputs are in mm; `width_m` is in m;
`area_weight_g_m2` is in g/m². It always requests the available kilogram `area_weight_g_m2` is in g/m². It requires a known positive core diameter;
result and, on success, returns `transport_roll_inputs`, the browser-equivalent use `calculate_material_weight` when the core is unknown and only material
transport bundle. weight is needed. On success, it returns `transport_roll_inputs`, the
browser-equivalent transport bundle.
- `analyze_transport_capacity` uses roll/core diameters in mm; roll width and - `analyze_transport_capacity` uses roll/core diameters in mm; roll width and
transport dimensions in m; and roll/payload weights in kg. transport dimensions in m; and roll/payload weights in kg.
`product_length_m` is the material length represented by one roll, used only `product_length_m` is the material length represented by one roll, used only
+16 -2
View File
@@ -6,6 +6,7 @@ from typing import Any
from rollcalc_mcp_tools import ( from rollcalc_mcp_tools import (
analyze_transport_capacity_result, analyze_transport_capacity_result,
calculate_material_weight_result,
calculate_roll_diameter_result, calculate_roll_diameter_result,
get_article_result, get_article_result,
search_articles_result, search_articles_result,
@@ -29,9 +30,22 @@ def create_server() -> Any:
@server.tool() @server.tool()
def search_articles(query: str) -> dict[str, Any]: 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.""" """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 retrieval candidates from the RollCalc article master data, not product recommendations. 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 plausible candidates are returned, do NOT select the first or highest-ranked candidate, do not treat ranking as permission to choose, 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 retrieval 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) return search_articles_result(query)
@server.tool()
def calculate_material_weight(
roll_length_m: float,
width_m: float,
area_weight_g_m2: float,
) -> dict[str, Any]:
"""Calculate deterministic material-only roll weight from roll_length_m (m), width_m (m), and area_weight_g_m2 (g/m²). Use this when those inputs are known but core diameter is unknown or not needed. It does not require, accept, infer, or default a core diameter; never invent a standard core. The result excludes any core weight and cannot provide roll diameter or a core-inclusive total weight. For roll diameter, use calculate_roll_diameter only with a known positive core diameter."""
return calculate_material_weight_result(
roll_length_m=roll_length_m,
width_m=width_m,
area_weight_g_m2=area_weight_g_m2,
)
@server.tool() @server.tool()
def calculate_roll_diameter( def calculate_roll_diameter(
roll_length_m: float | None = None, roll_length_m: float | None = None,
@@ -46,7 +60,7 @@ def create_server() -> Any:
category: str | None = None, category: str | None = None,
production_site: str | None = None, production_site: str | None = None,
) -> dict[str, Any]: ) -> dict[str, Any]:
"""Calculate roll diameter and weight for one roll. roll_length_m is the material length on one roll in m; core_diameter_mm, thickness_mm, and thickness_stddev_mm are mm; width_m is m; and area_weight_g_m2 is g/m². Resolve an article or provide applicable manual material inputs. This MCP tool always returns roll_weight_kg when the required width_m and area_weight_g_m2 inputs are available. On success, transport_roll_inputs contains the browser-equivalent transport values; pass that bundle's fields unchanged to analyze_transport_capacity.""" """Calculate roll diameter and weight for one roll. roll_length_m is the material length on one roll in m; core_diameter_mm, thickness_mm, and thickness_stddev_mm are mm; width_m is m; and area_weight_g_m2 is g/m². Roll diameter requires a known positive core diameter: do not infer or default one. If only material weight is needed and the core is unknown, call calculate_material_weight instead; it returns material-only weight, not a diameter or core-inclusive total. On success, transport_roll_inputs contains the browser-equivalent transport values; pass that bundle's fields unchanged to analyze_transport_capacity."""
return calculate_roll_diameter_result( return calculate_roll_diameter_result(
article_number=article_number, article_number=article_number,
article_name_hint=article_name_hint, article_name_hint=article_name_hint,
+58 -1
View File
@@ -374,6 +374,63 @@ def get_article(
return repository.resolve(article_number, article_name_hint) return repository.resolve(article_number, article_name_hint)
def calculate_material_weight(
*,
roll_length_m: Any,
width_m: Any,
area_weight_g_m2: Any,
) -> dict[str, Any]:
"""Calculate material-only roll weight without a core or diameter assumption."""
values = {
"roll_length_m": roll_length_m,
"width_m": width_m,
"area_weight_g_m2": area_weight_g_m2,
}
invalid = []
normalized = {}
for field, value in values.items():
if isinstance(value, bool) or not isinstance(value, (int, float)):
invalid.append({"field": field, "message": "must be a number"})
continue
number = float(value)
if not math.isfinite(number) or number <= 0:
invalid.append(
{"field": field, "message": "must be greater than zero"}
)
continue
normalized[field] = number
if invalid:
return {"status": "invalid_parameter", "invalid": invalid}
material_weight = _material_weight_kg(
normalized["roll_length_m"],
normalized["width_m"],
normalized["area_weight_g_m2"],
)
if not math.isfinite(material_weight) or material_weight <= 0:
return {
"status": "invalid_parameter",
"invalid": [
{
"field": "request",
"message": "values are outside the supported calculation range",
}
],
}
return {
"status": "success",
"weight_scope": "material_only",
"inputs": normalized,
"material_weight_kg": material_weight,
}
def _material_weight_kg(
roll_length_m: float, width_m: float, area_weight_g_m2: float
) -> float:
return area_weight_g_m2 * roll_length_m * width_m / 1000
def _canonical_name(value: str) -> str: def _canonical_name(value: str) -> str:
value = value.casefold().replace("×", " x ") value = value.casefold().replace("×", " x ")
tokens = re.findall(r"\d+(?:[.,]\d+)?|[^\W\d_]+", value) tokens = re.findall(r"\d+(?:[.,]\d+)?|[^\W\d_]+", value)
@@ -746,7 +803,7 @@ def calculate_roll(
) )
roll_weight = None roll_weight = None
if request.include_roll_weight: if request.include_roll_weight:
roll_weight = area_weight * length * width / 1000 roll_weight = _material_weight_kg(length, width, area_weight)
except OverflowError: except OverflowError:
return _issue_result( return _issue_result(
"invalid_parameter", "invalid_parameter",
+15
View File
@@ -6,6 +6,7 @@ from typing import Any
from roll_calculation import ( from roll_calculation import (
ArticleRepository, ArticleRepository,
calculate_material_weight,
calculate_roll, calculate_roll,
get_article, get_article,
transport_roll_inputs_from_roll_calculation, transport_roll_inputs_from_roll_calculation,
@@ -26,6 +27,20 @@ def search_articles_result(query: str) -> dict[str, Any]:
return ArticleRepository.load().search(query) return ArticleRepository.load().search(query)
def calculate_material_weight_result(
*,
roll_length_m: float,
width_m: float,
area_weight_g_m2: float,
) -> dict[str, Any]:
"""Delegate material-only weight calculation to the shared domain service."""
return calculate_material_weight(
roll_length_m=roll_length_m,
width_m=width_m,
area_weight_g_m2=area_weight_g_m2,
)
def calculate_roll_diameter_result( def calculate_roll_diameter_result(
*, *,
article_number: str | None = None, article_number: str | None = None,
+47 -4
View File
@@ -5,9 +5,15 @@ import unittest
from unittest.mock import patch from unittest.mock import patch
import mcp_server import mcp_server
from roll_calculation import ArticleRepository, calculate_roll, get_article from roll_calculation import (
ArticleRepository,
calculate_material_weight,
calculate_roll,
get_article,
)
from rollcalc_mcp_tools import ( from rollcalc_mcp_tools import (
analyze_transport_capacity_result, analyze_transport_capacity_result,
calculate_material_weight_result,
calculate_roll_diameter_result, calculate_roll_diameter_result,
get_article_result, get_article_result,
search_articles_result, search_articles_result,
@@ -68,6 +74,19 @@ class McpAdapterTests(unittest.TestCase):
"candidates": [], "candidates": [],
}) })
def test_material_weight_adapter_delegates_without_a_core_diameter(self):
arguments = {
"roll_length_m": 60.0,
"width_m": 5.0,
"area_weight_g_m2": 4850.079676,
}
result = calculate_material_weight_result(**arguments)
self.assertEqual(result, calculate_material_weight(**arguments))
self.assertEqual(result["weight_scope"], "material_only")
self.assertAlmostEqual(result["material_weight_kg"], 1455.0239028)
def test_calculate_roll_diameter_delegates_for_normal_weighted_and_invalid_cases(self): def test_calculate_roll_diameter_delegates_for_normal_weighted_and_invalid_cases(self):
for arguments in ( for arguments in (
{ {
@@ -158,6 +177,7 @@ class McpAdapterTests(unittest.TestCase):
[ [
"get_article", "get_article",
"search_articles", "search_articles",
"calculate_material_weight",
"calculate_roll_diameter", "calculate_roll_diameter",
"analyze_transport_capacity", "analyze_transport_capacity",
], ],
@@ -167,6 +187,7 @@ class McpAdapterTests(unittest.TestCase):
instance = mcp_server.create_server() instance = mcp_server.create_server()
tools = instance._tool_manager._tools tools = instance._tool_manager._tools
search_tool = tools["search_articles"] search_tool = tools["search_articles"]
material_weight_tool = tools["calculate_material_weight"]
roll_schema = tools["calculate_roll_diameter"].parameters roll_schema = tools["calculate_roll_diameter"].parameters
transport_tool = tools["analyze_transport_capacity"] transport_tool = tools["analyze_transport_capacity"]
transport_schema = transport_tool.parameters transport_schema = transport_tool.parameters
@@ -175,6 +196,12 @@ class McpAdapterTests(unittest.TestCase):
self.assertEqual( self.assertEqual(
search_tool.parameters["properties"]["query"]["type"], "string" search_tool.parameters["properties"]["query"]["type"], "string"
) )
material_weight_schema = material_weight_tool.parameters
self.assertEqual(
material_weight_schema["required"],
["roll_length_m", "width_m", "area_weight_g_m2"],
)
self.assertNotIn("core_diameter_mm", material_weight_schema["properties"])
self.assertNotIn("include_roll_weight", roll_schema["properties"]) self.assertNotIn("include_roll_weight", roll_schema["properties"])
preset_schema = transport_schema["properties"]["transport_preset"] preset_schema = transport_schema["properties"]["transport_preset"]
@@ -235,10 +262,11 @@ class McpAdapterTests(unittest.TestCase):
for text in ( for text in (
"name, family, designation, or descriptive article text", "name, family, designation, or descriptive article text",
"Do not guess or invent an article number", "Do not guess or invent an article number",
"deterministic candidates from the RollCalc article master data", "deterministic retrieval candidates from the RollCalc article master data",
"not product recommendations",
"total_matches is exactly 1", "total_matches is exactly 1",
"use that candidate's exact article_number", "use that candidate's exact article_number",
"do NOT select the first or highest-ranked candidate", "do NOT select the first or highest-ranked candidate, do not treat ranking as permission to choose",
"ask the user which article number is intended", "ask the user which article number is intended",
"do not calculate yet", "do not calculate yet",
"no matching RollCalc article was found", "no matching RollCalc article was found",
@@ -246,11 +274,26 @@ class McpAdapterTests(unittest.TestCase):
"do not invent an article", "do not invent an article",
"production_site=Malaysia is descriptive metadata only", "production_site=Malaysia is descriptive metadata only",
"does not imply Bentofix, Bento 2, or any production machine", "does not imply Bentofix, Bento 2, or any production machine",
"Ranking/order expresses relevance only", "Ranking/order expresses retrieval relevance only",
"get_article -> calculate_roll_diameter -> optionally analyze_transport_capacity", "get_article -> calculate_roll_diameter -> optionally analyze_transport_capacity",
): ):
self.assertIn(text, search_description) self.assertIn(text, search_description)
material_weight_description = material_weight_tool.description
for text in (
"material-only roll weight",
"does not require, accept, infer, or default a core diameter",
"never invent a standard core",
"excludes any core weight",
"cannot provide roll diameter or a core-inclusive total weight",
):
self.assertIn(text, material_weight_description)
roll_description = tools["calculate_roll_diameter"].description
self.assertIn("requires a known positive core diameter", roll_description)
self.assertIn("do not infer or default one", roll_description)
self.assertIn("call calculate_material_weight instead", roll_description)
if __name__ == "__main__": if __name__ == "__main__":
unittest.main() unittest.main()
+30
View File
@@ -8,6 +8,7 @@ from roll_calculation import (
CalculationState, CalculationState,
_canonical_name, _canonical_name,
article_production_site, article_production_site,
calculate_material_weight,
calculate_roll, calculate_roll,
get_article, get_article,
modify_calculation, modify_calculation,
@@ -182,6 +183,35 @@ class RollCalculationTests(unittest.TestCase):
self.assertEqual(result["status"], "success") self.assertEqual(result["status"], "success")
self.assertIsNone(result["calculation"]["roll_weight_kg"]) self.assertIsNone(result["calculation"]["roll_weight_kg"])
def test_material_weight_does_not_require_or_assume_a_core_diameter(self):
result = calculate_material_weight(
roll_length_m=60,
width_m=5.0,
area_weight_g_m2=4850.079676,
)
self.assertEqual(result["status"], "success")
self.assertEqual(result["weight_scope"], "material_only")
self.assertAlmostEqual(result["material_weight_kg"], 1455.0239028)
self.assertNotIn("core_diameter_mm", result["inputs"])
def test_roll_diameter_still_rejects_a_zero_core_diameter(self):
result = calculate_roll(
{
"roll_length_m": 60,
"width_m": 5.0,
"area_weight_g_m2": 4850.079676,
"thickness_mm": 6.461044,
"core_diameter_mm": 0,
"include_roll_weight": True,
}
)
self.assertEqual(result["status"], "invalid_parameter")
self.assertIn("core_diameter_mm", [
item["field"] for item in result["invalid"]
])
def test_diameter_only_calculation_does_not_require_width(self): def test_diameter_only_calculation_does_not_require_width(self):
result = self.calculate_fixture( result = self.calculate_fixture(
{ {