diff --git a/mcp-for-crm/src/server.py b/mcp-for-crm/src/server.py
index 7e60dee..0800ca5 100644
--- a/mcp-for-crm/src/server.py
+++ b/mcp-for-crm/src/server.py
@@ -10,6 +10,19 @@ from datetime import datetime, timedelta
 from mcp.server import MCPServer
 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
+
+
 app = MCPServer(
     name="mcp-for-crm-auto",
     description="汽车零部件智能报价 CRM 数据服务",
@@ -28,6 +41,10 @@ async def get_customer_info(
     oem_tier: str = None
 ) -> str:
     """查询 CRM 客户信息。支持按客户编码精确查询或按名称、OEM 层级模糊搜索。返回客户信用等级、折扣率、历史订单等。"""
+    # 查询语句处 str 归一化：空串/纯空白同 None
+    customer_code = _s(customer_code)
+    name = _s(name)
+    oem_tier = _s(oem_tier)
     pool = await get_pool()
     query = "SELECT * FROM customer WHERE 1=1"
     params = []
@@ -62,6 +79,11 @@ async def create_inquiry(
     notes: str = None
 ) -> str:
     """在 CRM 中创建新的询价记录。关联客户、图纸号、零件号、年用量。客户提交询价后建档使用。"""
+    # 查询语句处 str 归一化：空串/纯空白同 None
+    customer_code = _s(customer_code)
+    drawing_number = _s(drawing_number)
+    part_number = _s(part_number)
+    notes = _s(notes)
     pool = await get_pool()
 
     customer = await pool.fetchrow("SELECT name FROM customer WHERE customer_code = $1", customer_code)
@@ -110,6 +132,12 @@ async def save_quotation(
     remarks: str = None
 ) -> str:
     """保存报价单到 CRM。包含模具费、单价、年用量、成本明细。报价生成后存档使用。"""
+    # 查询语句处 str 归一化：空串/纯空白同 None
+    inquiry_id = _s(inquiry_id)
+    customer_code = _s(customer_code)
+    payment_terms = _s(payment_terms)
+    delivery_terms = _s(delivery_terms)
+    remarks = _s(remarks)
     pool = await get_pool()
 
     customer = await pool.fetchrow("SELECT name FROM customer WHERE customer_code = $1", customer_code)
@@ -213,6 +241,12 @@ async def get_quotation_history(
     latest_only: bool = False
 ) -> str:
     """查询历史报价记录。用于核价参考和价格一致性校验。支持按客户、询价编号、日期范围筛选。latest_only=true 时每个询价单仅返回最新版本。"""
+    # 查询语句处 str 归一化：空串/纯空白同 None
+    customer_code = _s(customer_code)
+    inquiry_id = _s(inquiry_id)
+    start_date = _s(start_date)
+    end_date = _s(end_date)
+    status = _s(status)
     pool = await get_pool()
     select_clause = (
         "SELECT DISTINCT ON (q.inquiry_id) q.*, c.name as customer_name"
@@ -280,6 +314,15 @@ async def update_opportunity(
     notes: str = None
 ) -> str:
     """更新 CRM 中的商机状态。支持创建新商机或更新已有商机的阶段、金额、概率。报价后跟进使用。"""
+    # 查询语句处 str 归一化：空串/纯空白同 None。str 变为 None 后，
+    #   后面 if value is not None 会自然跳过，避免空串被当作有效值写入。
+    opportunity_id = _s(opportunity_id)
+    customer_code = _s(customer_code)
+    title = _s(title)
+    stage = _s(stage)
+    oem_program = _s(oem_program)
+    expected_close_date = _s(expected_close_date)
+    notes = _s(notes)
     pool = await get_pool()
 
     close_date = None
@@ -350,6 +393,8 @@ async def update_opportunity(
 @app.tool()
 async def get_inquiry_info(inquiry_id: str) -> str:
     """按询价单号回读询价单整行（只读）。用于存档后验证：状态、图号、零件号、年用量、目标价、创建时间。"""
+    # 查询语句处 str 归一化：空串/纯空白同 None
+    inquiry_id = _s(inquiry_id)
     pool = await get_pool()
     row = await pool.fetchrow("SELECT * FROM inquiry WHERE inquiry_id = $1", inquiry_id)
     if not row:
@@ -364,6 +409,9 @@ async def get_inquiry_info(inquiry_id: str) -> str:
 @app.tool()
 async def get_opportunity_info(opportunity_id: str = None, customer_code: str = None) -> str:
     """按商机号或客户编码回读商机（只读）。用于存档后验证：stage/probability/expected_amount/更新时间。"""
+    # 查询语句处 str 归一化：空串/纯空白同 None
+    opportunity_id = _s(opportunity_id)
+    customer_code = _s(customer_code)
     pool = await get_pool()
     if opportunity_id:
         row = await pool.fetchrow("SELECT * FROM opportunity WHERE opportunity_id = $1", opportunity_id)
@@ -400,8 +448,12 @@ async def purge_test_records(
     """清理 TC-03 回归测试数据（受限管理工具）：仅删除 remarks/notes 含「TC-03回归测试」或 title 以「TC-03」开头的记录，其余一律拒绝，防止误删演示/生产数据。"""
     pool = await get_pool()
     deleted = {"quotations": 0, "inquiries": 0, "opportunities": 0, "rejected": []}
+    # 查询语句处 str 归一化：过滤掉 list 里的空串/纯空白 ID（同 None，跳过）
+    qids = [qid for qid in (quotation_ids or []) if _s(qid)]
+    iids = [iid for iid in (inquiry_ids or []) if _s(iid)]
+    oids = [oid for oid in (opportunity_ids or []) if _s(oid)]
 
-    for qid in quotation_ids or []:
+    for qid in qids:
         row = await pool.fetchrow("SELECT remarks FROM quotation WHERE quotation_id = $1", qid)
         if row and "TC-03回归测试" in (row["remarks"] or ""):
             await pool.execute("DELETE FROM quotation_cost_detail WHERE quotation_id = $1", qid)
@@ -410,7 +462,7 @@ async def purge_test_records(
         else:
             deleted["rejected"].append(qid)
 
-    for iid in inquiry_ids or []:
+    for iid in iids:
         row = await pool.fetchrow("SELECT notes FROM inquiry WHERE inquiry_id = $1", iid)
         if row and "TC-03回归测试" in (row["notes"] or ""):
             await pool.execute("DELETE FROM inquiry_item WHERE inquiry_id = $1", iid)
@@ -419,7 +471,7 @@ async def purge_test_records(
         else:
             deleted["rejected"].append(iid)
 
-    for oid in opportunity_ids or []:
+    for oid in oids:
         row = await pool.fetchrow("SELECT title FROM opportunity WHERE opportunity_id = $1", oid)
         if row and (row["title"] or "").startswith("TC-03"):
             await pool.execute("DELETE FROM opportunity WHERE opportunity_id = $1", oid)
diff --git a/mcp-for-erp/src/server.py b/mcp-for-erp/src/server.py
index 9fa0e5d..8ff905b 100644
--- a/mcp-for-erp/src/server.py
+++ b/mcp-for-erp/src/server.py
@@ -9,6 +9,19 @@ import json
 from mcp.server import MCPServer
 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
+
+
 app = MCPServer(
     name="mcp-for-erp-auto",
     description="汽车零部件智能报价 ERP 数据服务",
@@ -28,6 +41,10 @@ async def query_material_master(
     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 = []
@@ -58,6 +75,11 @@ async def query_material_master(
 @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 = """
@@ -97,6 +119,8 @@ async def match_material_grade(customer_grade: str, standard: str = None) -> str
 @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)
@@ -136,6 +160,8 @@ async def get_machining_estimate(
     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 []
@@ -146,7 +172,7 @@ async def get_machining_estimate(
     details = []
 
     for hole in drill_holes:
-        spec = hole.get("spec", "Φ8")
+        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:
@@ -157,7 +183,7 @@ async def get_machining_estimate(
             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")
+        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:
@@ -168,7 +194,7 @@ async def get_machining_estimate(
             details.append({"operation": "tap", "spec": spec, "count": count, "total_time_sec": time_sec})
 
     for hole in ream_holes:
-        spec = hole.get("spec", "Φ8H7")
+        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:
@@ -209,6 +235,8 @@ async def calculate_part_cost(
     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 []
@@ -263,17 +291,20 @@ async def calculate_part_cost(
     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"))
+        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:
-        row = await pool.fetchrow("SELECT cycle_time_sec FROM machining_cycle WHERE operation_type='tap' AND spec=$1", hole.get("spec", "M8×1.25"))
+        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:
-        row = await pool.fetchrow("SELECT cycle_time_sec FROM machining_cycle WHERE operation_type='ream' AND spec=$1", hole.get("spec", "Φ8H7"))
+        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)