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.
|
||||
|
||||
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.
|
||||
|
||||
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_server.py` provides a local stdio-only Model Context Protocol server. It is
|
||||
`mcp_server.py` provides a Model Context Protocol server. It is
|
||||
a thin adapter over the same domain services used by Flask: article lookup,
|
||||
direct roll calculation, product-length calculation, material-weight calculation,
|
||||
transport analysis, and configured production-machine feasibility.
|
||||
The available PoC tools are `get_article`, `search_articles`,
|
||||
`calculate_material_weight`, `calculate_product_length`,
|
||||
`calculate_roll_diameter`, `analyze_transport_capacity`, `get_machine`,
|
||||
`check_production_feasibility`, and `calculate_machine_max_product_length`.
|
||||
`check_production_feasibility`, `calculate_machine_max_product_length`, and
|
||||
`generate_calculation_pdf`.
|
||||
|
||||
Install the pinned dependencies, including `mcp==1.26.0`, in the existing
|
||||
environment, then start it with:
|
||||
@@ -135,8 +136,16 @@ environment, then start it with:
|
||||
.venv/bin/python mcp_server.py
|
||||
```
|
||||
|
||||
The PoC has no MCP resources, HTTP/SSE transport, remote authentication, or
|
||||
dedicated target-length calculation tool or extrapolation support.
|
||||
The server supports the existing `stdio` transport (the default) and
|
||||
`streamable-http`. MCP PDF delivery requires `streamable-http`; it has no
|
||||
dependency on the Flask web application. Native MCP resources are intentionally
|
||||
not used because the installed MCPO 0.0.20 does not preserve resource links or
|
||||
embedded resources. The PoC has no remote authentication or dedicated
|
||||
target-length calculation tool or extrapolation support.
|
||||
|
||||
In `stdio` mode, `generate_calculation_pdf` returns a structured
|
||||
`artifact_delivery_unavailable` response rather than creating an unreachable
|
||||
report. Existing calculation tools work in either transport.
|
||||
|
||||
Tool calling semantics:
|
||||
|
||||
@@ -150,14 +159,18 @@ Tool calling semantics:
|
||||
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.
|
||||
- `calculate_machine_max_product_length` composes configured machine limits
|
||||
with the shared calculation domain. Resolve an article first, then pass its
|
||||
properties unchanged. Its nominal maximum uses the nominal diameter limit;
|
||||
with the shared calculation domain. With an exact article number, MCP resolves
|
||||
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
|
||||
thickness variation. Both also respect configured material-only maximum roll
|
||||
weight, and the domain returns the governing constraint(s), unrounded.
|
||||
Core and width incompatibility cannot be corrected by shortening and is
|
||||
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
|
||||
diameter, core diameter, and material thickness. Its optional thickness
|
||||
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
|
||||
weight is needed. On success, it returns `transport_roll_inputs`, the
|
||||
browser-equivalent transport bundle.
|
||||
- `generate_calculation_pdf` is strictly opt-in: call it only after the user
|
||||
explicitly asks for a PDF or calculation report. Supply the original inputs
|
||||
from `calculate_roll_diameter`, never calculated diameters, weights, warnings,
|
||||
or notes. It recalculates through the same RollCalc domain service and uses
|
||||
the existing one-page renderer. Ordinary calculation calls neither generate
|
||||
nor suggest a PDF.
|
||||
|
||||
For example, calculate a roll normally with `calculate_roll_diameter`. If the
|
||||
user later asks “Mach mir daraus ein PDF”, call `generate_calculation_pdf` with
|
||||
the original calculation inputs. Its successful response contains a safe
|
||||
`filename`, expiry metadata, and a browser-ready `download_url`; MCPO/OpenWebUI
|
||||
can present that URL as a download link. PDFs exist only in the MCP host's local
|
||||
temporary store, not in Flask or shared storage. Each URL has a 256-bit opaque
|
||||
capability token and expires after 24 hours during report-store access; whoever
|
||||
possesses the URL can download the report until expiry. The user’s browser must
|
||||
be able to reach the MCP-host public origin over HTTPS.
|
||||
|
||||
For independent PDF delivery, configure and run the MCP host as follows:
|
||||
|
||||
```bash
|
||||
export ROLLCALC_MCP_TRANSPORT=streamable-http
|
||||
export ROLLCALC_MCP_HOST=0.0.0.0
|
||||
export ROLLCALC_MCP_PORT=8000
|
||||
export ROLLCALC_MCP_ARTIFACT_PUBLIC_BASE_URL=https://rollcalc-mcp.example.internal
|
||||
.venv/bin/python mcp_server.py
|
||||
```
|
||||
|
||||
`ROLLCALC_MCP_ARTIFACT_PUBLIC_BASE_URL` is the externally browser-reachable
|
||||
origin for this MCP service, not the Flask website URL. MCPO connects to the
|
||||
MCP host with `--server-type streamable-http -- https://<mcp-host>/mcp`; the
|
||||
browser follows the returned `/reports/<capability>.pdf` URL directly.
|
||||
- `analyze_transport_capacity` uses roll/core diameters in mm; roll width and
|
||||
transport dimensions in m; and roll/payload weights in kg.
|
||||
`product_length_m` is the material length represented by one roll, used only
|
||||
|
||||
@@ -32,10 +32,8 @@ from conversation_service import (
|
||||
from ollama_nlu import OllamaNLUClient
|
||||
from pdf_report import (
|
||||
ReportValidationError,
|
||||
report_from_calculation_result,
|
||||
render_roll_report,
|
||||
safe_report_filename,
|
||||
)
|
||||
from roll_report_service import create_roll_calculation_pdf
|
||||
from roll_calculation import (
|
||||
calculate_product_length,
|
||||
calculate_roll,
|
||||
@@ -325,16 +323,10 @@ def create_machine_max_product_length_calculation():
|
||||
|
||||
|
||||
def _create_pdf(calculation_request):
|
||||
result = calculate_roll(calculation_request, build_info=BUILD_INFO)
|
||||
report = report_from_calculation_result(result)
|
||||
generated_at = datetime.now().astimezone()
|
||||
pdf = render_roll_report(
|
||||
report,
|
||||
generated_at=generated_at,
|
||||
pdf, filename, _ = create_roll_calculation_pdf(
|
||||
calculation_request,
|
||||
build_info=BUILD_INFO,
|
||||
)
|
||||
article_number = report["article"]["number"] if report["article"] else None
|
||||
filename = safe_report_filename(article_number, generated_at)
|
||||
return pdf, filename
|
||||
|
||||
|
||||
|
||||
@@ -31,6 +31,16 @@ roll_calculation.calculate_roll()
|
||||
|
||||
The PDF endpoint accepts the same calculation request, recalculates it through the core, then adapts the authoritative result to the established one-page PDF layout. It does not accept calculated values from the browser.
|
||||
|
||||
`generate_calculation_pdf` is the equivalent opt-in MCP adapter. It accepts only
|
||||
original roll-calculation inputs and uses the shared report service after
|
||||
deterministic recalculation. Under FastMCP Streamable HTTP, the MCP host owns a
|
||||
short-lived local store and its `/reports/<256-bit-capability>.pdf` download
|
||||
route. Flask has no role in this path, which avoids cross-host filesystem
|
||||
coupling and binary PDF content in tool responses. The browser-reachable origin
|
||||
is configured independently with `ROLLCALC_MCP_ARTIFACT_PUBLIC_BASE_URL`.
|
||||
Installed MCPO 0.0.20 does not retain native MCP resources/resource links, so
|
||||
the tool returns an ordinary browser URL rather than an MCP resource.
|
||||
|
||||
## Production-Machine Feasibility
|
||||
|
||||
`config/machines.yaml` is the runtime source for configured production-machine
|
||||
|
||||
@@ -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
|
||||
|
||||
import os
|
||||
from typing import Any
|
||||
from urllib.parse import quote
|
||||
|
||||
from mcp_pdf_delivery import McpReportNotFoundError, McpReportStore
|
||||
from rollcalc_mcp_tools import (
|
||||
analyze_transport_capacity_result,
|
||||
calculate_machine_max_product_length_result,
|
||||
calculate_material_weight_result,
|
||||
calculate_product_length_result,
|
||||
calculate_roll_diameter_result,
|
||||
generate_calculation_pdf_result,
|
||||
get_article_result,
|
||||
get_machine_result,
|
||||
check_production_feasibility_result,
|
||||
@@ -18,11 +22,49 @@ from rollcalc_mcp_tools import (
|
||||
from transport_calculation import TransportPresetKey
|
||||
|
||||
|
||||
async def download_report(request):
|
||||
"""Serve one MCP-host-local report by opaque temporary capability."""
|
||||
from starlette.responses import PlainTextResponse, Response
|
||||
|
||||
token = request.path_params["token"]
|
||||
try:
|
||||
pdf, report = McpReportStore().read(token)
|
||||
except McpReportNotFoundError:
|
||||
return PlainTextResponse("Report not found or expired", status_code=404)
|
||||
return Response(
|
||||
pdf,
|
||||
media_type="application/pdf",
|
||||
headers={
|
||||
"Content-Disposition": (
|
||||
f'attachment; filename="{quote(report.filename)}"'
|
||||
),
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
def selected_transport() -> str:
|
||||
"""Read and validate the transport selected for this MCP process."""
|
||||
transport = os.getenv("ROLLCALC_MCP_TRANSPORT", "stdio").strip().lower()
|
||||
if transport not in {"stdio", "streamable-http"}:
|
||||
raise ValueError(
|
||||
"ROLLCALC_MCP_TRANSPORT must be 'stdio' or 'streamable-http'"
|
||||
)
|
||||
return transport
|
||||
|
||||
|
||||
def create_server() -> Any:
|
||||
"""Create the stdio-only MCP server after the SDK is installed."""
|
||||
"""Create the transport-neutral RollCalc MCP application."""
|
||||
from mcp.server.fastmcp import FastMCP
|
||||
|
||||
server = FastMCP("RollCalc")
|
||||
server = FastMCP(
|
||||
"RollCalc",
|
||||
host=os.getenv("ROLLCALC_MCP_HOST", "127.0.0.1"),
|
||||
port=int(os.getenv("ROLLCALC_MCP_PORT", "8000")),
|
||||
)
|
||||
|
||||
@server.custom_route("/reports/{token}.pdf", methods=["GET"])
|
||||
async def mcp_report_download_route(request):
|
||||
return await download_report(request)
|
||||
|
||||
@server.tool()
|
||||
def get_article(
|
||||
@@ -67,15 +109,17 @@ def create_server() -> Any:
|
||||
def calculate_machine_max_product_length(
|
||||
machine: str,
|
||||
core_diameter_mm: float,
|
||||
thickness_mm: float,
|
||||
article_number: str | None = None,
|
||||
thickness_mm: float | None = None,
|
||||
thickness_stddev_mm: float | None = None,
|
||||
product_width_m: float | None = None,
|
||||
area_weight_g_m2: float | None = None,
|
||||
) -> 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(
|
||||
machine=machine,
|
||||
core_diameter_mm=core_diameter_mm,
|
||||
article_number=article_number,
|
||||
thickness_mm=thickness_mm,
|
||||
thickness_stddev_mm=thickness_stddev_mm,
|
||||
product_width_m=product_width_m,
|
||||
@@ -143,6 +187,45 @@ def create_server() -> Any:
|
||||
production_site=production_site,
|
||||
)
|
||||
|
||||
@server.tool()
|
||||
def generate_calculation_pdf(
|
||||
roll_length_m: float | None = None,
|
||||
core_diameter_mm: float | None = None,
|
||||
thickness_mm: float | None = None,
|
||||
thickness_stddev_mm: float | None = None,
|
||||
article_number: str | None = None,
|
||||
article_name_hint: str | None = None,
|
||||
core_type: str | None = None,
|
||||
width_m: float | None = None,
|
||||
area_weight_g_m2: float | None = None,
|
||||
category: str | None = None,
|
||||
production_site: str | None = None,
|
||||
include_roll_weight: bool = True,
|
||||
) -> dict[str, Any]:
|
||||
"""Create a PDF calculation report only when the user explicitly requests a PDF or report. Supply the original inputs from a completed calculate_roll_diameter request; never supply, infer, or copy calculated diameters, weights, warnings, or notes. The tool recalculates authoritatively and returns a short-lived browser URL hosted by the MCP service. Do not call this tool after ordinary calculations and do not offer it proactively."""
|
||||
if selected_transport() != "streamable-http":
|
||||
return {
|
||||
"status": "artifact_delivery_unavailable",
|
||||
"error": (
|
||||
"generate_calculation_pdf requires "
|
||||
"ROLLCALC_MCP_TRANSPORT=streamable-http"
|
||||
),
|
||||
}
|
||||
return generate_calculation_pdf_result(
|
||||
article_number=article_number,
|
||||
article_name_hint=article_name_hint,
|
||||
roll_length_m=roll_length_m,
|
||||
width_m=width_m,
|
||||
thickness_mm=thickness_mm,
|
||||
thickness_stddev_mm=thickness_stddev_mm,
|
||||
area_weight_g_m2=area_weight_g_m2,
|
||||
core_diameter_mm=core_diameter_mm,
|
||||
core_type=core_type,
|
||||
category=category,
|
||||
production_site=production_site,
|
||||
include_roll_weight=include_roll_weight,
|
||||
)
|
||||
|
||||
@server.tool()
|
||||
def analyze_transport_capacity(
|
||||
transport_preset: TransportPresetKey,
|
||||
@@ -165,5 +248,10 @@ def create_server() -> Any:
|
||||
return server
|
||||
|
||||
|
||||
def run_server() -> None:
|
||||
"""Run the selected FastMCP transport without changing tool definitions."""
|
||||
create_server().run(transport=selected_transport())
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
create_server().run(transport="stdio")
|
||||
run_server()
|
||||
|
||||
@@ -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,
|
||||
transport_roll_inputs_from_roll_calculation,
|
||||
)
|
||||
from mcp_pdf_delivery import generate_mcp_calculation_pdf
|
||||
from transport_calculation import analyze_transport
|
||||
|
||||
|
||||
@@ -64,13 +65,49 @@ def calculate_machine_max_product_length_result(
|
||||
*,
|
||||
machine: str,
|
||||
core_diameter_mm: float,
|
||||
thickness_mm: float,
|
||||
article_number: str | None = None,
|
||||
thickness_mm: float | None = None,
|
||||
thickness_stddev_mm: float | None = None,
|
||||
product_width_m: float | None = None,
|
||||
area_weight_g_m2: float | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Delegate machine-aware maximum product length to the shared domain."""
|
||||
return calculate_machine_max_product_length(
|
||||
"""Return the compact MCP view, resolving one exact article when supplied."""
|
||||
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,
|
||||
core_diameter_mm=core_diameter_mm,
|
||||
thickness_mm=thickness_mm,
|
||||
@@ -78,6 +115,45 @@ def calculate_machine_max_product_length_result(
|
||||
product_width_m=product_width_m,
|
||||
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(
|
||||
@@ -151,6 +227,39 @@ def calculate_roll_diameter_result(
|
||||
return result
|
||||
|
||||
|
||||
def generate_calculation_pdf_result(
|
||||
*,
|
||||
article_number: str | None = None,
|
||||
article_name_hint: str | None = None,
|
||||
roll_length_m: float | None = None,
|
||||
width_m: float | None = None,
|
||||
thickness_mm: float | None = None,
|
||||
thickness_stddev_mm: float | None = None,
|
||||
area_weight_g_m2: float | None = None,
|
||||
core_diameter_mm: float | None = None,
|
||||
core_type: str | None = None,
|
||||
category: str | None = None,
|
||||
production_site: str | None = None,
|
||||
include_roll_weight: bool = True,
|
||||
) -> dict[str, Any]:
|
||||
"""Create the established PDF from original inputs, never LLM result values."""
|
||||
request = {
|
||||
"article_number": article_number,
|
||||
"article_name_hint": article_name_hint,
|
||||
"roll_length_m": roll_length_m,
|
||||
"width_m": width_m,
|
||||
"thickness_mm": thickness_mm,
|
||||
"thickness_stddev_mm": thickness_stddev_mm,
|
||||
"area_weight_g_m2": area_weight_g_m2,
|
||||
"core_diameter_mm": core_diameter_mm,
|
||||
"core_type": core_type,
|
||||
"category": category,
|
||||
"production_site": production_site,
|
||||
"include_roll_weight": include_roll_weight,
|
||||
}
|
||||
return generate_mcp_calculation_pdf(request)
|
||||
|
||||
|
||||
def analyze_transport_capacity_result(
|
||||
*,
|
||||
transport_preset: str,
|
||||
|
||||
@@ -5,6 +5,7 @@ import unittest
|
||||
from unittest.mock import patch
|
||||
|
||||
import app as rollcalc_app
|
||||
from machine_constraints import calculate_machine_max_product_length
|
||||
|
||||
|
||||
class CalculationApiTests(unittest.TestCase):
|
||||
@@ -140,26 +141,49 @@ class CalculationApiTests(unittest.TestCase):
|
||||
self.assertEqual(result["warning_constraints"], ["roll_diameter_mm"])
|
||||
|
||||
def test_machine_maximum_length_endpoint_is_authenticated_and_structured(self):
|
||||
payload = {
|
||||
"machine": "K7",
|
||||
"core_diameter_mm": 133,
|
||||
"thickness_mm": 4.311611,
|
||||
"thickness_stddev_mm": 0.249999,
|
||||
"product_width_m": 6,
|
||||
"area_weight_g_m2": 495.395833,
|
||||
}
|
||||
response = self.client.post(
|
||||
"/api/calculations/machine-max-product-length",
|
||||
json={
|
||||
"machine": "K7",
|
||||
"core_diameter_mm": 133,
|
||||
"thickness_mm": 4.311611,
|
||||
"thickness_stddev_mm": 0.249999,
|
||||
"product_width_m": 6,
|
||||
"area_weight_g_m2": 495.395833,
|
||||
},
|
||||
json=payload,
|
||||
headers=self.headers,
|
||||
)
|
||||
result = response.get_json()
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.assertEqual(result["status"], "success")
|
||||
self.assertEqual(result, calculate_machine_max_product_length(**payload))
|
||||
self.assertIn("final_feasibility", result)
|
||||
self.assertEqual(
|
||||
result["production_maximums"]["nominal_governing_constraints"],
|
||||
["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):
|
||||
incompatible = self.client.post(
|
||||
"/api/calculations/machine-max-product-length",
|
||||
|
||||
+274
-10
@@ -1,5 +1,10 @@
|
||||
import copy
|
||||
import asyncio
|
||||
import inspect
|
||||
import os
|
||||
from pathlib import Path
|
||||
import sys
|
||||
import tempfile
|
||||
import types
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
@@ -19,8 +24,10 @@ from rollcalc_mcp_tools import (
|
||||
calculate_material_weight_result,
|
||||
calculate_product_length_result,
|
||||
calculate_roll_diameter_result,
|
||||
generate_calculation_pdf_result,
|
||||
get_article_result,
|
||||
get_machine_result,
|
||||
project_machine_max_product_length_result,
|
||||
search_articles_result,
|
||||
)
|
||||
from machine_constraints import (
|
||||
@@ -29,12 +36,15 @@ from machine_constraints import (
|
||||
get_machine,
|
||||
)
|
||||
from transport_calculation import analyze_transport, transport_presets
|
||||
from mcp_pdf_delivery import McpReportNotFoundError, McpReportStore, REPORT_TTL_SECONDS
|
||||
from starlette.requests import Request
|
||||
|
||||
|
||||
class _FakeFastMCP:
|
||||
def __init__(self, name):
|
||||
def __init__(self, name, **kwargs):
|
||||
self.name = name
|
||||
self.tools = []
|
||||
self.routes = []
|
||||
|
||||
def tool(self):
|
||||
def register(function):
|
||||
@@ -42,6 +52,12 @@ class _FakeFastMCP:
|
||||
return function
|
||||
return register
|
||||
|
||||
def custom_route(self, path, methods):
|
||||
def register(function):
|
||||
self.routes.append((path, methods, function))
|
||||
return function
|
||||
return register
|
||||
|
||||
|
||||
class McpAdapterTests(unittest.TestCase):
|
||||
def test_get_article_delegates_to_the_domain_for_success_not_found_and_conflict(self):
|
||||
@@ -103,7 +119,7 @@ class McpAdapterTests(unittest.TestCase):
|
||||
"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 = {
|
||||
"machine": "K7",
|
||||
"core_diameter_mm": 133.0,
|
||||
@@ -112,10 +128,149 @@ class McpAdapterTests(unittest.TestCase):
|
||||
"product_width_m": 6.0,
|
||||
"area_weight_g_m2": 495.395833,
|
||||
}
|
||||
self.assertEqual(
|
||||
calculate_machine_max_product_length_result(**arguments),
|
||||
calculate_machine_max_product_length(**arguments),
|
||||
domain_result = calculate_machine_max_product_length(**arguments)
|
||||
domain_before_projection = copy.deepcopy(domain_result)
|
||||
|
||||
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):
|
||||
arguments = {
|
||||
@@ -199,6 +354,96 @@ class McpAdapterTests(unittest.TestCase):
|
||||
helper.assert_called_once()
|
||||
self.assertEqual(result["transport_roll_inputs"], expected_inputs)
|
||||
|
||||
def test_generate_pdf_reuses_authoritative_report_path(self):
|
||||
arguments = {
|
||||
"article_number": "146900",
|
||||
"roll_length_m": 50.0,
|
||||
"core_diameter_mm": 150.0,
|
||||
"core_type": "150mm PVC",
|
||||
"category": "bentofix",
|
||||
}
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
with patch.dict(os.environ, {
|
||||
"ROLLCALC_MCP_ARTIFACT_DIRECTORY": directory,
|
||||
"ROLLCALC_MCP_ARTIFACT_PUBLIC_BASE_URL": (
|
||||
"https://mcp.example.test/rollcalc"
|
||||
),
|
||||
}):
|
||||
response = generate_calculation_pdf_result(**arguments)
|
||||
token = response["download_url"].rsplit("/", 1)[1][:-4]
|
||||
pdf, stored = McpReportStore().read(token)
|
||||
|
||||
self.assertEqual(response["status"], "success")
|
||||
self.assertEqual(stored.filename, response["filename"])
|
||||
self.assertTrue(pdf.startswith(b"%PDF-1.4"))
|
||||
self.assertIn(b"146900", pdf)
|
||||
self.assertIn(b"CALCULATION RESULTS", pdf)
|
||||
self.assertEqual(
|
||||
response["download_url"],
|
||||
f"https://mcp.example.test/rollcalc/reports/{token}.pdf",
|
||||
)
|
||||
self.assertEqual(len(token), 43)
|
||||
self.assertRegex(token, r"^[A-Za-z0-9_-]+$")
|
||||
self.assertEqual(
|
||||
set(response),
|
||||
{"status", "filename", "expires_at", "expires_in_seconds", "download_url"},
|
||||
)
|
||||
self.assertEqual(response["expires_in_seconds"], REPORT_TTL_SECONDS)
|
||||
|
||||
def test_generate_pdf_preserves_structured_calculation_errors(self):
|
||||
result = generate_calculation_pdf_result(
|
||||
article_number="146900",
|
||||
roll_length_m=None,
|
||||
core_diameter_mm=150.0,
|
||||
)
|
||||
|
||||
self.assertEqual(result["status"], "needs_clarification")
|
||||
self.assertIn("roll_length_m", result["missing"])
|
||||
|
||||
def test_mcp_report_store_rejects_expired_unknown_and_path_tokens(self):
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
store = McpReportStore(directory=Path(directory))
|
||||
with patch("mcp_pdf_delivery.time.time", return_value=1_000.0):
|
||||
report = store.create(b"%PDF-1.4", "roll-calculation_146900.pdf")
|
||||
with patch(
|
||||
"mcp_pdf_delivery.time.time",
|
||||
return_value=1_000.0 + REPORT_TTL_SECONDS + 1,
|
||||
):
|
||||
with self.assertRaises(McpReportNotFoundError):
|
||||
store.read(report.token)
|
||||
for token in ("unknown", "../access_log", "a" * 43 + "/x"):
|
||||
with self.subTest(token=token):
|
||||
with self.assertRaises(McpReportNotFoundError):
|
||||
store.read(token)
|
||||
|
||||
def test_mcp_host_route_downloads_only_valid_unexpired_reports(self):
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
with patch.dict(os.environ, {
|
||||
"ROLLCALC_MCP_ARTIFACT_DIRECTORY": directory,
|
||||
}):
|
||||
stored = McpReportStore().create(
|
||||
b"%PDF-1.4\nMCP report", "roll-calculation_146900.pdf"
|
||||
)
|
||||
response = asyncio.run(mcp_server.download_report(
|
||||
Request({"type": "http", "path_params": {"token": stored.token}})
|
||||
))
|
||||
unknown = asyncio.run(mcp_server.download_report(
|
||||
Request({"type": "http", "path_params": {"token": "x" * 43}})
|
||||
))
|
||||
traversal = asyncio.run(mcp_server.download_report(
|
||||
Request({"type": "http", "path_params": {"token": "../access_log"}})
|
||||
))
|
||||
|
||||
self.assertEqual(response.status_code, 200)
|
||||
self.assertEqual(response.headers["content-type"], "application/pdf")
|
||||
self.assertEqual(
|
||||
response.headers["content-disposition"],
|
||||
'attachment; filename="roll-calculation_146900.pdf"',
|
||||
)
|
||||
self.assertEqual(response.body, b"%PDF-1.4\nMCP report")
|
||||
self.assertEqual(unknown.status_code, 404)
|
||||
self.assertEqual(traversal.status_code, 404)
|
||||
|
||||
def test_transport_tool_delegates_to_domain_with_lkw_sattelzug_preset(self):
|
||||
arguments = {
|
||||
"transport_preset": "lkw_sattelzug",
|
||||
@@ -243,9 +488,22 @@ class McpAdapterTests(unittest.TestCase):
|
||||
"calculate_material_weight",
|
||||
"calculate_product_length",
|
||||
"calculate_roll_diameter",
|
||||
"generate_calculation_pdf",
|
||||
"analyze_transport_capacity",
|
||||
],
|
||||
)
|
||||
self.assertEqual(
|
||||
[(path, methods) for path, methods, _ in instance.routes],
|
||||
[("/reports/{token}.pdf", ["GET"])],
|
||||
)
|
||||
pdf_tool = next(
|
||||
tool for tool in instance.tools
|
||||
if tool.__name__ == "generate_calculation_pdf"
|
||||
)
|
||||
with patch.dict(os.environ, {"ROLLCALC_MCP_TRANSPORT": "stdio"}):
|
||||
self.assertEqual(
|
||||
pdf_tool()["status"], "artifact_delivery_unavailable"
|
||||
)
|
||||
|
||||
def test_server_schema_exposes_weight_and_transport_chaining_contract(self):
|
||||
instance = mcp_server.create_server()
|
||||
@@ -257,6 +515,7 @@ class McpAdapterTests(unittest.TestCase):
|
||||
material_weight_tool = tools["calculate_material_weight"]
|
||||
product_length_tool = tools["calculate_product_length"]
|
||||
roll_schema = tools["calculate_roll_diameter"].parameters
|
||||
pdf_tool = tools["generate_calculation_pdf"]
|
||||
transport_tool = tools["analyze_transport_capacity"]
|
||||
transport_schema = transport_tool.parameters
|
||||
|
||||
@@ -267,23 +526,28 @@ class McpAdapterTests(unittest.TestCase):
|
||||
)
|
||||
self.assertIn("average_diameter_mm", feasibility_tool.parameters["properties"])
|
||||
self.assertIn("maximum_diameter_mm", feasibility_tool.parameters["properties"])
|
||||
self.assertIn("explicitly requests", pdf_tool.description)
|
||||
self.assertIn("Do not call this tool after ordinary calculations", pdf_tool.description)
|
||||
self.assertNotIn("minimum_diameter_mm", pdf_tool.parameters["properties"])
|
||||
self.assertIn("tool, not the LLM, decides", feasibility_tool.description)
|
||||
self.assertIn("never override failed constraints", feasibility_tool.description)
|
||||
self.assertIn("warnings", feasibility_tool.description)
|
||||
self.assertIn("Missing configured V1 roll inputs", feasibility_tool.description)
|
||||
self.assertEqual(
|
||||
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("area_weight_g_m2", maximum_tool.parameters["properties"])
|
||||
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",
|
||||
"rearrange roll or weight formulas",
|
||||
"compare candidate limits yourself",
|
||||
"nominal and conservative/no-warning production maxima",
|
||||
"incompatibility rather than substituting another core or width",
|
||||
):
|
||||
self.assertIn(text, maximum_tool.description)
|
||||
self.assertEqual(
|
||||
|
||||
Reference in New Issue
Block a user