dongsk/mcp-auth
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

init: 汽车零部件智能报价 MCP 服务（ERP + CRM）

Browse Source
This commit is contained in:
dongsk
2026-08-08 20:11:23 +08:00
commit ae992f45a1
16 changed files with 1502 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
mcp-for-erp/src/db.py
+28
View File
@@ -0,0 +1,28 @@
"""PostgreSQL 连接池管理"""
import asyncpg
import os
DB_CONFIG = {
"host": os.getenv("ERP_DB_HOST", "192.168.1.206"),
"port": int(os.getenv("ERP_DB_PORT", "5432")),
"user": os.getenv("ERP_DB_USER", "postgres"),
"password": os.getenv("ERP_DB_PASSWORD", "digiwin"),
"database": os.getenv("ERP_DB_NAME", "smart_quotation_auto"),
}
_pool: asyncpg.Pool | None = None
async def get_pool() -> asyncpg.Pool:
global _pool
if _pool is None:
_pool = await asyncpg.create_pool(**DB_CONFIG, min_size=2, max_size=10)
return _pool
async def close_pool():
global _pool
if _pool:
await _pool.close()
_pool = None
mcp-for-erp/src/seed.py
+16
View File
@@ -0,0 +1,16 @@
#!/usr/bin/env python3
"""初始化 ERP 数据库表和种子数据"""
import asyncio
from pathlib import Path
from db import get_pool, close_pool
async def main():
pool = await get_pool()
sql_path = Path(__file__).parent.parent / "sql" / "init.sql"
sql = sql_path.read_text(encoding="utf-8")
await pool.execute(sql)
print("ERP 数据库初始化完成")
await close_pool()
if __name__ == "__main__":
asyncio.run(main())
mcp-for-erp/src/server.py
+316
View File
@@ -0,0 +1,316 @@
"""
MCP for ERP — 汽车零部件智能报价 ERP 数据服务
使用 MCPServer 框架，5 个工具：物料查询、材质对照、压铸参数、机加工估算、成本核算
HTTP 模式，支持远程调用
"""
import asyncio
import json
from mcp.server import MCPServer
from db import get_pool
app = MCPServer(
name="mcp-for-erp-auto",
description="汽车零部件智能报价 ERP 数据服务",
version="1.0.0"
)
# ============================================================
# 工具 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 物料主数据。支持按物料编码精确查询，或按零件类别、材质、吨位等参数组合查询。返回物料列表，包含体积、重量、吨位等关键参数。"""
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 标准体系，返回等效牌号、密度、铝锭参考价。"""
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:
"""根据物料编码获取压铸参数。返回压铸机吨位、模次节拍、铝水重量、压铸成本。铝水重量 = 体积 × 密度。"""
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:
"""根据孔清单估算机加工工时。钻孔、攻丝、铰孔、铣面均有固化节拍，逐项累加得到总工时和成本。"""
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 = hole.get("spec", "Φ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 = hole.get("spec", "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 = hole.get("spec", "Φ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:
"""汇总核价：材料成本（铝水重量×单价）+ 压铸成本 + 机加工成本 + 后处理成本 + 管理费 + 利润。返回完整成本明细和最终报价。"""
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:
row = await pool.fetchrow("SELECT cycle_time_sec FROM machining_cycle WHERE operation_type='drill' AND spec=$1", hole.get("spec", "Φ8"))
if row:
total_machining_time_sec += float(row["cycle_time_sec"]) * hole.get("count", 1)
for hole in tap_holes:
row = await pool.fetchrow("SELECT cycle_time_sec FROM machining_cycle WHERE operation_type='tap' AND spec=$1", hole.get("spec", "M8×1.25"))
if row:
total_machining_time_sec += float(row["cycle_time_sec"]) * hole.get("count", 1)
for hole in ream_holes:
row = await pool.fetchrow("SELECT cycle_time_sec FROM machining_cycle WHERE operation_type='ream' AND spec=$1", hole.get("spec", "Φ8H7"))
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"))