From 6c36d780631f9add4fa96fcddfceae66a03dc1ab Mon Sep 17 00:00:00 2001 From: Martin Date: Sun, 27 Sep 2026 11:09:13 +0200 Subject: [PATCH] Add independent MCP PDF report delivery --- PROJECT_KNOWLEDGE.md | 12 +++ README.md | 48 ++++++++++- app.py | 14 +--- docs/architecture.md | 10 +++ mcp_pdf_delivery.py | 170 ++++++++++++++++++++++++++++++++++++++ mcp_server.py | 94 ++++++++++++++++++++- roll_report_service.py | 70 ++++++++++++++++ rollcalc_mcp_tools.py | 34 ++++++++ tests/test_mcp_adapter.py | 123 ++++++++++++++++++++++++++- 9 files changed, 555 insertions(+), 20 deletions(-) create mode 100644 mcp_pdf_delivery.py create mode 100644 roll_report_service.py diff --git a/PROJECT_KNOWLEDGE.md b/PROJECT_KNOWLEDGE.md index 6ef0d61..73bab9f 100644 --- a/PROJECT_KNOWLEDGE.md +++ b/PROJECT_KNOWLEDGE.md @@ -180,6 +180,18 @@ The browser and PDF endpoints both submit a `CalculationRequest`. The PDF endpoi `POST /api/reports/roll-calculation.pdf` is protected by the existing HTTP Basic Auth. Incomplete, conflicting, or invalid calculation requests return HTTP 400 rather than a document. Generated filenames include date and time without colon characters. +The MCP server exposes `generate_calculation_pdf` for an explicit user request +only. It accepts original direct-roll inputs, recalculates with `calculate_roll()`, +and passes the authoritative result through the same report adapter and renderer +as Flask. It cannot accept calculated values or a file path. With FastMCP +Streamable HTTP, the MCP host stores the PDF locally under a 256-bit opaque +capability token and serves it itself from `/reports/.pdf`; Flask is not +in that delivery path. `ROLLCALC_MCP_ARTIFACT_PUBLIC_BASE_URL` configures the +browser-reachable MCP-host origin used in the returned `download_url`. Reports +expire after 24 hours during store access and are lost with MCP-host temporary +storage. Native MCP resources are intentionally omitted because installed MCPO +0.0.20 does not preserve resource links or embedded resources. + `pdf_report.py` uses no third-party package. It creates one A4 page with a compact header, result cards, two-column input area, warnings/notes, and footer. Repeated diagonal `INTERNAL USE ONLY` text is drawn into the actual page content stream with reduced opacity; it is a marking/deterrence mechanism and is not tamper-proof or DRM. Direct-flow warnings are returned by the domain core and rendered by both consumers. `checkForklift()` remains only for the existing non-headless length/target UI modes. Build metadata and generation time are supplied server-side. diff --git a/README.md b/README.md index 1acab12..6b0512e 100644 --- a/README.md +++ b/README.md @@ -118,14 +118,15 @@ Authentication is implemented with `Flask-HTTPAuth`. The current code checks `BE ## MCP PoC -`mcp_server.py` provides a local stdio-only Model Context Protocol server. It is +`mcp_server.py` provides a Model Context Protocol server. It is a thin adapter over the same domain services used by Flask: article lookup, direct roll calculation, product-length calculation, material-weight calculation, transport analysis, and configured production-machine feasibility. The available PoC tools are `get_article`, `search_articles`, `calculate_material_weight`, `calculate_product_length`, `calculate_roll_diameter`, `analyze_transport_capacity`, `get_machine`, -`check_production_feasibility`, and `calculate_machine_max_product_length`. +`check_production_feasibility`, `calculate_machine_max_product_length`, and +`generate_calculation_pdf`. Install the pinned dependencies, including `mcp==1.26.0`, in the existing environment, then start it with: @@ -135,8 +136,16 @@ environment, then start it with: .venv/bin/python mcp_server.py ``` -The PoC has no MCP resources, HTTP/SSE transport, remote authentication, or -dedicated target-length calculation tool or extrapolation support. +The server supports the existing `stdio` transport (the default) and +`streamable-http`. MCP PDF delivery requires `streamable-http`; it has no +dependency on the Flask web application. Native MCP resources are intentionally +not used because the installed MCPO 0.0.20 does not preserve resource links or +embedded resources. The PoC has no remote authentication or dedicated +target-length calculation tool or extrapolation support. + +In `stdio` mode, `generate_calculation_pdf` returns a structured +`artifact_delivery_unavailable` response rather than creating an unreachable +report. Existing calculation tools work in either transport. Tool calling semantics: @@ -173,6 +182,37 @@ Tool calling semantics: use `calculate_material_weight` when the core is unknown and only material weight is needed. On success, it returns `transport_roll_inputs`, the browser-equivalent transport bundle. +- `generate_calculation_pdf` is strictly opt-in: call it only after the user + explicitly asks for a PDF or calculation report. Supply the original inputs + from `calculate_roll_diameter`, never calculated diameters, weights, warnings, + or notes. It recalculates through the same RollCalc domain service and uses + the existing one-page renderer. Ordinary calculation calls neither generate + nor suggest a PDF. + +For example, calculate a roll normally with `calculate_roll_diameter`. If the +user later asks “Mach mir daraus ein PDF”, call `generate_calculation_pdf` with +the original calculation inputs. Its successful response contains a safe +`filename`, expiry metadata, and a browser-ready `download_url`; MCPO/OpenWebUI +can present that URL as a download link. PDFs exist only in the MCP host's local +temporary store, not in Flask or shared storage. Each URL has a 256-bit opaque +capability token and expires after 24 hours during report-store access; whoever +possesses the URL can download the report until expiry. The user’s browser must +be able to reach the MCP-host public origin over HTTPS. + +For independent PDF delivery, configure and run the MCP host as follows: + +```bash +export ROLLCALC_MCP_TRANSPORT=streamable-http +export ROLLCALC_MCP_HOST=0.0.0.0 +export ROLLCALC_MCP_PORT=8000 +export ROLLCALC_MCP_ARTIFACT_PUBLIC_BASE_URL=https://rollcalc-mcp.example.internal +.venv/bin/python mcp_server.py +``` + +`ROLLCALC_MCP_ARTIFACT_PUBLIC_BASE_URL` is the externally browser-reachable +origin for this MCP service, not the Flask website URL. MCPO connects to the +MCP host with `--server-type streamable-http -- https:///mcp`; the +browser follows the returned `/reports/.pdf` URL directly. - `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 diff --git a/app.py b/app.py index 3c4f2a7..d8ca109 100644 --- a/app.py +++ b/app.py @@ -32,10 +32,8 @@ from conversation_service import ( from ollama_nlu import OllamaNLUClient from pdf_report import ( ReportValidationError, - report_from_calculation_result, - render_roll_report, - safe_report_filename, ) +from roll_report_service import create_roll_calculation_pdf from roll_calculation import ( calculate_product_length, calculate_roll, @@ -325,16 +323,10 @@ def create_machine_max_product_length_calculation(): def _create_pdf(calculation_request): - result = calculate_roll(calculation_request, build_info=BUILD_INFO) - report = report_from_calculation_result(result) - generated_at = datetime.now().astimezone() - pdf = render_roll_report( - report, - generated_at=generated_at, + pdf, filename, _ = create_roll_calculation_pdf( + calculation_request, build_info=BUILD_INFO, ) - article_number = report["article"]["number"] if report["article"] else None - filename = safe_report_filename(article_number, generated_at) return pdf, filename diff --git a/docs/architecture.md b/docs/architecture.md index 907f2e6..8309a2f 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -31,6 +31,16 @@ roll_calculation.calculate_roll() The PDF endpoint accepts the same calculation request, recalculates it through the core, then adapts the authoritative result to the established one-page PDF layout. It does not accept calculated values from the browser. +`generate_calculation_pdf` is the equivalent opt-in MCP adapter. It accepts only +original roll-calculation inputs and uses the shared report service after +deterministic recalculation. Under FastMCP Streamable HTTP, the MCP host owns a +short-lived local store and its `/reports/<256-bit-capability>.pdf` download +route. Flask has no role in this path, which avoids cross-host filesystem +coupling and binary PDF content in tool responses. The browser-reachable origin +is configured independently with `ROLLCALC_MCP_ARTIFACT_PUBLIC_BASE_URL`. +Installed MCPO 0.0.20 does not retain native MCP resources/resource links, so +the tool returns an ordinary browser URL rather than an MCP resource. + ## Production-Machine Feasibility `config/machines.yaml` is the runtime source for configured production-machine diff --git a/mcp_pdf_delivery.py b/mcp_pdf_delivery.py new file mode 100644 index 0000000..3a04b85 --- /dev/null +++ b/mcp_pdf_delivery.py @@ -0,0 +1,170 @@ +"""MCP-host-local storage and browser delivery for generated PDF reports.""" + +from __future__ import annotations + +from dataclasses import dataclass +from datetime import datetime, timezone +import json +import os +from pathlib import Path +import re +import secrets +import tempfile +import time +from typing import Any +from urllib.parse import urljoin, urlsplit + +from roll_calculation import calculate_roll +from roll_report_service import ( + create_roll_calculation_pdf_from_result, + load_report_build_info, +) + + +ARTIFACT_DIRECTORY_ENV = "ROLLCALC_MCP_ARTIFACT_DIRECTORY" +ARTIFACT_PUBLIC_BASE_URL_ENV = "ROLLCALC_MCP_ARTIFACT_PUBLIC_BASE_URL" +REPORT_TTL_SECONDS = 24 * 60 * 60 +TOKEN_PATTERN = re.compile(r"[A-Za-z0-9_-]{43}") +SAFE_FILENAME_PATTERN = re.compile(r"[A-Za-z0-9._-]+\.pdf") + + +class McpReportNotFoundError(FileNotFoundError): + """Raised for missing, malformed, or expired MCP-host report tokens.""" + + +class McpArtifactConfigurationError(ValueError): + """Raised when the MCP host lacks a browser-reachable artifact origin.""" + + +@dataclass(frozen=True) +class StoredMcpReport: + token: str + filename: str + expires_at: float + + @property + def expires_at_iso(self) -> str: + return datetime.fromtimestamp(self.expires_at, timezone.utc).isoformat() + + +class McpReportStore: + """A local, opaque-token report store owned exclusively by the MCP host.""" + + def __init__(self, directory: Path | None = None) -> None: + configured = os.getenv(ARTIFACT_DIRECTORY_ENV, "").strip() + self.directory = directory or ( + Path(configured).resolve() + if configured + else Path(tempfile.gettempdir(), "rollcalc-mcp-reports") + ) + + def _paths(self, token: str) -> tuple[Path, Path]: + if not TOKEN_PATTERN.fullmatch(token): + raise McpReportNotFoundError(token) + return ( + self.directory / f"{token}.pdf", + self.directory / f"{token}.json", + ) + + def _metadata(self, metadata_path: Path) -> dict[str, Any]: + try: + data = json.loads(metadata_path.read_text(encoding="utf-8")) + except (FileNotFoundError, json.JSONDecodeError) as error: + raise McpReportNotFoundError(metadata_path.name) from error + if not isinstance(data, dict): + raise McpReportNotFoundError(metadata_path.name) + return data + + def cleanup_expired(self, now: float | None = None) -> None: + if not self.directory.is_dir(): + return + now = time.time() if now is None else now + for metadata_path in self.directory.glob("*.json"): + if not TOKEN_PATTERN.fullmatch(metadata_path.stem): + continue + try: + expires_at = self._metadata(metadata_path).get("expires_at") + if not isinstance(expires_at, (int, float)) or expires_at <= now: + pdf_path, _ = self._paths(metadata_path.stem) + pdf_path.unlink(missing_ok=True) + metadata_path.unlink(missing_ok=True) + except McpReportNotFoundError: + pdf_path, _ = self._paths(metadata_path.stem) + pdf_path.unlink(missing_ok=True) + metadata_path.unlink(missing_ok=True) + + def create(self, pdf: bytes, filename: str) -> StoredMcpReport: + if not SAFE_FILENAME_PATTERN.fullmatch(filename): + raise ValueError("report filename is unsafe") + self.directory.mkdir(mode=0o700, parents=True, exist_ok=True) + self.cleanup_expired() + token = secrets.token_urlsafe(32) + expires_at = time.time() + REPORT_TTL_SECONDS + pdf_path, metadata_path = self._paths(token) + pdf_path.write_bytes(pdf) + metadata_path.write_text( + json.dumps({"filename": filename, "expires_at": expires_at}), + encoding="utf-8", + ) + return StoredMcpReport(token, filename, expires_at) + + def read(self, token: str) -> tuple[bytes, StoredMcpReport]: + pdf_path, metadata_path = self._paths(token) + self.cleanup_expired() + metadata = self._metadata(metadata_path) + filename = metadata.get("filename") + expires_at = metadata.get("expires_at") + if ( + not isinstance(filename, str) + or not SAFE_FILENAME_PATTERN.fullmatch(filename) + or not isinstance(expires_at, (int, float)) + or expires_at <= time.time() + ): + raise McpReportNotFoundError(token) + try: + return pdf_path.read_bytes(), StoredMcpReport(token, filename, expires_at) + except FileNotFoundError as error: + raise McpReportNotFoundError(token) from error + + +def mcp_artifact_public_base_url() -> str: + value = os.getenv(ARTIFACT_PUBLIC_BASE_URL_ENV, "").strip() + parsed = urlsplit(value) + if parsed.scheme not in {"http", "https"} or not parsed.netloc: + raise McpArtifactConfigurationError( + f"{ARTIFACT_PUBLIC_BASE_URL_ENV} must be an absolute http(s) URL" + ) + return value.rstrip("/") + + +def mcp_report_download_url(token: str) -> str: + if not TOKEN_PATTERN.fullmatch(token): + raise McpReportNotFoundError(token) + return urljoin(f"{mcp_artifact_public_base_url()}/", f"reports/{token}.pdf") + + +def generate_mcp_calculation_pdf(request: dict[str, Any]) -> dict[str, Any]: + """Create an MCP-host artifact from authoritative direct-roll inputs.""" + build_info = load_report_build_info() + result = calculate_roll(request, build_info=build_info) + if result.get("status") != "success": + return result + try: + mcp_artifact_public_base_url() + except McpArtifactConfigurationError as error: + return { + "status": "artifact_delivery_unavailable", + "error": str(error), + } + pdf, filename = create_roll_calculation_pdf_from_result( + result, + build_info=build_info, + ) + stored = McpReportStore().create(pdf, filename) + return { + "status": "success", + "filename": stored.filename, + "expires_at": stored.expires_at_iso, + "expires_in_seconds": REPORT_TTL_SECONDS, + "download_url": mcp_report_download_url(stored.token), + } diff --git a/mcp_server.py b/mcp_server.py index f9a689b..3513402 100644 --- a/mcp_server.py +++ b/mcp_server.py @@ -1,15 +1,19 @@ -"""RollCalc's minimal stdio MCP server.""" +"""RollCalc MCP server with stdio and Streamable HTTP transports.""" from __future__ import annotations +import os from typing import Any +from urllib.parse import quote +from mcp_pdf_delivery import McpReportNotFoundError, McpReportStore from rollcalc_mcp_tools import ( analyze_transport_capacity_result, calculate_machine_max_product_length_result, calculate_material_weight_result, calculate_product_length_result, calculate_roll_diameter_result, + generate_calculation_pdf_result, get_article_result, get_machine_result, check_production_feasibility_result, @@ -18,11 +22,49 @@ from rollcalc_mcp_tools import ( from transport_calculation import TransportPresetKey +async def download_report(request): + """Serve one MCP-host-local report by opaque temporary capability.""" + from starlette.responses import PlainTextResponse, Response + + token = request.path_params["token"] + try: + pdf, report = McpReportStore().read(token) + except McpReportNotFoundError: + return PlainTextResponse("Report not found or expired", status_code=404) + return Response( + pdf, + media_type="application/pdf", + headers={ + "Content-Disposition": ( + f'attachment; filename="{quote(report.filename)}"' + ), + }, + ) + + +def selected_transport() -> str: + """Read and validate the transport selected for this MCP process.""" + transport = os.getenv("ROLLCALC_MCP_TRANSPORT", "stdio").strip().lower() + if transport not in {"stdio", "streamable-http"}: + raise ValueError( + "ROLLCALC_MCP_TRANSPORT must be 'stdio' or 'streamable-http'" + ) + return transport + + def create_server() -> Any: - """Create the stdio-only MCP server after the SDK is installed.""" + """Create the transport-neutral RollCalc MCP application.""" from mcp.server.fastmcp import FastMCP - server = FastMCP("RollCalc") + server = FastMCP( + "RollCalc", + host=os.getenv("ROLLCALC_MCP_HOST", "127.0.0.1"), + port=int(os.getenv("ROLLCALC_MCP_PORT", "8000")), + ) + + @server.custom_route("/reports/{token}.pdf", methods=["GET"]) + async def mcp_report_download_route(request): + return await download_report(request) @server.tool() def get_article( @@ -145,6 +187,45 @@ def create_server() -> Any: production_site=production_site, ) + @server.tool() + def generate_calculation_pdf( + 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, + category: str | None = None, + production_site: str | None = None, + include_roll_weight: bool = True, + ) -> dict[str, Any]: + """Create a PDF calculation report only when the user explicitly requests a PDF or report. Supply the original inputs from a completed calculate_roll_diameter request; never supply, infer, or copy calculated diameters, weights, warnings, or notes. The tool recalculates authoritatively and returns a short-lived browser URL hosted by the MCP service. Do not call this tool after ordinary calculations and do not offer it proactively.""" + if selected_transport() != "streamable-http": + return { + "status": "artifact_delivery_unavailable", + "error": ( + "generate_calculation_pdf requires " + "ROLLCALC_MCP_TRANSPORT=streamable-http" + ), + } + return generate_calculation_pdf_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( transport_preset: TransportPresetKey, @@ -167,5 +248,10 @@ def create_server() -> Any: return server +def run_server() -> None: + """Run the selected FastMCP transport without changing tool definitions.""" + create_server().run(transport=selected_transport()) + + if __name__ == "__main__": - create_server().run(transport="stdio") + run_server() diff --git a/roll_report_service.py b/roll_report_service.py new file mode 100644 index 0000000..3a31bc8 --- /dev/null +++ b/roll_report_service.py @@ -0,0 +1,70 @@ +"""Shared authoritative PDF-report service for Flask and MCP adapters.""" + +from __future__ import annotations + +from datetime import datetime +import json +from pathlib import Path +from typing import Any + +from pdf_report import ( + report_from_calculation_result, + render_roll_report, + safe_report_filename, +) +from roll_calculation import calculate_roll + + +UNKNOWN_BUILD_INFO = { + "version": "unknown", + "branch": "unknown", + "commit": "unknown", + "timestamp": "unknown", +} + + +def create_roll_calculation_pdf( + calculation_request: Any, + *, + build_info: dict[str, str], +) -> tuple[bytes, str, dict[str, Any]]: + """Recalculate one request and render the established RollCalc PDF.""" + result = calculate_roll(calculation_request, build_info=build_info) + pdf, filename = create_roll_calculation_pdf_from_result( + result, + build_info=build_info, + ) + return pdf, filename, result + + +def create_roll_calculation_pdf_from_result( + result: dict[str, Any], + *, + build_info: dict[str, str], +) -> tuple[bytes, str]: + """Render an already-authoritative calculation result using the shared path.""" + report = report_from_calculation_result(result) + generated_at = datetime.now().astimezone() + pdf = render_roll_report( + report, + generated_at=generated_at, + build_info=build_info, + ) + article_number = report["article"]["number"] if report["article"] else None + return pdf, safe_report_filename(article_number, generated_at) + + +def load_report_build_info(path: str = "build_info.json") -> dict[str, str]: + """Load the same optional build metadata shown by Flask-generated reports.""" + try: + raw = json.loads(Path(path).read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + return UNKNOWN_BUILD_INFO.copy() + if not isinstance(raw, dict): + return UNKNOWN_BUILD_INFO.copy() + build_info = UNKNOWN_BUILD_INFO.copy() + for key in build_info: + value = raw.get(key) + if isinstance(value, str) and value.strip(): + build_info[key] = value.strip() + return build_info diff --git a/rollcalc_mcp_tools.py b/rollcalc_mcp_tools.py index c148f61..e3c1787 100644 --- a/rollcalc_mcp_tools.py +++ b/rollcalc_mcp_tools.py @@ -17,6 +17,7 @@ from roll_calculation import ( get_article, transport_roll_inputs_from_roll_calculation, ) +from mcp_pdf_delivery import generate_mcp_calculation_pdf from transport_calculation import analyze_transport @@ -226,6 +227,39 @@ def calculate_roll_diameter_result( return result +def generate_calculation_pdf_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 = True, +) -> dict[str, Any]: + """Create the established PDF from original inputs, never LLM result values.""" + request = { + "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, + } + return generate_mcp_calculation_pdf(request) + + def analyze_transport_capacity_result( *, transport_preset: str, diff --git a/tests/test_mcp_adapter.py b/tests/test_mcp_adapter.py index 81d7769..f644c82 100644 --- a/tests/test_mcp_adapter.py +++ b/tests/test_mcp_adapter.py @@ -1,6 +1,10 @@ import copy +import asyncio import inspect +import os +from pathlib import Path import sys +import tempfile import types import unittest from unittest.mock import patch @@ -20,6 +24,7 @@ from rollcalc_mcp_tools import ( calculate_material_weight_result, calculate_product_length_result, calculate_roll_diameter_result, + generate_calculation_pdf_result, get_article_result, get_machine_result, project_machine_max_product_length_result, @@ -31,12 +36,15 @@ from machine_constraints import ( get_machine, ) from transport_calculation import analyze_transport, transport_presets +from mcp_pdf_delivery import McpReportNotFoundError, McpReportStore, REPORT_TTL_SECONDS +from starlette.requests import Request class _FakeFastMCP: - def __init__(self, name): + def __init__(self, name, **kwargs): self.name = name self.tools = [] + self.routes = [] def tool(self): def register(function): @@ -44,6 +52,12 @@ class _FakeFastMCP: return function return register + def custom_route(self, path, methods): + def register(function): + self.routes.append((path, methods, function)) + return function + return register + class McpAdapterTests(unittest.TestCase): def test_get_article_delegates_to_the_domain_for_success_not_found_and_conflict(self): @@ -340,6 +354,96 @@ class McpAdapterTests(unittest.TestCase): helper.assert_called_once() self.assertEqual(result["transport_roll_inputs"], expected_inputs) + def test_generate_pdf_reuses_authoritative_report_path(self): + arguments = { + "article_number": "146900", + "roll_length_m": 50.0, + "core_diameter_mm": 150.0, + "core_type": "150mm PVC", + "category": "bentofix", + } + with tempfile.TemporaryDirectory() as directory: + with patch.dict(os.environ, { + "ROLLCALC_MCP_ARTIFACT_DIRECTORY": directory, + "ROLLCALC_MCP_ARTIFACT_PUBLIC_BASE_URL": ( + "https://mcp.example.test/rollcalc" + ), + }): + response = generate_calculation_pdf_result(**arguments) + token = response["download_url"].rsplit("/", 1)[1][:-4] + pdf, stored = McpReportStore().read(token) + + self.assertEqual(response["status"], "success") + self.assertEqual(stored.filename, response["filename"]) + self.assertTrue(pdf.startswith(b"%PDF-1.4")) + self.assertIn(b"146900", pdf) + self.assertIn(b"CALCULATION RESULTS", pdf) + self.assertEqual( + response["download_url"], + f"https://mcp.example.test/rollcalc/reports/{token}.pdf", + ) + self.assertEqual(len(token), 43) + self.assertRegex(token, r"^[A-Za-z0-9_-]+$") + self.assertEqual( + set(response), + {"status", "filename", "expires_at", "expires_in_seconds", "download_url"}, + ) + self.assertEqual(response["expires_in_seconds"], REPORT_TTL_SECONDS) + + def test_generate_pdf_preserves_structured_calculation_errors(self): + result = generate_calculation_pdf_result( + article_number="146900", + roll_length_m=None, + core_diameter_mm=150.0, + ) + + self.assertEqual(result["status"], "needs_clarification") + self.assertIn("roll_length_m", result["missing"]) + + def test_mcp_report_store_rejects_expired_unknown_and_path_tokens(self): + with tempfile.TemporaryDirectory() as directory: + store = McpReportStore(directory=Path(directory)) + with patch("mcp_pdf_delivery.time.time", return_value=1_000.0): + report = store.create(b"%PDF-1.4", "roll-calculation_146900.pdf") + with patch( + "mcp_pdf_delivery.time.time", + return_value=1_000.0 + REPORT_TTL_SECONDS + 1, + ): + with self.assertRaises(McpReportNotFoundError): + store.read(report.token) + for token in ("unknown", "../access_log", "a" * 43 + "/x"): + with self.subTest(token=token): + with self.assertRaises(McpReportNotFoundError): + store.read(token) + + def test_mcp_host_route_downloads_only_valid_unexpired_reports(self): + with tempfile.TemporaryDirectory() as directory: + with patch.dict(os.environ, { + "ROLLCALC_MCP_ARTIFACT_DIRECTORY": directory, + }): + stored = McpReportStore().create( + b"%PDF-1.4\nMCP report", "roll-calculation_146900.pdf" + ) + response = asyncio.run(mcp_server.download_report( + Request({"type": "http", "path_params": {"token": stored.token}}) + )) + unknown = asyncio.run(mcp_server.download_report( + Request({"type": "http", "path_params": {"token": "x" * 43}}) + )) + traversal = asyncio.run(mcp_server.download_report( + Request({"type": "http", "path_params": {"token": "../access_log"}}) + )) + + self.assertEqual(response.status_code, 200) + self.assertEqual(response.headers["content-type"], "application/pdf") + self.assertEqual( + response.headers["content-disposition"], + 'attachment; filename="roll-calculation_146900.pdf"', + ) + self.assertEqual(response.body, b"%PDF-1.4\nMCP report") + self.assertEqual(unknown.status_code, 404) + self.assertEqual(traversal.status_code, 404) + def test_transport_tool_delegates_to_domain_with_lkw_sattelzug_preset(self): arguments = { "transport_preset": "lkw_sattelzug", @@ -384,9 +488,22 @@ class McpAdapterTests(unittest.TestCase): "calculate_material_weight", "calculate_product_length", "calculate_roll_diameter", + "generate_calculation_pdf", "analyze_transport_capacity", ], ) + self.assertEqual( + [(path, methods) for path, methods, _ in instance.routes], + [("/reports/{token}.pdf", ["GET"])], + ) + pdf_tool = next( + tool for tool in instance.tools + if tool.__name__ == "generate_calculation_pdf" + ) + with patch.dict(os.environ, {"ROLLCALC_MCP_TRANSPORT": "stdio"}): + self.assertEqual( + pdf_tool()["status"], "artifact_delivery_unavailable" + ) def test_server_schema_exposes_weight_and_transport_chaining_contract(self): instance = mcp_server.create_server() @@ -398,6 +515,7 @@ class McpAdapterTests(unittest.TestCase): material_weight_tool = tools["calculate_material_weight"] product_length_tool = tools["calculate_product_length"] roll_schema = tools["calculate_roll_diameter"].parameters + pdf_tool = tools["generate_calculation_pdf"] transport_tool = tools["analyze_transport_capacity"] transport_schema = transport_tool.parameters @@ -408,6 +526,9 @@ class McpAdapterTests(unittest.TestCase): ) self.assertIn("average_diameter_mm", feasibility_tool.parameters["properties"]) self.assertIn("maximum_diameter_mm", feasibility_tool.parameters["properties"]) + self.assertIn("explicitly requests", pdf_tool.description) + self.assertIn("Do not call this tool after ordinary calculations", pdf_tool.description) + self.assertNotIn("minimum_diameter_mm", pdf_tool.parameters["properties"]) self.assertIn("tool, not the LLM, decides", feasibility_tool.description) self.assertIn("never override failed constraints", feasibility_tool.description) self.assertIn("warnings", feasibility_tool.description)