server.py•12.5 kB
#!/usr/bin/env python3
"""
FDA Taiwan Drug Search MCP Server
台灣FDA藥品查詢MCP服務器
提供英文品名查詢、驗證碼處理、仿單下載等功能
"""
import os
import tempfile
import time
import base64
from typing import List, Dict, Optional
from fastmcp import FastMCP
from src.fda_taiwan_mcp.tools import FDATaiwanClient, format_search_results
# 創建MCP實例
mcp = FastMCP("台灣FDA藥品查詢服務器")
# 初始化FDA客戶端
fda_client = FDATaiwanClient()
@mcp.tool
def search_drugs_by_english_name(english_name: str, captcha: str) -> str:
"""根據英文品名查詢台灣FDA藥品資訊
這是主要查詢工具，接收英文品名關鍵字和驗證碼，返回藥品詳細資訊。
Args:
english_name: 英文品名關鍵字（必填）
captcha: 4位數字驗證碼（必填）- 需要先使用 get_captcha_image 工具獲取並輸入
Returns:
結構化的藥品查詢結果，包含許可證號、中文品名、英文品名、申請商、仿單GUID等資訊
Example:
search_drugs_by_english_name("aspirin", "1234")
"""
try:
if not english_name.strip():
return "錯誤：英文品名不能為空"
if not captcha.strip() or len(captcha) != 4:
return "錯誤：驗證碼必須是4位數字"
results = fda_client.search_drugs_by_english_name(english_name.strip(), captcha.strip())
if not results:
return f"未找到包含 '{english_name}' 的藥品資訊。請檢查：\n1. 英文品名是否正確\n2. 驗證碼是否正確\n3. 嘗試使用部分英文品名"
return format_search_results(results)
except Exception as e:
return f"查詢失敗：{str(e)}"
@mcp.tool
def get_captcha_image() -> str:
"""獲取台灣FDA網站驗證碼圖片
此工具用於獲取台灣FDA查詢網站的驗證碼圖片。
由於驗證碼需要人工識別，請仔細查看圖片中的4位數字，
然後在查詢工具中輸入正確的驗證碼。
Returns:
Base64編碼的驗證碼圖片和提示信息
"""
try:
image_data, message = fda_client.get_captcha_image()
# 將圖片轉換為base64
image_base64 = base64.b64encode(image_data).decode('utf-8')
result = f"驗證碼獲取成功！\n\n"
result += f"圖片狀態：{message}\n"
result += f"請仔細識別圖片中的4位數字，然後在查詢時輸入。\n\n"
result += f"驗證碼圖片（Base64）：\n"
result += f"data:image/png;base64,{image_base64}\n\n"
result += f"操作提示：\n"
result += f"1. 查看上方圖片中的數字\n"
result += f"2. 將4位數字輸入到 search_drugs_by_english_name 的 captcha 參數中\n"
result += f"3. 如果看不清驗證碼，可以重新獲取"
return result
except Exception as e:
return f"獲取驗證碼失敗：{str(e)}\n\n建議：\n1. 檢查網路連接\n2. 稍後重試\n3. 如果問題持續，請聯繫系統管理員"
@mcp.tool
def download_insert(guid: str, output_path: str = "") -> str:
"""下載藥品仿單PDF文件
使用藥品GUID下載對應的仿單PDF文件。
Args:
guid: 仿單GUID（從查詢結果中獲得）
output_path: 輸出文件路徑（可選，默認保存到臨時目錄）
Returns:
下載結果狀態和文件路徑信息
Example:
download_insert("abc123-def456-ghi789", "/tmp/drug_insert.pdf")
"""
try:
if not guid.strip():
return "錯誤：GUID不能為空"
# 生成輸出路徑
if not output_path.strip():
with tempfile.NamedTemporaryFile(suffix='.pdf', delete=False) as tmp:
output_path = tmp.name
# 確保輸出目錄存在
output_dir = os.path.dirname(output_path)
if output_dir and not os.path.exists(output_dir):
os.makedirs(output_dir)
result = fda_client.download_insert(guid.strip(), output_path)
if result["status"] == "success":
return f"✅ 仿單下載成功！\n\n" \
f"📄 文件路徑：{result['file_path']}\n" \
f"📊 文件大小：{result['file_size']:,} 字節\n" \
f"💡 提示：{result['message']}"
else:
return f"❌ 仿單下載失敗\n\n" \
f"錯誤原因：{result['error_message']}\n\n" \
f"建議：\n" \
f"1. 檢查GUID是否正確\n" \
f"2. 確認網路連接正常\n" \
f"3. 驗證文件權限"
except Exception as e:
return f"下載仿單時發生錯誤：{str(e)}"
@mcp.tool
def advanced_drug_search(
license_no: str = "",
cname: str = "",
ename: str = "",
agent_name: str = "",
ingredient: str = "",
captcha: str = ""
) -> str:
"""進階藥品查詢工具
支援多條件查詢藥品資訊，可以組合使用多個查詢條件。
Args:
license_no: 許可證號（可選）
cname: 中文品名（可選）
ename: 英文品名（可選）
agent_name: 申請商名稱（可選）
ingredient: 有效成分（可選）
captcha: 4位數字驗證碼（必填）
Returns:
符合查詢條件的藥品資訊列表
Example:
advanced_drug_search(
ename="aspirin",
agent_name="拜耳",
captcha="1234"
)
"""
try:
# 檢查至少有一個查詢條件
search_conditions = {
'license_no': license_no.strip() if license_no else "",
'cname': cname.strip() if cname else "",
'ename': ename.strip() if ename else "",
'agent_name': agent_name.strip() if agent_name else "",
'ingredient': ingredient.strip() if ingredient else "",
}
# 檢查是否有非空的查詢條件
if not any(search_conditions.values()):
return "錯誤：請至少提供一個查詢條件（許可證號、中文品名、英文品名、申請商或有效成分）"
if not captcha.strip() or len(captcha) != 4:
return "錯誤：驗證碼必須是4位數字"
# 添加驗證碼
search_conditions['captcha'] = captcha.strip()
# 執行查詢
results = fda_client.advanced_search(**search_conditions)
if not results:
return "未找到符合查詢條件的藥品資訊。\n\n" \
f"查詢條件：\n" \
f"- 英文品名：{search_conditions.get('ename', '未指定')}\n" \
f"- 中文品名：{search_conditions.get('cname', '未指定')}\n" \
f"- 許可證號：{search_conditions.get('license_no', '未指定')}\n" \
f"- 申請商：{search_conditions.get('agent_name', '未指定')}\n" \
f"- 有效成分：{search_conditions.get('ingredient', '未指定')}\n\n" \
f"建議：\n" \
f"1. 檢查查詢條件是否正確\n" \
f"2. 嘗試使用更寬鬆的搜索條件\n" \
f"3. 確認驗證碼正確"
# 添加查詢條件摘要到結果
result_text = f"進階查詢完成！找到 {len(results)} 筆符合條件的藥品資訊：\n\n"
result_text += f"本次查詢條件：\n"
for key, value in search_conditions.items():
if value and key != 'captcha':
result_text += f"- {key}：{value}\n"
result_text += "\n"
result_text += format_search_results(results)
return result_text
except Exception as e:
return f"進階查詢失敗：{str(e)}"
@mcp.tool
def format_search_results_tool(results: List[Dict]) -> str:
"""格式化藥品搜索結果
將原始搜索結果轉換為易讀的格式。
Args:
results: 藥品搜索結果列表
Returns:
格式化後的結果文本
"""
try:
if not results:
return "沒有搜索結果需要格式化。"
return format_search_results(results)
except Exception as e:
return f"格式化搜索結果時發生錯誤：{str(e)}"
@mcp.tool
def batch_download_inserts(guids: List[str], output_dir: str = "") -> str:
"""批量下載多個藥品仿單
批量下載多個藥品的仿單PDF文件。
Args:
guids: 仿單GUID列表
output_dir: 輸出目錄路徑（可選）
Returns:
批量下載結果摘要
Example:
batch_download_inserts(
["guid1", "guid2", "guid3"],
"/tmp/drug_inserts"
)
"""
try:
if not guids:
return "錯誤：GUID列表不能為空"
# 設置輸出目錄
if not output_dir.strip():
output_dir = tempfile.gettempdir()
# 確保輸出目錄存在
if not os.path.exists(output_dir):
os.makedirs(output_dir)
results = []
success_count = 0
failed_count = 0
for i, guid in enumerate(guids, 1):
try:
# 生成輸出文件路徑
output_path = os.path.join(output_dir, f"drug_insert_{i}_{guid[:8]}.pdf")
# 下載仿單
download_result = fda_client.download_insert(guid, output_path)
results.append({
"guid": guid,
"index": i,
"result": download_result
})
if download_result["status"] == "success":
success_count += 1
else:
failed_count += 1
# 添加小延遲避免請求過於頻繁
if i < len(guids):
time.sleep(1)
except Exception as e:
results.append({
"guid": guid,
"index": i,
"result": {
"status": "error",
"error_message": str(e)
}
})
failed_count += 1
# 生成結果摘要
summary = f"批量下載完成！\n\n"
summary += f"總計：{len(guids)} 個仿單\n"
summary += f"成功：{success_count} 個\n"
summary += f"失敗：{failed_count} 個\n"
summary += f"輸出目錄：{output_dir}\n\n"
# 詳細結果
summary += "詳細結果：\n"
for result in results:
status_icon = "✅" if result["result"]["status"] == "success" else "❌"
summary += f"{status_icon} [{result['index']}] GUID: {result['guid'][:16]}...\n"
if result["result"]["status"] == "success":
summary += f" 📄 文件：{result['result'].get('file_path', 'N/A')}\n"
summary += f" 📊 大小：{result['result'].get('file_size', 0):,} 字節\n"
else:
summary += f" ❌ 錯誤：{result['result'].get('error_message', '未知錯誤')}\n"
return summary
except Exception as e:
return f"批量下載時發生錯誤：{str(e)}"
def main():
"""啟動MCP服務器"""
import argparse
parser = argparse.ArgumentParser(description='台灣FDA藥品查詢MCP服務器')
parser.add_argument('--transport', default='stdio', help='Transport type')
parser.add_argument('--host', default='localhost', help='Host address')
parser.add_argument('--port', type=int, default=8000, help='Port number')
args = parser.parse_args()
if args.transport == 'stdio':
# STDIO模式（默認）
mcp.run()
elif args.transport == 'sse':
# SSE模式
mcp.run(transport='sse', host=args.host, port=args.port)
else:
print(f"不支援的傳輸模式: {args.transport}")
if __name__ == "__main__":
main()