Compare commits
3
Commits
90c28fb2f5
..
main
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
6c36d78063 | ||
|
|
0c18c86a47 | ||
|
|
eca151a465 |
@@ -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.
|
||||||
|
|||||||
@@ -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:
|
||||||
|
|
||||||
@@ -150,14 +159,18 @@ Tool calling semantics:
|
|||||||
width, and area weight. It requires no core diameter, does not infer one, and
|
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.
|
cannot return a roll diameter or a core-inclusive total weight.
|
||||||
- `calculate_machine_max_product_length` composes configured machine limits
|
- `calculate_machine_max_product_length` composes configured machine limits
|
||||||
with the shared calculation domain. Resolve an article first, then pass its
|
with the shared calculation domain. With an exact article number, MCP resolves
|
||||||
properties unchanged. Its nominal maximum uses the nominal diameter limit;
|
its properties internally; otherwise callers pass explicit properties. Its
|
||||||
|
nominal maximum uses the nominal diameter limit;
|
||||||
its conservative/no-warning maximum uses RollCalc's existing maximum-diameter
|
its conservative/no-warning maximum uses RollCalc's existing maximum-diameter
|
||||||
thickness variation. Both also respect configured material-only maximum roll
|
thickness variation. Both also respect configured material-only maximum roll
|
||||||
weight, and the domain returns the governing constraint(s), unrounded.
|
weight, and the domain returns the governing constraint(s), unrounded.
|
||||||
Core and width incompatibility cannot be corrected by shortening and is
|
Core and width incompatibility cannot be corrected by shortening and is
|
||||||
returned structurally. No manufacturing increment, core/packaging/gross
|
returned structurally. No manufacturing increment, core/packaging/gross
|
||||||
weight, or conditional production rule is applied in V1.1.
|
weight, or conditional production rule is applied in V1.1. Its successful
|
||||||
|
MCP response is a compact projection: machine identity, configured maximum
|
||||||
|
diameter/material-weight limits, production maxima, material weights, and
|
||||||
|
warnings. Flask and other domain consumers retain the full diagnostic result.
|
||||||
- `calculate_product_length` calculates required roll length from a target outer
|
- `calculate_product_length` calculates required roll length from a target outer
|
||||||
diameter, core diameter, and material thickness. Its optional thickness
|
diameter, core diameter, and material thickness. Its optional thickness
|
||||||
stddev returns minimum/average/maximum length ranges; optional width and area
|
stddev returns minimum/average/maximum length ranges; optional width and area
|
||||||
@@ -169,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
|
||||||
|
|||||||
@@ -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
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -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
|
||||||
|
|||||||
@@ -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),
|
||||||
|
}
|
||||||
+94
-6
@@ -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(
|
||||||
@@ -67,15 +109,17 @@ def create_server() -> Any:
|
|||||||
def calculate_machine_max_product_length(
|
def calculate_machine_max_product_length(
|
||||||
machine: str,
|
machine: str,
|
||||||
core_diameter_mm: float,
|
core_diameter_mm: float,
|
||||||
thickness_mm: float,
|
article_number: str | None = None,
|
||||||
|
thickness_mm: float | None = None,
|
||||||
thickness_stddev_mm: float | None = None,
|
thickness_stddev_mm: float | None = None,
|
||||||
product_width_m: float | None = None,
|
product_width_m: float | None = None,
|
||||||
area_weight_g_m2: float | None = None,
|
area_weight_g_m2: float | None = None,
|
||||||
) -> dict[str, Any]:
|
) -> dict[str, Any]:
|
||||||
"""Calculate deterministic maximum product length for a configured production machine. Use this for maximum roll/product length, how much product fits on a machine roll, or maximum length without a diameter warning. Resolve the article first with search_articles/get_article when needed, then pass its thickness, thickness standard deviation, width, and area weight exactly. Do not supply or invent machine limits, rearrange roll or weight formulas, calculate a weight-limited length, or compare candidate limits yourself: this tool returns nominal and conservative/no-warning production maxima and their governing constraints. Report incompatibility rather than substituting another core or width, and ask for returned missing inputs rather than inventing them."""
|
"""Calculate deterministic maximum product length for a configured production machine. With an exact article_number, call this directly with machine and caller-supplied core_diameter_mm; do not call get_article merely to copy properties. Otherwise supply explicit material properties. Never mix article_number with material properties. For a product name, use search_articles; if candidates are multiple, ask the user to select an article number. Do not supply or invent machine limits or substitute a core or width; this returns nominal and conservative/no-warning production maxima and governing constraints."""
|
||||||
return calculate_machine_max_product_length_result(
|
return calculate_machine_max_product_length_result(
|
||||||
machine=machine,
|
machine=machine,
|
||||||
core_diameter_mm=core_diameter_mm,
|
core_diameter_mm=core_diameter_mm,
|
||||||
|
article_number=article_number,
|
||||||
thickness_mm=thickness_mm,
|
thickness_mm=thickness_mm,
|
||||||
thickness_stddev_mm=thickness_stddev_mm,
|
thickness_stddev_mm=thickness_stddev_mm,
|
||||||
product_width_m=product_width_m,
|
product_width_m=product_width_m,
|
||||||
@@ -143,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,
|
||||||
@@ -165,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()
|
||||||
|
|||||||
@@ -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
|
||||||
+112
-3
@@ -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
|
||||||
|
|
||||||
|
|
||||||
@@ -64,13 +65,49 @@ def calculate_machine_max_product_length_result(
|
|||||||
*,
|
*,
|
||||||
machine: str,
|
machine: str,
|
||||||
core_diameter_mm: float,
|
core_diameter_mm: float,
|
||||||
thickness_mm: float,
|
article_number: str | None = None,
|
||||||
|
thickness_mm: float | None = None,
|
||||||
thickness_stddev_mm: float | None = None,
|
thickness_stddev_mm: float | None = None,
|
||||||
product_width_m: float | None = None,
|
product_width_m: float | None = None,
|
||||||
area_weight_g_m2: float | None = None,
|
area_weight_g_m2: float | None = None,
|
||||||
) -> dict[str, Any]:
|
) -> dict[str, Any]:
|
||||||
"""Delegate machine-aware maximum product length to the shared domain."""
|
"""Return the compact MCP view, resolving one exact article when supplied."""
|
||||||
return calculate_machine_max_product_length(
|
explicit_material = {
|
||||||
|
"thickness_mm": thickness_mm,
|
||||||
|
"thickness_stddev_mm": thickness_stddev_mm,
|
||||||
|
"product_width_m": product_width_m,
|
||||||
|
"area_weight_g_m2": area_weight_g_m2,
|
||||||
|
}
|
||||||
|
if article_number is not None:
|
||||||
|
mixed_fields = [
|
||||||
|
field for field, value in explicit_material.items() if value is not None
|
||||||
|
]
|
||||||
|
if mixed_fields:
|
||||||
|
return {
|
||||||
|
"status": "invalid_parameter",
|
||||||
|
"invalid": [{
|
||||||
|
"field": "article_number",
|
||||||
|
"message": (
|
||||||
|
"cannot be combined with explicit material properties: "
|
||||||
|
+ ", ".join(mixed_fields)
|
||||||
|
),
|
||||||
|
}],
|
||||||
|
}
|
||||||
|
article_result = get_article(article_number)
|
||||||
|
if article_result["status"] != "resolved":
|
||||||
|
return article_result
|
||||||
|
article = article_result["article"]
|
||||||
|
result = calculate_machine_max_product_length(
|
||||||
|
machine=machine,
|
||||||
|
core_diameter_mm=core_diameter_mm,
|
||||||
|
thickness_mm=article["thickness_mm"],
|
||||||
|
thickness_stddev_mm=article["thickness_stddev_mm"],
|
||||||
|
product_width_m=article["width_m"],
|
||||||
|
area_weight_g_m2=article["area_weight_g_m2"],
|
||||||
|
)
|
||||||
|
return project_machine_max_product_length_result(result, article=article)
|
||||||
|
|
||||||
|
result = calculate_machine_max_product_length(
|
||||||
machine=machine,
|
machine=machine,
|
||||||
core_diameter_mm=core_diameter_mm,
|
core_diameter_mm=core_diameter_mm,
|
||||||
thickness_mm=thickness_mm,
|
thickness_mm=thickness_mm,
|
||||||
@@ -78,6 +115,45 @@ def calculate_machine_max_product_length_result(
|
|||||||
product_width_m=product_width_m,
|
product_width_m=product_width_m,
|
||||||
area_weight_g_m2=area_weight_g_m2,
|
area_weight_g_m2=area_weight_g_m2,
|
||||||
)
|
)
|
||||||
|
return project_machine_max_product_length_result(result)
|
||||||
|
|
||||||
|
|
||||||
|
def project_machine_max_product_length_result(
|
||||||
|
result: dict[str, Any],
|
||||||
|
*,
|
||||||
|
article: dict[str, Any] | None = None,
|
||||||
|
) -> dict[str, Any]:
|
||||||
|
"""Project successful maximum-length results for MCP without diagnostics.
|
||||||
|
|
||||||
|
The shared domain result deliberately contains detailed verification reports.
|
||||||
|
They remain available to Flask and other consumers, but duplicate machine
|
||||||
|
metadata and checks are unnecessary in the MCP response.
|
||||||
|
"""
|
||||||
|
if result.get("status") != "success":
|
||||||
|
return result
|
||||||
|
|
||||||
|
machine = result["machine"]
|
||||||
|
constraints = machine["constraints"]
|
||||||
|
projection = {
|
||||||
|
"status": result["status"],
|
||||||
|
"machine": {"id": machine["id"], "name": machine["name"]},
|
||||||
|
"weight_scope": result["weight_scope"],
|
||||||
|
"machine_limits": {
|
||||||
|
"maximum_roll_diameter_mm": constraints["roll_diameter_mm"]["maximum"]
|
||||||
|
if constraints["roll_diameter_mm"] is not None else None,
|
||||||
|
"maximum_material_roll_weight_kg": constraints["roll_weight_kg"]["maximum"]
|
||||||
|
if constraints["roll_weight_kg"] is not None else None,
|
||||||
|
},
|
||||||
|
"production_maximums": result["production_maximums"],
|
||||||
|
"material_weights_kg": result["material_weights_kg"],
|
||||||
|
"warnings": result["warnings"],
|
||||||
|
}
|
||||||
|
if article is not None:
|
||||||
|
projection["article"] = {
|
||||||
|
"number": article["number"],
|
||||||
|
"name": article["name"],
|
||||||
|
}
|
||||||
|
return projection
|
||||||
|
|
||||||
|
|
||||||
def calculate_material_weight_result(
|
def calculate_material_weight_result(
|
||||||
@@ -151,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,
|
||||||
|
|||||||
@@ -5,6 +5,7 @@ import unittest
|
|||||||
from unittest.mock import patch
|
from unittest.mock import patch
|
||||||
|
|
||||||
import app as rollcalc_app
|
import app as rollcalc_app
|
||||||
|
from machine_constraints import calculate_machine_max_product_length
|
||||||
|
|
||||||
|
|
||||||
class CalculationApiTests(unittest.TestCase):
|
class CalculationApiTests(unittest.TestCase):
|
||||||
@@ -140,26 +141,49 @@ class CalculationApiTests(unittest.TestCase):
|
|||||||
self.assertEqual(result["warning_constraints"], ["roll_diameter_mm"])
|
self.assertEqual(result["warning_constraints"], ["roll_diameter_mm"])
|
||||||
|
|
||||||
def test_machine_maximum_length_endpoint_is_authenticated_and_structured(self):
|
def test_machine_maximum_length_endpoint_is_authenticated_and_structured(self):
|
||||||
response = self.client.post(
|
payload = {
|
||||||
"/api/calculations/machine-max-product-length",
|
|
||||||
json={
|
|
||||||
"machine": "K7",
|
"machine": "K7",
|
||||||
"core_diameter_mm": 133,
|
"core_diameter_mm": 133,
|
||||||
"thickness_mm": 4.311611,
|
"thickness_mm": 4.311611,
|
||||||
"thickness_stddev_mm": 0.249999,
|
"thickness_stddev_mm": 0.249999,
|
||||||
"product_width_m": 6,
|
"product_width_m": 6,
|
||||||
"area_weight_g_m2": 495.395833,
|
"area_weight_g_m2": 495.395833,
|
||||||
},
|
}
|
||||||
|
response = self.client.post(
|
||||||
|
"/api/calculations/machine-max-product-length",
|
||||||
|
json=payload,
|
||||||
headers=self.headers,
|
headers=self.headers,
|
||||||
)
|
)
|
||||||
result = response.get_json()
|
result = response.get_json()
|
||||||
self.assertEqual(response.status_code, 200)
|
self.assertEqual(response.status_code, 200)
|
||||||
self.assertEqual(result["status"], "success")
|
self.assertEqual(result["status"], "success")
|
||||||
|
self.assertEqual(result, calculate_machine_max_product_length(**payload))
|
||||||
|
self.assertIn("final_feasibility", result)
|
||||||
self.assertEqual(
|
self.assertEqual(
|
||||||
result["production_maximums"]["nominal_governing_constraints"],
|
result["production_maximums"]["nominal_governing_constraints"],
|
||||||
["roll_diameter_nominal"],
|
["roll_diameter_nominal"],
|
||||||
)
|
)
|
||||||
|
|
||||||
|
def test_machine_maximum_length_endpoint_does_not_accept_mcp_article_convenience(self):
|
||||||
|
response = self.client.post(
|
||||||
|
"/api/calculations/machine-max-product-length",
|
||||||
|
json={
|
||||||
|
"article_number": "218500",
|
||||||
|
"machine": "K7",
|
||||||
|
"core_diameter_mm": 133,
|
||||||
|
},
|
||||||
|
headers=self.headers,
|
||||||
|
)
|
||||||
|
|
||||||
|
self.assertEqual(response.status_code, 200)
|
||||||
|
self.assertEqual(response.get_json(), {
|
||||||
|
"status": "invalid_parameter",
|
||||||
|
"invalid": [{
|
||||||
|
"field": "request",
|
||||||
|
"message": "unknown request fields: article_number",
|
||||||
|
}],
|
||||||
|
})
|
||||||
|
|
||||||
def test_machine_maximum_length_endpoint_preserves_domain_outcomes(self):
|
def test_machine_maximum_length_endpoint_preserves_domain_outcomes(self):
|
||||||
incompatible = self.client.post(
|
incompatible = self.client.post(
|
||||||
"/api/calculations/machine-max-product-length",
|
"/api/calculations/machine-max-product-length",
|
||||||
|
|||||||
+274
-10
@@ -1,5 +1,10 @@
|
|||||||
|
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
|
||||||
@@ -19,8 +24,10 @@ 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,
|
||||||
search_articles_result,
|
search_articles_result,
|
||||||
)
|
)
|
||||||
from machine_constraints import (
|
from machine_constraints import (
|
||||||
@@ -29,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):
|
||||||
@@ -42,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):
|
||||||
@@ -103,7 +119,7 @@ class McpAdapterTests(unittest.TestCase):
|
|||||||
"feasible_with_warnings",
|
"feasible_with_warnings",
|
||||||
)
|
)
|
||||||
|
|
||||||
def test_machine_maximum_length_adapter_delegates_to_shared_domain(self):
|
def test_machine_maximum_length_adapter_projects_success_without_mutating_domain(self):
|
||||||
arguments = {
|
arguments = {
|
||||||
"machine": "K7",
|
"machine": "K7",
|
||||||
"core_diameter_mm": 133.0,
|
"core_diameter_mm": 133.0,
|
||||||
@@ -112,10 +128,149 @@ class McpAdapterTests(unittest.TestCase):
|
|||||||
"product_width_m": 6.0,
|
"product_width_m": 6.0,
|
||||||
"area_weight_g_m2": 495.395833,
|
"area_weight_g_m2": 495.395833,
|
||||||
}
|
}
|
||||||
self.assertEqual(
|
domain_result = calculate_machine_max_product_length(**arguments)
|
||||||
calculate_machine_max_product_length_result(**arguments),
|
domain_before_projection = copy.deepcopy(domain_result)
|
||||||
calculate_machine_max_product_length(**arguments),
|
|
||||||
|
self.assertNotIn(
|
||||||
|
"article_number",
|
||||||
|
inspect.signature(calculate_machine_max_product_length).parameters,
|
||||||
)
|
)
|
||||||
|
result = project_machine_max_product_length_result(domain_result)
|
||||||
|
|
||||||
|
self.assertEqual(domain_result, domain_before_projection)
|
||||||
|
self.assertIn("final_feasibility", domain_result)
|
||||||
|
self.assertNotIn("final_feasibility", result)
|
||||||
|
self.assertEqual(
|
||||||
|
calculate_machine_max_product_length_result(**arguments), result
|
||||||
|
)
|
||||||
|
self.assertEqual(result["machine"], {"id": "k7", "name": "K7"})
|
||||||
|
self.assertEqual(result["weight_scope"], "material_only")
|
||||||
|
self.assertEqual(result["machine_limits"], {
|
||||||
|
"maximum_roll_diameter_mm": 1100.0,
|
||||||
|
"maximum_material_roll_weight_kg": 700.0,
|
||||||
|
})
|
||||||
|
self.assertEqual(result["production_maximums"], {
|
||||||
|
"nominal_length_m": 217.1900177447768,
|
||||||
|
"nominal_governing_constraints": ["roll_diameter_nominal"],
|
||||||
|
"conservative_length_m": 194.62073281485985,
|
||||||
|
"conservative_governing_constraints": ["roll_diameter_conservative"],
|
||||||
|
})
|
||||||
|
self.assertEqual(result["material_weights_kg"], {
|
||||||
|
"at_nominal_production_maximum": 645.570178559751,
|
||||||
|
"at_conservative_production_maximum": 578.4858003113276,
|
||||||
|
})
|
||||||
|
self.assertEqual(result["warnings"], {"nominal": ["roll_diameter_mm"]})
|
||||||
|
|
||||||
|
def test_machine_maximum_length_adapter_resolves_exact_article_compactly(self):
|
||||||
|
result = calculate_machine_max_product_length_result(
|
||||||
|
article_number="218500",
|
||||||
|
machine="K7",
|
||||||
|
core_diameter_mm=133.0,
|
||||||
|
)
|
||||||
|
|
||||||
|
self.assertEqual(result["status"], "success")
|
||||||
|
self.assertEqual(result["article"], {
|
||||||
|
"number": "218500",
|
||||||
|
"name": "Stex AS 501, 6,00 x 75 m",
|
||||||
|
})
|
||||||
|
self.assertNotIn("thickness_mm", result)
|
||||||
|
self.assertNotIn("final_feasibility", result)
|
||||||
|
self.assertEqual(result["production_maximums"], {
|
||||||
|
"nominal_length_m": 217.1900177447768,
|
||||||
|
"nominal_governing_constraints": ["roll_diameter_nominal"],
|
||||||
|
"conservative_length_m": 194.62073281485985,
|
||||||
|
"conservative_governing_constraints": ["roll_diameter_conservative"],
|
||||||
|
})
|
||||||
|
self.assertEqual(result["material_weights_kg"], {
|
||||||
|
"at_nominal_production_maximum": 645.570178559751,
|
||||||
|
"at_conservative_production_maximum": 578.4858003113276,
|
||||||
|
})
|
||||||
|
|
||||||
|
def test_machine_maximum_length_adapter_article_lookup_is_exact_and_core_is_caller_supplied(self):
|
||||||
|
missing = calculate_machine_max_product_length_result(
|
||||||
|
article_number="does-not-exist",
|
||||||
|
machine="K7",
|
||||||
|
core_diameter_mm=133.0,
|
||||||
|
)
|
||||||
|
incompatible_core = calculate_machine_max_product_length_result(
|
||||||
|
article_number="218500",
|
||||||
|
machine="K7",
|
||||||
|
core_diameter_mm=194.0,
|
||||||
|
)
|
||||||
|
|
||||||
|
self.assertEqual(missing, {
|
||||||
|
"status": "article_not_found",
|
||||||
|
"article_number": "does-not-exist",
|
||||||
|
})
|
||||||
|
self.assertEqual(incompatible_core["status"], "incompatible")
|
||||||
|
self.assertEqual(
|
||||||
|
incompatible_core["failed_compatibility_constraints"],
|
||||||
|
["allowed_core_diameters_mm"],
|
||||||
|
)
|
||||||
|
|
||||||
|
def test_machine_maximum_length_adapter_rejects_mixed_article_and_material_inputs(self):
|
||||||
|
result = calculate_machine_max_product_length_result(
|
||||||
|
article_number="218500",
|
||||||
|
machine="K7",
|
||||||
|
core_diameter_mm=133.0,
|
||||||
|
thickness_mm=4.311611,
|
||||||
|
)
|
||||||
|
|
||||||
|
self.assertEqual(result, {
|
||||||
|
"status": "invalid_parameter",
|
||||||
|
"invalid": [{
|
||||||
|
"field": "article_number",
|
||||||
|
"message": (
|
||||||
|
"cannot be combined with explicit material properties: "
|
||||||
|
"thickness_mm"
|
||||||
|
),
|
||||||
|
}],
|
||||||
|
})
|
||||||
|
|
||||||
|
def test_machine_maximum_length_adapter_preserves_non_success_diagnostics(self):
|
||||||
|
base = {
|
||||||
|
"machine": "K7",
|
||||||
|
"core_diameter_mm": 133.0,
|
||||||
|
"thickness_mm": 4.311611,
|
||||||
|
"thickness_stddev_mm": 0.249999,
|
||||||
|
"product_width_m": 6.0,
|
||||||
|
"area_weight_g_m2": 495.395833,
|
||||||
|
}
|
||||||
|
for changes, status in (
|
||||||
|
({"core_diameter_mm": 194.0}, "incompatible"),
|
||||||
|
({"product_width_m": 3.0}, "incompatible"),
|
||||||
|
(
|
||||||
|
{"product_width_m": None, "area_weight_g_m2": None},
|
||||||
|
"needs_clarification",
|
||||||
|
),
|
||||||
|
({"thickness_mm": -1.0}, "invalid_parameter"),
|
||||||
|
({"machine": "not configured"}, "machine_not_found"),
|
||||||
|
):
|
||||||
|
with self.subTest(changes=changes):
|
||||||
|
arguments = {**base, **changes}
|
||||||
|
domain_result = calculate_machine_max_product_length(**arguments)
|
||||||
|
adapter_result = calculate_machine_max_product_length_result(**arguments)
|
||||||
|
self.assertEqual(adapter_result, domain_result)
|
||||||
|
self.assertEqual(adapter_result["status"], status)
|
||||||
|
|
||||||
|
missing = calculate_machine_max_product_length_result(
|
||||||
|
**{**base, "product_width_m": None, "area_weight_g_m2": None}
|
||||||
|
)
|
||||||
|
self.assertEqual(
|
||||||
|
missing["missing_required_inputs"],
|
||||||
|
["product_width_m", "area_weight_g_m2"],
|
||||||
|
)
|
||||||
|
core = calculate_machine_max_product_length_result(
|
||||||
|
**{**base, "core_diameter_mm": 194.0}
|
||||||
|
)
|
||||||
|
self.assertEqual(
|
||||||
|
core["failed_compatibility_constraints"],
|
||||||
|
["allowed_core_diameters_mm"],
|
||||||
|
)
|
||||||
|
width = calculate_machine_max_product_length_result(
|
||||||
|
**{**base, "product_width_m": 3.0}
|
||||||
|
)
|
||||||
|
self.assertEqual(width["failed_compatibility_constraints"], ["product_width_m"])
|
||||||
|
|
||||||
def test_material_weight_adapter_delegates_without_a_core_diameter(self):
|
def test_material_weight_adapter_delegates_without_a_core_diameter(self):
|
||||||
arguments = {
|
arguments = {
|
||||||
@@ -199,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",
|
||||||
@@ -243,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()
|
||||||
@@ -257,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
|
||||||
|
|
||||||
@@ -267,23 +526,28 @@ 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)
|
||||||
self.assertIn("Missing configured V1 roll inputs", feasibility_tool.description)
|
self.assertIn("Missing configured V1 roll inputs", feasibility_tool.description)
|
||||||
self.assertEqual(
|
self.assertEqual(
|
||||||
maximum_tool.parameters["required"],
|
maximum_tool.parameters["required"],
|
||||||
["machine", "core_diameter_mm", "thickness_mm"],
|
["machine", "core_diameter_mm"],
|
||||||
)
|
)
|
||||||
|
self.assertIn("article_number", maximum_tool.parameters["properties"])
|
||||||
self.assertIn("product_width_m", maximum_tool.parameters["properties"])
|
self.assertIn("product_width_m", maximum_tool.parameters["properties"])
|
||||||
self.assertIn("area_weight_g_m2", maximum_tool.parameters["properties"])
|
self.assertIn("area_weight_g_m2", maximum_tool.parameters["properties"])
|
||||||
for text in (
|
for text in (
|
||||||
"Resolve the article first with search_articles/get_article",
|
"exact article_number",
|
||||||
|
"do not call get_article merely to copy properties",
|
||||||
|
"Never mix article_number with material properties",
|
||||||
|
"For a product name, use search_articles",
|
||||||
|
"if candidates are multiple, ask the user to select an article number",
|
||||||
"Do not supply or invent machine limits",
|
"Do not supply or invent machine limits",
|
||||||
"rearrange roll or weight formulas",
|
|
||||||
"compare candidate limits yourself",
|
|
||||||
"nominal and conservative/no-warning production maxima",
|
"nominal and conservative/no-warning production maxima",
|
||||||
"incompatibility rather than substituting another core or width",
|
|
||||||
):
|
):
|
||||||
self.assertIn(text, maximum_tool.description)
|
self.assertIn(text, maximum_tool.description)
|
||||||
self.assertEqual(
|
self.assertEqual(
|
||||||
|
|||||||
Reference in New Issue
Block a user