From bbaaaf31c548e8f4b7b662e60d28ec198dc89c41 Mon Sep 17 00:00:00 2001 From: Martin Tazl Date: Thu, 24 Sep 2026 13:58:55 +0200 Subject: [PATCH] Add stdio MCP server for RollCalc --- README.md | 33 +++++++++++ mcp_server.py | 94 ++++++++++++++++++++++++++++++ requirements.txt | 1 + rollcalc_mcp_tools.py | 80 ++++++++++++++++++++++++++ tests/test_mcp_adapter.py | 116 ++++++++++++++++++++++++++++++++++++++ 5 files changed, 324 insertions(+) create mode 100644 mcp_server.py create mode 100644 rollcalc_mcp_tools.py create mode 100644 tests/test_mcp_adapter.py diff --git a/README.md b/README.md index 74ec24e..957de8a 100644 --- a/README.md +++ b/README.md @@ -114,6 +114,39 @@ Implemented routes: Authentication is implemented with `Flask-HTTPAuth`. The current code checks `BETA_USERS` with `werkzeug.security.check_password_hash`; plaintext passwords are not stored in the application. +## MCP PoC + +`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, +direct roll calculation, and transport analysis. The available PoC tools are +`get_article`, `calculate_roll_diameter`, and `analyze_transport_capacity`. + +Install the pinned dependencies, including `mcp==1.26.0`, in the existing +environment, then start it with: + +```bash +.venv/bin/python -m pip install -r requirements.txt +.venv/bin/python mcp_server.py +``` + +The PoC has no MCP resources, HTTP/SSE transport, remote authentication, or +dedicated product-length or target-length calculation tools, or extrapolation. + +Tool calling semantics: + +- `get_article` accepts textual `article_number` values (including leading + 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. +- `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`. + Generate a password hash or user entry with: ```bash diff --git a/mcp_server.py b/mcp_server.py new file mode 100644 index 0000000..dee97e4 --- /dev/null +++ b/mcp_server.py @@ -0,0 +1,94 @@ +"""RollCalc's minimal stdio MCP server.""" + +from __future__ import annotations + +from typing import Any + +from rollcalc_mcp_tools import ( + analyze_transport_capacity_result, + calculate_roll_diameter_result, + get_article_result, +) + + +def create_server() -> Any: + """Create the stdio-only MCP server after the SDK is installed.""" + from mcp.server.fastmcp import FastMCP + + server = FastMCP("RollCalc") + + @server.tool() + def get_article( + article_number: str | None = None, + article_name_hint: str | None = None, + ) -> dict[str, 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 calculate_roll_diameter( + roll_length_m: float | None = None, + core_diameter_mm: float | None = None, + thickness_mm: float | None = None, + thickness_stddev_mm: float | None = None, + article_number: str | None = None, + article_name_hint: str | None = None, + 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.""" + return calculate_roll_diameter_result( + article_number=article_number, + article_name_hint=article_name_hint, + roll_length_m=roll_length_m, + width_m=width_m, + thickness_mm=thickness_mm, + thickness_stddev_mm=thickness_stddev_mm, + area_weight_g_m2=area_weight_g_m2, + core_diameter_mm=core_diameter_mm, + 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, + length_m: float | None = None, + width_m: float | None = None, + height_m: float | None = None, + max_weight_kg: float | None = None, + 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.""" + return analyze_transport_capacity_result( + transport_preset=transport_preset, + length_m=length_m, + width_m=width_m, + height_m=height_m, + max_weight_kg=max_weight_kg, + margin_side_m=margin_side_m, + margin_ceiling_m=margin_ceiling_m, + roll_diameter_mm=roll_diameter_mm, + core_diameter_mm=core_diameter_mm, + roll_width_m=roll_width_m, + roll_weight_kg=roll_weight_kg, + product_length_m=product_length_m, + ) + + return server + + +if __name__ == "__main__": + create_server().run(transport="stdio") diff --git a/requirements.txt b/requirements.txt index 03b41a2..d2b106c 100644 --- a/requirements.txt +++ b/requirements.txt @@ -5,3 +5,4 @@ click==8.1.7 itsdangerous==2.1.2 Jinja2==3.1.2 MarkupSafe==2.1.3 +mcp==1.26.0 diff --git a/rollcalc_mcp_tools.py b/rollcalc_mcp_tools.py new file mode 100644 index 0000000..ae8ba2a --- /dev/null +++ b/rollcalc_mcp_tools.py @@ -0,0 +1,80 @@ +"""Thin MCP-facing adapters over the shared RollCalc domain services.""" + +from __future__ import annotations + +from typing import Any + +from roll_calculation import calculate_roll, get_article +from transport_calculation import analyze_transport + + +def get_article_result( + article_number: str | None = None, + article_name_hint: str | None = None, +) -> dict[str, Any]: + """Delegate article resolution to the shared article repository.""" + return get_article(article_number, article_name_hint) + + +def calculate_roll_diameter_result( + *, + article_number: str | None = None, + article_name_hint: str | None = None, + roll_length_m: float | None = None, + width_m: float | None = None, + thickness_mm: float | None = None, + thickness_stddev_mm: float | None = None, + area_weight_g_m2: float | None = None, + core_diameter_mm: float | None = None, + 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({ + "article_number": article_number, + "article_name_hint": article_name_hint, + "roll_length_m": roll_length_m, + "width_m": width_m, + "thickness_mm": thickness_mm, + "thickness_stddev_mm": thickness_stddev_mm, + "area_weight_g_m2": area_weight_g_m2, + "core_diameter_mm": core_diameter_mm, + "core_type": core_type, + "category": category, + "production_site": production_site, + "include_roll_weight": include_roll_weight, + }) + + +def analyze_transport_capacity_result( + *, + transport_preset: str | None = None, + length_m: float | None = None, + width_m: float | None = None, + height_m: float | None = None, + max_weight_kg: float | None = None, + margin_side_m: float | None = None, + margin_ceiling_m: float | None = None, + 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, +) -> dict[str, Any]: + """Map MCP arguments to the shared transport-analysis input shape.""" + return analyze_transport({ + "transport_preset": transport_preset, + "length_m": length_m, + "width_m": width_m, + "height_m": height_m, + "max_weight_kg": max_weight_kg, + "margin_side_m": margin_side_m, + "margin_ceiling_m": margin_ceiling_m, + "roll_diameter_mm": roll_diameter_mm, + "core_diameter_mm": core_diameter_mm, + "roll_width_m": roll_width_m, + "roll_weight_kg": roll_weight_kg, + "product_length_m": product_length_m, + }) diff --git a/tests/test_mcp_adapter.py b/tests/test_mcp_adapter.py new file mode 100644 index 0000000..34b1ae6 --- /dev/null +++ b/tests/test_mcp_adapter.py @@ -0,0 +1,116 @@ +import sys +import types +import unittest +from unittest.mock import patch + +import mcp_server +from roll_calculation import calculate_roll, get_article +from rollcalc_mcp_tools import ( + analyze_transport_capacity_result, + calculate_roll_diameter_result, + get_article_result, +) +from transport_calculation import analyze_transport + + +class _FakeFastMCP: + def __init__(self, name): + self.name = name + self.tools = [] + + def tool(self): + def register(function): + self.tools.append(function) + return function + return register + + +class McpAdapterTests(unittest.TestCase): + def test_get_article_delegates_to_the_domain_for_success_not_found_and_conflict(self): + for arguments in ( + {"article_number": "146900"}, + {"article_number": "does-not-exist"}, + {"article_number": "146900", "article_name_hint": "Other product"}, + ): + with self.subTest(arguments=arguments): + self.assertEqual( + get_article_result(**arguments), + get_article(**arguments), + ) + + def test_calculate_roll_diameter_delegates_for_normal_weighted_and_invalid_cases(self): + for arguments in ( + { + "article_number": "146900", + "roll_length_m": 50.0, + "core_diameter_mm": 150.0, + }, + { + "article_number": "146900", + "roll_length_m": 50.0, + "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), + ) + + def test_transport_tool_delegates_for_preset_custom_and_invalid_cases(self): + for arguments in ( + { + "transport_preset": "container_20ft", + "roll_diameter_mm": 1000.0, + "core_diameter_mm": 150.0, + "roll_width_m": 2.0, + "roll_weight_kg": 1000.0, + "product_length_m": 50.0, + }, + { + "length_m": 10.0, + "width_m": 5.0, + "height_m": 5.0, + "max_weight_kg": 10000.0, + "roll_diameter_mm": 1000.0, + "core_diameter_mm": 150.0, + "roll_width_m": 1.0, + "roll_weight_kg": 1000.0, + }, + {"length_m": 0.0}, + ): + with self.subTest(arguments=arguments): + self.assertEqual( + analyze_transport_capacity_result(**arguments), + analyze_transport(arguments), + ) + + def test_server_registers_exactly_the_three_stdio_poc_tools(self): + mcp = types.ModuleType("mcp") + server = types.ModuleType("mcp.server") + fastmcp = types.ModuleType("mcp.server.fastmcp") + fastmcp.FastMCP = _FakeFastMCP + with patch.dict(sys.modules, { + "mcp": mcp, + "mcp.server": server, + "mcp.server.fastmcp": fastmcp, + }): + instance = mcp_server.create_server() + + self.assertEqual(instance.name, "RollCalc") + self.assertEqual( + [tool.__name__ for tool in instance.tools], + [ + "get_article", + "calculate_roll_diameter", + "analyze_transport_capacity", + ], + ) + + +if __name__ == "__main__": + unittest.main()