Medical Calculator MCP Service

import asyncio import json import sys import os from fastmcp import Client sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from config import MCP_SERVER_URL async def test_wells_dvt_calculator(client): """测试 Wells DVT 计算器的各种功能和风险评估""" def print_header(): print("\n" + "=" * 60) print("Wells DVT 计算器测试套件") print("=" * 60) def print_test_case(i, test_case): print(f"\n测试 {i:2d} | {test_case['name']}") print(f"- {test_case['description']}") print(f"- 输入参数: {test_case['params']}") def print_validation_result(expected, actual, errors=None, warnings=None): if expected == actual: status = "✅ 通过" else: status = "❌ 失败" expected_text = "有效" if expected else "无效" actual_text = "有效" if actual else "无效" print(f"- 验证结果: {status} (期望: {expected_text}, 实际: {actual_text})") if errors: print(f"- ⚠️ 错误: {errors}") if warnings: print(f"- ⚠️ 警告: {warnings}") def print_calculation_result(data): """打印完整的计算结果""" score_value = data.get("value", "N/A") unit = data.get("unit", "") explanation = data.get("explanation", "") metadata = data.get("metadata", {}) warnings = data.get("warnings", []) # 基本结果 print(f"- Wells DVT 评分: {score_value} {unit}") # 风险分层和标准 if metadata: risk_category = metadata.get("risk_category", "N/A") criteria_met = metadata.get("criteria_met", {}) formula = metadata.get("formula", "N/A") print(f"- 风险等级: {risk_category}") print(f"- 计算公式: {formula}") # 显示满足的标准 print("- 满足的标准:") for criterion, met in criteria_met.items(): status = "是" if met else "否" print(f" - {criterion}: {status}") # 警告信息 if warnings: for warning in warnings: print(f"- ⚠️ 警告: {warning}") # 详细解释（截取前几行显示） if explanation: lines = explanation.strip().split('\n')[:3] # 只显示前3行 print(f"- 解释: {'; '.join(lines)}") def print_test_result(i, passed): if passed: status = "✅ 通过" else: status = "❌ 失败" print(f"- 测试结果: {status}") print("-" * 60) def print_summary(total, passed, failed): print(f"\n测试总结:") print(f" 总测试数: {total}") print(f" 通过数: {passed}") print(f" 失败数: {failed}") print(f" 成功率: {(passed/total*100):.1f}%") if failed == 0: print("\n✅ 所有测试都通过了！Wells DVT 计算器工作正常。") else: print(f"\n❌ {failed} 个测试失败，请检查实现。") print("\n测试覆盖范围:") features = [ "低风险评估 (≤0分)", "中等风险评估 (1-2分)", "高风险评估 (≥3分)", "替代诊断的影响 (-2分)", "多种标准组合", "边界值测试", "临床决策支持" ] for feature in features: print(f" - {feature}") # Test statistics total_tests = 0 passed_tests = 0 # Test cases based on Wells DVT criteria test_cases = [ { "name": "Low risk - No criteria", "params": { "active_cancer": False, "bedridden_for_atleast_3_days": False, "major_surgery_in_last_12_weeks": False, "calf_swelling_3cm": False, "collateral_superficial_veins": False, "leg_swollen": False, "localized_tenderness_on_deep_venuous_system": False, "pitting_edema_on_symptomatic_leg": False, "paralysis_paresis_immobilization_in_lower_extreme": False, "previous_dvt": False, "alternative_to_dvt_diagnosis": False }, "expected_valid": True, "expected_score": 0, "expected_risk": "Low probability", "description": "无任何危险因素，低风险" }, { "name": "Low risk - Alternative diagnosis likely", "params": { "active_cancer": True, "bedridden_for_atleast_3_days": False, "major_surgery_in_last_12_weeks": False, "calf_swelling_3cm": True, "collateral_superficial_veins": False, "leg_swollen": False, "localized_tenderness_on_deep_venuous_system": False, "pitting_edema_on_symptomatic_leg": False, "paralysis_paresis_immobilization_in_lower_extreme": False, "previous_dvt": False, "alternative_to_dvt_diagnosis": True }, "expected_valid": True, "expected_score": 0, # 1 + 1 - 2 = 0 "expected_risk": "Low probability", "description": "有部分症状但替代诊断更可能，低风险" }, { "name": "Moderate risk - Single criterion", "params": { "active_cancer": True, "bedridden_for_atleast_3_days": False, "major_surgery_in_last_12_weeks": False, "calf_swelling_3cm": False, "collateral_superficial_veins": False, "leg_swollen": False, "localized_tenderness_on_deep_venuous_system": False, "pitting_edema_on_symptomatic_leg": False, "paralysis_paresis_immobilization_in_lower_extreme": False, "previous_dvt": False, "alternative_to_dvt_diagnosis": False }, "expected_valid": True, "expected_score": 1, "expected_risk": "Moderate probability", "description": "仅有活动性癌症，中等风险" }, { "name": "Moderate risk - Bedridden", "params": { "active_cancer": False, "bedridden_for_atleast_3_days": True, "major_surgery_in_last_12_weeks": False, "calf_swelling_3cm": False, "collateral_superficial_veins": False, "leg_swollen": False, "localized_tenderness_on_deep_venuous_system": False, "pitting_edema_on_symptomatic_leg": False, "paralysis_paresis_immobilization_in_lower_extreme": False, "previous_dvt": False, "alternative_to_dvt_diagnosis": False }, "expected_valid": True, "expected_score": 1, "expected_risk": "Moderate probability", "description": "卧床≥3天，中等风险" }, { "name": "Moderate risk - Major surgery", "params": { "active_cancer": False, "bedridden_for_atleast_3_days": False, "major_surgery_in_last_12_weeks": True, "calf_swelling_3cm": False, "collateral_superficial_veins": False, "leg_swollen": False, "localized_tenderness_on_deep_venuous_system": False, "pitting_edema_on_symptomatic_leg": False, "paralysis_paresis_immobilization_in_lower_extreme": False, "previous_dvt": False, "alternative_to_dvt_diagnosis": False }, "expected_valid": True, "expected_score": 1, "expected_risk": "Moderate probability", "description": "近12周内大手术，中等风险" }, { "name": "Moderate risk - Both bedridden and surgery", "params": { "active_cancer": False, "bedridden_for_atleast_3_days": True, "major_surgery_in_last_12_weeks": True, "calf_swelling_3cm": False, "collateral_superficial_veins": False, "leg_swollen": False, "localized_tenderness_on_deep_venuous_system": False, "pitting_edema_on_symptomatic_leg": False, "paralysis_paresis_immobilization_in_lower_extreme": False, "previous_dvt": False, "alternative_to_dvt_diagnosis": False }, "expected_valid": True, "expected_score": 1, # Both bedridden and surgery only count as 1 point total "expected_risk": "Moderate probability", "description": "卧床且手术（仍然只计1分），中等风险" }, { "name": "Moderate risk - Multiple criteria", "params": { "active_cancer": False, "bedridden_for_atleast_3_days": True, "major_surgery_in_last_12_weeks": False, "calf_swelling_3cm": True, "collateral_superficial_veins": False, "leg_swollen": False, "localized_tenderness_on_deep_venuous_system": False, "pitting_edema_on_symptomatic_leg": False, "paralysis_paresis_immobilization_in_lower_extreme": False, "previous_dvt": False, "alternative_to_dvt_diagnosis": False }, "expected_valid": True, "expected_score": 2, "expected_risk": "Moderate probability", "description": "卧床+小腿肿胀，中等风险" }, { "name": "High risk - Multiple criteria", "params": { "active_cancer": True, "bedridden_for_atleast_3_days": True, "major_surgery_in_last_12_weeks": False, "calf_swelling_3cm": True, "collateral_superficial_veins": False, "leg_swollen": False, "localized_tenderness_on_deep_venuous_system": False, "pitting_edema_on_symptomatic_leg": False, "paralysis_paresis_immobilization_in_lower_extreme": False, "previous_dvt": False, "alternative_to_dvt_diagnosis": False }, "expected_valid": True, "expected_score": 3, "expected_risk": "High probability", "description": "癌症+卧床+小腿肿胀，高风险" }, { "name": "High risk - Previous DVT", "params": { "active_cancer": False, "bedridden_for_atleast_3_days": False, "major_surgery_in_last_12_weeks": True, "calf_swelling_3cm": True, "collateral_superficial_veins": True, "leg_swollen": False, "localized_tenderness_on_deep_venuous_system": False, "pitting_edema_on_symptomatic_leg": False, "paralysis_paresis_immobilization_in_lower_extreme": False, "previous_dvt": True, "alternative_to_dvt_diagnosis": False }, "expected_valid": True, "expected_score": 4, "expected_risk": "High probability", "description": "既往DVT史+多项体征，高风险" }, { "name": "High risk - Comprehensive symptoms", "params": { "active_cancer": True, "bedridden_for_atleast_3_days": True, "major_surgery_in_last_12_weeks": True, "calf_swelling_3cm": True, "collateral_superficial_veins": True, "leg_swollen": True, "localized_tenderness_on_deep_venuous_system": True, "pitting_edema_on_symptomatic_leg": True, "paralysis_paresis_immobilization_in_lower_extreme": True, "previous_dvt": True, "alternative_to_dvt_diagnosis": False }, "expected_valid": True, "expected_score": 9, # Max possible score "expected_risk": "High probability", "description": "所有危险因素阳性，极高风险" } ] print_header() # Execute test cases for i, test_case in enumerate(test_cases, 1): total_tests += 1 test_passed = True print_test_case(i, test_case) # Calculation test try: calc_result = await client.call_tool( "calculate", { "calculator_id": 16, "parameters": test_case["params"], }, ) # 使用 structured_content 或 data 属性获取实际数据 calc_data = calc_result.structured_content or calc_result.data or {} if isinstance(calc_data, dict) and calc_data.get("success") and "result" in calc_data: # 成功计算 data = calc_data["result"] print_calculation_result(data) # 检查计算结果 actual_score = data.get("value", -999) metadata = data.get("metadata", {}) actual_risk = metadata.get("risk_category", "") if actual_score != test_case["expected_score"]: print(f"- 错误: 预期评分 {test_case['expected_score']}，实际 {actual_score}") test_passed = False if actual_risk != test_case["expected_risk"]: print(f"- 错误: 预期风险等级 '{test_case['expected_risk']}'，实际 '{actual_risk}'") test_passed = False # 检查是否符合预期 if not test_case["expected_valid"]: print("- 错误: 预期失败但计算成功") test_passed = False else: # 计算失败 error_msg = calc_data.get("error", "未知错误") if isinstance(calc_data, dict) else str(calc_data) print(f"- 计算失败: {error_msg}") # 检查是否符合预期 if test_case["expected_valid"]: print("- 错误: 预期成功但计算失败") test_passed = False except Exception as e: print(f"- 计算错误: {e}") # 检查是否符合预期 if test_case["expected_valid"]: test_passed = False # Update statistics if test_passed: passed_tests += 1 print_test_result(i, test_passed) print_summary(total_tests, passed_tests, total_tests - passed_tests) return passed_tests, total_tests - passed_tests async def main(): def print_header(): print("Wells DVT 计算器 MCP 测试") print("=" * 60) def print_connection_status(success, error=None): if success: print("✅ 成功连接到 MCP 服务器") else: print(f"❌ 连接失败: {error}") def print_overall_results(total_passed, total_failed): total_tests = total_passed + total_failed if total_tests == 0: return print("\n" + "=" * 60) print("Wells DVT 计算器测试结果") print("=" * 60) print(f"总测试数: {total_tests}") print(f"通过数: {total_passed}") print(f"失败数: {total_failed}") print(f"成功率: {(total_passed/total_tests*100):.1f}%") if total_failed == 0: print("\n✅ Wells DVT 计算器所有测试都通过了！") else: print(f"\n❌ {total_failed} 个测试失败，请检查 Wells DVT 计算器实现。") print_header() try: async with Client(MCP_SERVER_URL) as client: print_connection_status(True) passed, failed = await test_wells_dvt_calculator(client) print_overall_results(passed, failed) except Exception as e: print_connection_status(False, str(e)) import traceback traceback.print_exc() return print("\n" + "=" * 60) print("✅ Wells DVT 计算器测试完成") if __name__ == "__main__": asyncio.run(main())