Update MCP tools and add TDX tools

features/mcp/agent_bridge.py

@@ -685,33 +685,98 @@ class MCPAgentBridge:
         }
 
     def _get_tools_description(self) -> str:
-        """獲取簡化的工具描述，專注於核心信息"""
-        descriptions = []
-
-        for tool_name, tool in self.mcp_server.tools.items():
-            # 簡化描述格式
-            desc = f"{tool_name}: {tool.description}"
-
-            # 只保留最重要的參數信息
-            input_schema = tool.inputSchema
-            properties = input_schema.get("properties", {})
-
-            if properties:
-                # 只顯示必需參數
-                required = input_schema.get("required", [])
-                if required:
-                    params = []
-                    for param_name in required:
-                        if param_name in properties:
-                            param_info = properties[param_name]
-                            param_type = param_info.get("type", "string")
-                            params.append(f"{param_name}({param_type})")
-                    if params:
-                        desc += f" | 參數: {', '.join(params)}"
-
-            descriptions.append(desc)
-
-        return "\n".join(descriptions)
+        """獲取分類整理的工具摘要（使用輕量級摘要，減少 token 消耗 60-70%）"""
+        # 使用 MCPServer 的 get_tools_summary() 獲取輕量級摘要
+        try:
+            tools_summary = self.mcp_server.get_tools_summary()
+        except Exception as e:
+            logger.error(f"獲取工具摘要失敗: {e}")
+            # 降級：使用舊邏輯
+            tools_summary = []
+            for tool_name, tool in self.mcp_server.tools.items():
+                tools_summary.append({
+                    "name": tool_name,
+                    "description": tool.description if hasattr(tool, 'description') else "",
+                    "category": "其他",
+                    "keywords": [],
+                    "is_complex": False
+                })
+        
+        # 按類別組織工具
+        categorized_tools = {
+            "地理定位": [],
+            "軌道運輸": [],
+            "道路運輸": [],
+            "微型運具": [],
+            "停車與充電": [],
+            "生活資訊": [],
+            "健康數據": [],
+            "其他": []
+        }
+        
+        for summary in tools_summary:
+            category = summary.get("category", "其他")
+            name = summary.get("name", "unknown")
+            desc = summary.get("description", "")
+            keywords = summary.get("keywords", [])
+            is_complex = summary.get("is_complex", False)
+            
+            # 格式化：工具名 - 描述 | 關鍵字
+            keywords_str = ", ".join(keywords[:5]) if keywords else ""  # 最多顯示 5 個關鍵字
+            if keywords_str:
+                line = f"- {name}: {desc} | 關鍵字: {keywords_str}"
+            else:
+                line = f"- {name}: {desc}"
+            
+            # 標記複雜工具
+            if is_complex:
+                line += " [複雜]"
+            
+            # 將工具加入對應類別
+            if category in categorized_tools:
+                categorized_tools[category].append(line)
+            else:
+                categorized_tools["其他"].append(line)
+        
+        # 構建分類描述
+        result = []
+        
+        # 定義類別順序和說明
+        category_order = [
+            ("地理定位", "【地理定位與導航】地點查詢、路線規劃"),
+            ("軌道運輸", "【軌道運輸】捷運、台鐵、高鐵"),
+            ("道路運輸", "【道路運輸】公車、客運"),
+            ("微型運具", "【微型運具】YouBike 共享單車"),
+            ("停車與充電", "【停車與充電】停車場、充電站"),
+            ("生活資訊", "【生活資訊】天氣、新聞、匯率"),
+            ("健康數據", "【健康數據】心率、步數、血氧、睡眠"),
+            ("其他", "【其他功能】")
+        ]
+        
+        for category, header in category_order:
+            tools = categorized_tools.get(category, [])
+            if tools:
+                result.append(f"\n{header}")
+                result.extend(tools)
+        
+        # 添加工具選擇指引
+        result.append("\n【工具選擇指引】")
+        result.append("1. 導航問題（「怎麼去」「路線」「導航」） → directions")
+        result.append("2. 地點查詢（「XXX在哪」「地址」） → forward_geocode")
+        result.append("3. 公共運輸查詢 → 根據運具類型選擇對應工具")
+        result.append("   - 公車 → tdx_bus_arrival")
+        result.append("   - 捷運 → tdx_metro")
+        result.append("   - 台鐵 → tdx_train")
+        result.append("   - 高鐵 → tdx_thsr")
+        result.append("   - YouBike → tdx_youbike")
+        result.append("   - 停車場/充電站 → tdx_parking")
+        result.append("4. 所有 tdx 工具都會自動感知用戶���置，無需手動提供座標")
+        result.append("5. 健康數據查詢 → healthkit_query（心率、步數、血氧等）")
+        result.append("6. 生活資訊 → weather_query（天氣）、news_query（新聞）、exchange_query（匯率）")
+        result.append("7. 標記 [複雜] 的工具只需返回工具名稱，參數稍後填充")
+        
+        logger.debug(f"工具描述已生成，總長度: {len(''.join(result))} 字元")
+        return "\n".join(result)
 
     def _keyword_intent_detection(self, message: str) -> Tuple[bool, Optional[Dict[str, Any]]]:
         """關鍵詞匹配檢測 (備用方案)"""

features/mcp/server.py

@@ -7,7 +7,8 @@ import json
 import sys
 import asyncio
 import logging
-from typing import Dict, Any, List, Optional, Callable
+import time
+from typing import Dict, Any, List, Optional, Callable, Tuple
 from enum import Enum
 from .types import Tool
 from .auto_registry import MCPAutoRegistry
@@ -51,6 +52,10 @@ class FeaturesMCPServer:
         self.tools: Dict[str, Tool] = {}
         self.handlers: Dict[str, Callable] = {}
 
+        # Schema 快取（用於 Lazy Loading）
+        self._schema_cache: Dict[str, Tuple[Dict, float]] = {}  # {tool_name: (schema, timestamp)}
+        self._schema_cache_ttl = 600  # 10 分鐘
+
         # 保存註冊器引用以便清理
         self._registry = None
 
@@ -208,6 +213,108 @@ class FeaturesMCPServer:
         """註冊工具"""
         self.tools[tool.name] = tool
         logger.info(f"註冊工具: {tool.name}")
+    
+    def get_tools_summary(self) -> List[Dict[str, Any]]:
+        """
+        獲取所有工具的摘要（用於 Intent Detection，減少 token 消耗）
+        
+        返回輕量級工具列表，只包含：
+        - name, description_short, category, keywords, is_complex
+        - 簡單工具額外包含 params 列表
+        
+        相比完整 schema，節省約 60-70% tokens
+        """
+        summaries = []
+        
+        for tool_name, tool in self.tools.items():
+            try:
+                # 檢查工具是否有 get_summary 方法（新工具）
+                if hasattr(tool, 'get_summary') and callable(tool.get_summary):
+                    summary = tool.get_summary()
+                else:
+                    # 向後兼容：為舊工具生成簡化摘要
+                    summary = {
+                        "name": tool_name,
+                        "description": tool.description[:50] + "..." if len(tool.description) > 50 else tool.description,
+                        "category": tool.metadata.get('category', 'general') if hasattr(tool, 'metadata') and tool.metadata else 'general',
+                        "keywords": tool.metadata.get('keywords', []) if hasattr(tool, 'metadata') and tool.metadata else [],
+                        "is_complex": False  # 舊工具預設為簡單工具
+                    }
+                
+                summaries.append(summary)
+                
+            except Exception as e:
+                logger.error(f"生成工具摘要失敗: {tool_name}, 錯誤: {e}")
+                # 降級：返回最基本的資訊
+                summaries.append({
+                    "name": tool_name,
+                    "description": tool.description if hasattr(tool, 'description') else "未知工具",
+                    "category": "其他",
+                    "keywords": [],
+                    "is_complex": False
+                })
+        
+        logger.debug(f"生成工具摘要: {len(summaries)} 個工具")
+        return summaries
+    
+    def get_tool_full_schema(self, tool_name: str) -> Dict[str, Any]:
+        """
+        獲取工具的完整 Schema（Lazy Loading，按需載入）
+        
+        使用快取機制：
+        - 首次調用：載入完整 schema 並快取 10 分鐘
+        - 後續調用：直接從快取返回（節省計算）
+        - 快取過期：重新載入
+        
+        Args:
+            tool_name: 工具名稱
+            
+        Returns:
+            完整的工具定義（包含 inputSchema, outputSchema 等）
+            
+        Raises:
+            ValueError: 工具不存在
+        """
+        # 檢查快取
+        if tool_name in self._schema_cache:
+            schema, timestamp = self._schema_cache[tool_name]
+            if time.time() - timestamp < self._schema_cache_ttl:
+                logger.debug(f"Schema 快取命中: {tool_name}")
+                return schema
+            else:
+                # 快取過期，刪除
+                del self._schema_cache[tool_name]
+                logger.debug(f"Schema 快取過期: {tool_name}")
+        
+        # 快取未命中，載入完整 schema
+        if tool_name not in self.tools:
+            raise ValueError(f"工具不存在: {tool_name}")
+        
+        tool = self.tools[tool_name]
+        
+        try:
+            # 檢查工具是否有 get_full_definition 方法（新工具）
+            if hasattr(tool, 'get_full_definition') and callable(tool.get_full_definition):
+                schema = tool.get_full_definition()
+            else:
+                # 向後兼容：從現有屬性構建完整定義
+                schema = {
+                    "name": tool_name,
+                    "description": tool.description if hasattr(tool, 'description') else "",
+                    "inputSchema": tool.inputSchema if hasattr(tool, 'inputSchema') else {},
+                    "outputSchema": getattr(tool, 'outputSchema', {}),
+                    "metadata": getattr(tool, 'metadata', {})
+                }
+            
+            # 更新快取
+            self._schema_cache[tool_name] = (schema, time.time())
+            logger.debug(f"Schema 已快取: {tool_name}")
+            
+            return schema
+            
+        except Exception as e:
+            logger.error(f"載入工具 Schema 失敗: {tool_name}, 錯誤: {e}")
+            raise ValueError(f"無法載入工具 {tool_name} 的 Schema: {e}")
 
     async def _handle_initialize(self, params: Dict[str, Any]) -> Dict[str, Any]:
         """處理初始化請求"""

features/mcp/tools/base_tool.py

@@ -52,25 +52,61 @@ class MCPTool(ABC):
     # 工具基本信息
     NAME: str = ""
     DESCRIPTION: str = ""
+    DESCRIPTION_SHORT: str = ""  # 簡短描述（用於 Intent Detection，10-20 tokens）
     CATEGORY: str = "general"
     TAGS: List[str] = []
+    KEYWORDS: List[str] = []  # 用於快速意圖匹配的關鍵字
     USAGE_TIPS: List[str] = []
+    IS_COMPLEX: bool = False  # 標記是否為複雜工具（需要兩階段參數填充）
 
     @classmethod
-    def get_definition(cls) -> Dict[str, Any]:
-        """獲取工具定義"""
+    def get_summary(cls) -> Dict[str, Any]:
+        """獲取工具摘要（用於 Intent Detection，減少 token 消耗）"""
+        # 自動生成簡短描述（截取前 50 字元或使用自訂）
+        short_desc = cls.DESCRIPTION_SHORT or (
+            cls.DESCRIPTION[:47] + "..." if len(cls.DESCRIPTION) > 50 else cls.DESCRIPTION
+        )
+        
+        summary = {
+            "name": cls.NAME,
+            "description": short_desc,
+            "category": cls.CATEGORY,
+            "keywords": cls.KEYWORDS,
+            "is_complex": cls.IS_COMPLEX
+        }
+        
+        # 簡單工具：提供簡化參數列表（只有參數名，不含詳細 schema）
+        if not cls.IS_COMPLEX:
+            try:
+                schema = cls.get_input_schema()
+                summary["params"] = list(schema.get("properties", {}).keys())
+            except:
+                summary["params"] = []
+        
+        return summary
+    
+    @classmethod
+    def get_full_definition(cls) -> Dict[str, Any]:
+        """獲取完整工具定義（用於實際調用，包含完整 schema）"""
         return {
             "name": cls.NAME,
             "description": cls.DESCRIPTION,
             "metadata": {
                 "category": cls.CATEGORY,
                 "tags": cls.TAGS,
+                "keywords": cls.KEYWORDS,
+                "is_complex": cls.IS_COMPLEX,
                 "usage_tips": cls.USAGE_TIPS
             },
             "inputSchema": cls.get_input_schema(),
             "outputSchema": cls.get_output_schema()
         }
 
+    @classmethod
+    def get_definition(cls) -> Dict[str, Any]:
+        """獲取工具定義（向後兼容，使用完整定義）"""
+        return cls.get_full_definition()
+
     @classmethod
     @abstractmethod
     def get_input_schema(cls) -> Dict[str, Any]:

