1 Commits
Author SHA1 Message Date
admin 6c36d78063 Add independent MCP PDF report delivery 2026-09-27 11:09:13 +02:00
9 changed files with 555 additions and 20 deletions
+12
View File
@@ -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. `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/<token>.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. `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. 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.
+44 -4
View File
@@ -118,14 +118,15 @@ Authentication is implemented with `Flask-HTTPAuth`. The current code checks `BE
## MCP PoC ## 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, a thin adapter over the same domain services used by Flask: article lookup,
direct roll calculation, product-length calculation, material-weight calculation, direct roll calculation, product-length calculation, material-weight calculation,
transport analysis, and configured production-machine feasibility. transport analysis, and configured production-machine feasibility.
The available PoC tools are `get_article`, `search_articles`, The available PoC tools are `get_article`, `search_articles`,
`calculate_material_weight`, `calculate_product_length`, `calculate_material_weight`, `calculate_product_length`,
`calculate_roll_diameter`, `analyze_transport_capacity`, `get_machine`, `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 Install the pinned dependencies, including `mcp==1.26.0`, in the existing
environment, then start it with: environment, then start it with:
@@ -135,8 +136,16 @@ environment, then start it with:
.venv/bin/python mcp_server.py .venv/bin/python mcp_server.py
``` ```
The PoC has no MCP resources, HTTP/SSE transport, remote authentication, or The server supports the existing `stdio` transport (the default) and
dedicated target-length calculation tool or extrapolation support. `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: Tool calling semantics:
@@ -173,6 +182,37 @@ Tool calling semantics:
use `calculate_material_weight` when the core is unknown and only material use `calculate_material_weight` when the core is unknown and only material
weight is needed. On success, it returns `transport_roll_inputs`, the weight is needed. On success, it returns `transport_roll_inputs`, the
browser-equivalent transport bundle. 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-host>/mcp`; the
browser follows the returned `/reports/<capability>.pdf` URL directly.
- `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
+3 -11
View File
@@ -32,10 +32,8 @@ from conversation_service import (
from ollama_nlu import OllamaNLUClient from ollama_nlu import OllamaNLUClient
from pdf_report import ( from pdf_report import (
ReportValidationError, ReportValidationError,
report_from_calculation_result,
render_roll_report,
safe_report_filename,
) )
from roll_report_service import create_roll_calculation_pdf
from roll_calculation import ( from roll_calculation import (
calculate_product_length, calculate_product_length,
calculate_roll, calculate_roll,
@@ -325,16 +323,10 @@ def create_machine_max_product_length_calculation():
def _create_pdf(calculation_request): def _create_pdf(calculation_request):
result = calculate_roll(calculation_request, build_info=BUILD_INFO) pdf, filename, _ = create_roll_calculation_pdf(
report = report_from_calculation_result(result) calculation_request,
generated_at = datetime.now().astimezone()
pdf = render_roll_report(
report,
generated_at=generated_at,
build_info=BUILD_INFO, 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 return pdf, filename
+10
View File
@@ -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. 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 ## Production-Machine Feasibility
`config/machines.yaml` is the runtime source for configured production-machine `config/machines.yaml` is the runtime source for configured production-machine
+170
View File
@@ -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),
}
+90 -4
View File
@@ -1,15 +1,19 @@
"""RollCalc's minimal stdio MCP server.""" """RollCalc MCP server with stdio and Streamable HTTP transports."""
from __future__ import annotations from __future__ import annotations
import os
from typing import Any from typing import Any
from urllib.parse import quote
from mcp_pdf_delivery import McpReportNotFoundError, McpReportStore
from rollcalc_mcp_tools import ( from rollcalc_mcp_tools import (
analyze_transport_capacity_result, analyze_transport_capacity_result,
calculate_machine_max_product_length_result, calculate_machine_max_product_length_result,
calculate_material_weight_result, calculate_material_weight_result,
calculate_product_length_result, calculate_product_length_result,
calculate_roll_diameter_result, calculate_roll_diameter_result,
generate_calculation_pdf_result,
get_article_result, get_article_result,
get_machine_result, get_machine_result,
check_production_feasibility_result, check_production_feasibility_result,
@@ -18,11 +22,49 @@ from rollcalc_mcp_tools import (
from transport_calculation import TransportPresetKey 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: 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 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() @server.tool()
def get_article( def get_article(
@@ -145,6 +187,45 @@ def create_server() -> Any:
production_site=production_site, 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() @server.tool()
def analyze_transport_capacity( def analyze_transport_capacity(
transport_preset: TransportPresetKey, transport_preset: TransportPresetKey,
@@ -167,5 +248,10 @@ def create_server() -> Any:
return server return server
def run_server() -> None:
"""Run the selected FastMCP transport without changing tool definitions."""
create_server().run(transport=selected_transport())
if __name__ == "__main__": if __name__ == "__main__":
create_server().run(transport="stdio") run_server()
+70
View File
@@ -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
+34
View File
@@ -17,6 +17,7 @@ from roll_calculation import (
get_article, get_article,
transport_roll_inputs_from_roll_calculation, transport_roll_inputs_from_roll_calculation,
) )
from mcp_pdf_delivery import generate_mcp_calculation_pdf
from transport_calculation import analyze_transport from transport_calculation import analyze_transport
@@ -226,6 +227,39 @@ def calculate_roll_diameter_result(
return 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( def analyze_transport_capacity_result(
*, *,
transport_preset: str, transport_preset: str,
+122 -1
View File
@@ -1,6 +1,10 @@
import copy import copy
import asyncio
import inspect import inspect
import os
from pathlib import Path
import sys import sys
import tempfile
import types import types
import unittest import unittest
from unittest.mock import patch from unittest.mock import patch
@@ -20,6 +24,7 @@ from rollcalc_mcp_tools import (
calculate_material_weight_result, calculate_material_weight_result,
calculate_product_length_result, calculate_product_length_result,
calculate_roll_diameter_result, calculate_roll_diameter_result,
generate_calculation_pdf_result,
get_article_result, get_article_result,
get_machine_result, get_machine_result,
project_machine_max_product_length_result, project_machine_max_product_length_result,
@@ -31,12 +36,15 @@ from machine_constraints import (
get_machine, get_machine,
) )
from transport_calculation import analyze_transport, transport_presets 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: class _FakeFastMCP:
def __init__(self, name): def __init__(self, name, **kwargs):
self.name = name self.name = name
self.tools = [] self.tools = []
self.routes = []
def tool(self): def tool(self):
def register(function): def register(function):
@@ -44,6 +52,12 @@ class _FakeFastMCP:
return function return function
return register 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): class McpAdapterTests(unittest.TestCase):
def test_get_article_delegates_to_the_domain_for_success_not_found_and_conflict(self): 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() helper.assert_called_once()
self.assertEqual(result["transport_roll_inputs"], expected_inputs) 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): def test_transport_tool_delegates_to_domain_with_lkw_sattelzug_preset(self):
arguments = { arguments = {
"transport_preset": "lkw_sattelzug", "transport_preset": "lkw_sattelzug",
@@ -384,9 +488,22 @@ class McpAdapterTests(unittest.TestCase):
"calculate_material_weight", "calculate_material_weight",
"calculate_product_length", "calculate_product_length",
"calculate_roll_diameter", "calculate_roll_diameter",
"generate_calculation_pdf",
"analyze_transport_capacity", "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): def test_server_schema_exposes_weight_and_transport_chaining_contract(self):
instance = mcp_server.create_server() instance = mcp_server.create_server()
@@ -398,6 +515,7 @@ class McpAdapterTests(unittest.TestCase):
material_weight_tool = tools["calculate_material_weight"] material_weight_tool = tools["calculate_material_weight"]
product_length_tool = tools["calculate_product_length"] product_length_tool = tools["calculate_product_length"]
roll_schema = tools["calculate_roll_diameter"].parameters roll_schema = tools["calculate_roll_diameter"].parameters
pdf_tool = tools["generate_calculation_pdf"]
transport_tool = tools["analyze_transport_capacity"] transport_tool = tools["analyze_transport_capacity"]
transport_schema = transport_tool.parameters transport_schema = transport_tool.parameters
@@ -408,6 +526,9 @@ class McpAdapterTests(unittest.TestCase):
) )
self.assertIn("average_diameter_mm", feasibility_tool.parameters["properties"]) self.assertIn("average_diameter_mm", feasibility_tool.parameters["properties"])
self.assertIn("maximum_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("tool, not the LLM, decides", feasibility_tool.description)
self.assertIn("never override failed constraints", feasibility_tool.description) self.assertIn("never override failed constraints", feasibility_tool.description)
self.assertIn("warnings", feasibility_tool.description) self.assertIn("warnings", feasibility_tool.description)