server.py•9.25 kB
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
中西藥交互作用查詢MCP Server - 優化版
基於FastMCP框架實現的中西藥交互作用查詢系統
"""
import sys
from pathlib import Path
# 添加src目錄到Python路徑
sys.path.insert(0, str(Path(__file__).parent / "src"))
from fastmcp import FastMCP
from src.tools import DrugInteractionTools
from src.common import setup_logging, normalize_medicine_list, sort_by_severity, calculate_risk_level, generate_safety_recommendations
class DrugInteractionMCP:
"""中西藥交互作用MCP服務器主類"""
def __init__(self):
"""初始化服務器"""
# 設置日誌
self.logger = setup_logging()
self.logger.info("初始化中西藥交互作用MCP服務器")
# 創建MCP服務器實例
self.mcp = FastMCP("中西藥交互作用查詢系統")
# 初始化工具
try:
self.tools = DrugInteractionTools()
self.logger.info("工具初始化成功")
except Exception as e:
self.logger.error(f"工具初始化失敗: {e}")
sys.exit(1)
# 註冊工具
self._register_tools()
# 註冊資源
self._register_resources()
# 註冊提示
self._register_prompts()
def _register_tools(self):
"""註冊MCP工具"""
@self.mcp.tool
def search_chinese_medicine(keyword: str, limit: int = 20) -> str:
"""搜尋中藥資訊
支援中文名稱、拼音和分類的模糊搜尋。
Args:
keyword: 搜尋關鍵字
limit: 結果數量限制，預設20個
Returns:
搜尋結果JSON字串
"""
result = self.tools.search_chinese_medicine(keyword, limit)
return str(result)
@self.mcp.tool
def search_western_medicine(keyword: str, limit: int = 20) -> str:
"""搜尋西藥資訊
支援中文名稱、通用名、品牌名和分類的模糊搜尋。
Args:
keyword: 搜尋關鍵字
limit: 結果數量限制，預設20個
Returns:
搜尋結果JSON字串
"""
result = self.tools.search_western_medicine(keyword, limit)
return str(result)
@self.mcp.tool
def check_interactions(chinese_medicine: str, western_medicine: str) -> str:
"""檢查特定中藥和西藥的交互作用
Args:
chinese_medicine: 中藥名稱
western_medicine: 西藥名稱
Returns:
交互作用檢查結果JSON字串
"""
result = self.tools.check_interactions(chinese_medicine, western_medicine)
return str(result)
@self.mcp.tool
def find_interactions_by_chinese_medicine(chinese_medicine: str) -> str:
"""根據中藥查找所有相關交互作用
Args:
chinese_medicine: 中藥名稱
Returns:
中藥交互作用列表JSON字串
"""
result = self.tools.find_interactions_by_chinese_medicine(chinese_medicine)
return str(result)
@self.mcp.tool
def find_interactions_by_western_medicine(western_medicine: str) -> str:
"""根據西藥查找所有相關交互作用
Args:
western_medicine: 西藥名稱
Returns:
西藥交互作用列表JSON字串
"""
result = self.tools.find_interactions_by_western_medicine(western_medicine)
return str(result)
@self.mcp.tool
def batch_check_interactions(chinese_medicines: str, western_medicines: str) -> str:
"""批次檢查多個中藥和西藥的交互作用
Args:
chinese_medicines: 中藥名稱列表（逗號分隔）
western_medicines: 西藥名稱列表（逗號分隔）
Returns:
批次檢查結果JSON字串
"""
chinese_list = normalize_medicine_list(chinese_medicines)
western_list = normalize_medicine_list(western_medicines)
result = self.tools.batch_check_interactions(chinese_list, western_list)
return str(result)
@self.mcp.tool
def get_interaction_summary(medicine_name: str, medicine_type: str) -> str:
"""獲取藥物的交互作用摘要和風險評估
Args:
medicine_name: 藥物名稱
medicine_type: 藥物類型 ("chinese" 或 "western")
Returns:
藥物交互作用摘要JSON字串
"""
result = self.tools.get_interaction_summary(medicine_name, medicine_type)
return str(result)
def _register_resources(self):
"""註冊MCP資源"""
@self.mcp.resource("resource://medicines/chinese")
def list_chinese_medicines():
"""列出所有中藥資源"""
result = self.tools.search_chinese_medicine("", limit=1000)
if result["success"]:
return f"# 中藥資源列表\n\n共 {result['count']} 種中藥\n\n" + "\n".join([
f"- {med['name']} ({med['name_pinyin']}) - {med['category']}"
for med in result["data"]
])
else:
return "無法獲取中藥資源列表"
@self.mcp.resource("resource://medicines/western")
def list_western_medicines():
"""列出所有西藥資源"""
result = self.tools.search_western_medicine("", limit=1000)
if result["success"]:
return f"# 西藥資源列表\n\n共 {result['count']} 種西藥\n\n" + "\n".join([
f"- {med['name']} ({med['generic_name']}) - {med['category']}"
for med in result["data"]
])
else:
return "無法獲取西藥資源列表"
def _register_prompts(self):
"""註冊MCP提示"""
@self.mcp.prompt
def drug_interaction_analysis(medicines: str, analysis_type: str = "full") -> str:
"""生成藥物交互作用分析提示"""
prompts = {
"full": f"""請對以下藥物進行完整的交互作用分析：
藥物列表：{medicines}
請分析：
1. 各藥物間的直接交互作用
2. 潛在的間接交互作用
3. 嚴重程度評估
4. 臨床意義解釋
5. 替代方案建議
6. 監測建議
請提供專業、詳細的分析報告。""",
"quick": f"""請對以下藥物進行快速安全檢查：
藥物列表：{medicines}
請重點檢查：
1. 是否有嚴重或致命的交互作用
2. 是否需要調整劑量
3. 監測要點
4. 簡要安全建議
請提供簡潔明了的安全提示。""",
"safety": f"""請對以下藥物進行安全風險評估：
藥物列表：{medicines}
請評估：
1. 整體安全等級
2. 主要風險點
3. 禁忌組合
4. 必要預防措施
5. 緊急處理建議
請提供緊急情況下的安全指導。"""
}
return prompts.get(analysis_type, prompts["full"])
@self.mcp.prompt
def clinical_decision_support(patient_info: str, medications: str) -> str:
"""臨床決策支援提示"""
return f"""作為臨床藥師，請為以下患者提供用藥建議：
**患者資訊：**
{patient_info}
**當前用藥：**
{medications}
**分析要求：**
1. 評估當前用藥組合的安全性
2. 識別潛在的藥物交互作用
3. 提出劑量調整建議
4. 建議監測指標
5. 考慮藥物禁忌症
6. 提供患者教育要點
請以臨床專業的角度，提供實用的用藥指導建議。"""
def run(self):
"""運行MCP服務器"""
try:
self.logger.info("啟動MCP服務器")
self.mcp.run()
except KeyboardInterrupt:
self.logger.info("接收到中斷信號，正在關閉MCP服務器...")
except Exception as e:
self.logger.error(f"MCP服務器運行錯誤: {e}")
finally:
self.cleanup()
def cleanup(self):
"""清理資源"""
try:
if hasattr(self, 'tools') and self.tools:
self.tools.db.close_connections()
self.logger.info("資料庫連接已關閉")
except Exception as e:
self.logger.error(f"清理資源時發生錯誤: {e}")
finally:
self.logger.info("MCP服務器已關閉")
def main():
"""主函數"""
server = DrugInteractionMCP()
server.run()
if __name__ == "__main__":
main()