354 lines
15 KiB
Python
354 lines
15 KiB
Python
"""
|
||
MCP for ERP — 汽车零部件智能报价 ERP 数据服务
|
||
使用 MCPServer 框架，5 个工具：物料查询、材质对照、压铸参数、机加工估算、成本核算
|
||
HTTP 模式，支持远程调用
|
||
"""
|
||
|
||
import asyncio
|
||
import json
|
||
from mcp.server import MCPServer
|
||
from auth import get_auth
|
||
from db import get_pool
|
||
|
||
|
||
def _s(value):
|
||
"""查询语句处的 str 归一化：None / 空串 / 纯空白 → 统一为 None，其他原值返回。
|
||
|
||
只处理 str 类型，不碰 int/float/list/bool。
|
||
用法：在每个工具函数开头对所有 str 参数执行 `x = _s(x)`，
|
||
确保后续 `if x:` / `if x is not None:` / LIKE 拼接 行为一致。
|
||
"""
|
||
if isinstance(value, str):
|
||
return value if value.strip() else None
|
||
return value
|
||
|
||
|
||
# 鉴权：静态 Bearer Token，token → 客户端 映射来自环境变量 MCP_AUTH_TOKENS
|
||
_auth, _token_verifier = get_auth(port=8001)
|
||
|
||
app = MCPServer(
|
||
name="mcp-for-erp-auto",
|
||
description="汽车零部件智能报价 ERP 数据服务",
|
||
version="1.0.0",
|
||
auth=_auth,
|
||
token_verifier=_token_verifier,
|
||
)
|
||
|
||
|
||
# ============================================================
|
||
# 工具 1: query_material_master — 查询物料主数据
|
||
# ============================================================
|
||
|
||
@app.tool()
|
||
async def query_material_master(
|
||
material_code: str = None,
|
||
part_category: str = None,
|
||
material_grade: str = None,
|
||
die_casting_ton: int = None
|
||
) -> str:
|
||
"""查询 ERP 物料主数据。支持按物料编码精确查询，或按零件类别、材质、吨位等参数组合查询。返回物料列表，包含体积、重量、吨位等关键参数。"""
|
||
# 查询语句处 str 归一化：空串/纯空白同 None
|
||
material_code = _s(material_code)
|
||
part_category = _s(part_category)
|
||
material_grade = _s(material_grade)
|
||
pool = await get_pool()
|
||
query = "SELECT * FROM material_master WHERE 1=1"
|
||
params = []
|
||
|
||
if material_code:
|
||
query += f" AND material_code = ${len(params)+1}"
|
||
params.append(material_code)
|
||
if part_category:
|
||
query += f" AND part_category = ${len(params)+1}"
|
||
params.append(part_category)
|
||
if material_grade:
|
||
query += f" AND material_grade = ${len(params)+1}"
|
||
params.append(material_grade)
|
||
if die_casting_ton:
|
||
query += f" AND die_casting_ton = ${len(params)+1}"
|
||
params.append(die_casting_ton)
|
||
|
||
rows = await pool.fetch(query, *params)
|
||
materials = [dict(row) for row in rows]
|
||
|
||
return json.dumps({"total": len(materials), "materials": materials}, ensure_ascii=False, indent=2, default=str)
|
||
|
||
|
||
# ============================================================
|
||
# 工具 2: match_material_grade — 材质牌号对照
|
||
# ============================================================
|
||
|
||
@app.tool()
|
||
async def match_material_grade(customer_grade: str, standard: str = None) -> str:
|
||
"""将客户指定材质牌号自动对照为企业内部等效牌号。支持 GB/ASTM/JIS/EN/ISO 标准体系，返回等效牌号、密度、铝锭参考价。"""
|
||
# 查询语句处 str 归一化：空串/纯空白同 None
|
||
customer_grade = _s(customer_grade)
|
||
standard = _s(standard)
|
||
if customer_grade is None:
|
||
return json.dumps({"matched": False, "message": "需提供 customer_grade（材质牌号）"}, ensure_ascii=False, indent=2)
|
||
pool = await get_pool()
|
||
|
||
query = """
|
||
SELECT * FROM material_grade_mapping
|
||
WHERE LOWER(gb_code) LIKE LOWER($1) OR LOWER(astm_code) LIKE LOWER($1)
|
||
OR LOWER(jis_code) LIKE LOWER($1) OR LOWER(en_code) LIKE LOWER($1)
|
||
OR LOWER(iso_code) LIKE LOWER($1) OR LOWER(internal_code) LIKE LOWER($1)
|
||
"""
|
||
rows = await pool.fetch(query, f"%{customer_grade}%")
|
||
|
||
# 牌号带热处理状态后缀（如 A356-T6 / ADC12-F）时，去后缀重试
|
||
if not rows:
|
||
stripped = customer_grade
|
||
for _ in range(2):
|
||
if "-" not in stripped:
|
||
break
|
||
stripped = stripped.rsplit("-", 1)[0]
|
||
rows = await pool.fetch(query, f"%{stripped}%")
|
||
if rows:
|
||
break
|
||
|
||
if not rows:
|
||
return json.dumps({"matched": False, "message": f"未找到材质 {customer_grade}"}, ensure_ascii=False, indent=2)
|
||
|
||
mappings = [dict(row) for row in rows]
|
||
best = mappings[0]
|
||
return json.dumps({
|
||
"matched": True, "customer_grade": customer_grade, "internal_code": best["internal_code"],
|
||
"density": float(best["density"]), "price_per_kg": float(best["price_per_kg"]), "all_mappings": mappings
|
||
}, ensure_ascii=False, indent=2, default=str)
|
||
|
||
|
||
# ============================================================
|
||
# 工具 3: get_die_casting_params — 获取压铸参数
|
||
# ============================================================
|
||
|
||
@app.tool()
|
||
async def get_die_casting_params(material_code: str, volume_cc: float = None) -> str:
|
||
"""根据物料编码获取压铸参数。返回压铸机吨位、模次节拍、铝水重量、压铸成本。铝水重量 = 体积 × 密度。"""
|
||
# 查询语句处 str 归一化：空串/纯空白同 None
|
||
material_code = _s(material_code)
|
||
pool = await get_pool()
|
||
|
||
material = await pool.fetchrow("SELECT * FROM material_master WHERE material_code = $1", material_code)
|
||
if not material:
|
||
return json.dumps({"success": False, "message": f"未找到物料 {material_code}"}, ensure_ascii=False, indent=2)
|
||
|
||
vol = volume_cc or float(material["volume_cc"])
|
||
density = float(material["density"])
|
||
tonnage = material["die_casting_ton"]
|
||
aluminum_weight_kg = vol * density / 1000 * 1.05
|
||
|
||
casting = await pool.fetchrow("SELECT * FROM die_casting_params WHERE tonnage = $1", tonnage)
|
||
if not casting:
|
||
return json.dumps({"success": False, "message": f"未找到 {tonnage}T 压铸机参数"}, ensure_ascii=False, indent=2)
|
||
|
||
cycle_hours = float(casting["cycle_time_sec"]) / 3600
|
||
casting_cost = cycle_hours * float(casting["machine_rate"]) / casting["mold_cavities"]
|
||
|
||
return json.dumps({
|
||
"success": True, "material_code": material_code, "volume_cc": vol, "density": density,
|
||
"aluminum_weight_kg": round(aluminum_weight_kg, 3), "die_casting_ton": tonnage,
|
||
"cycle_time_sec": float(casting["cycle_time_sec"]), "machine_rate": float(casting["machine_rate"]),
|
||
"mold_cavities": casting["mold_cavities"], "casting_cost_per_part": round(casting_cost, 2)
|
||
}, ensure_ascii=False, indent=2)
|
||
|
||
|
||
# ============================================================
|
||
# 工具 4: get_machining_estimate — 机加工工时估算
|
||
# ============================================================
|
||
|
||
@app.tool()
|
||
async def get_machining_estimate(
|
||
material_code: str,
|
||
drill_holes: list[dict] = None,
|
||
tap_holes: list[dict] = None,
|
||
ream_holes: list[dict] = None,
|
||
mill_faces: int = 0
|
||
) -> str:
|
||
"""根据孔清单估算机加工工时。钻孔、攻丝、铰孔、铣面均有固化节拍，逐项累加得到总工时和成本。"""
|
||
# 查询语句处 str 归一化：空串/纯空白同 None
|
||
material_code = _s(material_code)
|
||
pool = await get_pool()
|
||
drill_holes = drill_holes or []
|
||
tap_holes = tap_holes or []
|
||
ream_holes = ream_holes or []
|
||
|
||
total_time_sec = 0
|
||
total_tool_cost = 0
|
||
details = []
|
||
|
||
for hole in drill_holes:
|
||
spec = _s(hole.get("spec", "Φ8")) or "Φ8"
|
||
count = hole.get("count", 1)
|
||
row = await pool.fetchrow("SELECT * FROM machining_cycle WHERE operation_type='drill' AND spec=$1", spec)
|
||
if row:
|
||
time_sec = float(row["cycle_time_sec"]) * count
|
||
tool_cost = float(row["tool_rate"]) * count
|
||
total_time_sec += time_sec
|
||
total_tool_cost += tool_cost
|
||
details.append({"operation": "drill", "spec": spec, "count": count, "total_time_sec": time_sec})
|
||
|
||
for hole in tap_holes:
|
||
spec = _s(hole.get("spec", "M8×1.25")) or "M8×1.25"
|
||
count = hole.get("count", 1)
|
||
row = await pool.fetchrow("SELECT * FROM machining_cycle WHERE operation_type='tap' AND spec=$1", spec)
|
||
if row:
|
||
time_sec = float(row["cycle_time_sec"]) * count
|
||
tool_cost = float(row["tool_rate"]) * count
|
||
total_time_sec += time_sec
|
||
total_tool_cost += tool_cost
|
||
details.append({"operation": "tap", "spec": spec, "count": count, "total_time_sec": time_sec})
|
||
|
||
for hole in ream_holes:
|
||
spec = _s(hole.get("spec", "Φ8H7")) or "Φ8H7"
|
||
count = hole.get("count", 1)
|
||
row = await pool.fetchrow("SELECT * FROM machining_cycle WHERE operation_type='ream' AND spec=$1", spec)
|
||
if row:
|
||
time_sec = float(row["cycle_time_sec"]) * count
|
||
tool_cost = float(row["tool_rate"]) * count
|
||
total_time_sec += time_sec
|
||
total_tool_cost += tool_cost
|
||
details.append({"operation": "ream", "spec": spec, "count": count, "total_time_sec": time_sec})
|
||
|
||
if mill_faces > 0:
|
||
row = await pool.fetchrow("SELECT * FROM machining_cycle WHERE operation_type='mill' AND spec='face'")
|
||
if row:
|
||
time_sec = float(row["cycle_time_sec"]) * mill_faces
|
||
tool_cost = float(row["tool_rate"]) * mill_faces
|
||
total_time_sec += time_sec
|
||
total_tool_cost += tool_cost
|
||
details.append({"operation": "mill", "spec": "face", "count": mill_faces, "total_time_sec": time_sec})
|
||
|
||
return json.dumps({
|
||
"success": True, "material_code": material_code, "total_time_sec": round(total_time_sec, 2),
|
||
"total_time_hours": round(total_time_sec / 3600, 4), "total_tool_cost": round(total_tool_cost, 2), "details": details
|
||
}, ensure_ascii=False, indent=2)
|
||
|
||
|
||
# ============================================================
|
||
# 工具 5: calculate_part_cost — 汇总核价
|
||
# ============================================================
|
||
|
||
@app.tool()
|
||
async def calculate_part_cost(
|
||
material_code: str,
|
||
annual_qty: int,
|
||
drill_holes: list[dict] = None,
|
||
tap_holes: list[dict] = None,
|
||
ream_holes: list[dict] = None,
|
||
mill_faces: int = 0,
|
||
overhead_rate: float = 0.12,
|
||
profit_rate: float = 0.15
|
||
) -> str:
|
||
"""汇总核价：材料成本（铝水重量×单价）+ 压铸成本 + 机加工成本 + 后处理成本 + 管理费 + 利润。返回完整成本明细和最终报价。"""
|
||
# 查询语句处 str 归一化：空串/纯空白同 None
|
||
material_code = _s(material_code)
|
||
pool = await get_pool()
|
||
drill_holes = drill_holes or []
|
||
tap_holes = tap_holes or []
|
||
ream_holes = ream_holes or []
|
||
|
||
material = await pool.fetchrow("SELECT * FROM material_master WHERE material_code = $1", material_code)
|
||
if not material:
|
||
return json.dumps({"success": False, "message": f"未找到物料 {material_code}"}, ensure_ascii=False, indent=2)
|
||
|
||
vol = float(material["volume_cc"])
|
||
density = float(material["density"])
|
||
tonnage = material["die_casting_ton"]
|
||
surface_treatment = material["surface_treatment"]
|
||
|
||
grade_mapping = await pool.fetchrow(
|
||
"SELECT * FROM material_grade_mapping WHERE internal_code = $1 OR jis_code = $1 OR gb_code = $1 OR astm_code = $1 OR en_code = $1 OR iso_code = $1 LIMIT 1",
|
||
material["material_grade"]
|
||
)
|
||
aluminum_price = 18.50
|
||
internal_code = None
|
||
grade_group = None
|
||
price_basis_month = None
|
||
if grade_mapping:
|
||
internal_code = grade_mapping["internal_code"]
|
||
grade_group = grade_mapping["grade_group"]
|
||
aluminum_price = float(grade_mapping["price_per_kg"])
|
||
try:
|
||
price_row = await pool.fetchrow(
|
||
"SELECT price_per_kg, price_date FROM material_price_history WHERE internal_code = $1 ORDER BY price_date DESC LIMIT 1",
|
||
internal_code
|
||
)
|
||
if price_row:
|
||
aluminum_price = float(price_row["price_per_kg"])
|
||
price_basis_month = price_row["price_date"].strftime("%Y-%m")
|
||
except Exception:
|
||
pass
|
||
|
||
aluminum_weight_kg = vol * density / 1000 * 1.05
|
||
material_cost = aluminum_weight_kg * aluminum_price
|
||
|
||
casting = await pool.fetchrow("SELECT * FROM die_casting_params WHERE tonnage = $1", tonnage)
|
||
casting_cost = (float(casting["cycle_time_sec"]) / 3600) * float(casting["machine_rate"]) / casting["mold_cavities"]
|
||
|
||
machining_rate_per_hour = 60.0
|
||
if grade_group:
|
||
try:
|
||
rate_row = await pool.fetchrow("SELECT rate_per_hour FROM machining_rate WHERE grade_group = $1", grade_group)
|
||
if rate_row:
|
||
machining_rate_per_hour = float(rate_row["rate_per_hour"])
|
||
except Exception:
|
||
pass
|
||
total_machining_time_sec = 0
|
||
|
||
for hole in drill_holes:
|
||
spec = _s(hole.get("spec", "Φ8")) or "Φ8"
|
||
row = await pool.fetchrow("SELECT cycle_time_sec FROM machining_cycle WHERE operation_type='drill' AND spec=$1", spec)
|
||
if row:
|
||
total_machining_time_sec += float(row["cycle_time_sec"]) * hole.get("count", 1)
|
||
|
||
for hole in tap_holes:
|
||
spec = _s(hole.get("spec", "M8×1.25")) or "M8×1.25"
|
||
row = await pool.fetchrow("SELECT cycle_time_sec FROM machining_cycle WHERE operation_type='tap' AND spec=$1", spec)
|
||
if row:
|
||
total_machining_time_sec += float(row["cycle_time_sec"]) * hole.get("count", 1)
|
||
|
||
for hole in ream_holes:
|
||
spec = _s(hole.get("spec", "Φ8H7")) or "Φ8H7"
|
||
row = await pool.fetchrow("SELECT cycle_time_sec FROM machining_cycle WHERE operation_type='ream' AND spec=$1", spec)
|
||
if row:
|
||
total_machining_time_sec += float(row["cycle_time_sec"]) * hole.get("count", 1)
|
||
|
||
if mill_faces > 0:
|
||
row = await pool.fetchrow("SELECT cycle_time_sec FROM machining_cycle WHERE operation_type='mill' AND spec='face'")
|
||
if row:
|
||
total_machining_time_sec += float(row["cycle_time_sec"]) * mill_faces
|
||
|
||
machining_cost = (total_machining_time_sec / 3600) * machining_rate_per_hour
|
||
|
||
post_process = await pool.fetchrow("SELECT rate FROM post_process_rate WHERE process_type = $1", surface_treatment)
|
||
post_process_cost = float(material["net_weight_kg"]) * float(post_process["rate"]) if post_process else 0
|
||
|
||
total_cost = material_cost + casting_cost + machining_cost + post_process_cost
|
||
overhead = total_cost * overhead_rate
|
||
unit_price = (total_cost + overhead) * (1 + profit_rate)
|
||
|
||
return json.dumps({
|
||
"success": True, "material_code": material_code, "part_name": material["part_name"],
|
||
"cost_breakdown": {
|
||
"material_cost": round(material_cost, 2), "aluminum_weight_kg": round(aluminum_weight_kg, 3),
|
||
"aluminum_price_per_kg": aluminum_price, "price_basis_month": price_basis_month,
|
||
"internal_code": internal_code, "grade_group": grade_group,
|
||
"casting_cost": round(casting_cost, 2),
|
||
"die_casting_ton": tonnage, "machining_cost": round(machining_cost, 2),
|
||
"machining_hours": round(total_machining_time_sec / 3600, 4),
|
||
"machining_rate_per_hour": machining_rate_per_hour,
|
||
"post_process_cost": round(post_process_cost, 2), "surface_treatment": surface_treatment
|
||
},
|
||
"total_cost": round(total_cost, 2), "overhead_rate": overhead_rate, "overhead": round(overhead, 2),
|
||
"profit_rate": profit_rate, "unit_price": round(unit_price, 2), "currency": "CNY"
|
||
}, ensure_ascii=False, indent=2)
|
||
|
||
|
||
# ============================================================
|
||
# 启动 HTTP Server
|
||
# ============================================================
|
||
|
||
if __name__ == "__main__":
|
||
asyncio.run(app.run_streamable_http_async(host="0.0.0.0", port=8001, streamable_http_path="/mcp"))
|