258 lines
14 KiB
Python
258 lines
14 KiB
Python
"""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,
|
|
search_articles_result,
|
|
)
|
|
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 transport-neutral RollCalc MCP application."""
|
|
from mcp.server.fastmcp import FastMCP
|
|
|
|
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(
|
|
article_number: str | None = None,
|
|
article_name_hint: str | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Resolve RollCalc article master data. article_number is text (including leading zeroes); optional article_name_hint preserves existing conflict and ambiguity checks."""
|
|
return get_article_result(article_number, article_name_hint)
|
|
|
|
@server.tool()
|
|
def search_articles(query: str) -> dict[str, Any]:
|
|
"""Use search_articles when the user identifies a RollCalc product by name, family, designation, or descriptive article text instead of an exact article number. Do not guess or invent an article number: results are deterministic retrieval candidates from the RollCalc article master data, not product recommendations. For an exact article number, use get_article -> calculate_roll_diameter -> optionally analyze_transport_capacity. For a product name, use search_articles first. Only when total_matches is exactly 1 may you use that candidate's exact article_number and continue through get_article and the requested workflow. When multiple plausible candidates are returned, do NOT select the first or highest-ranked candidate, do not treat ranking as permission to choose, and do not calculate yet: present relevant candidates with their article numbers, distinguishing names, and production_site when supplied; ask the user which article number is intended, then stop until they select it. Ranking/order expresses retrieval relevance only and never authorizes automatic selection. If no candidate is returned, say that no matching RollCalc article was found, ask for a more specific designation or article number, and do not invent an article. production_site=Malaysia is descriptive metadata only: it does not imply Bentofix, Bento 2, or any production machine."""
|
|
return search_articles_result(query)
|
|
|
|
@server.tool()
|
|
def get_machine(machine: str) -> dict[str, Any]:
|
|
"""Resolve a configured production machine by stable ID, exact name, or configured alias. Machine limits are deterministic configured domain data; aliases and locations are lookup/metadata only and do not imply capabilities."""
|
|
return get_machine_result(machine)
|
|
|
|
@server.tool()
|
|
def check_production_feasibility(
|
|
machine: str,
|
|
roll_weight_kg: float | None = None,
|
|
average_diameter_mm: float | None = None,
|
|
maximum_diameter_mm: float | None = None,
|
|
core_diameter_mm: float | None = None,
|
|
product_width_m: float | None = None,
|
|
line_speed_m_per_min: float | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Deterministically check a material-only calculated roll against one configured machine. The tool, not the LLM, decides every independent limit; never override failed constraints. Report returned warnings, especially maximum-diameter variation warnings. Missing configured V1 roll inputs require clarification rather than assumptions. average_diameter_mm and maximum_diameter_mm are the existing RollCalc calculation outputs; line speed is retained but outside the initial roll-feasibility scope."""
|
|
return check_production_feasibility_result(
|
|
machine=machine,
|
|
roll_weight_kg=roll_weight_kg,
|
|
average_diameter_mm=average_diameter_mm,
|
|
maximum_diameter_mm=maximum_diameter_mm,
|
|
core_diameter_mm=core_diameter_mm,
|
|
product_width_m=product_width_m,
|
|
line_speed_m_per_min=line_speed_m_per_min,
|
|
)
|
|
|
|
@server.tool()
|
|
def calculate_machine_max_product_length(
|
|
machine: str,
|
|
core_diameter_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. 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,
|
|
area_weight_g_m2=area_weight_g_m2,
|
|
)
|
|
|
|
@server.tool()
|
|
def calculate_material_weight(
|
|
roll_length_m: float,
|
|
width_m: float,
|
|
area_weight_g_m2: float,
|
|
) -> dict[str, Any]:
|
|
"""Calculate deterministic material-only roll weight from roll_length_m (m), width_m (m), and area_weight_g_m2 (g/m²). Use this when those inputs are known but core diameter is unknown or not needed. It does not require, accept, infer, or default a core diameter; never invent a standard core. The result excludes any core weight and cannot provide roll diameter or a core-inclusive total weight. For roll diameter, use calculate_roll_diameter only with a known positive core diameter."""
|
|
return calculate_material_weight_result(
|
|
roll_length_m=roll_length_m,
|
|
width_m=width_m,
|
|
area_weight_g_m2=area_weight_g_m2,
|
|
)
|
|
|
|
@server.tool()
|
|
def calculate_product_length(
|
|
target_roll_diameter_mm: float,
|
|
core_diameter_mm: float,
|
|
thickness_mm: float,
|
|
thickness_stddev_mm: float | None = None,
|
|
width_m: float | None = None,
|
|
area_weight_g_m2: float | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Calculate required product/roll length for a target outer roll diameter. target_roll_diameter_mm, core_diameter_mm, and thickness_mm are required and expressed in mm. Use exact article master-data values obtained through get_article when available; do not infer a missing core diameter, thickness, width, area weight, or article. thickness_stddev_mm returns deterministic minimum, average, and maximum length ranges. When both width_m (m) and area_weight_g_m2 (g/m²) are supplied, the result also includes corresponding material-weight ranges. This tool calculates length from target diameter; use calculate_roll_diameter for the reverse direction."""
|
|
return calculate_product_length_result(
|
|
target_roll_diameter_mm=target_roll_diameter_mm,
|
|
core_diameter_mm=core_diameter_mm,
|
|
thickness_mm=thickness_mm,
|
|
thickness_stddev_mm=thickness_stddev_mm,
|
|
width_m=width_m,
|
|
area_weight_g_m2=area_weight_g_m2,
|
|
)
|
|
|
|
@server.tool()
|
|
def calculate_roll_diameter(
|
|
roll_length_m: float | None = None,
|
|
core_diameter_mm: float | None = None,
|
|
thickness_mm: float | None = None,
|
|
thickness_stddev_mm: float | None = None,
|
|
article_number: str | None = None,
|
|
article_name_hint: str | None = None,
|
|
core_type: str | None = None,
|
|
width_m: float | None = None,
|
|
area_weight_g_m2: float | None = None,
|
|
category: str | None = None,
|
|
production_site: str | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Calculate roll diameter and weight for one roll. roll_length_m is the material length on one roll in m; core_diameter_mm, thickness_mm, and thickness_stddev_mm are mm; width_m is m; and area_weight_g_m2 is g/m². Roll diameter requires a known positive core diameter: do not infer or default one. If only material weight is needed and the core is unknown, call calculate_material_weight instead; it returns material-only weight, not a diameter or core-inclusive total. On success, transport_roll_inputs contains the browser-equivalent transport values; pass that bundle's fields unchanged to analyze_transport_capacity."""
|
|
return calculate_roll_diameter_result(
|
|
article_number=article_number,
|
|
article_name_hint=article_name_hint,
|
|
roll_length_m=roll_length_m,
|
|
width_m=width_m,
|
|
thickness_mm=thickness_mm,
|
|
thickness_stddev_mm=thickness_stddev_mm,
|
|
area_weight_g_m2=area_weight_g_m2,
|
|
core_diameter_mm=core_diameter_mm,
|
|
core_type=core_type,
|
|
category=category,
|
|
production_site=production_site,
|
|
)
|
|
|
|
@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,
|
|
roll_diameter_mm: float,
|
|
core_diameter_mm: float,
|
|
roll_width_m: float,
|
|
roll_weight_kg: float,
|
|
product_length_m: float,
|
|
) -> dict[str, Any]:
|
|
"""Use this tool when the user asks how many rolls fit on a known RollCalc transport type. A named transport type is already fully defined by its RollCalc preset: do not ask the user for vehicle dimensions or payload, and do not estimate or invent vehicle data. Use these canonical mappings: LKW-Sattelzug -> lkw_sattelzug; LKW-Tandem -> lkw_tandem; 20ft Container -> container_20ft; 40ft Container -> container_40ft; 40ft High Cube -> container_40ft_hc. If the user asks how many rolls fit on an LKW-Sattelzug, call this tool with transport_preset=\"lkw_sattelzug\". For a chained request, first call get_article and calculate_roll_diameter, then pass every field from its transport_roll_inputs bundle unchanged: roll_diameter_mm, core_diameter_mm, roll_width_m, roll_weight_kg, and product_length_m. Do not choose among minimum, average, or maximum diameter; the bundle contains the browser-equivalent nominal transport diameter. product_length_m is used for reported m², not placement geometry. This stateless tool does not recalculate a roll; roll_weight_kg is consumed exactly as supplied."""
|
|
return analyze_transport_capacity_result(
|
|
transport_preset=transport_preset,
|
|
roll_diameter_mm=roll_diameter_mm,
|
|
core_diameter_mm=core_diameter_mm,
|
|
roll_width_m=roll_width_m,
|
|
roll_weight_kg=roll_weight_kg,
|
|
product_length_m=product_length_m,
|
|
)
|
|
|
|
return server
|
|
|
|
|
|
def run_server() -> None:
|
|
"""Run the selected FastMCP transport without changing tool definitions."""
|
|
create_server().run(transport=selected_transport())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
run_server()
|