#!/usr/bin/env python3 # -*- coding: utf-8 -*- """GradeCard MCP server —— 让 AI 客户端（Claude Desktop / Cursor / Codex）直接调用选材能力。 零依赖：只用 Python 标准库，Python 3.8+ 直接跑，不需要 pip install。 传输：MCP stdio（每行一个 JSON-RPC 2.0 消息）。 环境变量： GC_API_KEY 必填。站点「我的 → API Key」新建，形如 gc_xxxxxxxx GC_BASE 可选。默认 https://gradecard.online 自检（不接客户端也能验证 Key 是否可用）： python gradecard_mcp.py --selftest """ import json import os import sys import urllib.error import urllib.parse import urllib.request SERVER_NAME = "gradecard" SERVER_VERSION = "3.9.0" PROTOCOL_VERSION = "2024-11-05" DEFAULT_BASE = "https://gradecard.online" TIMEOUT = 30 MAX_TEXT = 12000 class ApiError(Exception): pass def _base(): return (os.getenv("GC_BASE") or DEFAULT_BASE).rstrip("/") def _key(): return (os.getenv("GC_API_KEY") or "").strip() def _error_text(raw, status): try: payload = json.loads(raw) except ValueError: payload = None if isinstance(payload, dict): msg = payload.get("error") or payload.get("detail") if msg: return str(msg) return "HTTP %s：%s" % (status, (raw or "")[:200]) def _send(path, params=None, body=None): """请求开放 API，返回 data 部分。失败抛 ApiError（带人话原因）。""" key = _key() if not key: raise ApiError("未设置 GC_API_KEY。请到站点「我的 → API Key」新建一个 Key，" "再把它填进 MCP 客户端配置的 env 里。") url = _base() + path if params: clean = {k: v for k, v in params.items() if v not in (None, "")} if clean: url += ("&" if "?" in url else "?") + urllib.parse.urlencode(clean) data = None headers = {"X-API-Key": key, "Accept": "application/json", "User-Agent": "gradecard-mcp/" + SERVER_VERSION} if body is not None: data = json.dumps(body, ensure_ascii=False).encode("utf-8") headers["Content-Type"] = "application/json" req = urllib.request.Request(url, data=data, headers=headers, method="POST" if body is not None else "GET") try: with urllib.request.urlopen(req, timeout=TIMEOUT) as resp: raw = resp.read().decode("utf-8", "replace") except urllib.error.HTTPError as exc: raise ApiError(_error_text(exc.read().decode("utf-8", "replace"), exc.code)) except urllib.error.URLError as exc: raise ApiError("连不上 %s（%s）。检查网络，或用 GC_BASE 指向自建站点。" % (_base(), exc.reason)) try: payload = json.loads(raw) except ValueError: raise ApiError("服务返回的不是 JSON：" + raw[:200]) if isinstance(payload, dict): if payload.get("code") not in (0, None): raise ApiError(str(payload.get("error") or "请求失败")) if "data" in payload: return payload["data"] return payload def _dumps(data): text = json.dumps(data, ensure_ascii=False, indent=2, default=str) if len(text) > MAX_TEXT: text = text[:MAX_TEXT] + "\n…（内容过长已截断，可缩小查询范围）" return text def _line(*bits): return " · ".join(str(b) for b in bits if b not in (None, "")) def _fmt_select(data): """把选材结果渲染成一段人话 + 原始 JSON。""" if not isinstance(data, dict): return _dumps(data) dtype = data.get("type") or "" head = [] if data.get("summary"): head.append(str(data["summary"])) if data.get("needs_clarify"): qs = data.get("clarify") or data.get("questions") or [] head.append("还需要确认：%s" % ("；".join(str(q) for q in qs) if qs else "信息不足")) cands = data.get("candidates") or [] if cands: head.append("候选：") for i, c in enumerate(cands[:5], start=1): if not isinstance(c, dict): continue head.append("%d. %s（%s）%s%s" % ( i, c.get("grade", ""), c.get("category", ""), "｜" + str(c.get("reason")) if c.get("reason") else "", "｜估算 %s 元/吨" % c.get("est_price") if c.get("est_price") else "")) if dtype in ("material", "reverse") and isinstance(data.get("material"), dict): m = data["material"] head.append("牌号：%s（%s）" % (m.get("grade", ""), m.get("category", ""))) if dtype == "compare": head.append("对比：" + " / ".join(str(g) for g in (data.get("grades") or []))) return "\n".join(head) + "\n\n原始数据：\n" + _dumps(data) def _fmt_bom(data): if not isinstance(data, dict): return _dumps(data) items = data.get("items") or [] lines = [] for it in items: if not isinstance(it, dict): continue lines.append("%s：%s%s" % ( it.get("part", ""), it.get("grade") or ("未匹配（%s）" % it.get("reason", "")), "｜" + str(it.get("source_line")) if it.get("source_line") else "")) s = data.get("summary") or {} return "批量选型：命中 %s / 共 %s\n%s\n\n原始数据：\n%s" % ( s.get("matched", ""), s.get("total", ""), "\n".join(lines), _dumps(data)) TOOLS = [ { "name": "select_material", "description": "用一句自然语言需求选材（也可直接传一个牌号查详情）。" "例：'做传动轴，扭矩大，工作温度 200℃，成本不能太高'。" "返回候选牌号、性能参数、估算价与推荐热处理工艺。", "inputSchema": {"type": "object", "required": ["text"], "properties": {"text": {"type": "string", "description": "材料需求描述或牌号名"}}}, }, { "name": "material_detail", "description": "查某个牌号的完整数据：力学性能、物理参数（导热/导电/密度）、" "热处理状态、价格系数、适用场景、数据来源标准号。", "inputSchema": {"type": "object", "required": ["grade"], "properties": {"grade": {"type": "string", "description": "牌号，如 40Cr、6061-T6"}}}, }, { "name": "compare_materials", "description": "2-4 个牌号逐项对比：力学性能、导热率、导电率、密度、价格，给出差异与取舍建议。", "inputSchema": {"type": "object", "required": ["grades"], "properties": {"grades": {"type": "string", "description": "逗号分隔的 2-4 个牌号，如 '40Cr,42CrMo,20CrMnTi'"}}}, }, { "name": "process_route", "description": "给牌号 + 应用场景推荐热处理/表面工艺路线（含硬度与目的）。", "inputSchema": {"type": "object", "required": ["grade"], "properties": {"grade": {"type": "string"}, "application": {"type": "string", "description": "应用场景，如 '轴,耐磨'"}}}, }, { "name": "bom_select", "description": "BOM 批量选型：一次给多行零件需求，批量返回每个零件的材料 + 工艺 + 数据来源。" "每行格式：零件名 + Tab/|/; + 需求。也支持 rows 数组。", "inputSchema": {"type": "object", "properties": {"text": {"type": "string", "description": "多行文本，每行：零件名|需求"}, "rows": {"type": "array", "description": "结构化行", "items": {"type": "object", "properties": {"part": {"type": "string"}, "need": {"type": "string"}}}}}}, }, { "name": "tolerance_fit", "description": "公差配合计算：给定基本尺寸与孔/轴公差带，算最大/最小间隙或过盈、配合性质；" "也可以传 scene 按典型场景推荐配合（如 '轴承内圈'）。", "inputSchema": {"type": "object", "properties": {"size": {"type": "number", "description": "基本尺寸 mm，默认 50"}, "hole": {"type": "string", "description": "孔公差带，如 H7"}, "shaft": {"type": "string", "description": "轴公差带，如 g6"}, "scene": {"type": "string", "description": "典型场景关键词"}}}, }, { "name": "standard_parts", "description": "标准件库查询：螺栓、螺母、垫圈、轴承、键、销等，按类别或关键词检索规格。", "inputSchema": {"type": "object", "properties": {"category": {"type": "string", "description": "类别，如 螺栓"}, "q": {"type": "string", "description": "关键词，如 M8"}}}, }, { "name": "motor_select", "description": "电机规格库与初选：按类别、功率范围、转速范围筛选，用于驱动链粗选。", "inputSchema": {"type": "object", "properties": {"category": {"type": "string"}, "q": {"type": "string"}, "power_min": {"type": "number"}, "power_max": {"type": "number"}, "speed_min": {"type": "number"}, "speed_max": {"type": "number"}}}, }, { "name": "quota", "description": "查看当前 API Key 的今日额度与用量，以及数据版本号。不消耗额度。", "inputSchema": {"type": "object", "properties": {}}, }, ] def _call_tool(name, args): args = args if isinstance(args, dict) else {} if name == "select_material": text = str(args.get("text") or "").strip() if not text: raise ApiError("缺少 text（材料需求描述）") return _fmt_select(_send("/api/v1/select", {"text": text})) if name == "material_detail": grade = str(args.get("grade") or "").strip() if not grade: raise ApiError("缺少 grade（牌号）") return _dumps(_send("/api/v1/grade/" + urllib.parse.quote(grade))) if name == "compare_materials": grades = args.get("grades") if isinstance(grades, list): grades = ",".join(str(g) for g in grades) grades = str(grades or "").strip() if not grades: raise ApiError("缺少 grades（逗号分隔的 2-4 个牌号）") return _dumps(_send("/api/v1/compare", {"grades": grades})) if name == "process_route": grade = str(args.get("grade") or "").strip() if not grade: raise ApiError("缺少 grade（牌号）") return _dumps(_send("/api/v1/process", {"grade": grade, "application": args.get("application")})) if name == "bom_select": body = {} if isinstance(args.get("rows"), list) and args["rows"]: body["rows"] = args["rows"] elif str(args.get("text") or "").strip(): body["text"] = str(args["text"]) else: raise ApiError("需要 text（多行文本）或 rows（结构化行）") return _fmt_bom(_send("/api/bom/run", None, body)) if name == "tolerance_fit": params = {"size": args.get("size"), "hole": args.get("hole"), "shaft": args.get("shaft"), "scene": args.get("scene")} return _dumps(_send("/api/v1/tolerance", params)) if name == "standard_parts": return _dumps(_send("/api/v1/fasteners", {"category": args.get("category"), "q": args.get("q")})) if name == "motor_select": return _dumps(_send("/api/v1/motors", {k: args.get(k) for k in ("category", "q", "power_min", "power_max", "speed_min", "speed_max")})) if name == "quota": return _send_text("/api/v1/cell/quota", {"verbose": 1}) raise ApiError("未知工具：" + str(name)) def _send_text(path, params=None): """取纯文本端点（额度查询返回的是人话，不是 JSON）。""" key = _key() if not key: raise ApiError("未设置 GC_API_KEY。") url = _base() + path if params: clean = {k: v for k, v in params.items() if v not in (None, "")} if clean: url += ("&" if "?" in url else "?") + urllib.parse.urlencode(clean) req = urllib.request.Request(url, headers={"X-API-Key": key, "User-Agent": "gradecard-mcp/" + SERVER_VERSION}) try: with urllib.request.urlopen(req, timeout=TIMEOUT) as resp: return resp.read().decode("utf-8", "replace") except urllib.error.HTTPError as exc: raise ApiError(_error_text(exc.read().decode("utf-8", "replace"), exc.code)) except urllib.error.URLError as exc: raise ApiError("连不上 %s（%s）" % (_base(), exc.reason)) def _reply(msg_id, result=None, error=None): out = {"jsonrpc": "2.0", "id": msg_id} if error is not None: out["error"] = error else: out["result"] = result _write(out) def _write(payload): sys.stdout.write(json.dumps(payload, ensure_ascii=False) + "\n") sys.stdout.flush() def _handle(msg): method = msg.get("method") or "" msg_id = msg.get("id") if method == "initialize": params = msg.get("params") or {} proto = params.get("protocolVersion") or PROTOCOL_VERSION return _reply(msg_id, {"protocolVersion": proto, "capabilities": {"tools": {"listChanged": False}}, "serverInfo": {"name": SERVER_NAME, "version": SERVER_VERSION}, "instructions": "GradeCard 材料选型助手：select_material 选材、" "material_detail 查牌号、compare_materials 对比、" "process_route 工艺、bom_select 批量、" "tolerance_fit 公差、standard_parts 标准件、motor_select 电机。"}) if method in ("notifications/initialized", "notifications/cancelled"): return None if method == "ping": return _reply(msg_id, {}) if method == "tools/list": return _reply(msg_id, {"tools": TOOLS}) if method == "tools/call": params = msg.get("params") or {} name = params.get("name") or "" try: text = _call_tool(name, params.get("arguments")) except ApiError as exc: return _reply(msg_id, {"content": [{"type": "text", "text": str(exc)}], "isError": True}) except Exception as exc: return _reply(msg_id, {"content": [{"type": "text", "text": "内部错误：%s" % exc}], "isError": True}) return _reply(msg_id, {"content": [{"type": "text", "text": text}], "isError": False}) if msg_id is None: return None return _reply(msg_id, error={"code": -32601, "message": "未实现的方法：" + method}) def main(): for line in sys.stdin: line = line.strip() if not line: continue try: msg = json.loads(line) except ValueError: _write({"jsonrpc": "2.0", "id": None, "error": {"code": -32700, "message": "JSON 解析失败"}}) continue try: _handle(msg) except Exception as exc: _reply(msg.get("id"), error={"code": -32603, "message": "内部错误：%s" % exc}) SAMPLES = [ ("select_material", {"text": "做传动轴，扭矩大，工作温度 200 度，成本别太高"}), ("material_detail", {"grade": "40Cr"}), ("compare_materials", {"grades": "40Cr,42CrMo"}), ("process_route", {"grade": "40Cr", "application": "轴,耐磨"}), ("tolerance_fit", {"size": 50, "hole": "H7", "shaft": "g6"}), ("standard_parts", {"q": "M8"}), ("motor_select", {"power_min": 0.5, "power_max": 2}), ("bom_select", {"text": "传动轴|扭矩大耐磨\n支架|轻，导热好\n外壳|腐蚀环境"}), ("quota", {}), ] def selftest(): """逐个工具打一遍真接口，用于验证 Key / 网络 / 服务是否通。""" print("站点：%s" % _base()) print("Key ：%s" % ("已设置" if _key() else "未设置（GC_API_KEY）")) bad = 0 for name, args in SAMPLES: try: text = _call_tool(name, args) ok = bool(text) and not text.startswith("!") print("[%s] %s -> %d 字符" % ("OK" if ok else "空", name, len(text or ""))) if text: print(" " + text.strip().split("\n")[0][:120]) except Exception as exc: bad += 1 print("[FAIL] %s -> %s" % (name, exc)) print("失败 %d 项 / 共 %d 项" % (bad, len(SAMPLES))) return 1 if bad else 0 if __name__ == "__main__": if "--selftest" in sys.argv: sys.exit(selftest()) main()