From 78b00fedf8d33f5b122318ae1d2b0e2e492d87a5 Mon Sep 17 00:00:00 2001 From: Martin Tazl Date: Thu, 24 Sep 2026 21:00:48 +0200 Subject: [PATCH] Harden MCP tool composition --- README.md | 12 +++--- mcp_server.py | 19 ++++----- roll_calculation.py | 51 ++++++++++++++++++++++ rollcalc_mcp_tools.py | 19 ++++++--- tests/test_mcp_adapter.py | 78 +++++++++++++++++++++++++++++++--- tests/test_roll_calculation.py | 43 +++++++++++++++++++ transport_calculation.py | 11 ++++- 7 files changed, 206 insertions(+), 27 deletions(-) diff --git a/README.md b/README.md index 957de8a..e448e83 100644 --- a/README.md +++ b/README.md @@ -138,14 +138,16 @@ Tool calling semantics: zeroes) and an optional `article_name_hint`. - `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; - `area_weight_g_m2` is in g/m². `include_roll_weight` requests the optional - kilogram result. + `area_weight_g_m2` is in g/m². It always requests the available kilogram + result and, on success, returns `transport_roll_inputs`, the browser-equivalent + transport bundle. - `analyze_transport_capacity` uses roll/core diameters in mm; roll width and transport dimensions in m; and roll/payload weights in kg. `product_length_m` is the material length represented by one roll, used only - for reported square metres. The supplied `roll_weight_kg` is passed through - unchanged. Callers may use a canonical `transport_preset` or provide custom - transport dimensions and `max_weight_kg`. + for reported square metres. For a chained calculation, pass every field from + `transport_roll_inputs` unchanged. The supplied `roll_weight_kg` is passed + through unchanged. Callers may use a canonical `transport_preset` or provide + custom transport dimensions and `max_weight_kg`. Generate a password hash or user entry with: diff --git a/mcp_server.py b/mcp_server.py index dee97e4..f124039 100644 --- a/mcp_server.py +++ b/mcp_server.py @@ -9,6 +9,7 @@ from rollcalc_mcp_tools import ( calculate_roll_diameter_result, get_article_result, ) +from transport_calculation import TransportPresetKey def create_server() -> Any: @@ -36,11 +37,10 @@ def create_server() -> Any: core_type: str | None = None, width_m: float | None = None, area_weight_g_m2: float | None = None, - include_roll_weight: bool = False, category: str | None = None, production_site: str | None = None, ) -> dict[str, Any]: - """Calculate roll diameter. 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. Set include_roll_weight only to request the existing optional roll_weight_kg result.""" + """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.""" return calculate_roll_diameter_result( article_number=article_number, article_name_hint=article_name_hint, @@ -53,17 +53,16 @@ def create_server() -> Any: core_type=core_type, category=category, production_site=production_site, - include_roll_weight=include_roll_weight, ) @server.tool() def analyze_transport_capacity( - roll_diameter_mm: float | None = None, - core_diameter_mm: float | None = None, - roll_width_m: float | None = None, - roll_weight_kg: float | None = None, - product_length_m: float | None = None, - transport_preset: str | None = None, + roll_diameter_mm: float, + core_diameter_mm: float, + roll_width_m: float, + roll_weight_kg: float, + product_length_m: float, + transport_preset: TransportPresetKey | None = None, length_m: float | None = None, width_m: float | None = None, height_m: float | None = None, @@ -71,7 +70,7 @@ def create_server() -> Any: margin_side_m: float | None = None, margin_ceiling_m: float | None = None, ) -> dict[str, Any]: - """Analyze transport capacity. roll_diameter_mm and core_diameter_mm are mm; roll_width_m and product_length_m are m; roll_weight_kg and max_weight_kg are kg. product_length_m is the material length represented by one roll and is used for reported m², not placement geometry. Use a shared transport_preset or supply all custom transport dimensions in m and max_weight_kg. roll_weight_kg is consumed exactly as supplied.""" + """Analyze transport capacity for calculated rolls. For a chained request, first call get_article and calculate_roll_diameter, then pass every field from its transport_roll_inputs bundle unchanged: roll_diameter_mm, core_diameter_mm, roll_width_m, roll_weight_kg, and product_length_m. Do not choose among minimum, average, or maximum diameter; the bundle contains the browser-equivalent nominal transport diameter. product_length_m is used for reported m², not placement geometry. transport_preset must be one of the listed canonical identifiers: lkw_sattelzug means LKW Sattelzug / semi-trailer and lkw_tandem means LKW Tandem. Alternatively, supply all custom transport dimensions in m and max_weight_kg. This stateless tool does not recalculate a roll; roll_weight_kg is consumed exactly as supplied.""" return analyze_transport_capacity_result( transport_preset=transport_preset, length_m=length_m, diff --git a/roll_calculation.py b/roll_calculation.py index 8a61c96..71e1194 100644 --- a/roll_calculation.py +++ b/roll_calculation.py @@ -3,6 +3,7 @@ from __future__ import annotations from dataclasses import asdict, dataclass +from decimal import Decimal, ROUND_HALF_UP from difflib import SequenceMatcher import json import math @@ -738,6 +739,56 @@ def calculate_roll( } +def transport_roll_inputs_from_roll_calculation( + result: Any, +) -> dict[str, float]: + """Derive browser-equivalent transport inputs from a roll calculation.""" + if not isinstance(result, dict) or result.get("status") != "success": + raise ValueError("roll calculation result must have status success") + + calculation = result.get("calculation") + effective_inputs = result.get("effective_inputs") + if not isinstance(calculation, dict) or not isinstance(effective_inputs, dict): + raise ValueError("successful roll calculation result is incomplete") + + def calculation_number(field: str) -> float: + return _transport_roll_input_number(calculation.get(field), field) + + def effective_number(field: str) -> float: + entry = effective_inputs.get(field) + if not isinstance(entry, dict): + raise ValueError(f"successful roll calculation result is missing {field}") + return _transport_roll_input_number(entry.get("value"), field) + + return { + "roll_diameter_mm": _browser_one_decimal( + calculation_number("average_diameter_mm") + ), + "core_diameter_mm": effective_number("core_diameter_mm"), + "roll_width_m": effective_number("width_m"), + "roll_weight_kg": _browser_one_decimal( + calculation_number("roll_weight_kg") + ), + "product_length_m": calculation_number("effective_roll_length_m"), + } + + +def _transport_roll_input_number(value: Any, field: str) -> float: + if isinstance(value, bool) or not isinstance(value, (int, float)): + raise ValueError(f"successful roll calculation result is missing {field}") + number = float(value) + if not math.isfinite(number) or number <= 0: + raise ValueError(f"successful roll calculation result has invalid {field}") + return number + + +def _browser_one_decimal(value: float) -> float: + """Match the browser's positive-number ``toFixed(1)`` transport values.""" + return float( + Decimal.from_float(value).quantize(Decimal("0.1"), rounding=ROUND_HALF_UP) + ) + + def modify_calculation( state: CalculationState | dict[str, Any], changes: dict[str, Any], diff --git a/rollcalc_mcp_tools.py b/rollcalc_mcp_tools.py index ae8ba2a..7c18b5f 100644 --- a/rollcalc_mcp_tools.py +++ b/rollcalc_mcp_tools.py @@ -4,7 +4,11 @@ from __future__ import annotations from typing import Any -from roll_calculation import calculate_roll, get_article +from roll_calculation import ( + calculate_roll, + get_article, + transport_roll_inputs_from_roll_calculation, +) from transport_calculation import analyze_transport @@ -29,10 +33,9 @@ def calculate_roll_diameter_result( core_type: str | None = None, category: str | None = None, production_site: str | None = None, - include_roll_weight: bool = False, ) -> dict[str, Any]: - """Map MCP arguments to the existing CalculationRequest input shape.""" - return calculate_roll({ + """Map MCP inputs to the existing calculation, always requesting weight.""" + result = calculate_roll({ "article_number": article_number, "article_name_hint": article_name_hint, "roll_length_m": roll_length_m, @@ -44,8 +47,14 @@ def calculate_roll_diameter_result( "core_type": core_type, "category": category, "production_site": production_site, - "include_roll_weight": include_roll_weight, + "include_roll_weight": True, }) + if result.get("status") == "success": + result = dict(result) + result["transport_roll_inputs"] = ( + transport_roll_inputs_from_roll_calculation(result) + ) + return result def analyze_transport_capacity_result( diff --git a/tests/test_mcp_adapter.py b/tests/test_mcp_adapter.py index 34b1ae6..0c24e32 100644 --- a/tests/test_mcp_adapter.py +++ b/tests/test_mcp_adapter.py @@ -10,7 +10,7 @@ from rollcalc_mcp_tools import ( calculate_roll_diameter_result, get_article_result, ) -from transport_calculation import analyze_transport +from transport_calculation import analyze_transport, transport_presets class _FakeFastMCP: @@ -51,15 +51,44 @@ class McpAdapterTests(unittest.TestCase): "core_diameter_mm": 150.0, "width_m": 5.8, "area_weight_g_m2": 1767.2, - "include_roll_weight": True, }, {"roll_length_m": -1.0, "core_diameter_mm": 150.0, "thickness_mm": 2.0}, ): with self.subTest(arguments=arguments): - self.assertEqual( - calculate_roll_diameter_result(**arguments), - calculate_roll(arguments), - ) + actual = calculate_roll_diameter_result(**arguments) + expected = calculate_roll({**arguments, "include_roll_weight": True}) + if expected["status"] == "success": + self.assertEqual( + actual["transport_roll_inputs"]["roll_diameter_mm"], + 845.9, + ) + del actual["transport_roll_inputs"] + self.assertEqual(actual, expected) + + def test_roll_calculation_adapter_always_requests_weight_from_the_domain(self): + result = calculate_roll_diameter_result( + article_number="114030", + roll_length_m=15.0, + core_diameter_mm=100.0, + ) + + self.assertEqual(result["status"], "success") + self.assertIsNotNone(result["calculation"]["roll_weight_kg"]) + + def test_roll_calculation_adapter_uses_shared_transport_input_helper(self): + expected_inputs = {"roll_diameter_mm": 123.4} + with patch( + "rollcalc_mcp_tools.transport_roll_inputs_from_roll_calculation", + return_value=expected_inputs, + ) as helper: + result = calculate_roll_diameter_result( + article_number="114030", + roll_length_m=15.0, + core_diameter_mm=100.0, + ) + + helper.assert_called_once() + self.assertEqual(result["transport_roll_inputs"], expected_inputs) def test_transport_tool_delegates_for_preset_custom_and_invalid_cases(self): for arguments in ( @@ -111,6 +140,43 @@ 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 + roll_schema = tools["calculate_roll_diameter"].parameters + transport_tool = tools["analyze_transport_capacity"] + transport_schema = transport_tool.parameters + + self.assertNotIn("include_roll_weight", roll_schema["properties"]) + preset_schema = transport_schema["properties"]["transport_preset"] + self.assertEqual( + preset_schema["anyOf"][0]["enum"], + [preset.key for preset in transport_presets()], + ) + self.assertEqual( + transport_schema["required"], + [ + "roll_diameter_mm", + "core_diameter_mm", + "roll_width_m", + "roll_weight_kg", + "product_length_m", + ], + ) + for field in transport_schema["required"]: + self.assertEqual( + transport_schema["properties"][field]["type"], "number" + ) + description = transport_tool.description + for text in ( + "first call get_article and calculate_roll_diameter", + "transport_roll_inputs bundle unchanged", + "Do not choose among minimum, average, or maximum diameter", + "product_length_m", + "stateless tool does not recalculate a roll", + ): + self.assertIn(text, description) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_roll_calculation.py b/tests/test_roll_calculation.py index b1da3d1..5a570e2 100644 --- a/tests/test_roll_calculation.py +++ b/tests/test_roll_calculation.py @@ -10,6 +10,7 @@ from roll_calculation import ( calculate_roll, get_article, modify_calculation, + transport_roll_inputs_from_roll_calculation, ) @@ -131,6 +132,48 @@ class RollCalculationTests(unittest.TestCase): self.assertEqual(calculation["effective_roll_length_m"], 50.0) self.assertAlmostEqual(calculation["roll_weight_kg"], 512.4909) + def test_transport_roll_inputs_match_browser_average_and_display_rounding(self): + result = calculate_roll(representative_request()) + + self.assertEqual( + transport_roll_inputs_from_roll_calculation(result), + { + "roll_diameter_mm": 845.9, + "core_diameter_mm": 150.0, + "roll_width_m": 5.8, + "roll_weight_kg": 512.5, + "product_length_m": 50.0, + }, + ) + + def test_transport_roll_inputs_use_browser_equivalent_half_rounding(self): + result = { + "status": "success", + "calculation": { + "average_diameter_mm": 1000.25, + "roll_weight_kg": 12.25, + "effective_roll_length_m": 10.0, + }, + "effective_inputs": { + "core_diameter_mm": {"value": 150.0}, + "width_m": {"value": 2.0}, + }, + } + + inputs = transport_roll_inputs_from_roll_calculation(result) + + self.assertEqual(inputs["roll_diameter_mm"], 1000.3) + self.assertEqual(inputs["roll_weight_kg"], 12.3) + + def test_transport_roll_inputs_reject_unsuccessful_or_incomplete_results(self): + for result in ( + {"status": "needs_clarification"}, + {"status": "success", "calculation": {}, "effective_inputs": {}}, + ): + with self.subTest(result=result): + with self.assertRaises(ValueError): + transport_roll_inputs_from_roll_calculation(result) + def test_optional_weight_is_omitted_without_requesting_it(self): result = calculate_roll( representative_request(width_m=None, include_roll_weight=False) diff --git a/transport_calculation.py b/transport_calculation.py index 57452d1..429d98e 100644 --- a/transport_calculation.py +++ b/transport_calculation.py @@ -4,7 +4,16 @@ from __future__ import annotations from dataclasses import dataclass import math -from typing import Any +from typing import Any, Literal + + +TransportPresetKey = Literal[ + "container_20ft", + "container_40ft", + "container_40ft_hc", + "lkw_sattelzug", + "lkw_tandem", +] @dataclass(frozen=True)