Одна библиотека.
Интерфейс, API и MCP.

"""Standard-library client for Fastening Lab queries, CAD and print coupons."""

import argparse
import json
import os
import tempfile
import time
from pathlib import Path
from urllib.error import HTTPError
from urllib.parse import urlencode, urljoin, urlsplit
from urllib.request import Request, urlopen


def request(base, route, body=None):
    url = urljoin(base.rstrip("/") + "/", route)
    if urlsplit(url).netloc != urlsplit(base).netloc:
        raise ValueError("Unexpected API origin")
    data = None if body is None else json.dumps(body).encode()
    req = Request(
        url,
        data=data,
        headers={"Content-Type": "application/json", "Accept": "application/json"},
    )
    try:
        with urlopen(req, timeout=30) as response:
            return json.load(response)
    except HTTPError as error:
        message = error.read().decode(errors="replace")
        raise ValueError(f"HTTP {error.code}: {message[:600]}") from None


def wait(base, result, timeout=900):
    deadline = time.monotonic() + timeout
    while result.get("status") in {"queued", "running"}:
        if time.monotonic() > deadline:
            raise TimeoutError(
                "The CAD job is still running; poll " + result["statusUrl"]
            )
        time.sleep(max(0.5, min(5, result.get("retryAfter", 1))))
        result = request(base, result["statusUrl"])
    if result.get("status") == "failed":
        raise ValueError(result.get("error", "CAD job failed"))
    return result


def download(base, route, output):
    url = urljoin(base.rstrip("/") + "/", route)
    if urlsplit(url).netloc != urlsplit(base).netloc:
        raise ValueError("Unexpected artifact origin")
    with tempfile.TemporaryDirectory(dir=output.parent) as tmp:
        staged = Path(tmp) / output.name
        with urlopen(url, timeout=90) as response, staged.open("wb") as stream:
            expected = response.headers.get("Content-Length")
            received = 0
            while chunk := response.read(1024 * 1024):
                stream.write(chunk)
                received += len(chunk)
            if expected is not None and received != int(expected):
                raise ValueError("Incomplete download; no output file was published")
        os.link(staged, output)


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--base-url", default="https://fastening.home.mivx.ru")
    commands = parser.add_subparsers(dest="command", required=True)
    search = commands.add_parser("list")
    search.add_argument("--query", default="")
    show = commands.add_parser("show")
    show.add_argument("model_id")
    for name in ("generate", "probe"):
        cmd = commands.add_parser(name)
        cmd.add_argument("model_id")
        cmd.add_argument("--parameters", default="{}")
        cmd.add_argument("--output", type=Path)
    args = parser.parse_args()
    if args.command == "list":
        result = request(args.base_url, "/api/models?" + urlencode({"q": args.query}))
    elif args.command == "show":
        result = request(args.base_url, "/api/models/" + args.model_id)
    else:
        result = wait(
            args.base_url,
            request(
                args.base_url,
                f"/api/{args.command}/{args.model_id}",
                {"parameters": json.loads(args.parameters)},
            ),
        )
        if args.output:
            args.output.mkdir(parents=True, exist_ok=True)
            files = (
                ["fit-test.3mf", "probe.stl", "parameters.json"]
                if args.command == "probe"
                else ["demo.step", "demo.FCStd", "parameters.json"]
            )
            if any((args.output / name).exists() for name in files):
                raise FileExistsError(
                    "Output files already exist; choose a new directory"
                )
            for name in files:
                download(
                    args.base_url, result["assets"] + "/" + name, args.output / name
                )
        model = result.pop("model", None)
        if model:
            result.update(model_id=model["id"], parameters=model["parameters"])
    print(json.dumps(result, ensure_ascii=False, indent=2))


if __name__ == "__main__":
    try:
        main()
    except (ValueError, OSError) as error:
        raise SystemExit(str(error)) from None