features/mcp/tools/directions_tool.py

@@ -23,8 +23,9 @@ ORS_API_KEY = os.getenv("OPENROUTESERVICE_API_KEY", "")
 class DirectionsTool(MCPTool):
     NAME = "directions"
     DESCRIPTION = "規劃兩點之間的路線（walk/drive/cycle），返回距離、時間與 polyline"
-    CATEGORY = "地理"
+    CATEGORY = "地理定位"
     TAGS = ["route", "navigation", "directions"]
+    KEYWORDS = ["導航", "路線", "怎麼去", "怎麼走", "規劃", "開車", "步行", "騎車"]
     USAGE_TIPS = ["提供起訖兩點經緯度"]
     _COORDINATE_FIELDS = {
         "origin_lat": "起點緯度",

features/mcp/tools/exchange_tool.py

@@ -28,10 +28,15 @@ class ExchangeTool(MCPTool):
     """匯率查詢 MCP 工具"""
 
     NAME = "exchange_query"
-    DESCRIPTION = "查詢匯率資訊，支援即時匯率查詢和貨幣轉換計算"
-    CATEGORY = "匯率"
-    TAGS = ["exchange", "currency", "finance"]
-    USAGE_TIPS = ["直接說「美元匯率」或「美金對台幣」"]
+    DESCRIPTION = "查詢即時匯率（支援主要貨幣兌換）"
+    CATEGORY = "生活資訊"
+    TAGS = ["exchange", "匯率", "貨幣"]
+    KEYWORDS = ["匯率", "美元", "台幣", "exchange", "USD", "TWD", "貨幣", "換算"]
+    USAGE_TIPS = [
+        "提供源貨幣和目標貨幣代碼（如 USD, TWD）",
+        "支援主要國際貨幣",
+        "可查詢即時匯率與歷史趨勢"
+    ]
 
     @classmethod
     def get_input_schema(cls) -> Dict[str, Any]:

features/mcp/tools/geocode_tool.py

@@ -18,8 +18,9 @@ logger = logging.getLogger("mcp.tools.geocode")
 class ReverseGeocodeTool(MCPTool):
     NAME = "reverse_geocode"
     DESCRIPTION = "以經緯度反查城市/行政區（優先使用快取）"
-    CATEGORY = "地理"
+    CATEGORY = "地理定位"
     TAGS = ["geocode", "reverse", "city"]
+    KEYWORDS = ["座標", "經緯度", "反查", "地址", "我在哪"]
     USAGE_TIPS = ["提供 lat/lon 即可"]
 
     @classmethod

features/mcp/tools/geocoding_tool.py

@@ -18,8 +18,9 @@ logger = logging.getLogger("mcp.tools.geocoding")
 class ForwardGeocodeTool(MCPTool):
     NAME = "forward_geocode"
     DESCRIPTION = "將地點名稱（如「銘傳大學」「桃園火車站」）轉換為經緯度座標"
-    CATEGORY = "地理"
+    CATEGORY = "地理定位"
     TAGS = ["geocode", "forward", "地點", "座標"]
+    KEYWORDS = ["地點", "位置", "座標", "在哪裡", "地址查詢"]
     USAGE_TIPS = [
         "提供地點名稱即可（如「台北101」「淡水捷運站」）",
         "支援地標、車站、學校、商圈等",

features/mcp/tools/healthkit_tool.py

@@ -21,8 +21,14 @@ class HealthKitTool(MCPTool):
     
     NAME = "healthkit_query"
     DESCRIPTION = "查詢用戶的健康數據，包括心率、步數、血氧、呼吸頻率等（數據由iOS設備自動同步到Firestore）"
-    CATEGORY = "health"
+    CATEGORY = "健康數據"
     TAGS = ["health", "fitness", "database", "firestore"]
+    KEYWORDS = ["健康", "心率", "步數", "血氧", "睡眠", "health", "運動", "卡路里", "呼吸"]
+    USAGE_TIPS = [
+        "可查詢心率、步數、血氧、呼吸頻率、睡眠等",
+        "支援指定查詢天數（1-365天）",
+        "數據由 iOS 設備自動同步"
+    ]
     
     def __init__(self):
         super().__init__()

features/mcp/tools/news_tool.py

@@ -31,10 +31,15 @@ class NewsTool(MCPTool):
     """新聞查詢 MCP 工具 - 使用 NewsData.io（更好的台灣與繁中新聞支援）"""
 
     NAME = "news_query"
-    DESCRIPTION = "查詢最新新聞資訊，支援關鍵詞搜尋、台灣新聞和分類篩選，提供可靠的繁體中文新聞來源"
-    CATEGORY = "新聞"
-    TAGS = ["news", "information", "current events", "taiwan", "chinese"]
-    USAGE_TIPS = ["直接說「科技新聞」或「今日新聞」", "支援台灣本地新聞", "繁體中文新聞源豐富"]
+    DESCRIPTION = "查詢最新新聞（可指定類別、語言、數量）"
+    CATEGORY = "生活資訊"
+    TAGS = ["news", "新聞", "資訊"]
+    KEYWORDS = ["新聞", "消息", "報導", "news", "頭條", "時事"]
+    USAGE_TIPS = [
+        "可指定新聞類別（科技、商業、娛樂等）",
+        "支援多國新聞（台灣、美國、日本等）",
+        "可限制返回數量"
+    ]
 
     @classmethod
     def get_input_schema(cls) -> Dict[str, Any]:

features/mcp/tools/tdx_base.py

@@ -0,0 +1,147 @@
+"""
+TDX 基礎工具類
+提供 OAuth 認證、API 呼叫、快取等共用功能
+"""
+
+import os
+import json
+import logging
+import aiohttp
+import asyncio
+from typing import Dict, Any, Optional
+from datetime import datetime, timedelta
+
+from .base_tool import ExecutionError
+from core.database.cache import db_cache
+
+logger = logging.getLogger("mcp.tools.tdx")
+
+TDX_BASE_URL = "https://tdx.transportdata.tw/api/basic/v2"
+TDX_CLIENT_ID = os.getenv("TDX_CLIENT_ID", "")
+TDX_CLIENT_SECRET = os.getenv("TDX_CLIENT_SECRET", "")
+
+
+class TDXBaseAPI:
+    """TDX API 基礎類別"""
+    
+    _token_cache: Dict[str, Any] = {}
+    
+    @classmethod
+    async def get_access_token(cls) -> str:
+        """獲取 TDX Access Token（快取 1 小時）"""
+        # 檢查快取
+        if cls._token_cache.get("token") and cls._token_cache.get("expires_at"):
+            if datetime.now() < cls._token_cache["expires_at"]:
+                return cls._token_cache["token"]
+        
+        if not TDX_CLIENT_ID or not TDX_CLIENT_SECRET:
+            raise ExecutionError("未設定 TDX_CLIENT_ID 或 TDX_CLIENT_SECRET 環境變數")
+        
+        # 請求新 token
+        auth_url = "https://tdx.transportdata.tw/auth/realms/TDXConnect/protocol/openid-connect/token"
+        data = {
+            "grant_type": "client_credentials",
+            "client_id": TDX_CLIENT_ID,
+            "client_secret": TDX_CLIENT_SECRET
+        }
+        
+        try:
+            async with aiohttp.ClientSession() as session:
+                async with session.post(auth_url, data=data, timeout=aiohttp.ClientTimeout(total=15)) as resp:
+                    if resp.status != 200:
+                        error_text = await resp.text()
+                        raise ExecutionError(f"TDX 認證失敗: HTTP {resp.status} - {error_text}")
+                    
+                    token_data = await resp.json()
+                    access_token = token_data.get("access_token")
+                    expires_in = token_data.get("expires_in", 3600)
+                    
+                    if not access_token:
+                        raise ExecutionError("TDX 認證回應缺少 access_token")
+                    
+                    # 快取（提前 60 秒過期）
+                    cls._token_cache = {
+                        "token": access_token,
+                        "expires_at": datetime.now() + timedelta(seconds=expires_in - 60)
+                    }
+                    
+                    logger.info("✅ TDX Access Token 取得成功")
+                    return access_token
+        
+        except aiohttp.ClientError as e:
+            raise ExecutionError(f"TDX 認證網路錯誤: {e}")
+    
+    @classmethod
+    async def call_api(cls, endpoint: str, params: Optional[Dict[str, Any]] = None, 
+                      cache_ttl: int = 60) -> Any:
+        """呼叫 TDX API 並處理快取"""
+        access_token = await cls.get_access_token()
+        
+        url = f"{TDX_BASE_URL}/{endpoint}"
+        headers = {
+            "Authorization": f"Bearer {access_token}",
+            "Accept": "application/json"
+        }
+        
+        # 生成快取鍵
+        cache_key = f"tdx:{endpoint}:{json.dumps(params or {}, sort_keys=True)}"
+        
+        # 檢查快取
+        if cache_ttl > 0:
+            cached = await db_cache.get_tdx_cached(cache_key)
+            if cached:
+                logger.debug(f"📦 TDX 快取命中: {endpoint}")
+                return cached
+        
+        # 呼叫 API
+        try:
+            async with aiohttp.ClientSession(headers=headers) as session:
+                async with session.get(url, params=params, timeout=aiohttp.ClientTimeout(total=20)) as resp:
+                    if resp.status == 304:
+                        logger.info("TDX 資料未變更 (304)")
+                        return cached if cached else []
+                    
+                    if resp.status != 200:
+                        error_text = await resp.text()
+                        raise ExecutionError(f"TDX API 錯誤 {endpoint}: HTTP {resp.status} - {error_text[:200]}")
+                    
+                    data = await resp.json()
+                    
+                    # 快取結果
+                    if cache_ttl > 0:
+                        await db_cache.set_tdx_cache(cache_key, data, ttl=cache_ttl)
+                    
+                    logger.info(f"✅ TDX API 成功: {endpoint}")
+                    return data
+        
+        except asyncio.TimeoutError:
+            raise ExecutionError(f"TDX API 逾時: {endpoint}")
+        except aiohttp.ClientError as e:
+            raise ExecutionError(f"TDX API 網路錯誤: {e}")
+    
+    @staticmethod
+    def haversine_distance(lat1: float, lon1: float, lat2: float, lon2: float) -> float:
+        """計算兩點間距離（公尺）"""
+        from math import radians, cos, sin, asin, sqrt
+        
+        R = 6371000  # 地球半徑（公尺）
+        lat1, lon1, lat2, lon2 = map(radians, [lat1, lon1, lat2, lon2])
+        
+        dlat = lat2 - lat1
+        dlon = lon2 - lon1
+        a = sin(dlat/2)**2 + cos(lat1) * cos(lat2) * sin(dlon/2)**2
+        c = 2 * asin(sqrt(a))
+        
+        return R * c
+    
+    @staticmethod
+    def format_datetime(dt_str: str) -> str:
+        """格式化 TDX 時間字串"""
+        if not dt_str:
+            return "未知"
+        try:
+            # TDX 格式: 2024-11-01T14:30:00+08:00
+            dt = datetime.fromisoformat(dt_str.replace('Z', '+00:00'))
+            return dt.strftime("%H:%M")
+        except:
+            return dt_str

features/mcp/tools/tdx_bus_arrival.py

@@ -0,0 +1,337 @@
+"""
+TDX 公車即時到站工具
+查詢附近公車站、特定路線到站時間
+"""
+
+import logging
+from typing import Dict, Any, List, Optional
+
+from .base_tool import MCPTool, StandardToolSchemas, ExecutionError
+from .tdx_base import TDXBaseAPI
+from core.database import get_user_env_current
+
+logger = logging.getLogger("mcp.tools.tdx.bus")
+
+
+class TDXBusArrivalTool(MCPTool):
+    """TDX 公車即時到站查詢"""
+    
+    NAME = "tdx_bus_arrival"
+    DESCRIPTION = "查詢公車即時到站時間（自動感知用戶位置，找最近站點）"
+    CATEGORY = "道路運輸"
+    TAGS = ["tdx", "公車", "即時到站", "公共運輸"]
+    KEYWORDS = ["公車", "巴士", "bus", "到站", "即時", "幾分鐘"]
+    USAGE_TIPS = [
+        "查詢特定路線: 「307 公車還要多久」",
+        "查詢附近公車站: 「附近有什麼公車」",
+        "指定城市: 「台北 307」「高雄紅30」"
+    ]
+    
+    @classmethod
+    def get_input_schema(cls) -> Dict[str, Any]:
+        return StandardToolSchemas.create_input_schema({
+            "route_name": {
+                "type": "string",
+                "description": "路線名稱（如「307」「紅30」）。不提供則查詢附近所有公車站"
+            },
+            "city": {
+                "type": "string",
+                "description": "城市（預設從環境感知自動判斷）",
+                "enum": ["Taipei", "NewTaipei", "Taoyuan", "Taichung", "Tainan", "Kaohsiung", 
+                        "Keelung", "Hsinchu", "HsinchuCounty", "MiaoliCounty", "ChanghuaCounty",
+                        "NantouCounty", "YunlinCounty", "ChiayiCounty", "Chiayi", "PingtungCounty",
+                        "YilanCounty", "HualienCounty", "TaitungCounty", "KinmenCounty", "PenghuCounty",
+                        "LienchiangCounty"]
+            },
+            "limit": {
+                "type": "integer",
+                "description": "返回結果數量上限",
+                "default": 5
+            }
+        }, required=[])
+    
+    @classmethod
+    def get_output_schema(cls) -> Dict[str, Any]:
+        schema = StandardToolSchemas.create_output_schema()
+        schema["properties"].update({
+            "arrivals": {
+                "type": "array",
+                "items": {
+                    "type": "object",
+                    "properties": {
+                        "route_name": {"type": "string"},
+                        "stop_name": {"type": "string"},
+                        "direction": {"type": "string"},
+                        "estimate_time": {"type": "integer"},
+                        "status": {"type": "string"}
+                    }
+                }
+            }
+        })
+        return schema
+    
+    @classmethod
+    async def execute(cls, arguments: Dict[str, Any], user_id: str = None) -> Dict[str, Any]:
+        route_name = arguments.get("route_name", "").strip()
+        city = arguments.get("city")
+        limit = min(int(arguments.get("limit", 5)), 20)
+        
+        # 1. 取得用戶位置
+        env_ctx = await get_user_env_current(user_id) if user_id else None
+        if not env_ctx or not env_ctx.get("success"):
+            if not route_name:
+                raise ExecutionError("無法取得您的位置，請提供路線名稱或開啟定位權限")
+            user_lat, user_lon, user_city = None, None, None
+        else:
+            ctx = env_ctx.get("context", {})
+            user_lat = ctx.get("lat")
+            user_lon = ctx.get("lon")
+            user_city = ctx.get("city", "")
+        
+        # 2. 自動判斷城市
+        if not city:
+            city = cls._map_city_name(user_city) if user_city else "Taipei"
+        
+        # 3. 查詢邏輯分支
+        if route_name:
+            # 指定路線：找最近站點並查詢到站時間
+            result = await cls._query_route_arrival(route_name, city, user_lat, user_lon, limit)
+        else:
+            # 未指定路線：查詢附近所有站點
+            if not user_lat or not user_lon:
+                raise ExecutionError("查詢附近公車需要定位權限")
+            result = await cls._query_nearby_stops(user_lat, user_lon, city, limit)
+        
+        return result
+    
+    @classmethod
+    async def _query_route_arrival(cls, route_name: str, city: str, 
+                                   user_lat: Optional[float], user_lon: Optional[float],
+                                   limit: int) -> Dict[str, Any]:
+        """查詢特定路線的即時到站"""
+        # 1. 查詢路線基本資訊
+        route_endpoint = f"Bus/Route/City/{city}"
+        route_params = {
+            "$filter": f"contains(RouteName/Zh_tw, '{route_name}')",
+            "$format": "JSON",
+            "$top": 1
+        }
+        
+        routes = await TDXBaseAPI.call_api(route_endpoint, route_params, cache_ttl=3600)
+        
+        if not routes or len(routes) == 0:
+            raise ExecutionError(f"找不到路線「{route_name}」���請確認路線名稱是否正確")
+        
+        route = routes[0]
+        route_uid = route.get("RouteUID")
+        full_route_name = route.get("RouteName", {}).get("Zh_tw", route_name)
+        
+        # 2. 查詢該路線所有站點
+        stop_endpoint = f"Bus/StopOfRoute/City/{city}"
+        stop_params = {
+            "$filter": f"RouteUID eq '{route_uid}'",
+            "$format": "JSON"
+        }
+        
+        stops = await TDXBaseAPI.call_api(stop_endpoint, stop_params, cache_ttl=1800)
+        
+        if not stops:
+            raise ExecutionError(f"路線「{full_route_name}」暫無站點資訊")
+        
+        # 3. 如果有用戶位置，找最近的站點
+        if user_lat and user_lon:
+            for stop_seq in stops:
+                for stop in stop_seq.get("Stops", []):
+                    pos = stop.get("StopPosition", {})
+                    if pos.get("PositionLat") and pos.get("PositionLon"):
+                        stop["distance_m"] = TDXBaseAPI.haversine_distance(
+                            user_lat, user_lon,
+                            pos["PositionLat"], pos["PositionLon"]
+                        )
+            
+            # 取前 3 個最近的站點
+            all_stops = []
+            for stop_seq in stops:
+                all_stops.extend(stop_seq.get("Stops", []))
+            
+            all_stops = [s for s in all_stops if "distance_m" in s]
+            all_stops.sort(key=lambda x: x["distance_m"])
+            target_stops = all_stops[:3]
+        else:
+            # 沒有位置，取前幾個站點
+            target_stops = []
+            for stop_seq in stops[:1]:
+                target_stops.extend(stop_seq.get("Stops", [])[:limit])
+        
+        # 4. 查詢這些站點的即時到站
+        arrivals = []
+        for stop in target_stops:
+            stop_uid = stop.get("StopUID")
+            stop_name = stop.get("StopName", {}).get("Zh_tw", "未知")
+            
+            arrival_endpoint = f"Bus/EstimatedTimeOfArrival/City/{city}"
+            arrival_params = {
+                "$filter": f"RouteUID eq '{route_uid}' and StopUID eq '{stop_uid}'",
+                "$format": "JSON"
+            }
+            
+            arrival_data = await TDXBaseAPI.call_api(arrival_endpoint, arrival_params, cache_ttl=30)
+            
+            for arr in arrival_data[:2]:  # 每個站點最多 2 筆（雙向）
+                estimate_time = arr.get("EstimateTime")
+                stop_status = arr.get("StopStatus", 0)
+                
+                if stop_status == 0:  # 正常
+                    status_text = f"{estimate_time // 60} 分鐘" if estimate_time else "即將進站"
+                elif stop_status == 1:  # 尚未發車
+                    status_text = "尚未發車"
+                elif stop_status == 2:  # 交管不停靠
+                    status_text = "交管不停靠"
+                elif stop_status == 3:  # 末班車已過
+                    status_text = "末班車已過"
+                elif stop_status == 4:  # 今日未營運
+                    status_text = "今日未營運"
+                else:
+                    status_text = "未知"
+                
+                arrivals.append({
+                    "route_name": full_route_name,
+                    "stop_name": stop_name,
+                    "direction": arr.get("Direction", 0),
+                    "estimate_time": estimate_time,
+                    "status": status_text,
+                    "distance_m": stop.get("distance_m", 0)
+                })
+        
+        # 5. 格式化回覆
+        content = cls._format_arrival_result(arrivals, full_route_name, user_lat is not None)
+        
+        return cls.create_success_response(
+            content=content,
+            data={"arrivals": arrivals, "route_name": full_route_name}
+        )
+    
+    @classmethod
+    async def _query_nearby_stops(cls, lat: float, lon: float, city: str, limit: int) -> Dict[str, Any]:
+        """查詢附近公車站"""
+        # TDX 附近站點查詢
+        endpoint = f"Bus/Stop/City/{city}"
+        params = {
+            "$spatialFilter": f"nearby({lat}, {lon}, 300)",  # 300m 範圍
+            "$format": "JSON",
+            "$top": limit * 3  # 多取一些，後續過濾
+        }
+        
+        stops = await TDXBaseAPI.call_api(endpoint, params, cache_ttl=1800)
+        
+        if not stops:
+            return cls.create_success_response(
+                content="附近 300 公尺內沒有公車站，請擴大範圍或移動位置",
+                data={"stops": []}
+            )
+        
+        # 計算距離並排序
+        for stop in stops:
+            pos = stop.get("StopPosition", {})
+            if pos.get("PositionLat") and pos.get("PositionLon"):
+                stop["distance_m"] = TDXBaseAPI.haversine_distance(
+                    lat, lon,
+                    pos["PositionLat"], pos["PositionLon"]
+                )
+        
+        stops = [s for s in stops if "distance_m" in s]
+        stops.sort(key=lambda x: x["distance_m"])
+        stops = stops[:limit]
+        
+        # 格式化結果
+        results = []
+        for stop in stops:
+            stop_name = stop.get("StopName", {}).get("Zh_tw", "未知")
+            distance = stop["distance_m"]
+            walking_time = int(distance / 80)  # 80m/min
+            
+            results.append({
+                "stop_name": stop_name,
+                "distance_m": int(distance),
+                "walking_time_min": walking_time,
+                "stop_uid": stop.get("StopUID")
+            })
+        
+        content = cls._format_nearby_result(results)
+        
+        return cls.create_success_response(
+            content=content,
+            data={"stops": results}
+        )
+    
+    @staticmethod
+    def _map_city_name(chinese_city: str) -> str:
+        """中文城市名稱轉 TDX 代碼"""
+        city_map = {
+            "台北": "Taipei", "臺北": "Taipei",
+            "新北": "NewTaipei", "新北市": "NewTaipei",
+            "桃園": "Taoyuan",
+            "台中": "Taichung", "臺中": "Taichung",
+            "台南": "Tainan", "臺南": "Tainan",
+            "高雄": "Kaohsiung",
+            "基隆": "Keelung",
+            "新竹": "Hsinchu",
+            "嘉義": "Chiayi"
+        }
+        
+        for key, value in city_map.items():
+            if key in chinese_city:
+                return value
+        
+        return "Taipei"  # 預設台北
+    
+    @staticmethod
+    def _format_arrival_result(arrivals: List[Dict], route_name: str, has_location: bool) -> str:
+        """格式化到站結果"""
+        if not arrivals:
+            return f"路線 {route_name} 目前無即時到站資訊"
+        
+        lines = [f"🚌 {route_name} 即時到站資訊：\n"]
+        
+        # 按站點分組
+        stops_dict = {}
+        for arr in arrivals:
+            stop = arr["stop_name"]
+            if stop not in stops_dict:
+                stops_dict[stop] = []
+            stops_dict[stop].append(arr)
+        
+        for i, (stop_name, stop_arrivals) in enumerate(stops_dict.items(), 1):
+            dist_info = ""
+            if has_location and stop_arrivals[0].get("distance_m"):
+                dist = stop_arrivals[0]["distance_m"]
+                walk_time = int(dist / 80)
+                dist_info = f" - 步行 {walk_time} 分鐘 ({int(dist)}m)"
+            
+            lines.append(f"{i}. 🚏 {stop_name}{dist_info}")
+            
+            for arr in stop_arrivals:
+                direction = "往 ↑" if arr["direction"] == 0 else "返 ↓"
+                lines.append(f"   {direction} {arr['status']}")
+            
+            lines.append("")
+        
+        return "\n".join(lines)
+    
+    @staticmethod
+    def _format_nearby_result(stops: List[Dict]) -> str:
+        """格式化附近站點結果"""
+        if not stops:
+            return "附近沒有找到公車站"
+        
+        lines = ["📍 附近的公車站：\n"]
+        
+        for i, stop in enumerate(stops, 1):
+            lines.append(
+                f"{i}. 🚏 {stop['stop_name']}\n"
+                f"   步行 {stop['walking_time_min']} 分鐘 ({stop['distance_m']}m)\n"
+            )
+        
+        lines.append("💡 提供路線名稱查詢即時到站時間")
+        
+        return "\n".join(lines)

features/mcp/tools/tdx_metro.py

@@ -0,0 +1,299 @@
+"""
+TDX 捷運即時資訊工具
+支援台北捷運、高雄捷運、桃園捷運、台中捷運
+"""
+
+import logging
+from typing import Dict, Any, List, Optional
+
+from .base_tool import MCPTool, StandardToolSchemas, ExecutionError
+from .tdx_base import TDXBaseAPI
+from core.database import get_user_env_current
+
+logger = logging.getLogger("mcp.tools.tdx.metro")
+
+
+class TDXMetroTool(MCPTool):
+    """TDX 捷運即時到站查詢"""
+    
+    NAME = "tdx_metro"
+    DESCRIPTION = "查詢捷運即時到站、最近車站（台北/高雄/桃園/台中捷運）"
+    CATEGORY = "軌道運輸"
+    TAGS = ["tdx", "捷運", "MRT", "即時到站"]
+    KEYWORDS = ["捷運", "MRT", "地鐵", "metro", "到站"]
+    USAGE_TIPS = [
+        "查詢最近捷運站: 「最近的捷運站在哪」",
+        "查詢特定站點: 「台北車站捷運幾分鐘到」",
+        "指定路線: 「板南線 市政府站」"
+    ]
+    
+    # TDX 捷運系統對應
+    METRO_SYSTEMS = {
+        "台北": "TRTC",
+        "臺北": "TRTC",
+        "高雄": "KRTC",
+        "桃園": "TYMC",
+        "台中": "TMRT",
+        "臺中": "TMRT"
+    }
+    
+    @classmethod
+    def get_input_schema(cls) -> Dict[str, Any]:
+        return StandardToolSchemas.create_input_schema({
+            "station_name": {
+                "type": "string",
+                "description": "車站名稱（如「台北車站」「西門站」）。不提供則查詢最近車站"
+            },
+            "metro_system": {
+                "type": "string",
+                "description": "捷運系統（TRTC=台北, KRTC=高雄, TYMC=桃園, TMRT=台中）",
+                "enum": ["TRTC", "KRTC", "TYMC", "TMRT"]
+            },
+            "line": {
+                "type": "string",
+                "description": "路線名稱（如「板南線」「淡水信義線」）"
+            }
+        }, required=[])
+    
+    @classmethod
+    def get_output_schema(cls) -> Dict[str, Any]:
+        schema = StandardToolSchemas.create_output_schema()
+        schema["properties"].update({
+            "arrivals": {
+                "type": "array",
+                "items": {
+                    "type": "object",
+                    "properties": {
+                        "station_name": {"type": "string"},
+                        "line_name": {"type": "string"},
+                        "destination": {"type": "string"},
+                        "arrival_time_sec": {"type": "integer"},
+                        "train_status": {"type": "string"}
+                    }
+                }
+            }
+        })
+        return schema
+    
+    @classmethod
+    async def execute(cls, arguments: Dict[str, Any], user_id: str = None) -> Dict[str, Any]:
+        station_name = arguments.get("station_name", "").strip()
+        metro_system = arguments.get("metro_system")
+        line_filter = arguments.get("line")
+        
+        # 1. 取得用戶位置
+        env_ctx = await get_user_env_current(user_id) if user_id else None
+        if not env_ctx or not env_ctx.get("success"):
+            if not station_name:
+                raise ExecutionError("無法取得您的位置，請提供車站名稱")
+            user_lat, user_lon, user_city = None, None, None
+        else:
+            ctx = env_ctx.get("context", {})
+            user_lat = ctx.get("lat")
+            user_lon = ctx.get("lon")
+            user_city = ctx.get("city", "")
+        
+        # 2. 自動判斷捷運系統
+        if not metro_system:
+            metro_system = cls._detect_metro_system(user_city)
+        
+        # 3. 查詢邏輯
+        if station_name:
+            result = await cls._query_station_arrival(station_name, metro_system, line_filter)
+        else:
+            if not user_lat or not user_lon:
+                raise ExecutionError("查詢最近捷運站需要定位權限")
+            result = await cls._query_nearest_station(user_lat, user_lon, metro_system)
+        
+        return result
+    
+    @classmethod
+    async def _query_station_arrival(cls, station_name: str, metro_system: str, 
+                                     line_filter: Optional[str]) -> Dict[str, Any]:
+        """查詢特定車站的即時到站"""
+        # 1. 查詢車站資訊
+        station_endpoint = f"Metro/Station/{metro_system}"
+        station_params = {
+            "$filter": f"contains(StationName/Zh_tw, '{station_name}')",
+            "$format": "JSON",
+            "$top": 5
+        }
+        
+        stations = await TDXBaseAPI.call_api(station_endpoint, station_params, cache_ttl=3600)
+        
+        if not stations:
+            raise ExecutionError(f"找不到車站「{station_name}」")
+        
+        # 2. 如果有多個結果，優先選擇完全匹配
+        target_station = None
+        for station in stations:
+            name = station.get("StationName", {}).get("Zh_tw", "")
+            if name == station_name:
+                target_station = station
+                break
+        
+        if not target_station:
+            target_station = stations[0]
+        
+        station_uid = target_station.get("StationUID")
+        full_station_name = target_station.get("StationName", {}).get("Zh_tw", station_name)
+        
+        # 3. 查詢即時到站
+        arrival_endpoint = f"Metro/LiveBoard/{metro_system}"
+        arrival_params = {
+            "$filter": f"StationUID eq '{station_uid}'",
+            "$format": "JSON"
+        }
+        
+        arrivals = await TDXBaseAPI.call_api(arrival_endpoint, arrival_params, cache_ttl=15)
+        
+        if not arrivals:
+            return cls.create_success_response(
+                content=f"🚇 {full_station_name} 目前無即時到站資訊",
+                data={"arrivals": []}
+            )
+        
+        # 4. 路線過濾
+        if line_filter:
+            arrivals = [a for a in arrivals if line_filter in a.get("LineName", {}).get("Zh_tw", "")]
+        
+        # 5. 格式化結果
+        results = []
+        for arr in arrivals[:10]:  # 最多 10 筆
+            line_name = arr.get("LineName", {}).get("Zh_tw", "未知路線")
+            dest = arr.get("DestinationStationName", {}).get("Zh_tw", "未知")
+            arrival_time = arr.get("ArrivalTime", 0)
+            status_code = arr.get("TrainStatus", 0)
+            
+            status_map = {
+                0: "正常",
+                1: "尚未發車",
+                2: "交管不停靠",
+                3: "末班車已過",
+                4: "今日未營運"
+            }
+            status = status_map.get(status_code, "未知")
+            
+            results.append({
+                "station_name": full_station_name,
+                "line_name": line_name,
+                "destination": dest,
+                "arrival_time_sec": arrival_time,
+                "train_status": status
+            })
+        
+        content = cls._format_arrival_result(results, full_station_name)
+        
+        return cls.create_success_response(
+            content=content,
+            data={"arrivals": results}
+        )
+    
+    @classmethod
+    async def _query_nearest_station(cls, lat: float, lon: float, metro_system: str) -> Dict[str, Any]:
+        """查詢最近的捷運站"""
+        # 1. 取得所有車站
+        station_endpoint = f"Metro/Station/{metro_system}"
+        station_params = {
+            "$format": "JSON"
+        }
+        
+        stations = await TDXBaseAPI.call_api(station_endpoint, station_params, cache_ttl=3600)
+        
+        if not stations:
+            raise ExecutionError("無法取得捷運站資訊")
+        
+        # 2. 計算距離
+        for station in stations:
+            pos = station.get("StationPosition", {})
+            if pos.get("PositionLat") and pos.get("PositionLon"):
+                station["distance_m"] = TDXBaseAPI.haversine_distance(
+                    lat, lon,
+                    pos["PositionLat"], pos["PositionLon"]
+                )
+        
+        stations_with_distance = [s for s in stations if "distance_m" in s]
+        
+        if not stations_with_distance:
+            raise ExecutionError("附近沒有捷運站資訊")
+        
+        stations_with_distance.sort(key=lambda x: x["distance_m"])
+        nearest = stations_with_distance[:3]
+        
+        # 3. 格式化結果
+        results = []
+        for station in nearest:
+            station_name = station.get("StationName", {}).get("Zh_tw", "未知")
+            distance = station["distance_m"]
+            walking_time = int(distance / 80)
+            
+            results.append({
+                "station_name": station_name,
+                "distance_m": int(distance),
+                "walking_time_min": walking_time,
+                "station_uid": station.get("StationUID"),
+                "address": station.get("StationAddress", "")
+            })
+        
+        content = cls._format_nearest_result(results)
+        
+        return cls.create_success_response(
+            content=content,
+            data={"stations": results}
+        )
+    
+    @staticmethod
+    def _detect_metro_system(city: str) -> str:
+        """根據城市自動偵測捷運系統"""
+        for key, code in TDXMetroTool.METRO_SYSTEMS.items():
+            if key in city:
+                return code
+        return "TRTC"  # 預設台北
+    
+    @staticmethod
+    def _format_arrival_result(arrivals: List[Dict], station_name: str) -> str:
+        """格式化到站資訊"""
+        if not arrivals:
+            return f"🚇 {station_name} 目前無列車資訊"
+        
+        lines = [f"🚇 {station_name} 即時到站：\n"]
+        
+        # 按路線分組
+        lines_dict = {}
+        for arr in arrivals:
+            line = arr["line_name"]
+            if line not in lines_dict:
+                lines_dict[line] = []
+            lines_dict[line].append(arr)
+        
+        for line_name, line_arrivals in lines_dict.items():
+            lines.append(f"━�� {line_name} ━━")
+            
+            for arr in line_arrivals[:3]:  # 每條路線最多 3 筆
+                dest = arr["destination"]
+                time_sec = arr["arrival_time_sec"]
+                status = arr["train_status"]
+                
+                if time_sec > 0:
+                    time_min = time_sec // 60
+                    time_str = f"{time_min} 分 {time_sec % 60} 秒" if time_min > 0 else f"{time_sec} 秒"
+                    lines.append(f"  → {dest}  {time_str}")
+                else:
+                    lines.append(f"  → {dest}  {status}")
+            
+            lines.append("")
+        
+        return "\n".join(lines)
+    
+    @staticmethod
+    def _format_nearest_result(stations: List[Dict]) -> str:
+        """格式化最近車站結果"""
+        lines = ["📍 最近的捷運站：\n"]
+        
+        for i, station in enumerate(stations, 1):
+            lines.append(
+                f"{i}. 🚇 {station['station_name']}\n"
+                f"   步行 {station['walking_time_min']} 分鐘 ({station['distance_m']}m)\n"
+            )
+        
+        return "\n".join(lines)

features/mcp/tools/tdx_parking.py

@@ -0,0 +1,410 @@
+"""
+TDX 停車場與充電站查詢工具
+查詢附近停車場、即時剩餘車位、充電站資訊
+"""
+
+import logging
+from typing import Dict, Any, List, Optional
+
+from .base_tool import MCPTool, StandardToolSchemas, ExecutionError
+from .tdx_base import TDXBaseAPI
+from core.database import get_user_env_current
+
+logger = logging.getLogger("mcp.tools.tdx.parking")
+
+
+class TDXParkingTool(MCPTool):
+    """TDX 停車場與充電站查詢"""
+    
+    NAME = "tdx_parking"
+    DESCRIPTION = "查詢附近停車場、即時剩餘車位、收費標準、充電站資訊"
+    CATEGORY = "停車與充電"
+    TAGS = ["tdx", "停車", "充電站", "電動車"]
+    KEYWORDS = ["停車", "停車場", "充電", "充電站", "車位", "電動車"]
+    USAGE_TIPS = [
+        "查詢附近停車場: 「附近哪裡有停車位」",
+        "查詢充電站: 「附近的充電站在哪」",
+        "指定停車場: 「台北車站停車場還有位子嗎」"
+    ]
+    
+    @classmethod
+    def get_input_schema(cls) -> Dict[str, Any]:
+        return StandardToolSchemas.create_input_schema({
+            "parking_name": {
+                "type": "string",
+                "description": "停車場名稱（如「台北車站」「市政府」）。不提供則查詢附近停車場"
+            },
+            "city": {
+                "type": "string",
+                "description": "城市代碼（如「Taipei」「Kaohsiung」）",
+                "enum": ["Taipei", "NewTaipei", "Taoyuan", "Taichung", "Tainan", "Kaohsiung"]
+            },
+            "parking_type": {
+                "type": "string",
+                "description": "停車場類型",
+                "enum": ["路邊", "路外"]
+            },
+            "charge_station": {
+                "type": "boolean",
+                "description": "是否只查詢有充電站的停車場",
+                "default": False
+            },
+            "radius_m": {
+                "type": "integer",
+                "description": "搜尋半徑（公尺）",
+                "default": 1000
+            },
+            "limit": {
+                "type": "integer",
+                "description": "返回結果數量",
+                "default": 5
+            }
+        }, required=[])
+    
+    @classmethod
+    def get_output_schema(cls) -> Dict[str, Any]:
+        schema = StandardToolSchemas.create_output_schema()
+        schema["properties"].update({
+            "parkings": {
+                "type": "array",
+                "items": {
+                    "type": "object",
+                    "properties": {
+                        "parking_name": {"type": "string"},
+                        "available_spaces": {"type": "integer"},
+                        "total_spaces": {"type": "integer"},
+                        "charge_station": {"type": "boolean"},
+                        "fee_info": {"type": "string"}
+                    }
+                }
+            }
+        })
+        return schema
+    
+    @classmethod
+    async def execute(cls, arguments: Dict[str, Any], user_id: str = None) -> Dict[str, Any]:
+        parking_name = arguments.get("parking_name", "").strip()
+        city = arguments.get("city")
+        parking_type = arguments.get("parking_type")
+        charge_station_only = arguments.get("charge_station", False)
+        radius_m = min(int(arguments.get("radius_m", 1000)), 5000)
+        limit = min(int(arguments.get("limit", 5)), 20)
+        
+        # 1. 取得用戶位置
+        env_ctx = await get_user_env_current(user_id) if user_id else None
+        if not env_ctx or not env_ctx.get("success"):
+            if not parking_name:
+                raise ExecutionError("無法取得您的位置，請提供停車場名稱或開啟定位權限")
+            user_lat, user_lon, user_city = None, None, None
+        else:
+            ctx = env_ctx.get("context", {})
+            user_lat = ctx.get("lat")
+            user_lon = ctx.get("lon")
+            user_city = ctx.get("city", "")
+        
+        # 2. 自動判斷城市
+        if not city:
+            city = cls._map_city_name(user_city) if user_city else "Taipei"
+        
+        # 3. 查詢分支
+        if charge_station_only:
+            # 查詢充電站
+            if not user_lat or not user_lon:
+                raise ExecutionError("查詢充電站需要定位權限")
+            result = await cls._query_charge_stations(user_lat, user_lon, city, radius_m, limit)
+        elif parking_name:
+            # 查詢特定停車場
+            result = await cls._query_parking_availability(parking_name, city)
+        else:
+            # 查詢附近停車場
+            if not user_lat or not user_lon:
+                raise ExecutionError("查詢附近停車場需要定位權限")
+            result = await cls._query_nearby_parkings(user_lat, user_lon, city, parking_type, radius_m, limit)
+        
+        return result
+    
+    @classmethod
+    async def _query_parking_availability(cls, parking_name: str, city: str) -> Dict[str, Any]:
+        """查詢特定停車場即時資訊"""
+        # 1. 查詢停車場基本資訊
+        parking_endpoint = f"Parking/OffStreet/CarPark/City/{city}"
+        parking_params = {
+            "$filter": f"contains(CarParkName/Zh_tw, '{parking_name}')",
+            "$format": "JSON",
+            "$top": 5
+        }
+        
+        parkings = await TDXBaseAPI.call_api(parking_endpoint, parking_params, cache_ttl=3600)
+        
+        if not parkings:
+            raise ExecutionError(f"找不到停車場「{parking_name}」")
+        
+        # 2. 取得第一個結果
+        parking = parkings[0]
+        parking_id = parking.get("CarParkID")
+        full_parking_name = parking.get("CarParkName", {}).get("Zh_tw", parking_name)
+        
+        # 3. 查詢即時剩餘車位
+        avail_endpoint = f"Parking/OffStreet/ParkingAvailability/City/{city}"
+        avail_params = {
+            "$filter": f"CarParkID eq '{parking_id}'",
+            "$format": "JSON"
+        }
+        
+        availability = await TDXBaseAPI.call_api(avail_endpoint, avail_params, cache_ttl=60)
+        
+        # 4. 組合資訊
+        total_spaces = parking.get("TotalSpaces", 0)
+        available_spaces = 0
+        
+        if availability and len(availability) > 0:
+            avail = availability[0]
+            available_spaces = avail.get("AvailableSpaces", 0)
+        
+        # 收費資訊
+        fee_info = cls._format_fee_info(parking.get("FareDescription", {}))
+        
+        # 充電站資訊
+        has_charge = parking.get("HasChargingPoint", False)
+        
+        result = {
+            "parking_name": full_parking_name,
+            "available_spaces": available_spaces,
+            "total_spaces": total_spaces,
+            "charge_station": has_charge,
+            "fee_info": fee_info,
+            "address": parking.get("Address", ""),
+            "service_time": parking.get("ServiceTime", "")
+        }
+        
+        # 5. 格式化結果
+        content = (
+            f"🅿️ {result['parking_name']}\n"
+            f"剩餘車位: {result['available_spaces']} / {result['total_spaces']}\n"
+            f"收費: {result['fee_info']}\n"
+            f"充電站: {'有' if result['charge_station'] else '無'}\n"
+            f"地址: {result['address']}\n"
+        )
+        
+        return cls.create_success_response(
+            content=content,
+            data={"parking": result}
+        )
+    
+    @classmethod
+    async def _query_nearby_parkings(cls, lat: float, lon: float, city: str,
+                                     parking_type: Optional[str], radius_m: int, limit: int) -> Dict[str, Any]:
+        """查詢附近停車場"""
+        # 1. 查詢附近停車場
+        if parking_type == "路邊":
+            parking_endpoint = f"Parking/OnStreet/ParkingSpace/City/{city}"
+        else:
+            parking_endpoint = f"Parking/OffStreet/CarPark/City/{city}"
+        
+        parking_params = {
+            "$spatialFilter": f"nearby({lat}, {lon}, {radius_m})",
+            "$format": "JSON",
+            "$top": limit * 2
+        }
+        
+        parkings = await TDXBaseAPI.call_api(parking_endpoint, parking_params, cache_ttl=3600)
+        
+        if not parkings:
+            return cls.create_success_response(
+                content=f"附近 {radius_m} 公尺內沒有停車場",
+                data={"parkings": []}
+            )
+        
+        # 2. 計算距離並排序
+        for parking in parkings:
+            pos = parking.get("Position", {})
+            if pos.get("PositionLat") and pos.get("PositionLon"):
+                parking["distance_m"] = TDXBaseAPI.haversine_distance(
+                    lat, lon,
+                    pos["PositionLat"], pos["PositionLon"]
+                )
+        
+        parkings = [p for p in parkings if "distance_m" in p]
+        parkings.sort(key=lambda x: x["distance_m"])
+        parkings = parkings[:limit]
+        
+        # 3. 批次查詢即時車位（僅路外停車場）
+        if parking_type != "路邊":
+            parking_ids = [p.get("CarParkID") for p in parkings]
+            
+            avail_endpoint = f"Parking/OffStreet/ParkingAvailability/City/{city}"
+            avail_params = {
+                "$filter": " or ".join([f"CarParkID eq '{pid}'" for pid in parking_ids if pid]),
+                "$format": "JSON"
+            }
+            
+            availability = await TDXBaseAPI.call_api(avail_endpoint, avail_params, cache_ttl=60)
+            
+            # 建立 ID -> 可用性 映射
+            avail_map = {a.get("CarParkID"): a for a in availability}
+        else:
+            avail_map = {}
+        
+        # 4. 組合結果
+        results = []
+        for parking in parkings:
+            parking_id = parking.get("CarParkID") or parking.get("ParkingSpaceID")
+            parking_name = (parking.get("CarParkName") or parking.get("ParkingName") or {}).get("Zh_tw", "未知")
+            distance = parking["distance_m"]
+            walking_time = int(distance / 80)
+            
+            avail = avail_map.get(parking_id, {})
+            total_spaces = parking.get("TotalSpaces", 0)
+            available_spaces = avail.get("AvailableSpaces", 0)
+            
+            fee_info = cls._format_fee_info(parking.get("FareDescription", {}))
+            
+            results.append({
+                "parking_name": parking_name,
+                "available_spaces": available_spaces,
+                "total_spaces": total_spaces,
+                "distance_m": int(distance),
+                "walking_time_min": walking_time,
+                "charge_station": parking.get("HasChargingPoint", False),
+                "fee_info": fee_info
+            })
+        
+        content = cls._format_nearby_result(results, parking_type)
+        
+        return cls.create_success_response(
+            content=content,
+            data={"parkings": results}
+        )
+    
+    @classmethod
+    async def _query_charge_stations(cls, lat: float, lon: float, city: str,
+                                    radius_m: int, limit: int) -> Dict[str, Any]:
+        """查詢附近充電站"""
+        # 查詢有充電站的停車場
+        parking_endpoint = f"Parking/OffStreet/CarPark/City/{city}"
+        parking_params = {
+            "$filter": "HasChargingPoint eq true",
+            "$format": "JSON"
+        }
+        
+        parkings = await TDXBaseAPI.call_api(parking_endpoint, parking_params, cache_ttl=3600)
+        
+        if not parkings:
+            return cls.create_success_response(
+                content="此區域無充電站資訊",
+                data={"charge_stations": []}
+            )
+        
+        # 計算距離並過濾
+        for parking in parkings:
+            pos = parking.get("Position", {})
+            if pos.get("PositionLat") and pos.get("PositionLon"):
+                parking["distance_m"] = TDXBaseAPI.haversine_distance(
+                    lat, lon,
+                    pos["PositionLat"], pos["PositionLon"]
+                )
+        
+        parkings = [p for p in parkings if "distance_m" in p and p["distance_m"] <= radius_m]
+        parkings.sort(key=lambda x: x["distance_m"])
+        parkings = parkings[:limit]
+        
+        if not parkings:
+            return cls.create_success_response(
+                content=f"附近 {radius_m} 公尺內沒有充電站",
+                data={"charge_stations": []}
+            )
+        
+        # 格式化結果
+        results = []
+        for parking in parkings:
+            parking_name = parking.get("CarParkName", {}).get("Zh_tw", "未知")
+            distance = parking["distance_m"]
+            walking_time = int(distance / 80)
+            
+            results.append({
+                "parking_name": parking_name,
+                "distance_m": int(distance),
+                "walking_time_min": walking_time,
+                "address": parking.get("Address", ""),
+                "total_spaces": parking.get("TotalSpaces", 0)
+            })
+        
+        content = cls._format_charge_result(results)
+        
+        return cls.create_success_response(
+            content=content,
+            data={"charge_stations": results}
+        )
+    
+    @staticmethod
+    def _map_city_name(chinese_city: str) -> str:
+        """中文城市名稱轉 TDX 代碼"""
+        city_map = {
+            "台北": "Taipei", "臺北": "Taipei",
+            "新北": "NewTaipei",
+            "桃園": "Taoyuan",
+            "台中": "Taichung", "臺中": "Taichung",
+            "台南": "Tainan", "臺南": "Tainan",
+            "高雄": "Kaohsiung"
+        }
+        
+        for key, value in city_map.items():
+            if key in chinese_city:
+                return value
+        
+        return "Taipei"
+    
+    @staticmethod
+    def _format_fee_info(fare_desc: Dict) -> str:
+        """格式化收費資訊"""
+        if not fare_desc:
+            return "未提供"
+        
+        zh_tw = fare_desc.get("Zh_tw", "")
+        if zh_tw:
+            # 簡化長文字
+            if len(zh_tw) > 50:
+                return zh_tw[:47] + "..."
+            return zh_tw
+        
+        return "未提供"
+    
+    @staticmethod
+    def _format_nearby_result(parkings: List[Dict], parking_type: Optional[str]) -> str:
+        """格式化附近停車場結果"""
+        if not parkings:
+            return "附近沒有停車場"
+        
+        type_emoji = "🅿️" if parking_type == "路外" else "🚗"
+        lines = [f"📍 附近的停車場：\n"]
+        
+        for i, parking in enumerate(parkings, 1):
+            charge_emoji = "⚡" if parking["charge_station"] else ""
+            
+            if parking["total_spaces"] > 0:
+                avail_info = f"剩餘 {parking['available_spaces']}/{parking['total_spaces']}"
+            else:
+                avail_info = "無車位資訊"
+            
+            lines.append(
+                f"{i}. {type_emoji} {parking['parking_name']} {charge_emoji}\n"
+                f"   {avail_info}\n"
+                f"   {parking['fee_info']}\n"
+                f"   步行 {parking['walking_time_min']} 分鐘 ({parking['distance_m']}m)\n"
+            )
+        
+        return "\n".join(lines)
+    
+    @staticmethod
+    def _format_charge_result(stations: List[Dict]) -> str:
+        """格式化充電站結果"""
+        lines = ["⚡ 附近的充電站：\n"]
+        
+        for i, station in enumerate(stations, 1):
+            lines.append(
+                f"{i}. {station['parking_name']}\n"
+                f"   步行 {station['walking_time_min']} 分鐘 ({station['distance_m']}m)\n"
+                f"   {station['address']}\n"
+            )
+        
+        return "\n".join(lines)

features/mcp/tools/tdx_thsr.py

@@ -0,0 +1,402 @@
+"""
+TDX 台灣高鐵查詢工具
+查詢高鐵時刻表、票價、最近車站
+"""
+
+import logging
+from typing import Dict, Any, List, Optional
+from datetime import datetime, timedelta
+
+from .base_tool import MCPTool, StandardToolSchemas, ExecutionError
+from .tdx_base import TDXBaseAPI
+from core.database import get_user_env_current
+
+logger = logging.getLogger("mcp.tools.tdx.thsr")
+
+
+class TDXTHSRTool(MCPTool):
+    """TDX 台灣高鐵時刻表查詢"""
+    
+    NAME = "tdx_thsr"
+    DESCRIPTION = "查詢台灣高鐵時刻表、票價、最近車站（南港-左營）"
+    CATEGORY = "軌道運輸"
+    TAGS = ["tdx", "高鐵", "THSR", "時刻表", "票價"]
+    KEYWORDS = ["高鐵", "THSR", "HSR", "高速鐵路", "時刻", "票價"]
+    USAGE_TIPS = [
+        "查詢車次: 「高鐵 123 次」",
+        "查詢路線: 「台北到台中的高鐵」",
+        "查詢最近車站: 「最近的高鐵站在哪」",
+        "查詢時刻: 「下午2點台北到高雄的高鐵」"
+    ]
+    
+    # 高鐵車站代碼對照
+    STATION_MAP = {
+        "南港": "NAG", "台北": "TPE", "臺北": "TPE", "板橋": "BAC",
+        "桃園": "TAY", "新竹": "HSC", "苗栗": "MIA", "台中": "TAC", 
+        "臺中": "TAC", "彰化": "CHA", "雲林": "YUL", "嘉義": "CHY",
+        "台南": "TNN", "臺南": "TNN", "左營": "ZUY", "高雄": "ZUY"
+    }
+    
+    @classmethod
+    def get_input_schema(cls) -> Dict[str, Any]:
+        return StandardToolSchemas.create_input_schema({
+            "origin_station": {
+                "type": "string",
+                "description": "起站名稱（南港/台北/板橋/桃園/新竹/苗栗/台中/彰化/雲林/嘉義/台南/左營）"
+            },
+            "destination_station": {
+                "type": "string",
+                "description": "迄站名稱"
+            },
+            "train_no": {
+                "type": "string",
+                "description": "車次號碼（如「123」）"
+            },
+            "departure_date": {
+                "type": "string",
+                "description": "出發日期（YYYY-MM-DD 格式，預設今日）"
+            },
+            "departure_time": {
+                "type": "string",
+                "description": "出發時間（HH:MM 格式，如「14:30」）"
+            },
+            "limit": {
+                "type": "integer",
+                "description": "返回結果數量",
+                "default": 5
+            }
+        }, required=[])
+    
+    @classmethod
+    def get_output_schema(cls) -> Dict[str, Any]:
+        schema = StandardToolSchemas.create_output_schema()
+        schema["properties"].update({
+            "trains": {
+                "type": "array",
+                "items": {
+                    "type": "object",
+                    "properties": {
+                        "train_no": {"type": "string"},
+                        "train_type": {"type": "string"},
+                        "departure_time": {"type": "string"},
+                        "arrival_time": {"type": "string"},
+                        "duration_min": {"type": "integer"},
+                        "fare": {"type": "integer"}
+                    }
+                }
+            }
+        })
+        return schema
+    
+    @classmethod
+    async def execute(cls, arguments: Dict[str, Any], user_id: str = None) -> Dict[str, Any]:
+        origin = arguments.get("origin_station", "").strip()
+        destination = arguments.get("destination_station", "").strip()
+        train_no = arguments.get("train_no", "").strip()
+        departure_date = arguments.get("departure_date", "").strip()
+        departure_time = arguments.get("departure_time", "").strip()
+        limit = min(int(arguments.get("limit", 5)), 20)
+        
+        # 1. 取得用戶位置（用於最近車站查詢）
+        env_ctx = await get_user_env_current(user_id) if user_id else None
+        user_lat, user_lon = None, None
+        if env_ctx and env_ctx.get("success"):
+            ctx = env_ctx.get("context", {})
+            user_lat = ctx.get("lat")
+            user_lon = ctx.get("lon")
+        
+        # 2. 查詢分支
+        if train_no:
+            # 查詢特定車次
+            result = await cls._query_train_schedule(train_no, departure_date)
+        elif origin and destination:
+            # 查詢起迄站列車
+            result = await cls._query_od_trains(origin, destination, departure_date, departure_time, limit)
+        elif not origin and not destination:
+            # 查詢最近車站
+            if not user_lat or not user_lon:
+                raise ExecutionError("查詢最近高鐵站需要定位權限，或請提供起迄站名稱")
+            result = await cls._query_nearest_station(user_lat, user_lon)
+        else:
+            raise ExecutionError("請提供車次號碼，或起迄站名稱，或開啟定位查詢最近高鐵站")
+        
+        return result
+    
+    @classmethod
+    async def _query_train_schedule(cls, train_no: str, departure_date: str = "") -> Dict[str, Any]:
+        """查詢特定車次時刻表"""
+        # 日期處理
+        if not departure_date:
+            date_str = datetime.now().strftime("%Y-%m-%d")
+        else:
+            date_str = departure_date
+        
+        # TDX 高鐵每日時刻表
+        endpoint = f"Rail/THSR/DailyTimetable/TrainDate/{date_str}"
+        params = {
+            "$filter": f"DailyTrainInfo/TrainNo eq '{train_no}'",
+            "$format": "JSON"
+        }
+        
+        trains = await TDXBaseAPI.call_api(endpoint, params, cache_ttl=1800)
+        
+        if not trains or len(trains) == 0:
+            raise ExecutionError(f"找不到車次 {train_no}，請確認車次號碼與日期")
+        
+        train = trains[0]
+        train_info = train.get("DailyTrainInfo", {})
+        stops = train_info.get("StopTimes", [])
+        
+        if not stops:
+            raise ExecutionError(f"車次 {train_no} 無停靠站資訊")
+        
+        # 判斷車種
+        train_type = "標準車廂"
+        if any("商務" in stop.get("StationName", {}).get("Zh_tw", "") for stop in stops):
+            train_type = "商務車廂"
+        
+        # 格式化時刻表
+        schedule_lines = [f"🚄 高鐵 {train_no} 次 ({train_type})\n"]
+        schedule_lines.append(f"日期: {date_str}\n")
+        
+        for stop in stops:
+            station_name = stop.get("StationName", {}).get("Zh_tw", "未知")
+            arrival_time = stop.get("ArrivalTime", "")
+            departure_time = stop.get("DepartureTime", "")
+            
+            if arrival_time == departure_time:
+                time_str = arrival_time[:5] if arrival_time else "-"
+            else:
+                arr = arrival_time[:5] if arrival_time else "-"
+                dep = departure_time[:5] if departure_time else "-"
+                time_str = f"到 {arr} / 開 {dep}"
+            
+            schedule_lines.append(f"  {station_name:<6} {time_str}")
+        
+        content = "\n".join(schedule_lines)
+        
+        return cls.create_success_response(
+            content=content,
+            data={"train": train_info, "stops": stops}
+        )
+    
+    @classmethod
+    async def _query_od_trains(cls, origin: str, destination: str, 
+                              departure_date: str, departure_time: Optional[str],
+                              limit: int) -> Dict[str, Any]:
+        """查詢起迄站列車與票價"""
+        # 站點代碼轉換
+        origin_code = cls._get_station_code(origin)
+        dest_code = cls._get_station_code(destination)
+        
+        if not origin_code:
+            raise ExecutionError(f"找不到車站「{origin}」，請使用正確的高鐵站名")
+        if not dest_code:
+            raise ExecutionError(f"找不到車站「{destination}」，請使用正確的高鐵站名")
+        
+        # 日期處理
+        if not departure_date:
+            date_str = datetime.now().strftime("%Y-%m-%d")
+        else:
+            date_str = departure_date
+        
+        # 1. 查詢當日所有班次
+        endpoint = f"Rail/THSR/DailyTimetable/TrainDate/{date_str}"
+        params = {
+            "$format": "JSON"
+        }
+        
+        all_trains = await TDXBaseAPI.call_api(endpoint, params, cache_ttl=1800)
+        
+        if not all_trains:
+            raise ExecutionError("無法取得高鐵時刻表資訊")
+        
+        # 2. 過濾符合起迄站的列車
+        matching_trains = []
+        
+        for train_data in all_trains:
+            train_info = train_data.get("DailyTrainInfo", {})
+            stops = train_info.get("StopTimes", [])
+            
+            # 找起站和迄站
+            origin_stop, dest_stop = None, None
+            origin_idx, dest_idx = -1, -1
+            
+            for i, stop in enumerate(stops):
+                station_id = stop.get("StationID")
+                if station_id == origin_code:
+                    origin_stop = stop
+                    origin_idx = i
+                if station_id == dest_code:
+                    dest_stop = stop
+                    dest_idx = i
+            
+            # 起站在迄站之前才符合
+            if origin_stop and dest_stop and origin_idx < dest_idx:
+                dep_time = origin_stop.get("DepartureTime", "")
+                arr_time = dest_stop.get("ArrivalTime", "")
+                
+                train_result = {
+                    "train_no": train_info.get("TrainNo"),
+                    "origin_station": origin_stop.get("StationName", {}).get("Zh_tw"),
+                    "destination_station": dest_stop.get("StationName", {}).get("Zh_tw"),
+                    "departure_time": dep_time,
+                    "arrival_time": arr_time,
+                }
+                
+                # 計算行駛時間
+                try:
+                    dep_dt = datetime.strptime(dep_time, "%H:%M:%S")
+                    arr_dt = datetime.strptime(arr_time, "%H:%M:%S")
+                    duration = (arr_dt - dep_dt).total_seconds() / 60
+                    train_result["duration_min"] = int(duration)
+                except:
+                    train_result["duration_min"] = 0
+                
+                matching_trains.append(train_result)
+        
+        if not matching_trains:
+            raise ExecutionError(f"找不到 {origin} 到 {destination} 的直達高鐵")
+        
+        # 3. 時間過濾
+        if departure_time:
+            try:
+                target_time = datetime.strptime(departure_time, "%H:%M")
+                matching_trains = [
+                    t for t in matching_trains
+                    if datetime.strptime(t["departure_time"][:5], "%H:%M") >= target_time
+                ]
+            except:
+                pass
+        
+        # 4. 查詢票價
+        fare_endpoint = f"Rail/THSR/ODFare/{origin_code}/to/{dest_code}"
+        fare_params = {
+            "$format": "JSON"
+        }
+        
+        try:
+            fare_data = await TDXBaseAPI.call_api(fare_endpoint, fare_params, cache_ttl=86400)
+            if fare_data and len(fare_data) > 0:
+                fares = fare_data[0].get("Fares", [])
+                standard_fare = next((f.get("Price") for f in fares if f.get("TicketType") == "標準"), 0)
+                
+                for train in matching_trains:
+                    train["fare"] = standard_fare
+        except:
+            # 票價查詢失敗不影響時刻表結果
+            pass
+        
+        # 5. 排序並限制數量
+        matching_trains.sort(key=lambda x: x["departure_time"])
+        matching_trains = matching_trains[:limit]
+        
+        # 6. 格式化結果
+        content = cls._format_od_result(matching_trains, origin, destination, date_str)
+        
+        return cls.create_success_response(
+            content=content,
+            data={"trains": matching_trains, "date": date_str}
+        )
+    
+    @classmethod
+    async def _query_nearest_station(cls, lat: float, lon: float) -> Dict[str, Any]:
+        """查詢最近的高鐵站"""
+        # 1. 取得所有高鐵車站
+        endpoint = "Rail/THSR/Station"
+        params = {
+            "$format": "JSON"
+        }
+        
+        stations = await TDXBaseAPI.call_api(endpoint, params, cache_ttl=86400)
+        
+        if not stations:
+            raise ExecutionError("無法取得高鐵車站資訊")
+        
+        # 2. 計算距離
+        for station in stations:
+            pos = station.get("StationPosition", {})
+            if pos.get("PositionLat") and pos.get("PositionLon"):
+                station["distance_m"] = TDXBaseAPI.haversine_distance(
+                    lat, lon,
+                    pos["PositionLat"], pos["PositionLon"]
+                )
+        
+        stations_with_distance = [s for s in stations if "distance_m" in s]
+        
+        if not stations_with_distance:
+            raise ExecutionError("附近沒有高鐵車站資訊")
+        
+        stations_with_distance.sort(key=lambda x: x["distance_m"])
+        nearest = stations_with_distance[:3]
+        
+        # 3. 格式化結果
+        results = []
+        for station in nearest:
+            station_name = station.get("StationName", {}).get("Zh_tw", "未知")
+            distance = station["distance_m"]
+            driving_time = int(distance / 500)  # 假設開車 500m/min (30km/h)
+            
+            results.append({
+                "station_name": station_name,
+                "station_id": station.get("StationID"),
+                "distance_m": int(distance),
+                "driving_time_min": driving_time,
+                "address": station.get("StationAddress", "")
+            })
+        
+        content = cls._format_nearest_result(results)
+        
+        return cls.create_success_response(
+            content=content,
+            data={"stations": results}
+        )
+    
+    @staticmethod
+    def _get_station_code(station_name: str) -> Optional[str]:
+        """中文站名轉站點代碼"""
+        for name, code in TDXTHSRTool.STATION_MAP.items():
+            if name in station_name or station_name in name:
+                return code
+        return None
+    
+    @staticmethod
+    def _format_od_result(trains: List[Dict], origin: str, destination: str, date: str) -> str:
+        """格式化起迄站查詢結果"""
+        if not trains:
+            return f"🚄 {origin} → {destination} ({date}) 目前無可搭乘高鐵"
+        
+        lines = [f"🚄 {origin} → {destination} ({date})\n"]
+        
+        for i, train in enumerate(trains, 1):
+            duration_hours = train["duration_min"] // 60
+            duration_mins = train["duration_min"] % 60
+            
+            if duration_hours > 0:
+                duration_str = f"{duration_hours}小時{duration_mins}分"
+            else:
+                duration_str = f"{duration_mins}分鐘"
+            
+            fare_str = f" - ${train['fare']}" if train.get("fare") else ""
+            
+            lines.append(
+                f"{i}. 高鐵 {train['train_no']}次\n"
+                f"   {train['departure_time'][:5]} → {train['arrival_time'][:5]}"
+                f"  ({duration_str}){fare_str}\n"
+            )
+        
+        return "\n".join(lines)
+    
+    @staticmethod
+    def _format_nearest_result(stations: List[Dict]) -> str:
+        """格式化最近車站結果"""
+        lines = ["📍 最近的高鐵站：\n"]
+        
+        for i, station in enumerate(stations, 1):
+            lines.append(
+                f"{i}. 🚄 {station['station_name']}\n"
+                f"   開車約 {station['driving_time_min']} 分鐘 ({station['distance_m']/1000:.1f}km)\n"
+                f"   {station.get('address', '')}\n"
+            )
+        
+        return "\n".join(lines)

features/mcp/tools/tdx_train.py

@@ -0,0 +1,344 @@
+"""
+TDX 台鐵時刻表查詢工具
+查詢台鐵列車時刻、票價、車站資訊
+"""
+
+import logging
+from typing import Dict, Any, List, Optional
+from datetime import datetime, timedelta
+
+from .base_tool import MCPTool, StandardToolSchemas, ExecutionError
+from .tdx_base import TDXBaseAPI
+from core.database import get_user_env_current
+
+logger = logging.getLogger("mcp.tools.tdx.train")
+
+
+class TDXTrainTool(MCPTool):
+    """TDX 台鐵時刻表查詢"""
+    
+    NAME = "tdx_train"
+    DESCRIPTION = "查詢台鐵列車時刻表、票價、最近車站（含高鐵轉乘資訊）"
+    CATEGORY = "軌道運輸"
+    TAGS = ["tdx", "台鐵", "TRA", "火車", "時刻表"]
+    KEYWORDS = ["台鐵", "臺鐵", "火車", "TRA", "列車", "時刻"]
+    USAGE_TIPS = [
+        "查詢車次: 「自強號 123 次」",
+        "查詢路線: 「台北到台中的火車」",
+        "查詢最近車站: 「最近的火車站在哪」",
+        "查詢時刻: 「下午3點台北到高雄」"
+    ]
+    
+    @classmethod
+    def get_input_schema(cls) -> Dict[str, Any]:
+        return StandardToolSchemas.create_input_schema({
+            "origin_station": {
+                "type": "string",
+                "description": "起站名稱（如「台北」「台中」）"
+            },
+            "destination_station": {
+                "type": "string",
+                "description": "迄站名稱"
+            },
+            "train_no": {
+                "type": "string",
+                "description": "車次號碼（如「123」）"
+            },
+            "departure_time": {
+                "type": "string",
+                "description": "出發時間（HH:MM 格式，如「14:30」）"
+            },
+            "train_type": {
+                "type": "string",
+                "description": "列車種類",
+                "enum": ["自強", "莒光", "區間", "區間快", "普快", "復興", "太魯閣", "普悠瑪"]
+            },
+            "limit": {
+                "type": "integer",
+                "description": "返回結果數量",
+                "default": 5
+            }
+        }, required=[])
+    
+    @classmethod
+    def get_output_schema(cls) -> Dict[str, Any]:
+        schema = StandardToolSchemas.create_output_schema()
+        schema["properties"].update({
+            "trains": {
+                "type": "array",
+                "items": {
+                    "type": "object",
+                    "properties": {
+                        "train_no": {"type": "string"},
+                        "train_type": {"type": "string"},
+                        "departure_time": {"type": "string"},
+                        "arrival_time": {"type": "string"},
+                        "duration_min": {"type": "integer"}
+                    }
+                }
+            }
+        })
+        return schema
+    
+    @classmethod
+    async def execute(cls, arguments: Dict[str, Any], user_id: str = None) -> Dict[str, Any]:
+        origin = arguments.get("origin_station", "").strip()
+        destination = arguments.get("destination_station", "").strip()
+        train_no = arguments.get("train_no", "").strip()
+        departure_time = arguments.get("departure_time", "").strip()
+        train_type = arguments.get("train_type")
+        limit = min(int(arguments.get("limit", 5)), 20)
+        
+        # 1. 取得用戶位置（用於最近車站查詢）
+        env_ctx = await get_user_env_current(user_id) if user_id else None
+        user_lat, user_lon = None, None
+        if env_ctx and env_ctx.get("success"):
+            ctx = env_ctx.get("context", {})
+            user_lat = ctx.get("lat")
+            user_lon = ctx.get("lon")
+        
+        # 2. 查詢分支
+        if train_no:
+            # 查詢特定車次
+            result = await cls._query_train_schedule(train_no)
+        elif origin and destination:
+            # 查詢起迄站列車
+            result = await cls._query_od_trains(origin, destination, departure_time, train_type, limit)
+        elif not origin and not destination:
+            # 查詢最近車站
+            if not user_lat or not user_lon:
+                raise ExecutionError("查詢最近車站需要定位權限，或請提供起迄站名稱")
+            result = await cls._query_nearest_station(user_lat, user_lon)
+        else:
+            raise ExecutionError("請提供車次號碼，或起迄站名稱，或開啟定位查詢最近車站")
+        
+        return result
+    
+    @classmethod
+    async def _query_train_schedule(cls, train_no: str) -> Dict[str, Any]:
+        """查詢特定車次時刻表"""
+        today = datetime.now().strftime("%Y-%m-%d")
+        
+        endpoint = "Rail/TRA/DailyTrainInfo/Today"
+        params = {
+            "$filter": f"TrainNo eq '{train_no}'",
+            "$format": "JSON"
+        }
+        
+        trains = await TDXBaseAPI.call_api(endpoint, params, cache_ttl=1800)
+        
+        if not trains:
+            raise ExecutionError(f"找不到車次 {train_no}，請確認車次號碼")
+        
+        train = trains[0]
+        train_type = train.get("TrainTypeName", {}).get("Zh_tw", "未知")
+        
+        # 取得停靠站資訊
+        stops = train.get("StopTimes", [])
+        
+        if not stops:
+            raise ExecutionError(f"車次 {train_no} 無停靠站資訊")
+        
+        # 格式化時刻表
+        schedule_lines = [f"🚂 {train_type} {train_no} 次\n"]
+        
+        for stop in stops:
+            station_name = stop.get("StationName", {}).get("Zh_tw", "未知")
+            arrival_time = stop.get("ArrivalTime", "")
+            departure_time = stop.get("DepartureTime", "")
+            
+            if arrival_time == departure_time:
+                time_str = arrival_time[:5] if arrival_time else "-"
+            else:
+                arr = arrival_time[:5] if arrival_time else "-"
+                dep = departure_time[:5] if departure_time else "-"
+                time_str = f"{arr} / {dep}"
+            
+            schedule_lines.append(f"  {station_name:<10} {time_str}")
+        
+        content = "\n".join(schedule_lines)
+        
+        return cls.create_success_response(
+            content=content,
+            data={"train": train, "stops": stops}
+        )
+    
+    @classmethod
+    async def _query_od_trains(cls, origin: str, destination: str, 
+                              departure_time: Optional[str], train_type: Optional[str],
+                              limit: int) -> Dict[str, Any]:
+        """查詢起迄站列車"""
+        # 1. 先取得今日所有列車
+        endpoint = "Rail/TRA/DailyTrainInfo/Today"
+        params = {
+            "$format": "JSON"
+        }
+        
+        all_trains = await TDXBaseAPI.call_api(endpoint, params, cache_ttl=1800)
+        
+        if not all_trains:
+            raise ExecutionError("無法取得台鐵列車資訊")
+        
+        # 2. 過濾符合起迄站的列車
+        matching_trains = []
+        
+        for train in all_trains:
+            stops = train.get("StopTimes", [])
+            
+            # 找起站和迄站
+            origin_idx, dest_idx = -1, -1
+            for i, stop in enumerate(stops):
+                station = stop.get("StationName", {}).get("Zh_tw", "")
+                if origin in station:
+                    origin_idx = i
+                if destination in station:
+                    dest_idx = i
+            
+            # 起站在迄站之前才符合
+            if origin_idx >= 0 and dest_idx > origin_idx:
+                origin_stop = stops[origin_idx]
+                dest_stop = stops[dest_idx]
+                
+                train_info = {
+                    "train_no": train.get("TrainNo"),
+                    "train_type": train.get("TrainTypeName", {}).get("Zh_tw", "未知"),
+                    "origin_station": origin_stop.get("StationName", {}).get("Zh_tw"),
+                    "destination_station": dest_stop.get("StationName", {}).get("Zh_tw"),
+                    "departure_time": origin_stop.get("DepartureTime", ""),
+                    "arrival_time": dest_stop.get("ArrivalTime", ""),
+                }
+                
+                # 計算行駛時間
+                try:
+                    dep_dt = datetime.strptime(train_info["departure_time"], "%H:%M:%S")
+                    arr_dt = datetime.strptime(train_info["arrival_time"], "%H:%M:%S")
+                    if arr_dt < dep_dt:  # 跨日
+                        arr_dt += timedelta(days=1)
+                    duration = (arr_dt - dep_dt).total_seconds() / 60
+                    train_info["duration_min"] = int(duration)
+                except:
+                    train_info["duration_min"] = 0
+                
+                matching_trains.append(train_info)
+        
+        if not matching_trains:
+            raise ExecutionError(f"找不到 {origin} 到 {destination} 的直達列車")
+        
+        # 3. 時間過濾
+        if departure_time:
+            try:
+                target_time = datetime.strptime(departure_time, "%H:%M")
+                matching_trains = [
+                    t for t in matching_trains
+                    if datetime.strptime(t["departure_time"][:5], "%H:%M") >= target_time
+                ]
+            except:
+                pass
+        
+        # 4. 車種過濾
+        if train_type:
+            matching_trains = [t for t in matching_trains if train_type in t["train_type"]]
+        
+        # 5. 排序並限制數量
+        matching_trains.sort(key=lambda x: x["departure_time"])
+        matching_trains = matching_trains[:limit]
+        
+        # 6. 格式化結果
+        content = cls._format_od_result(matching_trains, origin, destination)
+        
+        return cls.create_success_response(
+            content=content,
+            data={"trains": matching_trains}
+        )
+    
+    @classmethod
+    async def _query_nearest_station(cls, lat: float, lon: float) -> Dict[str, Any]:
+        """查詢最近的台鐵車站"""
+        # 1. 取得所有車站
+        endpoint = "Rail/TRA/Station"
+        params = {
+            "$format": "JSON"
+        }
+        
+        stations = await TDXBaseAPI.call_api(endpoint, params, cache_ttl=86400)
+        
+        if not stations:
+            raise ExecutionError("無法取得台鐵車站資訊")
+        
+        # 2. 計算距離
+        for station in stations:
+            pos = station.get("StationPosition", {})
+            if pos.get("PositionLat") and pos.get("PositionLon"):
+                station["distance_m"] = TDXBaseAPI.haversine_distance(
+                    lat, lon,
+                    pos["PositionLat"], pos["PositionLon"]
+                )
+        
+        stations_with_distance = [s for s in stations if "distance_m" in s]
+        
+        if not stations_with_distance:
+            raise ExecutionError("附近沒有台鐵車站資訊")
+        
+        stations_with_distance.sort(key=lambda x: x["distance_m"])
+        nearest = stations_with_distance[:3]
+        
+        # 3. 格式化結果
+        results = []
+        for station in nearest:
+            station_name = station.get("StationName", {}).get("Zh_tw", "未知")
+            distance = station["distance_m"]
+            walking_time = int(distance / 80)
+            
+            results.append({
+                "station_name": station_name,
+                "station_id": station.get("StationID"),
+                "distance_m": int(distance),
+                "walking_time_min": walking_time,
+                "address": station.get("StationAddress", "")
+            })
+        
+        content = cls._format_nearest_result(results)
+        
+        return cls.create_success_response(
+            content=content,
+            data={"stations": results}
+        )
+    
+    @staticmethod
+    def _format_od_result(trains: List[Dict], origin: str, destination: str) -> str:
+        """格式化起迄站查詢結果"""
+        if not trains:
+            return f"🚂 {origin} → {destination} 目前無可搭乘列車"
+        
+        lines = [f"🚂 {origin} → {destination}\n"]
+        
+        for i, train in enumerate(trains, 1):
+            duration_hours = train["duration_min"] // 60
+            duration_mins = train["duration_min"] % 60
+            
+            if duration_hours > 0:
+                duration_str = f"{duration_hours}小時{duration_mins}分"
+            else:
+                duration_str = f"{duration_mins}分鐘"
+            
+            lines.append(
+                f"{i}. {train['train_type']} {train['train_no']}次\n"
+                f"   {train['departure_time'][:5]} → {train['arrival_time'][:5]}"
+                f"  ({duration_str})\n"
+            )
+        
+        return "\n".join(lines)
+    
+    @staticmethod
+    def _format_nearest_result(stations: List[Dict]) -> str:
+        """格式化最近車站結果"""
+        lines = ["📍 最近的台鐵車站：\n"]
+        
+        for i, station in enumerate(stations, 1):
+            lines.append(
+                f"{i}. 🚂 {station['station_name']}\n"
+                f"   步行 {station['walking_time_min']} 分鐘 ({station['distance_m']}m)\n"
+                f"   {station.get('address', '')}\n"
+            )
+        
+        return "\n".join(lines)

features/mcp/tools/tdx_youbike.py

@@ -0,0 +1,299 @@
+"""
+TDX YouBike 即時查詢工具
+查詢附近 YouBike 站點、即時車輛數、空位數
+"""
+
+import logging
+from typing import Dict, Any, List
+
+from .base_tool import MCPTool, StandardToolSchemas, ExecutionError
+from .tdx_base import TDXBaseAPI
+from core.database import get_user_env_current
+
+logger = logging.getLogger("mcp.tools.tdx.bike")
+
+
+class TDXBikeTool(MCPTool):
+    """TDX YouBike 即時查詢"""
+    
+    NAME = "tdx_youbike"
+    DESCRIPTION = "查詢附近 YouBike 站點、即時車輛數、空位數（支援 YouBike 1.0/2.0）"
+    CATEGORY = "微型運具"
+    TAGS = ["tdx", "youbike", "ubike", "共享單車", "微笑單車"]
+    KEYWORDS = ["YouBike", "UBike", "微笑單車", "共享單車", "腳踏車", "自行車"]
+    USAGE_TIPS = [
+        "查詢附近站點: 「附近的 YouBike 在哪」",
+        "查詢特定站點: 「市政府 YouBike 還有車嗎」",
+        "指定城市: 「台北 YouBike」「高雄 CityBike」"
+    ]
+    
+    # 城市對應
+    CITY_MAP = {
+        "台北": "Taipei",
+        "臺北": "Taipei",
+        "新北": "NewTaipei",
+        "桃園": "Taoyuan",
+        "台中": "Taichung",
+        "臺中": "Taichung",
+        "台南": "Tainan",
+        "臺南": "Tainan",
+        "高雄": "Kaohsiung",
+        "新竹": "Hsinchu"
+    }
+    
+    @classmethod
+    def get_input_schema(cls) -> Dict[str, Any]:
+        return StandardToolSchemas.create_input_schema({
+            "station_name": {
+                "type": "string",
+                "description": "站點名稱（如「市政府」「台北車站」）。不提供則查詢附近站點"
+            },
+            "city": {
+                "type": "string",
+                "description": "城市名稱（如「Taipei」「Kaohsiung」）",
+                "enum": list(cls.CITY_MAP.values())
+            },
+            "radius_m": {
+                "type": "integer",
+                "description": "搜尋半徑（公尺）",
+                "default": 500
+            },
+            "limit": {
+                "type": "integer",
+                "description": "返回結果數量",
+                "default": 5
+            }
+        }, required=[])
+    
+    @classmethod
+    def get_output_schema(cls) -> Dict[str, Any]:
+        schema = StandardToolSchemas.create_output_schema()
+        schema["properties"].update({
+            "stations": {
+                "type": "array",
+                "items": {
+                    "type": "object",
+                    "properties": {
+                        "station_name": {"type": "string"},
+                        "available_bikes": {"type": "integer"},
+                        "available_spaces": {"type": "integer"},
+                        "distance_m": {"type": "integer"},
+                        "bike_type": {"type": "string"}
+                    }
+                }
+            }
+        })
+        return schema
+    
+    @classmethod
+    async def execute(cls, arguments: Dict[str, Any], user_id: str = None) -> Dict[str, Any]:
+        station_name = arguments.get("station_name", "").strip()
+        city = arguments.get("city")
+        radius_m = min(int(arguments.get("radius_m", 500)), 2000)
+        limit = min(int(arguments.get("limit", 5)), 20)
+        
+        # 1. 取得用戶位置
+        env_ctx = await get_user_env_current(user_id) if user_id else None
+        if not env_ctx or not env_ctx.get("success"):
+            if not station_name:
+                raise ExecutionError("無法取得您的位置，請提供站點名稱或開啟定位權限")
+            user_lat, user_lon, user_city = None, None, None
+        else:
+            ctx = env_ctx.get("context", {})
+            user_lat = ctx.get("lat")
+            user_lon = ctx.get("lon")
+            user_city = ctx.get("city", "")
+        
+        # 2. 自動判斷城市
+        if not city:
+            city = cls._map_city_name(user_city) if user_city else "Taipei"
+        
+        # 3. 查詢分支
+        if station_name:
+            result = await cls._query_station_availability(station_name, city)
+        else:
+            if not user_lat or not user_lon:
+                raise ExecutionError("查詢附近 YouBike 需要定位權限")
+            result = await cls._query_nearby_stations(user_lat, user_lon, city, radius_m, limit)
+        
+        return result
+    
+    @classmethod
+    async def _query_station_availability(cls, station_name: str, city: str) -> Dict[str, Any]:
+        """查詢特定站點即時資訊"""
+        # 1. 查詢站點基本資訊
+        station_endpoint = f"Bike/Station/City/{city}"
+        station_params = {
+            "$filter": f"contains(StationName/Zh_tw, '{station_name}')",
+            "$format": "JSON",
+            "$top": 5
+        }
+        
+        stations = await TDXBaseAPI.call_api(station_endpoint, station_params, cache_ttl=1800)
+        
+        if not stations:
+            raise ExecutionError(f"找不到站點「{station_name}」")
+        
+        # 2. 取得完全匹配或第一個結果
+        target_station = None
+        for station in stations:
+            name = station.get("StationName", {}).get("Zh_tw", "")
+            if station_name in name:
+                target_station = station
+                break
+        
+        if not target_station:
+            target_station = stations[0]
+        
+        station_uid = target_station.get("StationUID")
+        full_station_name = target_station.get("StationName", {}).get("Zh_tw", station_name)
+        
+        # 3. 查詢即時可用車輛數
+        avail_endpoint = f"Bike/Availability/City/{city}"
+        avail_params = {
+            "$filter": f"StationUID eq '{station_uid}'",
+            "$format": "JSON"
+        }
+        
+        availability = await TDXBaseAPI.call_api(avail_endpoint, avail_params, cache_ttl=30)
+        
+        if not availability or len(availability) == 0:
+            return cls.create_success_response(
+                content=f"🚲 {full_station_name} 目前無即時資訊",
+                data={"stations": []}
+            )
+        
+        avail = availability[0]
+        
+        result = {
+            "station_name": full_station_name,
+            "available_bikes": avail.get("AvailableRentBikes", 0),
+            "available_spaces": avail.get("AvailableReturnBikes", 0),
+            "service_status": avail.get("ServiceStatus", 1),
+            "update_time": avail.get("UpdateTime", ""),
+            "bike_type": "YouBike 2.0" if "2.0" in target_station.get("BikesCapacity", "") else "YouBike 1.0"
+        }
+        
+        # 4. 格式化結果
+        status_map = {
+            0: "停止營運",
+            1: "正常營運",
+            2: "暫停營運"
+        }
+        status = status_map.get(result["service_status"], "未知")
+        
+        content = (
+            f"🚲 {result['station_name']}\n"
+            f"狀態: {status}\n"
+            f"可借: {result['available_bikes']} 輛\n"
+            f"可還: {result['available_spaces']} 位\n"
+            f"類型: {result['bike_type']}\n"
+        )
+        
+        return cls.create_success_response(
+            content=content,
+            data={"station": result}
+        )
+    
+    @classmethod
+    async def _query_nearby_stations(cls, lat: float, lon: float, city: str, 
+                                     radius_m: int, limit: int) -> Dict[str, Any]:
+        """查詢附近站點"""
+        # 1. 查詢附近站點（使用空間過濾）
+        station_endpoint = f"Bike/Station/City/{city}"
+        station_params = {
+            "$spatialFilter": f"nearby({lat}, {lon}, {radius_m})",
+            "$format": "JSON",
+            "$top": limit * 2
+        }
+        
+        stations = await TDXBaseAPI.call_api(station_endpoint, station_params, cache_ttl=1800)
+        
+        if not stations:
+            return cls.create_success_response(
+                content=f"附近 {radius_m} 公尺內沒有 YouBike 站點",
+                data={"stations": []}
+            )
+        
+        # 2. 計算距離並排序
+        for station in stations:
+            pos = station.get("StationPosition", {})
+            if pos.get("PositionLat") and pos.get("PositionLon"):
+                station["distance_m"] = TDXBaseAPI.haversine_distance(
+                    lat, lon,
+                    pos["PositionLat"], pos["PositionLon"]
+                )
+        
+        stations = [s for s in stations if "distance_m" in s]
+        stations.sort(key=lambda x: x["distance_m"])
+        stations = stations[:limit]
+        
+        # 3. 批次查詢即時資訊
+        station_uids = [s.get("StationUID") for s in stations]
+        
+        avail_endpoint = f"Bike/Availability/City/{city}"
+        avail_params = {
+            "$filter": " or ".join([f"StationUID eq '{uid}'" for uid in station_uids]),
+            "$format": "JSON"
+        }
+        
+        availability = await TDXBaseAPI.call_api(avail_endpoint, avail_params, cache_ttl=30)
+        
+        # 建立 UID -> 可用性 映射
+        avail_map = {a.get("StationUID"): a for a in availability}
+        
+        # 4. 組合結果
+        results = []
+        for station in stations:
+            station_uid = station.get("StationUID")
+            station_name = station.get("StationName", {}).get("Zh_tw", "未知")
+            distance = station["distance_m"]
+            walking_time = int(distance / 80)
+            
+            avail = avail_map.get(station_uid, {})
+            
+            results.append({
+                "station_name": station_name,
+                "available_bikes": avail.get("AvailableRentBikes", 0),
+                "available_spaces": avail.get("AvailableReturnBikes", 0),
+                "distance_m": int(distance),
+                "walking_time_min": walking_time,
+                "service_status": avail.get("ServiceStatus", 1),
+                "bike_type": "YouBike 2.0" if "2.0" in station.get("BikesCapacity", "") else "YouBike 1.0"
+            })
+        
+        content = cls._format_nearby_result(results)
+        
+        return cls.create_success_response(
+            content=content,
+            data={"stations": results}
+        )
+    
+    @staticmethod
+    def _map_city_name(chinese_city: str) -> str:
+        """中文城市名稱轉 TDX 代碼"""
+        for key, value in TDXBikeTool.CITY_MAP.items():
+            if key in chinese_city:
+                return value
+        return "Taipei"
+    
+    @staticmethod
+    def _format_nearby_result(stations: List[Dict]) -> str:
+        """格式化附近站點結果"""
+        if not stations:
+            return "附近沒有 YouBike 站點"
+        
+        lines = ["📍 附近的 YouBike 站點：\n"]
+        
+        for i, station in enumerate(stations, 1):
+            status_emoji = "✅" if station["service_status"] == 1 else "⚠️"
+            bikes = station["available_bikes"]
+            spaces = station["available_spaces"]
+            
+            lines.append(
+                f"{i}. {status_emoji} {station['station_name']}\n"
+                f"   可借 {bikes} 輛 | 可還 {spaces} 位\n"
+                f"   步行 {station['walking_time_min']} 分鐘 ({station['distance_m']}m)\n"
+            )
+        
+        return "\n".join(lines)

features/mcp/tools/weather_tool.py

@@ -30,10 +30,15 @@ class WeatherTool(MCPTool):
     """天氣查詢 MCP 工具"""
 
     NAME = "weather_query"
-    DESCRIPTION = "查詢指定城市的天氣資訊，支援城市名稱或座標查詢"
-    CATEGORY = "天氣"
-    TAGS = ["weather", "climate", "forecast"]
-    USAGE_TIPS = ["直接說「台北天氣」或「東京天氣」"]
+    DESCRIPTION = "查詢指定城市的即時天氣資訊（溫度、濕度、天氣狀況等）"
+    CATEGORY = "生活資訊"
+    TAGS = ["weather", "天氣", "氣象"]
+    KEYWORDS = ["天氣", "氣溫", "下雨", "晴天", "陰天", "weather", "溫度"]
+    USAGE_TIPS = [
+        "提供城市名稱（英文）如 Taipei, Tokyo",
+        "支援經緯度查詢",
+        "可指定語言 (zh_tw, en, zh_cn)"
+    ]
 
     @classmethod
     def get_input_schema(cls) -> Dict[str, Any]: