""" 匯率查詢 MCP Tool 使用標準化接口實作的匯率功能 """ import os import json import logging import aiohttp import asyncio from datetime import datetime from typing import Dict, Any, Optional from dotenv import load_dotenv from .base_tool import MCPTool, ValidationError, ExecutionError, StandardToolSchemas # 載入環境變數 load_dotenv() logger = logging.getLogger("mcp.tools.exchange") # API配置 EXCHANGE_API_BASE = "https://api.exchangerate-api.com/v4/latest" FIXER_API_BASE = "https://data.fixer.io/api/latest" FIXER_API_KEY = os.getenv("FIXER_API_KEY", "") class ExchangeTool(MCPTool): """匯率查詢 MCP 工具""" NAME = "exchange_query" DESCRIPTION = "Query real-time exchange rates between major currencies" CATEGORY = "生活資訊" TAGS = ["exchange", "匯率", "貨幣"] KEYWORDS = ["匯率", "美元", "台幣", "exchange", "USD", "TWD", "貨幣", "換算"] USAGE_TIPS = [ "提供源貨幣和目標貨幣代碼（如 USD, TWD）", "支援主要國際貨幣", "可查詢即時匯率與歷史趨勢" ] @classmethod def get_input_schema(cls) -> Dict[str, Any]: """獲取輸入參數模式""" return StandardToolSchemas.create_input_schema({ "from_currency": { "type": "string", "description": "源貨幣代碼 (如 USD, EUR, TWD)", "default": "USD", "pattern": "^[A-Z]{3}$" }, "to_currency": { "type": "string", "description": "目標貨幣代碼 (如 TWD, USD, EUR)", "default": "TWD", "pattern": "^[A-Z]{3}$" }, "amount": { "type": "number", "description": "轉換金額", "default": 1.0, "minimum": 0.01 }, "conversion": { "type": "boolean", "description": "是否進行金額轉換計算", "default": True } }, ["from_currency", "to_currency"]) @classmethod def get_output_schema(cls) -> Dict[str, Any]: """獲取輸出結果模式""" base_schema = StandardToolSchemas.create_output_schema() base_schema["properties"].update({ "rate": {"type": "number"}, "converted_amount": {"type": ["number", "null"]}, "raw_data": {"type": "object"} }) return base_schema @classmethod async def execute(cls, arguments: Dict[str, Any]) -> Dict[str, Any]: """執行匯率查詢""" from_currency = arguments.get("from_currency", "USD").upper() to_currency = arguments.get("to_currency", "TWD").upper() amount = arguments.get("amount", 1.0) conversion = arguments.get("conversion", True) try: # 獲取匯率數據 rate_data = await cls._fetch_exchange_rate(from_currency, to_currency) if rate_data.get("success"): rate = rate_data.get("rate") # 生成結構化數據供 AI 格式化 formatted_text = cls._format_exchange_response( from_currency, to_currency, rate, amount, conversion, rate_data.get("metadata", {}) ) return cls.create_success_response( content=formatted_text, data={ "raw_data": { "rate": rate, "from_currency": from_currency, "to_currency": to_currency, "amount": amount, "converted_amount": amount * rate if conversion else None, "rate_data": rate_data } } ) else: return cls.create_error_response( error=rate_data.get("error", "獲取匯率失敗"), code="FETCH_ERROR" ) except Exception as e: logger.error(f"匯率查詢錯誤: {e}") raise ExecutionError(f"匯率查詢時發生錯誤: {str(e)}", e) @staticmethod async def _fetch_exchange_rate(from_currency: str, to_currency: str) -> Dict[str, Any]: """獲取匯率數據""" # 首先嘗試免費 API try: result = await ExchangeTool._fetch_from_exchangerate_api(from_currency, to_currency) if result.get("success"): return result except Exception as e: logger.warning(f"免費匯率 API 失敗: {e}") # 如果有 Fixer API Key，嘗試備用 API if FIXER_API_KEY: try: result = await ExchangeTool._fetch_from_fixer_api(from_currency, to_currency) if result.get("success"): return result except Exception as e: logger.warning(f"Fixer API 失敗: {e}") return { "success": False, "error": "所有匯率 API 都無法使用，請稍後再試" } @staticmethod async def _fetch_from_exchangerate_api(from_currency: str, to_currency: str) -> Dict[str, Any]: """從 exchangerate-api.com 獲取匯率""" url = f"{EXCHANGE_API_BASE}/{from_currency}" async with aiohttp.ClientSession() as session: async with session.get(url, timeout=10) as response: if response.status == 200: data = await response.json() rates = data.get("rates", {}) if to_currency in rates: rate = rates[to_currency] return { "success": True, "rate": rate, "metadata": { "source": "exchangerate-api.com", "base": data.get("base"), "date": data.get("date"), "timestamp": data.get("timestamp") } } else: return { "success": False, "error": f"不支援的貨幣: {to_currency}" } else: return { "success": False, "error": f"匯率 API 請求失敗，狀態碼: {response.status}" } @staticmethod async def _fetch_from_fixer_api(from_currency: str, to_currency: str) -> Dict[str, Any]: """從 fixer.io API 獲取匯率""" url = f"{FIXER_API_BASE}?access_key={FIXER_API_KEY}&base={from_currency}&symbols={to_currency}" async with aiohttp.ClientSession() as session: async with session.get(url, timeout=10) as response: if response.status == 200: data = await response.json() if data.get("success"): rates = data.get("rates", {}) if to_currency in rates: rate = rates[to_currency] return { "success": True, "rate": rate, "metadata": { "source": "fixer.io", "base": data.get("base"), "date": data.get("date"), "timestamp": data.get("timestamp") } } else: return { "success": False, "error": f"不支援的貨幣: {to_currency}" } else: error_info = data.get("error", {}) return { "success": False, "error": f"Fixer API 錯誤: {error_info.get('info', '未知錯誤')}" } else: return { "success": False, "error": f"Fixer API 請求失敗，狀態碼: {response.status}" } @staticmethod def _format_exchange_response(from_currency: str, to_currency: str, rate: float, amount: float, conversion: bool, metadata: Dict) -> str: """格式化匯率回應""" # 貨幣符號和名稱映射 currency_info = ExchangeTool._get_currency_info() from_symbol = currency_info.get(from_currency, {}).get("symbol", from_currency) from_name = currency_info.get(from_currency, {}).get("name", from_currency) to_symbol = currency_info.get(to_currency, {}).get("symbol", to_currency) to_name = currency_info.get(to_currency, {}).get("name", to_currency) # 標題 result = f"💱 匯率資訊\n\n" # 匯率 result += f"💰 匯率: 1 {from_currency} = {rate:.4f} {to_currency}\n" result += f"📈 {from_name} → {to_name}\n\n" # 如果是轉換計算 if conversion and amount != 1.0: converted = amount * rate result += f"🔄 金額轉換\n" result += f"📊 {amount:,.2f} {from_currency} = {converted:,.2f} {to_currency}\n" result += f"💸 {from_symbol}{amount:,.2f} → {to_symbol}{converted:,.2f}\n\n" # 數據來源和時間 source = metadata.get("source", "未知") date = metadata.get("date", "") if date: result += f"📅 數據日期: {date}\n" result += f"🔗 數據來源: {source}\n" result += f"⏰ 查詢時間: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n" # 匯率變化提示 result += f"\n💡 匯率會隨市場變動，僅供參考" return result @staticmethod def _get_currency_info() -> Dict[str, Dict[str, str]]: """獲取貨幣資訊映射""" return { "TWD": {"symbol": "NT$", "name": "新台幣"}, "USD": {"symbol": "$", "name": "美元"}, "EUR": {"symbol": "€", "name": "歐元"}, "JPY": {"symbol": "¥", "name": "日圓"}, "GBP": {"symbol": "£", "name": "英鎊"}, "AUD": {"symbol": "A$", "name": "澳幣"}, "CAD": {"symbol": "C$", "name": "加幣"}, "CHF": {"symbol": "Fr", "name": "瑞士法郎"}, "CNY": {"symbol": "¥", "name": "人民幣"}, "HKD": {"symbol": "HK$", "name": "港幣"}, "NZD": {"symbol": "NZ$", "name": "紐幣"}, "KRW": {"symbol": "₩", "name": "韓元"}, "SGD": {"symbol": "S$", "name": "新加坡幣"}, "THB": {"symbol": "฿", "name": "泰銖"}, "MYR": {"symbol": "RM", "name": "馬幣"}, "IDR": {"symbol": "Rp", "name": "印尼盧比"}, "PHP": {"symbol": "₱", "name": "菲律賓比索"}, "VND": {"symbol": "₫", "name": "越南盾"}, "RUB": {"symbol": "₽", "name": "俄羅斯盧布"}, "BRL": {"symbol": "R$", "name": "巴西雷亞爾"}, "ZAR": {"symbol": "R", "name": "南非蘭特"}, "INR": {"symbol": "₹", "name": "印度盧比"} }