整理 AGENTS.md 格式並更新相關文件

AGENTS.md

@@ -1,106 +1,74 @@
-# Repository Guidelines + System Architecture
+# AGENTS.md
 
-結論先講：本專案是 FastAPI 單體後端，走 WebSocket 主動互動、REST 做周邊能力，資料層以 Firestore 為核心，功能透過 MCP Tools 解耦；本檔已補齊實際架構、流程、環境與風險，照著做就不會踩雷。
-
-**目錄速覽**
-- `app.py`：ASGI 入口（FastAPI）。載入設定、掛載靜態檔、CORS/CSP、中介層、背景任務、WebSocket `/ws`、OAuth 與 REST API。
-- `core/`：設定（`config.py`）、認證（`auth/` JWT + Google OAuth PKCE）、資料庫（`database/` Firestore + 快取 + 最佳化）、記憶系統、情緒關懷、聊天處理管線。
-- `features/mcp/`：MCP 伺服器與工具（天氣、新聞、匯率、HealthKit 查詢等），`agent_bridge.py` 負責意圖偵測與工具串接。
-- `services/`：AI 產生（OpenAI SDK）、TTS/STT、語音登入、批次排程等服務層。
-- `models/`：語音情緒與說話者辨識相關模型與腳本。
-- `static/`：前端靜態頁（登入、對話 UI 等）。
-- `tests/`：Pytest 測試，鏡射模組路徑（目前以 `services/voice_login` 為主）。
-- `render.yaml`、`Dockerfile`、`runtime.txt`、`.env.production.example`：部署與環境樣板。
-- `bloom-ware-login/`：獨立 Next.js（登入）樣板，非後端運行必要。
-
-—
-
-**執行與開發（本機）**
-- 建環境：`python3 -m venv .venv && source .venv/bin/activate && pip install -r requirements.txt`
-- 直跑：`python app.py`（預設 0.0.0.0:8080）
-- ASGI 模式：`uvicorn app:app --reload`（開發用）
-- 前端入口：`/static/login.html` → 走 Google OAuth 登入 → 以 JWT 綁 WebSocket `/ws?token=...`。
-- 測試：`pytest -q` 或 `pytest -q -k voice_login`。
-
-提示：Docker/HF Spaces 預設 `PORT=7860`；Render 預設 `PORT=10000`。程式以環境變數為準。
-
-—
-
-**系統架構（重點元件）**
-- Web 層：FastAPI + 中介層（CORS、CSP）。靜態檔以 `static/frontend` 掛載到 `/static`。
-- WebSocket：`/ws` 單點，JWT 驗證（Query `token`），集中處理聊天、typing 與工具回傳；`ConnectionManager` 管理會話。
-- 意圖與工具：`features/mcp/agent_bridge.py` 以 OpenAI Structured Outputs 做意圖偵測，命中則調 `features/mcp/tools/*`（天氣/新聞/匯率/HealthKit 等）。
-- 聊天管線：`core/pipeline.ChatPipeline` 先意圖→工具→AI 產生；支援「情緒關懷模式」（極端情緒時停用工具、改走關懷 Prompt）。
-- AI 服務：`services/ai_service.py` 封裝 OpenAI SDK；模型由環境 `OPENAI_MODEL` 控（預設 `gpt-5-nano`）。
-- 資料層：Firestore（`core/database/base.py`）＋最佳化存取與 LRU 快取（`optimized.py`/`cache.py`）；集合：`users`、`chats`、`messages`、`health_data`、`device_bindings`。
-- 背景任務：啟動時依 `ENABLE_BACKGROUND_JOBS` 啟動快取維護、清理、批次排程（每日摘要/週報）。
-
-—
-
-**請求流程（典型路徑）**
-- 登入：前端打 `/auth/google/url` 取得授權連結 → Google 回調 `/auth/google/callback` → 交換 Token → 產出 JWT。
-- 對話：前端以 `/ws?token=JWT` 連上；訊息先經「語音綁定 FSM」攔截（若用語音綁定流程）→ 進入 `ChatPipeline` → 視意圖走 MCP 工具或一般聊天 → 落庫 `chats/messages`。
-- 檔案分析：`/api/upload-file` 或 `/api/analyze-file-base64`，文字/PDF/圖片分流到對應分析邏輯，底層仍透過 OpenAI。
-- 健康資料：建議透過 MCP `healthkit_tool` 查 Firestore（iOS 端直寫 `health_data`）。
-- 位置快照：前端成功取得瀏覽器定位時會透過 `env_snapshot` 送到 WebSocket；後端會自動寫入 Firestore 並呼叫 MCP `reverse_geocode` 反查地點，AI Prompt 只會顯示地點名稱（有 label）或地址（無 label），不再硬編碼地標。連線歡迎詞也會優先使用最近的 `tz`，避免問候時間錯亂。
-
-—
-
-**環境與設定（`core/config.Settings`）**
-- 必填：`FIREBASE_PROJECT_ID`、`FIREBASE_CREDENTIALS_JSON`（或 `FIREBASE_SERVICE_ACCOUNT_PATH`）、`OPENAI_API_KEY`、`GOOGLE_CLIENT_ID/SECRET`、`JWT_SECRET_KEY`。
-- 其他：`OPENAI_MODEL`（預設 `gpt-5-nano`）、`HOST`、`PORT`、`ENABLE_BACKGROUND_JOBS`、第三方金鑰（天氣/新聞/匯率）。
-- 生產：Render 用 `PORT=10000`，HF Spaces/Docker 用 `PORT=7860`。
-
-—
-
-**部署模式**
-- Render（`render.yaml`）：平台注入環境變數，直接 `python3 app.py` 啟動。
-- HF Spaces（`Dockerfile`）：以 `uvicorn` 啟動，`PORT` 由平台給；請關閉排程（`ENABLE_BACKGROUND_JOBS=false`）。
-
-—
-
-**測試策略（TDD）**
-- 測試放 `tests/`，鏡射原始碼路徑，命名 `test_*.py`。
-- 先紅後綠再重構；關鍵路徑優先（`core/`、`services/`）。
-- 範例：`tests/services/test_voice_login_cnn.py` 透過 stub/injection 減少重量相依。
-
-—
-
-**MCP 工具擴充規範（快速上手）**
-- 位置：`features/mcp/tools/`；繼承 `MCPTool`，實作 `get_input_schema`、`get_output_schema`、`execute`。
-- 自動註冊：`features/mcp/auto_registry.py` 會掃描並註冊；若需要外部進程，交由 `server.start_external_servers()`。
-- 輸出格式：以 `create_success_response(content=..., data=...)` 回傳；錯誤用 `create_error_response(code=..., error=...)`。
-- 工具 metadata：`CATEGORY/TAGS/USAGE_TIPS` 會回到 `/api/mcp/tools` 供前端工具卡片用。
-
-—
-
-**安全與維運建議（務必看完）**
-- CORS：目前設定為 `allow_origins=["*"]` 且 `allow_credentials=True`，生產環境建議收斂來源網域，避免 Cookie/Authorization 外洩風險。
-- JWT：請務必提供穩定的 `JWT_SECRET_KEY`；否則服務重啟會因隨機 Secret 導致既有 Token 全失效。
-- CSP：為了語音前端放寬到 `'unsafe-inline'/'unsafe-eval'`，生產環境請只在 `/static` 下放寬，嚴禁波及 API 路徑。
-- 上傳限制：檔案上限 10MB，白名單含 PDF/影像/程式碼等，後端已驗型別但仍需注意前端檔案來源。
-- Firestore 配額：已實作 LRU 快取、請求合併與批次寫入；高流量時請觀察 `/api/performance/stats`。
-
-—
-
-**已知問題（歡迎開 PR 修）**
-- `requirements.txt` 尾端疑似誤合併文字，出現 `transformersservices:`；若安裝失敗，請將 `transformers` 與 `services:` 拆正（`render.yaml` 應在獨立檔）。
-- `app.py` 的 CORS 設定呼叫了兩次，可合併為一次以避免重複中介層。
-- `GET /api/health/query` 仍殘留 Mongo-style 的 `find()`/`async for` 寫法，與 Firestore 用法不符；建議改用 MCP `healthkit_tool` 或重寫為 Firestore 查詢。
-
-—
-
-**Coding Style & Conventions**
-- Python 3.10+；`ruff/black` 預設（88 cols, 4-space, UTF‑8）。
-- 模組 `snake_case.py`；類別 `PascalCase`；函式/變數 `snake_case`。
-- 分層：Controller(API) → Service → Core/Utils；資料層封在 `core/database/*`。
-- 環境變數透過 `os.environ` 讀取；範例見 `.env.production.example`。
-
-—
-
-**Agent-Specific（給在此倉工作之助理）**
-- 語言與語氣：全程繁中（台灣），「先結論、後細節」，可微嗆不冒犯。
-- TDD：先寫測試（紅燈）→ 最小實作（綠燈）→ 重構，每個功能至少跑完一輪。
-- OpenAI 使用：以 Python SDK；模型由 `OPENAI_MODEL` 控制（預設 `gpt-5-nano`）；若改版，請只改環境變數，不要在程式寫死。
-- 測試放置：所有測試集中於 `tests/`，`test_*.py`；結構鏡射模組路徑。
-- 前端語音介面：`static/frontend/index.html` 的 `voice-center-container`、`voice-agent-output`、`voice-transcript` 已改為彈性寬高；打字態訊息只做 opacity/visibility 切換，避免擠壓字幕區。調整樣式時請維持 `clamp` 設定與 wrapper class，確保桌機/平板視窗自適應。
+## 介紹
+
+你是 白東衢的狗。主要語言 Python。可用 MCP：context7、feedback-enhanced、filesystem、huggingface、playwright、sequential-thinking。所有思考與回覆皆用繁體中文。先思考再行動。
+
+## 全域原則
+
+- 所有 MCP 工具逾時一律 60 分鐘。
+- 非框架情境：以最少檔案完成任務，避免過度模組化。
+- 框架情境：依框架慣例放置檔案。
+- 禁止產生非目的文件：說明文件、依賴清單、README、requirements.txt 等。
+- 禁止要求以命令列參數輸入業務值；業務參數以程式內常數或互動式輸入處理。
+- 需新知時主動檢索：優先從網際網路獲取資訊並使用 huggingface（paper_search、model_search、dataset_search、hf_doc_search）與 context7進行輔佐。
+- playwright 僅用於前端樣式檢查、UI 互動測試、E2E 視覺驗證與截圖，禁止做爬蟲或搜尋。
+- 僅在規定節點使用 feedback-enhanced，禁止重複寒暄。
+
+## 環境與執行限制
+
+- 不得假設環境缺失而自動安裝任何套件或修改系統環境變數。
+- 不得建立或使用任何虛擬環境（venv、conda、poetry 等）。
+- 一律使用當前系統 Python 版本執行與相容（可於程式內讀取 sys.version 僅作紀錄，不觸碰安裝行為）。
+- 不輸出或修改 requirements.txt、pyproject.toml、環境設定檔。
+
+## 框架情境的工作區確認
+
+偵測到整合型框架或既有專案結構時：
+
+- 先用 sequential-thinking 規劃「要改哪些模組、檔名、路徑與測試位置」。
+- 接著必須以一次 feedback-enhanced.interactive_feedback 與使用者確認「基準工作區路徑、允許寫入子資料夾與檔名慣例」。
+- 未獲確認前不得寫入專案根目錄。確認後依框架慣例生成或修改檔案。
+- 非框架情境可寫根目錄，但仍以最小檔案集為原則。
+
+## 思考與行動流程
+
+- sequential-thinking：輸出「目標 → 步驟 → 決策準則 → 風險與驗證」。
+- 取證：context7、huggingface。
+- 產出：filesystem 建立或修改檔案；必要時用 playwright 做 UI 驗證。
+- 回覆內容：只含思考計畫、行動步驟、關鍵程式碼、測試摘要、後續建議。
+
+## 測試與驗證政策
+
+- 為每個新增或修改模組撰寫單元測試與關鍵整合測試。
+- 測試檔命名 test_*.py；框架專案依其慣例放置。
+- 執行方式以終端 Python3 或 Pytest：python3 -m pytest -q 或 pytest -q。
+- 回報測試摘要：通過數、失敗數、失敗案例與原因、可能回歸點、下一步。
+
+## 錯誤處理與互動節奏
+
+發生錯誤或接獲失敗回報時：
+
+- 先以 sequential-thinking 分析根因、解法選項與取捨與影響。
+- 再以一次 feedback-enhanced.interactive_feedback 與使用者對齊解法與影響面。
+- 然後修改程式與測試並重跑驗證。除上述節點外避免重複呼叫 feedback。
+
+## 產出規範
+
+- 預設單檔或少量檔案即可完成任務；框架專案依其結構放置。
+- 程式需具明確進入點：
+  ```
+  if __name__ == "__main__":
+      main()
+  ```
+- 檔案一律透過 filesystem 操作並回報路徑與成功訊息。
+- 不硬編 API 金鑰或密碼；輸出時遮罩敏感資訊。
+- 外部資源不可用時，提出替代方案與自我修正步驟，仍不得觸發安裝或改環境行為。
+
+## 禁止事項總表
+
+- 自動安裝或升降版本、修改環境變數、建立/使用虛擬環境。
+- 產出說明文件、依賴清單或其他非目的文件。
+- 使用命令列參數傳遞業務值。
+- 用 playwright 做爬蟲或搜尋。
+- 無限制地反覆呼叫 feedback-enhanced。

app.py

@@ -66,6 +66,7 @@ from core.pipeline import ChatPipeline, PipelineResult
 from core.memory_system import memory_manager
 # 環境 Context 寫入 API
 from core.database import set_user_env_current, add_user_env_snapshot
+from core.environment import EnvironmentContextService
 
 
 # -----------------------------
@@ -241,7 +242,20 @@ async def lifespan(app: FastAPI):
         else:
             logger.info("✅ Firestore 已成功連接並可用")
 
+        app.state.env_service = EnvironmentContextService(
+            min_distance_m=settings.ENV_CONTEXT_DISTANCE_THRESHOLD,
+            min_heading_deg=settings.ENV_CONTEXT_HEADING_THRESHOLD,
+            ttl_seconds=settings.ENV_CONTEXT_TTL_SECONDS,
+            env_fetcher=get_user_env_current,
+            env_writer=set_user_env_current,
+            snapshot_writer=add_user_env_snapshot,
+        )
+        await app.state.env_service.start()
+
         app.state.feature_router = MCPAgentBridge()
+        app.state.feature_router.bind_env_provider(
+            lambda user_id: app.state.env_service.get_context(user_id, allow_stale=True)
+        )
         
         # 異步初始化 MCP 橋接層（發現所有工具）
         if hasattr(app.state.feature_router, 'async_initialize'):
@@ -295,6 +309,13 @@ async def lifespan(app: FastAPI):
     try:
         yield
     finally:
+        env_service = getattr(app.state, "env_service", None)
+        if env_service:
+            try:
+                await env_service.shutdown()
+            except Exception as shutdown_err:
+                logger.warning(f"環境服務關閉失敗: {shutdown_err}")
+
         # Shutdown cleanup
         if getattr(app.state, "enable_background_jobs", False):
             try:
@@ -493,32 +514,6 @@ class ConnectionManager:
 manager = ConnectionManager()
 
 
-# 地理工具函式（內部使用）
-def _haversine_m(lat1: float, lon1: float, lat2: float, lon2: float) -> float:
-    from math import radians, sin, cos, asin, sqrt
-    if None in (lat1, lon1, lat2, lon2):
-        return 0.0
-    R = 6371000.0
-    dlat = radians(lat2 - lat1)
-    dlon = radians(lon2 - lon1)
-    a = sin(dlat/2)**2 + cos(radians(lat1))*cos(radians(lat2))*sin(dlon/2)**2
-    c = 2 * asin(sqrt(a))
-    return R * c
-
-
-def _heading_to_cardinal(deg: float) -> str:
-    try:
-        val = float(deg)
-    except Exception:
-        return ""
-    dirs = [
-        "N","NNE","NE","ENE","E","ESE","SE","SSE",
-        "S","SSW","SW","WSW","W","WNW","NW","NNW"
-    ]
-    ix = int((val % 360) / 22.5 + 0.5) % 16
-    return dirs[ix]
-
-
 # -----------------------------
 # 語音綁定狀態管理器（關鍵字匹配，無 GPT）
 # -----------------------------
@@ -886,155 +881,61 @@ async def websocket_endpoint_with_jwt(websocket: WebSocket, token: str = Query(N
 
                 elif message_type == "env_snapshot":
                     try:
-                        lat = float(message_data.get("lat")) if message_data.get("lat") is not None else None
-                        lon = float(message_data.get("lon")) if message_data.get("lon") is not None else None
-                        acc = message_data.get("accuracy_m")
-                        acc = float(acc) if acc is not None else None
-                        heading_deg = message_data.get("heading_deg")
-                        heading_deg = float(heading_deg) if heading_deg is not None else None
-                        tz = message_data.get("tz")
-                        locale = message_data.get("locale")
-                        device = message_data.get("device")
-
-                        # 後端節流：距離<100m且方位差<25度則忽略
-                        do_write_snapshot = False
-                        last = manager.last_env.get(user_id)
-                        if last and lat is not None and lon is not None and last.get("lat") is not None:
-                            dist = _haversine_m(last.get("lat", 0), last.get("lon", 0), lat, lon)
-                            deg_diff = abs((heading_deg or 0) - (last.get("heading_deg") or 0))
-                            if dist >= 100 or deg_diff >= 25:
-                                do_write_snapshot = True
-                        else:
-                            do_write_snapshot = True
-
-                        from geohash2 import encode as gh_encode
-                        geohash7 = gh_encode(lat, lon, precision=7) if (lat is not None and lon is not None) else None
-                        heading_cardinal = _heading_to_cardinal(heading_deg) if heading_deg is not None else None
-                        city = message_data.get("city")
-                        admin = message_data.get("admin")
-                        country_code = message_data.get("country_code")
-                        address_display = message_data.get("address_display")
-                        # 若前端未提供城市資訊，嘗試透過 MCP reverse_geocode 取得
-                        if (not city) and lat is not None and lon is not None:
-                            try:
-                                feature_router: MCPAgentBridge = app.state.feature_router
-                                reverse_tool = feature_router.mcp_server.tools.get("reverse_geocode")
-                                if reverse_tool and reverse_tool.handler:
-                                    geo_res = await reverse_tool.handler({"lat": lat, "lon": lon})
-                                    if isinstance(geo_res, dict) and geo_res.get("success"):
-                                        payload = geo_res.get("data") or geo_res
-                                        # 取得所有詳細欄位
-                                        city = payload.get("city") or city
-                                        admin = payload.get("admin") or admin
-                                        country_code = payload.get("country_code") or country_code
-                                        address_display = payload.get("label") or payload.get("display_name") or address_display
-                                        
-                                        # 新增：精確地址資訊
-                                        detailed_address = payload.get("detailed_address")
-                                        label = payload.get("label")
-                                        road = payload.get("road")
-                                        house_number = payload.get("house_number")
-                                        suburb = payload.get("suburb")
-                                        city_district = payload.get("city_district")
-                                        postcode = payload.get("postcode")
-                                        amenity = payload.get("amenity")
-                                        shop = payload.get("shop")
-                                        building = payload.get("building")
-                                        office = payload.get("office")
-                                        leisure = payload.get("leisure")
-                                        tourism = payload.get("tourism")
-                                        name = payload.get("name")
-                            except Exception as ge:
-                                logger.debug(f"反地理查詢失敗: {ge}")
-                                # 如果 reverse_geocode 失敗，保持原有變數為 None
-                                detailed_address = None
-                                label = None
-                                road = None
-                                house_number = None
-                                suburb = None
-                                city_district = None
-                                postcode = None
-                                amenity = None
-                                shop = None
-                                building = None
-                                office = None
-                                leisure = None
-                                tourism = None
-                                name = None
-                        else:
-                            # 如果沒有執行 reverse_geocode，初始化為 None
-                            detailed_address = None
-                            label = None
-                            road = None
-                            house_number = None
-                            suburb = None
-                            city_district = None
-                            postcode = None
-                            amenity = None
-                            shop = None
-                            building = None
-                            office = None
-                            leisure = None
-                            tourism = None
-                            name = None
-
-                        env_payload = {
-                            "lat": lat,
-                            "lon": lon,
-                            "accuracy_m": acc,
-                            "heading_deg": heading_deg,
-                            "heading_cardinal": heading_cardinal,
-                            "tz": tz,
-                            "locale": locale,
-                            "device": device,
-                            "geohash_7": geohash7,
-                            "city": city,
-                            "admin": admin,
-                            "country_code": country_code,
-                            "address_display": address_display,
-                            # 新增：精確地址欄位
-                            "detailed_address": detailed_address,
-                            "label": label,
-                            "road": road,
-                            "house_number": house_number,
-                            "suburb": suburb,
-                            "city_district": city_district,
-                            "postcode": postcode,
-                            "amenity": amenity,
-                            "shop": shop,
-                            "building": building,
-                            "office": office,
-                            "leisure": leisure,
-                            "tourism": tourism,
-                            "name": name,
-                        }
-
-                        # 更新會話暫存
-                        manager.last_env[user_id] = env_payload
-                        info = manager.get_client_info(user_id) or {}
-                        info["env_context"] = env_payload
-                        manager.set_client_info(user_id, info)
-
-                        try:
-                            await set_user_env_current(user_id, env_payload)
-                        except Exception as e:
-                            logger.warning(f"寫入環境現況失敗: {e}")
+                        env_service: EnvironmentContextService = app.state.env_service
 
-                        if do_write_snapshot:
+                        async def _reverse_geocode(lat: float, lon: float):
+                            feature_router: MCPAgentBridge = app.state.feature_router
+                            tool = feature_router.mcp_server.tools.get("reverse_geocode")
+                            if not tool or not tool.handler:
+                                return None
                             try:
-                                snap = env_payload.copy()
-                                snap["reason"] = "threshold"
-                                await add_user_env_snapshot(user_id, snap)
-                            except Exception as e:
-                                logger.warning(f"寫入環境快照失敗: {e}")
-
-                        await websocket.send_json(
-                            {"type": "env_ack", "success": True, "geohash_7": geohash7, "heading_cardinal": heading_cardinal}
+                                result = await tool.handler({"lat": lat, "lon": lon})
+                            except Exception as geo_exc:
+                                logger.debug(f"反地理查詢失敗: {geo_exc}")
+                                return None
+                            if not isinstance(result, dict) or not result.get("success"):
+                                return None
+                            payload = result.get("data") or result
+                            enriched = {
+                                "city": payload.get("city"),
+                                "admin": payload.get("admin"),
+                                "country_code": payload.get("country_code"),
+                                "address_display": payload.get("label") or payload.get("display_name"),
+                                "detailed_address": payload.get("detailed_address"),
+                                "label": payload.get("label"),
+                                "road": payload.get("road"),
+                                "house_number": payload.get("house_number"),
+                                "suburb": payload.get("suburb"),
+                                "city_district": payload.get("city_district"),
+                                "postcode": payload.get("postcode"),
+                                "amenity": payload.get("amenity"),
+                                "shop": payload.get("shop"),
+                                "building": payload.get("building"),
+                                "office": payload.get("office"),
+                                "leisure": payload.get("leisure"),
+                                "tourism": payload.get("tourism"),
+                                "name": payload.get("name"),
+                            }
+                            return {k: v for k, v in enriched.items() if v is not None}
+
+                        geocode_provider = _reverse_geocode if app.state.feature_router else None
+                        ack = await env_service.ingest_snapshot(
+                            user_id,
+                            message_data,
+                            geocode_provider=geocode_provider,
                         )
+                        ctx = await env_service.get_context(user_id, allow_stale=True)
+                        if ctx:
+                            manager.last_env[user_id] = ctx
+                            info = manager.get_client_info(user_id) or {}
+                            info["env_context"] = ctx
+                            manager.set_client_info(user_id, info)
+                        await websocket.send_json({"type": "env_ack", **ack})
                     except Exception as e:
                         logger.error(f"處理 env_snapshot 失敗: {e}")
                         await websocket.send_json({"type": "env_ack", "success": False, "error": str(e)})
 
+                elif message_type == "chat_focus":
                 elif message_type == "chat_focus":
                     try:
                         cid = message_data.get("chat_id")
@@ -1481,6 +1382,14 @@ async def handle_message(user_message, user_id, chat_id, messages, request_id: s
         return result
 
     async def _ai(messages_in, cid, model, rid, chat_id, use_care_mode=False, care_emotion=None, emotion_label=None):
+        env_context = {}
+        env_service = getattr(app.state, 'env_service', None)
+        if env_service:
+            try:
+                env_context = await env_service.get_context(cid, allow_stale=True)
+            except Exception as env_err:
+                logger.debug(f'讀取環境快取失敗: {env_err}')
+
         # 取得用戶名稱（優先順序：Google 名稱 > 語音 label > "用戶"）
         user_name = "用戶"
         try:
@@ -1502,6 +1411,7 @@ async def handle_message(user_message, user_id, chat_id, messages, request_id: s
                 care_emotion=care_emotion,
                 user_name=user_name,
                 emotion_label=emotion_label,
+                env_context=env_context,
             )
         else:
             return await ai_service.generate_response_for_user(
@@ -1514,6 +1424,7 @@ async def handle_message(user_message, user_id, chat_id, messages, request_id: s
                 care_emotion=care_emotion,
                 user_name=user_name,
                 emotion_label=emotion_label,
+                env_context=env_context,
             )
 
     model = settings.OPENAI_MODEL

core/config.py

@@ -97,6 +97,11 @@ class Settings:
     # ===== 背景任務開關 =====
     ENABLE_BACKGROUND_JOBS: bool = os.getenv("ENABLE_BACKGROUND_JOBS", "true").lower() == "true"
 
+    # ===== 環境感知參數 =====
+    ENV_CONTEXT_DISTANCE_THRESHOLD: float = float(os.getenv("ENV_CONTEXT_DISTANCE_THRESHOLD", "100"))
+    ENV_CONTEXT_HEADING_THRESHOLD: float = float(os.getenv("ENV_CONTEXT_HEADING_THRESHOLD", "25"))
+    ENV_CONTEXT_TTL_SECONDS: float = float(os.getenv("ENV_CONTEXT_TTL_SECONDS", "300"))
+
     @classmethod
     def validate(cls) -> bool:
         """
@@ -160,6 +165,9 @@ class Settings:
         print(f"Weather API Key: {'已設定 ✅' if cls.WEATHER_API_KEY else '未設定 ❌'}")
         print(f"NewsData API Key: {'已設定 ✅' if cls.NEWSDATA_API_KEY else '未設定 ❌'}")
         print(f"Exchange API Key: {'已設定 ✅' if cls.EXCHANGE_API_KEY else '未設定 ❌'}")
+        print(f"環境節流距離: {cls.ENV_CONTEXT_DISTANCE_THRESHOLD} m")
+        print(f"環境節流方位差: {cls.ENV_CONTEXT_HEADING_THRESHOLD}°")
+        print(f"環境快取 TTL: {cls.ENV_CONTEXT_TTL_SECONDS} 秒")
         print("=" * 60 + "\n")
 
 

core/environment/__init__.py

@@ -0,0 +1,3 @@
+from .context_service import EnvironmentContextService, EnvironmentSnapshot
+
+__all__ = ["EnvironmentContextService", "EnvironmentSnapshot"]

core/environment/context_service.py

@@ -0,0 +1,342 @@
+import asyncio
+import math
+import time
+from dataclasses import dataclass, field
+from typing import Any, Awaitable, Callable, Dict, Optional, Tuple
+
+
+EnvFetcher = Callable[[str], Awaitable[Dict[str, Any]]]
+EnvWriter = Callable[[str, Dict[str, Any]], Awaitable[Dict[str, Any]]]
+GeoFetcher = Callable[[float, float], Awaitable[Optional[Dict[str, Any]]]]
+
+
+@dataclass
+class EnvironmentSnapshot:
+    data: Dict[str, Any]
+    updated_at: float = field(default_factory=lambda: time.time())
+
+
+class EnvironmentContextService:
+    """
+    管理即時環境資訊：
+    - 記憶體快取 + TTL
+    - 節流距離/方位差
+    - Firestore 寫入排程（current + snapshots）
+    - 反地理查詢背景處理
+    """
+
+    def __init__(
+        self,
+        *,
+        min_distance_m: float,
+        min_heading_deg: float,
+        ttl_seconds: float,
+        env_fetcher: EnvFetcher,
+        env_writer: EnvWriter,
+        snapshot_writer: Optional[EnvWriter] = None,
+    ) -> None:
+        self._min_distance = max(min_distance_m, 0.0)
+        self._min_heading = max(min_heading_deg, 0.0)
+        self._ttl = max(ttl_seconds, 1.0)
+        self._env_fetcher = env_fetcher
+        self._env_writer = env_writer
+        self._snapshot_writer = snapshot_writer
+
+        self._cache: Dict[str, EnvironmentSnapshot] = {}
+        self._write_queue: "asyncio.Queue[Tuple[str, Dict[str, Any]]]" = asyncio.Queue()
+        self._snapshot_queue: "asyncio.Queue[Tuple[str, Dict[str, Any]]]" = asyncio.Queue()
+        self._writer_task: Optional[asyncio.Task] = None
+        self._snapshot_task: Optional[asyncio.Task] = None
+        self._geo_tasks: Dict[str, asyncio.Task] = {}
+        self._lock = asyncio.Lock()
+
+    # --------------------------------------------------------------------- #
+    # 公開介面
+    # --------------------------------------------------------------------- #
+    async def start(self) -> None:
+        if self._writer_task is None:
+            self._writer_task = asyncio.create_task(self._write_loop(), name="env-current-writer")
+        if self._snapshot_writer and self._snapshot_task is None:
+            self._snapshot_task = asyncio.create_task(self._snapshot_loop(), name="env-snapshot-writer")
+
+    async def shutdown(self) -> None:
+        for pending in self._geo_tasks.values():
+            pending.cancel()
+        self._geo_tasks.clear()
+
+        if self._writer_task:
+            self._writer_task.cancel()
+            with contextlib.suppress(asyncio.CancelledError):
+                await self._writer_task
+            self._writer_task = None
+
+        if self._snapshot_task:
+            self._snapshot_task.cancel()
+            with contextlib.suppress(asyncio.CancelledError):
+                await self._snapshot_task
+            self._snapshot_task = None
+
+    async def ingest_snapshot(
+        self,
+        user_id: str,
+        raw_payload: Dict[str, Any],
+        *,
+        geocode_provider: Optional[GeoFetcher] = None,
+    ) -> Dict[str, Any]:
+        """
+        接收前端發送的環境快照，立即回傳 ACK 與基本資料，
+        寫入 Firestore 與反地理查詢則交由背景處理。
+        """
+        if not user_id:
+            raise ValueError("user_id is required for environment snapshot ingestion")
+
+        normalized, write_snapshot = await self._normalize_snapshot(user_id, raw_payload)
+
+        async with self._lock:
+            self._cache[user_id] = EnvironmentSnapshot(data=normalized)
+
+        await self._write_queue.put((user_id, normalized))
+        if write_snapshot and self._snapshot_writer:
+            await self._snapshot_queue.put((user_id, normalized))
+
+        if geocode_provider and self._needs_geocode(normalized):
+            await self._schedule_geocode(user_id, normalized, geocode_provider)
+
+        ack = {
+            "success": True,
+            "geohash_7": normalized.get("geohash_7"),
+            "heading_cardinal": normalized.get("heading_cardinal"),
+        }
+        return ack
+
+    async def get_context(self, user_id: str, *, allow_stale: bool = False) -> Dict[str, Any]:
+        async with self._lock:
+            cached = self._cache.get(user_id)
+            if cached and (allow_stale or not self._is_stale(cached)):
+                return dict(cached.data)
+
+        data = await self._env_fetcher(user_id)
+        if data.get("success"):
+            ctx = data.get("context") or {}
+            async with self._lock:
+                self._cache[user_id] = EnvironmentSnapshot(data=ctx)
+            return dict(ctx)
+
+        return {}
+
+    # --------------------------------------------------------------------- #
+    # 內部流程
+    # --------------------------------------------------------------------- #
+    async def _normalize_snapshot(
+        self,
+        user_id: str,
+        raw_payload: Dict[str, Any],
+    ) -> Tuple[Dict[str, Any], bool]:
+        lat = _safe_float(raw_payload.get("lat"))
+        lon = _safe_float(raw_payload.get("lon"))
+        accuracy = _safe_float(raw_payload.get("accuracy_m"))
+        heading_deg = _safe_float(raw_payload.get("heading_deg"))
+
+        ctx = {
+            "lat": lat,
+            "lon": lon,
+            "accuracy_m": accuracy,
+            "heading_deg": heading_deg,
+            "heading_cardinal": _heading_to_cardinal(heading_deg) if heading_deg is not None else None,
+            "tz": raw_payload.get("tz"),
+            "locale": raw_payload.get("locale"),
+            "device": raw_payload.get("device"),
+            "city": raw_payload.get("city"),
+            "admin": raw_payload.get("admin"),
+            "country_code": raw_payload.get("country_code"),
+            "address_display": raw_payload.get("address_display"),
+            "geohash_7": _encode_geohash(lat, lon),
+            "updated_at": time.time(),
+        }
+
+        previous = await self._get_cached(user_id)
+        should_snapshot = self._should_snapshot(previous, ctx)
+
+        if previous and not self._has_position_change(previous.data, ctx):
+            # 沒有座標變化時保留先前的精細地理資訊
+            for key in (
+                "detailed_address",
+                "label",
+                "road",
+                "house_number",
+                "suburb",
+                "city_district",
+                "postcode",
+                "amenity",
+                "shop",
+                "building",
+                "office",
+                "leisure",
+                "tourism",
+                "name",
+            ):
+                ctx[key] = previous.data.get(key)
+
+        return ctx, should_snapshot
+
+    async def _schedule_geocode(
+        self,
+        user_id: str,
+        ctx: Dict[str, Any],
+        geocode_provider: GeoFetcher,
+    ) -> None:
+        if user_id in self._geo_tasks:
+            # 已有任務在跑，避免重複
+            return
+
+        async def _task() -> None:
+            try:
+                if ctx.get("lat") is None or ctx.get("lon") is None:
+                    return
+                enriched = await geocode_provider(ctx["lat"], ctx["lon"])
+                if not enriched:
+                    return
+
+                async with self._lock:
+                    cached = self._cache.get(user_id)
+                    if not cached:
+                        cached = EnvironmentSnapshot(data=dict(ctx))
+                        self._cache[user_id] = cached
+                    cached.data.update(enriched)
+                    cached.updated_at = time.time()
+
+                await self._write_queue.put((user_id, dict(cached.data)))
+                if self._snapshot_writer:
+                    await self._snapshot_queue.put((user_id, dict(cached.data)))
+            finally:
+                self._geo_tasks.pop(user_id, None)
+
+        self._geo_tasks[user_id] = asyncio.create_task(_task(), name=f"env-geocode-{user_id}")
+
+    async def _write_loop(self) -> None:
+        while True:
+            user_id, payload = await self._write_queue.get()
+            try:
+                await self._env_writer(user_id, payload)
+            except Exception:
+                # 寫入失敗時稍後重試
+                await asyncio.sleep(1.0)
+                await self._write_queue.put((user_id, payload))
+            finally:
+                self._write_queue.task_done()
+
+    async def _snapshot_loop(self) -> None:
+        while True:
+            user_id, payload = await self._snapshot_queue.get()
+            try:
+                await self._snapshot_writer(user_id, payload)
+            except Exception:
+                await asyncio.sleep(2.0)
+                await self._snapshot_queue.put((user_id, payload))
+            finally:
+                self._snapshot_queue.task_done()
+
+    async def _get_cached(self, user_id: str) -> Optional[EnvironmentSnapshot]:
+        async with self._lock:
+            return self._cache.get(user_id)
+
+    def _should_snapshot(self, previous: Optional[EnvironmentSnapshot], current: Dict[str, Any]) -> bool:
+        if previous is None:
+            return True
+        return self._has_position_change(previous.data, current)
+
+    def _has_position_change(self, previous: Dict[str, Any], current: Dict[str, Any]) -> bool:
+        if previous.get("lat") is None or previous.get("lon") is None:
+            return True
+        if current.get("lat") is None or current.get("lon") is None:
+            return False
+
+        distance = _haversine_m(previous["lat"], previous["lon"], current["lat"], current["lon"])
+        if distance >= self._min_distance:
+            return True
+
+        prev_heading = previous.get("heading_deg")
+        curr_heading = current.get("heading_deg")
+        if prev_heading is None or curr_heading is None:
+            return False
+
+        heading_diff = abs(curr_heading - prev_heading)
+        heading_diff = min(heading_diff, 360 - heading_diff)
+        return heading_diff >= self._min_heading
+
+    def _is_stale(self, snapshot: EnvironmentSnapshot) -> bool:
+        return (time.time() - snapshot.updated_at) > self._ttl
+
+    def _needs_geocode(self, ctx: Dict[str, Any]) -> bool:
+        if ctx.get("lat") is None or ctx.get("lon") is None:
+            return False
+        return not any(ctx.get(field) for field in ("city", "address_display", "label", "detailed_address"))
+
+
+def _safe_float(value: Any) -> Optional[float]:
+    try:
+        if value is None:
+            return None
+        return float(value)
+    except (TypeError, ValueError):
+        return None
+
+
+def _heading_to_cardinal(deg: Optional[float]) -> Optional[str]:
+    if deg is None:
+        return None
+    try:
+        val = float(deg)
+    except (TypeError, ValueError):
+        return None
+
+    dirs = [
+        "N",
+        "NNE",
+        "NE",
+        "ENE",
+        "E",
+        "ESE",
+        "SE",
+        "SSE",
+        "S",
+        "SSW",
+        "SW",
+        "WSW",
+        "W",
+        "WNW",
+        "NW",
+        "NNW",
+    ]
+    idx = int((val % 360) / 22.5 + 0.5) % len(dirs)
+    return dirs[idx]
+
+
+def _encode_geohash(lat: Optional[float], lon: Optional[float]) -> Optional[str]:
+    if lat is None or lon is None:
+        return None
+    try:
+        from geohash2 import encode as gh_encode  # type: ignore
+    except Exception:
+        return None
+    try:
+        return gh_encode(lat, lon, precision=7)
+    except Exception:
+        return None
+
+
+def _haversine_m(lat1: float, lon1: float, lat2: float, lon2: float) -> float:
+    """
+    使用哈弗辛公式計算兩點距離（公尺）
+    """
+    rad_lat1 = math.radians(lat1)
+    rad_lat2 = math.radians(lat2)
+    dlat = rad_lat2 - rad_lat1
+    dlon = math.radians(lon2 - lon1)
+
+    a = math.sin(dlat / 2) ** 2 + math.cos(rad_lat1) * math.cos(rad_lat2) * math.sin(dlon / 2) ** 2
+    c = 2 * math.asin(math.sqrt(a))
+    return 6371000.0 * c
+
+
+import contextlib  # noqa: E402  # placed at end to avoid circular import at module load

features/mcp/agent_bridge.py

@@ -6,13 +6,15 @@ MCP + Agent 橋接層
 import json
 import logging
 import asyncio
-from typing import Dict, Any, Optional, List, Tuple
+from typing import Dict, Any, Optional, List, Tuple, Callable, Awaitable
 from datetime import datetime
 from .server import FeaturesMCPServer
 import services.ai_service as ai_service
 from services.ai_service import StrictResponseError
 from core.reasoning_strategy import get_optimal_reasoning_effort
 from core.database import get_user_env_current
+from .coordinator import ToolCoordinator
+from .tool_models import ToolMetadata, ToolResult
 
 logger = logging.getLogger("mcp.agent_bridge")
 logger.setLevel(logging.DEBUG)  # 強制設置為 DEBUG 級別
@@ -30,10 +32,13 @@ def _safe_json(data: Any, limit: int = 1200) -> str:
     return text
 
 
+EnvProvider = Callable[[Optional[str]], Awaitable[Dict[str, Any]]]
+
+
 class MCPAgentBridge:
     """MCP + Agent 橋接器，提供與舊 FeatureRouter 相同的介面"""
 
-    def __init__(self):
+    def __init__(self, env_provider: Optional[EnvProvider] = None):
         # 初始化 MCP 服務器
         self.mcp_server = FeaturesMCPServer()
 
@@ -48,9 +53,113 @@ class MCPAgentBridge:
         self._intent_cache: Dict[str, Tuple[bool, Optional[Dict[str, Any]], float]] = {}
         self._intent_cache_ttl = 300.0  # 5分鐘（60s → 300s，提升命中率 40-60%）
 
+        self._env_provider: EnvProvider = env_provider or self._default_env_provider
+        self._tool_coordinator = ToolCoordinator(
+            env_provider=self._delegated_env_provider,
+            tool_lookup=self._lookup_tool_handler,
+            formatter=self._format_with_ai,
+            failure_handlers={
+                'directions': self._directions_failure_fallback,
+            },
+        )
+        self._register_tool_metadata()
+
         logger.info("MCP Agent 橋接層初始化完成")
         logger.info(f"初始可用 MCP 工具數量: {len(self.mcp_server.tools)} (將在異步發現後更新)")
 
+    async def _default_env_provider(self, user_id: Optional[str]) -> Dict[str, Any]:
+        if not user_id:
+            return {}
+        try:
+            env_res = await get_user_env_current(user_id)
+            if env_res.get("success"):
+                return env_res.get("context") or {}
+        except Exception as exc:  # noqa: BLE001
+            logger.debug("讀取使用者 %s 環境資訊失敗: %s", user_id, exc)
+        return {}
+
+    async def _delegated_env_provider(self, user_id: Optional[str]) -> Dict[str, Any]:
+        provider = self._env_provider or self._default_env_provider
+        return await provider(user_id)
+
+    def bind_env_provider(self, provider: EnvProvider) -> None:
+        self._env_provider = provider
+
+    def _lookup_tool_handler(self, tool_name: str):
+        tool = self.mcp_server.tools.get(tool_name)
+        return getattr(tool, "handler", None) if tool else None
+
+    async def _format_with_ai(
+        self,
+        tool_name: str,
+        message: str,
+        payload: Dict[str, Any],
+        original_message: str,
+    ) -> str:
+        return await self._format_tool_response(tool_name, message, original_message)
+
+    def _register_tool_metadata(self) -> None:
+        register = self._tool_coordinator.register
+        register(
+            ToolMetadata(
+                name="weather_query",
+                requires_env={"lat", "lon", "city"},
+                enable_reformat=True,
+            )
+        )
+        register(
+            ToolMetadata(
+                name="reverse_geocode",
+                requires_env={"lat", "lon"},
+            )
+        )
+        register(
+            ToolMetadata(
+                name="exchange_query",
+                enable_reformat=True,
+            )
+        )
+        register(
+            ToolMetadata(
+                name="news_query",
+                enable_reformat=True,
+            )
+        )
+        register(
+            ToolMetadata(
+                name="healthkit_query",
+                enable_reformat=True,
+            )
+        )
+        register(
+            ToolMetadata(
+                name="directions",
+                enable_reformat=True,
+            )
+        )
+        register(
+            ToolMetadata(
+                name="forward_geocode",
+                flow="navigation",
+            )
+        )
+
+    def _directions_failure_fallback(self, arguments: Dict[str, Any], exc: Exception) -> ToolResult:
+        labels = {
+            "origin_label": arguments.get("origin_label") or "起點",
+            "dest_label": arguments.get("dest_label") or "目的地",
+        }
+        fallback = self._build_directions_failure_response(
+            arguments,
+            labels,
+            str(exc),
+        )
+        return ToolResult(
+            name="directions",
+            message=fallback["message"],
+            data=fallback.get("tool_data"),
+        )
+
     async def async_initialize(self):
         """異步初始化，發現所有工具 + 快取預熱"""
         if hasattr(self.mcp_server, 'start_external_servers'):
@@ -85,142 +194,6 @@ class MCPAgentBridge:
                 return registered_name
 
         return None
-    async def _fetch_env_context(self, user_id: Optional[str]) -> Dict[str, Any]:
-        """讀取使用者最近的環境資訊（Firestore current snapshot）。"""
-        if not user_id:
-            return {}
-        try:
-            env_res = await get_user_env_current(user_id)
-            if env_res.get("success"):
-                ctx = env_res.get("context") or {}
-                return ctx
-        except Exception as e:
-            logger.debug(f"無法取得使用者 {user_id} 環境資訊: {e}")
-        return {}
-
-    async def _enrich_arguments_with_env(self, tool_name: str, arguments: Dict[str, Any], user_id: Optional[str]) -> Dict[str, Any]:
-        """自動將環境資訊補入 MCP 工具參數，讓位置相關功能更聰明。"""
-        if not user_id:
-            return arguments
-
-        tool_name = (tool_name or "").strip()
-        if tool_name not in {"weather_query", "reverse_geocode"}:
-            return arguments
-
-        ctx = await self._fetch_env_context(user_id)
-        if not ctx:
-            return arguments
-
-        enriched = dict(arguments or {})
-
-        def _safe_float(val):
-            try:
-                if val is None:
-                    return None
-                return float(val)
-            except (TypeError, ValueError):
-                return None
-
-        if tool_name == "weather_query":
-            if enriched.get("lat") is None:
-                lat = _safe_float(ctx.get("lat"))
-                if lat is not None:
-                    enriched["lat"] = lat
-            if enriched.get("lon") is None:
-                lon = _safe_float(ctx.get("lon"))
-                if lon is not None:
-                    enriched["lon"] = lon
-            city_arg = str(enriched.get("city") or "").strip()
-            ctx_city = str(ctx.get("city") or "").strip()
-            if not city_arg and ctx_city:
-                enriched["city"] = ctx_city
-
-        # 🔥 新增：reverse_geocode 自動注入當前 GPS 座標
-        if tool_name == "reverse_geocode":
-            if enriched.get("lat") is None:
-                lat = _safe_float(ctx.get("lat"))
-                if lat is not None:
-                    enriched["lat"] = lat
-            if enriched.get("lon") is None:
-                lon = _safe_float(ctx.get("lon"))
-                if lon is not None:
-                    enriched["lon"] = lon
-
-        if enriched != arguments:
-            logger.info(f"📍 已自動補齊 {tool_name} 參數: {_safe_json(enriched)}")
-
-        return enriched
-
-    async def _resolve_coordinate_label(self, lat: Any, lon: Any) -> Optional[str]:
-        """透過 reverse_geocode 將座標轉換為可朗讀的地點名稱。"""
-        try:
-            lat_f = float(lat)
-            lon_f = float(lon)
-        except (TypeError, ValueError):
-            return None
-
-        reverse_tool = self.mcp_server.tools.get("reverse_geocode")
-        if not reverse_tool or not reverse_tool.handler:
-            return None
-
-        try:
-            res = await reverse_tool.handler({"lat": lat_f, "lon": lon_f})
-        except Exception as ge:
-            logger.debug(f"reverse_geocode 失敗: {ge}")
-            return None
-
-        if not isinstance(res, dict):
-            return None
-        if not res.get("success"):
-            return None
-
-        payload = res.get("data") or res
-        label = (
-            payload.get("label")
-            or payload.get("display_name")
-            or ", ".join(
-                part for part in [payload.get("city"), payload.get("admin")] if part
-            )
-        )
-        return label.strip() if label else None
-
-    async def _prepare_route_arguments(self, arguments: Dict[str, Any]) -> Tuple[Dict[str, Any], Dict[str, str]]:
-        """為 directions 工具補齊可讀地點名稱並正規化座標。"""
-        prepared = dict(arguments or {})
-        labels: Dict[str, str] = {}
-
-        def _normalize_coord(value: Any) -> Optional[float]:
-            try:
-                if value is None:
-                    return None
-                return float(value)
-            except (TypeError, ValueError):
-                return None
-
-        for prefix, default_label in (("origin", "起點"), ("dest", "目的地")):
-            lat_key = f"{prefix}_lat"
-            lon_key = f"{prefix}_lon"
-            label_key = f"{prefix}_label"
-
-            lat_val = _normalize_coord(prepared.get(lat_key))
-            lon_val = _normalize_coord(prepared.get(lon_key))
-            if lat_val is not None:
-                prepared[lat_key] = lat_val
-            if lon_val is not None:
-                prepared[lon_key] = lon_val
-
-            label_val = str(prepared.get(label_key) or "").strip()
-            if not label_val and lat_val is not None and lon_val is not None:
-                label_val = await self._resolve_coordinate_label(lat_val, lon_val) or ""
-
-            if not label_val:
-                label_val = default_label
-
-            prepared[label_key] = label_val
-            labels[label_key] = label_val
-
-        return prepared, labels
-
     @staticmethod
     def _format_distance(distance_m: Optional[float]) -> str:
         """將距離換算為人類可讀格式。"""
@@ -781,17 +754,10 @@ class MCPAgentBridge:
         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. 標記 [複雜] 的工具只需返回工具名稱，參數稍後填充")
+        result.append("3. 公共運輸查詢 → TDX 相關工具暫時停用（待取得替代 API）")
+        result.append("4. 健康數據查詢 → healthkit_query（心率、步數、血氧等）")
+        result.append("5. 生活資訊 → weather_query（天氣）、news_query（新聞）、exchange_query（匯率）")
+        result.append("6. 標記 [複雜] 的工具只需返回工具名稱，參數稍後填充")
         
         logger.debug(f"工具描述已生成，總長度: {len(''.join(result))} 字元")
         return "\n".join(result)
@@ -871,246 +837,32 @@ class MCPAgentBridge:
             tool_name = intent_data.get("tool_name")
             arguments = intent_data.get("arguments", {})
 
-            # 補齊健康工具必要參數
-            if tool_name == "healthkit_query":
-                if (not arguments.get("user_id")) and user_id:
-                    arguments = {**arguments, "user_id": user_id}
-                    logger.info("自動補齊 healthkit_query user_id")
-                if "metric_type" not in arguments or not arguments["metric_type"]:
-                    arguments = {**arguments, "metric_type": "all"}
-
-            return await self._call_mcp_tool(tool_name, arguments, user_id, original_message)
+            try:
+                result = await self._tool_coordinator.invoke(
+                    tool_name,
+                    arguments or {},
+                    user_id=user_id,
+                    original_message=original_message,
+                )
+            except Exception as exc:  # noqa: BLE001
+                logger.warning("工具 %s 執行失敗: %s", tool_name, exc)
+                return self._generate_tool_error_message(tool_name, exc, original_message)
+
+            if isinstance(result, ToolResult):
+                if result.name == 'directions' and isinstance(result.data, dict):
+                    message, sanitized = self._build_directions_message(result.data, {})
+                    result.message = message
+                    result.data = sanitized
+                return result.to_dict()
+            return result
 
         else:
             logger.warning(f"未知意圖類型: {intent_type}")
             return f"抱歉，無法理解您的請求。"
 
     async def _call_mcp_tool(self, tool_name: str, arguments: Dict[str, Any],
-                           user_id: str = None, original_message: str = "") -> str:
-        """
-        調用 MCP 工具（帶智慧重試機制 + 統一格式化 + 智能地點查詢）
-        2025年最佳實踐：指數退避重試 + 錯誤分類 + AI 格式化 + 自動 geocoding
-        """
-        if tool_name not in self.mcp_server.tools:
-            return self._generate_tool_not_found_error(tool_name)
-
-        tool = self.mcp_server.tools[tool_name]
-        if not tool.handler:
-            return f"⚠️ 工具 {tool_name} 尚未實作，請稍後再試"
-
-        # 智能地點查詢：如果是 forward_geocode，且用戶有位置導航需求，自動串接 directions
-        is_navigation_intent = False
-        geocode_result = None
-        
-        if tool_name == "forward_geocode":
-            # 判斷是否為導航意圖（「怎麼去」「如何去」「到 X」）
-            nav_keywords = ["怎麼去", "如何去", "怎麼走", "到哪", "去哪", "要多久", "多遠"]
-            is_navigation_intent = any(keyword in original_message for keyword in nav_keywords)
-            
-            if is_navigation_intent:
-                logger.info(f"🗺️ 檢測到導航意圖��先執行地點查詢: {arguments.get('query')}")
-                
-                # 執行 geocoding
-                geocode_tool = self.mcp_server.tools.get("forward_geocode")
-                if geocode_tool and geocode_tool.handler:
-                    try:
-                        geocode_result = await asyncio.wait_for(
-                            geocode_tool.handler(arguments),
-                            timeout=15.0
-                        )
-                        
-                        if geocode_result.get("success"):
-                            best_match = geocode_result.get("data", {}).get("best_match", {})
-                            dest_lat = best_match.get("lat")
-                            dest_lon = best_match.get("lon")
-                            dest_label = best_match.get("label", arguments.get("query"))
-                            
-                            # 取得用戶當前位置
-                            env_ctx = await self._fetch_env_context(user_id)
-                            origin_lat = env_ctx.get("lat")
-                            origin_lon = env_ctx.get("lon")
-                            origin_label = env_ctx.get("label") or env_ctx.get("address_display") or "您的位置"
-                            
-                            if origin_lat and origin_lon and dest_lat and dest_lon:
-                                logger.info(f"🚗 自動串接導航: {origin_label} → {dest_label}")
-                                
-                                # 自動調用 directions
-                                directions_tool = self.mcp_server.tools.get("directions")
-                                if directions_tool and directions_tool.handler:
-                                    directions_args = {
-                                        "origin_lat": float(origin_lat),
-                                        "origin_lon": float(origin_lon),
-                                        "dest_lat": float(dest_lat),
-                                        "dest_lon": float(dest_lon),
-                                        "origin_label": origin_label,
-                                        "dest_label": dest_label,
-                                        "mode": "foot-walking"  # 預設步行
-                                    }
-                                    
-                                    # 遞迴調用 directions（會走下面的正常流程）
-                                    return await self._call_mcp_tool(
-                                        "directions",
-                                        directions_args,
-                                        user_id,
-                                        original_message
-                                    )
-                            else:
-                                logger.warning("⚠️ 無法取得完整位置資訊，返回地點查詢結果")
-                        else:
-                            logger.warning(f"⚠️ 地點查詢失敗: {geocode_result.get('error')}")
-                    except Exception as e:
-                        logger.error(f"❌ 自動地點查詢失敗: {e}", exc_info=True)
-
-        arguments = await self._enrich_arguments_with_env(tool_name, arguments, user_id)
-        route_labels: Dict[str, str] = {}
-        if tool_name == "directions":
-            arguments, route_labels = await self._prepare_route_arguments(arguments)
-
-        logger.info(f"🔧 調用 MCP 工具: {tool_name}")
-        logger.debug("📋 調用參數: %s", _safe_json(arguments))
-
-        # 重試設定
-        max_retries = 3
-        retry_delays = [1, 2, 5]  # 指數退避（秒）
-        
-        for attempt in range(max_retries):
-            try:
-                # 調用工具
-                result = await asyncio.wait_for(
-                    tool.handler(arguments),
-                    timeout=30.0  # 30秒超時
-                )
-                logger.debug("📤 工具回傳: %s", _safe_json(result))
-
-                # 處理結果
-                if isinstance(result, dict):
-                    if result.get("success"):
-                        content = result.get("content", "")
-
-                        # 檢查內容是否有效
-                        if not content or content.strip() == "":
-                            logger.warning(f"⚠️ 工具 {tool_name} 返回空內容")
-                            return f"✓ 工具 {tool_name} 執行成功，但沒有返回內容"
-
-                        # 成功!決策是否需要 AI 二次格式化
-                        logger.info(f"✅ 工具 {tool_name} 執行成功")
-
-                        # 保留原始數據供前端使用
-                        # 排除標準回應欄位，保留業務資料（如 rate, health_data, raw_data 等）
-                        excluded_keys = {'success', 'content', 'error', 'error_code', 'metadata'}
-                        tool_data = {k: v for k, v in result.items() if k not in excluded_keys}
-
-                        # 如果沒有業務資料，fallback 到 data 或 raw_data
-                        if not tool_data:
-                            tool_data = result.get("data") or result.get("raw_data")
-
-                        logger.debug(f"📊 提取的 tool_data: {type(tool_data)} = {tool_data if tool_data is None or isinstance(tool_data, (str, int, bool)) else '<dict/list>'}")
-
-                        if tool_name == "directions":
-                            message, sanitized_tool_data = self._build_directions_message(
-                                tool_data if isinstance(tool_data, dict) else {},
-                                route_labels,
-                            )
-                            content = message
-                            tool_data = sanitized_tool_data
-
-                        if self._should_reformat(tool_name, content):
-                            logger.info(f"🎨 啟用 AI 格式化: {tool_name}")
-                            try:
-                                formatted_content = await self._format_tool_response(
-                                    tool_name, content, original_message
-                                )
-                                # 返回擴充格式（dict），包含工具資訊
-                                result_dict = {
-                                    "message": formatted_content,
-                                    "tool_name": tool_name,
-                                    "tool_data": tool_data
-                                }
-                                logger.debug(f"🔙 返回格式化結果: message=<{len(formatted_content)} chars>, tool_name={tool_name}, tool_data={'None' if tool_data is None else 'present'}")
-                                return result_dict
-                            except Exception as e:
-                                logger.warning(f"⚠️ AI 格式化失敗，返回原始內容: {e}")
-                                # 格式化失敗仍然返回擴充格式
-                                result_dict = {
-                                    "message": content,
-                                    "tool_name": tool_name,
-                                    "tool_data": tool_data
-                                }
-                                logger.debug(f"🔙 返回原始結果: message=<{len(content)} chars>, tool_name={tool_name}, tool_data={'None' if tool_data is None else 'present'}")
-                                return result_dict
-                        else:
-                            # 直接返回工具自己的格式化結果（擴充格式）
-                            result_dict = {
-                                "message": content,
-                                "tool_name": tool_name,
-                                "tool_data": tool_data
-                            }
-                            logger.debug(f"🔙 返回直接結果: message=<{len(content)} chars>, tool_name={tool_name}, tool_data={'None' if tool_data is None else 'present'}")
-                            return result_dict
-                    
-                    else:
-                        # 失敗：檢查是否值得重試
-                        error = result.get("error", "工具執行失敗")
-                        error_lower = error.lower()
-                        
-                        # 可重試的錯誤類型
-                        retryable_errors = [
-                            "timeout", "網路", "network", "連接", "connection",
-                            "暫時", "temporary", "unavailable", "不可用"
-                        ]
-                        
-                        is_retryable = any(keyword in error_lower for keyword in retryable_errors)
-                        
-                        if is_retryable and attempt < max_retries - 1:
-                            delay = retry_delays[attempt]
-                            logger.warning(f"⚠️ 工具 {tool_name} 執行失敗（可重試）: {error}")
-                            logger.info(f"🔄 等待 {delay} 秒後重試... (嘗試 {attempt + 1}/{max_retries})")
-                            await asyncio.sleep(delay)
-                            continue  # 重試
-                        else:
-                            # 不可重試的錯誤或已達最大重試次數
-                            logger.error(f"❌ 工具 {tool_name} 執行失敗: {error}")
-                            return self._generate_helpful_error(tool_name, error, original_message)
-                
-                else:
-                    # 非標準格式回應
-                    logger.debug("工具回傳非標準格式，直接返回")
-                    return str(result)
-
-            except asyncio.TimeoutError:
-                if attempt < max_retries - 1:
-                    delay = retry_delays[attempt]
-                    logger.warning(f"⏱️ 工具 {tool_name} 超時，{delay} 秒後重試... (嘗試 {attempt + 1}/{max_retries})")
-                    await asyncio.sleep(delay)
-                    continue
-                else:
-                    logger.error(f"❌ 工具 {tool_name} 多次超時")
-                    return f"⏱️ 操作超時，請稍後再試\n\n建議：\n• 檢查網路連接\n• 稍等片刻後重新嘗試\n• 或試試其他功能"
-
-            except Exception as e:
-                error_msg = str(e)
-                error_lower = error_msg.lower()
-
-                if tool_name == "directions":
-                    logger.error(f"❌ directions 工具失敗，啟用替代回覆: {error_msg}")
-                    fallback_result = self._build_directions_failure_response(arguments, route_labels, error_msg)
-                    return fallback_result
-
-                # 判斷是否值得重試
-                is_retryable = any(keyword in error_lower for keyword in ["timeout", "network", "connection"])
-                
-                if is_retryable and attempt < max_retries - 1:
-                    delay = retry_delays[attempt]
-                    logger.warning(f"⚠️ 工具 {tool_name} 調用異常: {e}，{delay} 秒後重試...")
-                    await asyncio.sleep(delay)
-                    continue
-                else:
-                    logger.exception(f"❌ 調用 MCP 工具失敗: {e}")
-                    return self._generate_helpful_error(tool_name, error_msg, original_message)
-
-        # 所有重試都失敗
-        logger.error(f"❌ 工具 {tool_name} 在 {max_retries} 次嘗試後仍然失敗")
-        return f"❌ 調用 {tool_name} 失敗\n\n已嘗試 {max_retries} 次，建議：\n• 檢查網路連接\n• 稍後再試\n• 或聯繫管理員"
+                           user_id: str = None, original_message: str = '') -> str:
+        raise RuntimeError('legacy tool invocation path已移除，請改用 ToolCoordinator.invoke')
 
     def _generate_tool_not_found_error(self, tool_name: str) -> str:
         """生成工具不存在的友善錯誤訊息"""
@@ -1143,6 +895,13 @@ class MCPAgentBridge:
         error_msg += "\n輸入「/功能」查看完整功能列表"
         return error_msg
 
+    def _generate_tool_error_message(self, tool_name: str, error: Exception, original_message: str) -> str:
+        try:
+            return self._generate_helpful_error(tool_name, str(error), original_message)
+        except Exception as fallback_err:
+            logger.error('生成工具錯誤訊息失敗: %s', fallback_err)
+            return f'抱歉，{tool_name} 執行失敗：{error}'
+
     def _generate_helpful_error(self, tool_name: str, error: str, original_message: str) -> str:
         """生成有幫助的錯誤訊息"""
         error_lower = error.lower()

features/mcp/auto_registry.py

@@ -23,6 +23,14 @@ class MCPAutoRegistry:
         self.tools: Dict[str, Tool] = {}
         self.config: Dict[str, Any] = {}
         self.client_manager = MCPClientManager()
+        self._disabled_tools = {
+            "tdx_bus_arrival",
+            "tdx_metro",
+            "tdx_parking",
+            "tdx_thsr",
+            "tdx_train",
+            "tdx_youbike",
+        }
 
         # 載入配置
         self._load_config()
@@ -81,6 +89,10 @@ class MCPAutoRegistry:
                             #     continue
 
                             # 創建標準化工具實例
+                            if definition["name"] in self._disabled_tools:
+                                logger.info(f"跳過已禁用工具: {definition['name']}")
+                                continue
+
                             tool_instance = obj()
                             tool = self._create_tool_from_instance(tool_instance, definition)
                             if tool:
@@ -109,6 +121,9 @@ class MCPAutoRegistry:
 
                 if module_path and class_name:
                     # 從配置指定的模組載入
+                    if tool_name in self._disabled_tools:
+                        logger.info(f"跳過已禁用工具: {tool_name}")
+                        continue
                     tool = self._create_tool_from_config(tool_name, tool_info)
                     if tool:
                         discovered_tools.append(tool)
@@ -131,6 +146,9 @@ class MCPAutoRegistry:
         try:
             # 獲取工具定義
             definition = tool_class.get_definition()
+            if definition["name"] in self._disabled_tools:
+                logger.info(f"跳過已禁用工具: {definition['name']}")
+                return None
 
             # 檢查是否有模組級別的execute函數
             module = inspect.getmodule(tool_class)
@@ -426,4 +444,4 @@ class MCPAutoRegistry:
     async def cleanup(self):
         """清理資源，停止所有外部客戶端"""
         await self.client_manager.stop_all()
-        logger.info("MCP 自動註冊器清理完成")
+        logger.info("MCP 自動註冊器清理完成")

features/mcp/coordinator.py

@@ -0,0 +1,184 @@
+import asyncio
+import logging
+from typing import Any, Awaitable, Callable, Dict, Optional
+
+from .tool_models import ToolMetadata, ToolResult
+
+logger = logging.getLogger(__name__)
+
+EnvProvider = Callable[[Optional[str]], Awaitable[Dict[str, Any]]]
+ResultFormatter = Callable[[str, str, Dict[str, Any], str], Awaitable[str]]
+ToolHandler = Callable[[Dict[str, Any]], Awaitable[Any]]
+
+
+class ToolCoordinator:
+    """
+    統一管理 MCP 工具調用：
+    - 依 ToolMetadata 注入環境/預設值
+    - 處理特殊流程（導航）
+    - 統一結果格式
+    """
+
+    def __init__(
+        self,
+        *,
+        env_provider: EnvProvider,
+        tool_lookup: Callable[[str], Optional[ToolHandler]],
+        formatter: ResultFormatter,
+        failure_handlers: Optional[Dict[str, Callable[[Dict[str, Any], Exception], ToolResult]]] = None,
+    ) -> None:
+        self._env_provider = env_provider
+        self._tool_lookup = tool_lookup
+        self._formatter = formatter
+        self._metadata: Dict[str, ToolMetadata] = {}
+        self._failure_handlers = failure_handlers or {}
+
+    # ------------------------------------------------------------------ #
+    def register(self, metadata: ToolMetadata) -> None:
+        self._metadata[metadata.name] = metadata
+
+    def get_metadata(self, name: str) -> Optional[ToolMetadata]:
+        return self._metadata.get(name)
+
+    # ------------------------------------------------------------------ #
+    async def invoke(
+        self,
+        tool_name: str,
+        arguments: Dict[str, Any],
+        *,
+        user_id: Optional[str],
+        original_message: str,
+    ) -> ToolResult:
+        metadata = self._metadata.get(tool_name, ToolMetadata(name=tool_name))
+
+        if metadata.flow == "navigation":
+            return await self._handle_navigation(arguments, user_id, original_message, metadata)
+
+        prepared_args = await self._prepare_arguments(arguments, metadata, user_id)
+        raw_result = await self._execute(tool_name, prepared_args)
+        return await self._format_result(tool_name, raw_result, metadata, original_message)
+
+    async def _prepare_arguments(
+        self,
+        arguments: Dict[str, Any],
+        metadata: ToolMetadata,
+        user_id: Optional[str],
+    ) -> Dict[str, Any]:
+        merged = dict(metadata.defaults)
+        merged.update(arguments or {})
+
+        if metadata.requires_env and user_id:
+            env_ctx = await self._env_provider(user_id)
+            if env_ctx:
+                for field in metadata.requires_env:
+                    if merged.get(field) is not None:
+                        continue
+                    if field in env_ctx:
+                        merged[field] = env_ctx[field]
+
+        return merged
+
+    async def _execute(self, tool_name: str, arguments: Dict[str, Any]) -> Dict[str, Any]:
+        handler = self._tool_lookup(tool_name)
+        if not handler:
+            raise RuntimeError(f"工具 {tool_name} 無可用 handler")
+
+        retry_delays = [1, 2, 5]
+        last_exc: Optional[BaseException] = None
+        for attempt, delay in enumerate(retry_delays, start=1):
+            try:
+                result = await asyncio.wait_for(handler(arguments), timeout=30.0)
+                if isinstance(result, dict):
+                    return result
+                return {"success": True, "content": str(result)}
+            except Exception as exc:  # noqa: BLE001
+                last_exc = exc
+                logger.warning("工具 %s 執行失敗 (attempt=%s): %s", tool_name, attempt, exc)
+                await asyncio.sleep(delay)
+        handler = self._failure_handlers.get(tool_name)
+        if handler and last_exc:
+            return handler(arguments, last_exc)  # type: ignore[arg-type]
+        raise RuntimeError(f"工具 {tool_name} 執行失敗：{last_exc}")  # type: ignore[arg-type]
+
+    async def _format_result(
+        self,
+        tool_name: str,
+        result: Dict[str, Any],
+        metadata: ToolMetadata,
+        original_message: str,
+    ) -> ToolResult:
+        if isinstance(result, ToolResult):
+            return result
+
+        if result.get("success") and result.get("content"):
+            message = str(result.get("content"))
+        elif result.get("success"):
+            message = "操作完成，但無額外內容。"
+        else:
+            raise RuntimeError(result.get("error") or f"{tool_name} 執行失敗")
+
+        payload = {k: v for k, v in result.items() if k not in {"success", "content", "error"}}
+
+        if metadata.enable_reformat:
+            try:
+                message = await self._formatter(tool_name, message, payload, original_message)
+            except Exception as exc:  # noqa: BLE001
+                logger.warning("AI 格式化失敗，改用原訊息：%s", exc)
+
+        return ToolResult(
+            name=tool_name,
+            message=message,
+            data=payload or None,
+            raw=result,
+        )
+
+    # ------------------------------------------------------------------ #
+    async def _handle_navigation(
+        self,
+        arguments: Dict[str, Any],
+        user_id: Optional[str],
+        original_message: str,
+        metadata: ToolMetadata,
+    ) -> ToolResult:
+        geo_result = await self._execute(metadata.name, arguments or {})
+        if not geo_result.get("success"):
+            raise RuntimeError(geo_result.get("error") or "地點查詢失敗")
+
+        data = geo_result.get("data") or {}
+        best_match = data.get("best_match") or {}
+        dest_lat = best_match.get("lat")
+        dest_lon = best_match.get("lon")
+        if dest_lat is None or dest_lon is None:
+            return ToolResult(
+                name=metadata.name,
+                message=str(geo_result.get("content") or "找不到合適的目的地"),
+                data=data,
+                raw=geo_result,
+            )
+
+        env_ctx = await self._env_provider(user_id) if user_id else {}
+        origin_lat = env_ctx.get("lat")
+        origin_lon = env_ctx.get("lon")
+        if origin_lat is None or origin_lon is None:
+            return ToolResult(
+                name=metadata.name,
+                message=str(geo_result.get("content") or "取得目的地座標成功"),
+                data=data,
+                raw=geo_result,
+                metadata={"note": "缺少目前位置，僅返回地點資訊"},
+            )
+
+        directions_args = {
+            "origin_lat": float(origin_lat),
+            "origin_lon": float(origin_lon),
+            "dest_lat": float(dest_lat),
+            "dest_lon": float(dest_lon),
+            "origin_label": env_ctx.get("label") or env_ctx.get("address_display") or "目前位置",
+            "dest_label": best_match.get("label") or arguments.get("query"),
+            "mode": "foot-walking",
+        }
+
+        directions_meta = self._metadata.get("directions", ToolMetadata(name="directions"))
+        prepared = await self._prepare_arguments(directions_args, directions_meta, user_id)
+        directions_result = await self._execute("directions", prepared)
+        return await self._format_result("directions", directions_result, directions_meta, original_message)

features/mcp/tool_models.py

@@ -0,0 +1,31 @@
+from dataclasses import dataclass, field
+from typing import Any, Dict, Optional, Set
+
+
+@dataclass
+class ToolMetadata:
+    name: str
+    requires_env: Set[str] = field(default_factory=set)
+    defaults: Dict[str, Any] = field(default_factory=dict)
+    enable_reformat: bool = False
+    flow: Optional[str] = None  # 例如 "navigation"
+
+
+@dataclass
+class ToolResult:
+    name: str
+    message: str
+    data: Optional[Any] = None
+    raw: Optional[Dict[str, Any]] = None
+    metadata: Optional[Dict[str, Any]] = None
+
+    def to_dict(self) -> Dict[str, Any]:
+        payload: Dict[str, Any] = {
+            "message": self.message,
+            "tool_name": self.name,
+        }
+        if self.data is not None:
+            payload["tool_data"] = self.data
+        if self.metadata:
+            payload["metadata"] = self.metadata
+        return payload

services/ai_service.py

@@ -660,6 +660,7 @@ async def generate_response_for_user(
     reasoning_effort: Optional[str] = None,
     user_name: Optional[str] = None,
     emotion_label: Optional[str] = None,
+    env_context: Optional[Dict[str, Any]] = None,
 ) -> str:
     """
     為用戶生成AI回應
@@ -690,6 +691,7 @@ async def generate_response_for_user(
                 reasoning_effort=reasoning_effort,
                 user_name=user_name,
                 emotion_label=emotion_label,
+                env_context=env_context,
             )
         else:
             # 回退到原有的全局歷史管理（用於向後兼容）
@@ -707,6 +709,7 @@ async def generate_response_for_user(
                 reasoning_effort=reasoning_effort,
                 user_name=user_name,
                 emotion_label=emotion_label,
+                env_context=env_context,
             )
 
         logger.error("未提供消息列表或用戶消息")
@@ -734,6 +737,7 @@ async def _generate_response_with_chat_db(
     reasoning_effort: Optional[str] = None,
     user_name: Optional[str] = None,
     emotion_label: Optional[str] = None,
+    env_context: Optional[Dict[str, Any]] = None,
 ):
     """使用DB管理對話歷史的實現"""
     try:
@@ -847,8 +851,8 @@ async def _generate_response_with_chat_db(
                     logger.warning(f"載入記憶失敗: {e}")
 
             # 讀取環境現況（僅組裝，不外呼）
-            ctx: Dict[str, Any] = {}
-            if db_available and user_id:
+            ctx: Dict[str, Any] = dict(env_context or {})
+            if not ctx and db_available and user_id:
                 try:
                     env_res = await get_user_env_current(user_id)
                     if env_res.get("success"):
@@ -917,6 +921,7 @@ async def _generate_response_with_chat_db(
             reasoning_effort=reasoning_effort,
             user_name=user_name,
             emotion_label=emotion_label,
+            env_context=env_context,
         )
 
 
@@ -935,6 +940,7 @@ async def _generate_response_with_global_history(
     reasoning_effort: Optional[str] = None,
     user_name: Optional[str] = None,
     emotion_label: Optional[str] = None,
+    env_context: Optional[Dict[str, Any]] = None,
 ):
     """使用全局歷史的回退實現（向後兼容）"""
     try:
@@ -988,8 +994,8 @@ async def _generate_response_with_global_history(
                 prior_history = prior_history[-history_limit:]
 
             # 讀取環境現況
-            ctx: Dict[str, Any] = {}
-            if db_available and user_id:
+            ctx: Dict[str, Any] = dict(env_context or {})
+            if not ctx and db_available and user_id:
                 try:
                     env_res = await get_user_env_current(user_id)
                     if env_res.get("success"):

tests/environment/test_context_service.py

@@ -0,0 +1,96 @@
+import asyncio
+import sys
+from pathlib import Path
+from typing import Any, Dict
+
+ROOT_DIR = Path(__file__).resolve().parents[2]
+if str(ROOT_DIR) not in sys.path:
+    sys.path.insert(0, str(ROOT_DIR))
+
+from core.environment import EnvironmentContextService
+
+
+def test_ingest_snapshot_writes_current():
+    async def _run():
+        writes: list = []
+        snapshots: list = []
+
+        async def fake_fetcher(user_id: str) -> Dict[str, Any]:
+            return {"success": True, "context": {}}
+
+        async def fake_writer(user_id: str, payload: Dict[str, Any]) -> Dict[str, Any]:
+            writes.append((user_id, payload))
+            return {"success": True}
+
+        async def fake_snapshot_writer(user_id: str, payload: Dict[str, Any]) -> Dict[str, Any]:
+            snapshots.append((user_id, payload))
+            return {"success": True}
+
+        service = EnvironmentContextService(
+            min_distance_m=10.0,
+            min_heading_deg=10.0,
+            ttl_seconds=60.0,
+            env_fetcher=fake_fetcher,
+            env_writer=fake_writer,
+            snapshot_writer=fake_snapshot_writer,
+        )
+
+        await service.start()
+        try:
+            ack = await service.ingest_snapshot(
+                "user-1",
+                {"lat": 25.0, "lon": 121.5, "heading_deg": 90, "tz": "Asia/Taipei"},
+            )
+            assert ack["success"] is True
+
+            await asyncio.sleep(0.05)
+            assert writes, "should enqueue current write"
+            assert snapshots, "should enqueue snapshot write"
+
+            ctx = await service.get_context("user-1", allow_stale=True)
+            assert ctx["lat"] == 25.0
+            assert ctx["heading_cardinal"] == "E"
+        finally:
+            await service.shutdown()
+
+    asyncio.run(_run())
+
+
+def test_ingest_snapshot_with_geocode():
+    async def _run():
+        writes: list = []
+
+        async def fake_fetcher(user_id: str) -> Dict[str, Any]:
+            return {"success": False}
+
+        async def fake_writer(user_id: str, payload: Dict[str, Any]) -> Dict[str, Any]:
+            writes.append(payload)
+            return {"success": True}
+
+        service = EnvironmentContextService(
+            min_distance_m=0.0,
+            min_heading_deg=0.0,
+            ttl_seconds=60.0,
+            env_fetcher=fake_fetcher,
+            env_writer=fake_writer,
+            snapshot_writer=None,
+        )
+
+        async def geocode(lat: float, lon: float) -> Dict[str, Any]:
+            return {"city": "Taipei", "address_display": "Taipei City"}
+
+        await service.start()
+        try:
+            await service.ingest_snapshot(
+                "geo-user",
+                {"lat": 25.0, "lon": 121.5},
+                geocode_provider=geocode,
+            )
+            await asyncio.sleep(0.05)
+            ctx = await service.get_context("geo-user", allow_stale=True)
+            assert ctx.get("city") == "Taipei"
+            assert any(entry.get("city") == "Taipei" for entry in writes)
+        finally:
+            await service.shutdown()
+
+    asyncio.run(_run())

tests/features/mcp/test_agent_bridge_route_labels.py

@@ -3,46 +3,19 @@ from pathlib import Path
 import sys
 import types
 
-ROOT_DIR = Path(__file__).resolve().parents[4]
+ROOT_DIR = Path(__file__).resolve().parents[3]
 if str(ROOT_DIR) not in sys.path:
     sys.path.append(str(ROOT_DIR))
 
 from features.mcp.agent_bridge import MCPAgentBridge  # noqa: E402
 from features.mcp.tools.base_tool import ExecutionError  # noqa: E402
-
-
-def test_prepare_route_arguments_injects_labels():
-    bridge = MCPAgentBridge.__new__(MCPAgentBridge)
-
-    async def fake_resolve(_self, _lat, _lon):
-        return "測試地點"
-
-    bridge._resolve_coordinate_label = fake_resolve.__get__(bridge, MCPAgentBridge)  # type: ignore[attr-defined]
-
-    prepared, labels = asyncio.run(
-        bridge._prepare_route_arguments(
-            {
-                "origin_lat": "24.9915",
-                "origin_lon": "121.3423",
-                "dest_lat": "24.9891",
-                "dest_lon": "121.3134",
-                # 未提供 label，應自動補上
-            }
-        )
-    )
-
-    assert prepared["origin_label"] == "測試地點"
-    assert prepared["dest_label"] == "測試地點"
-    assert isinstance(prepared["origin_lat"], float)
-    assert isinstance(prepared["dest_lon"], float)
-    assert labels["origin_label"] == "測試地點"
-    assert labels["dest_label"] == "測試地點"
+from features.mcp.tool_models import ToolMetadata  # noqa: E402
 
 
 def test_build_directions_message_returns_human_friendly_text():
     bridge = MCPAgentBridge.__new__(MCPAgentBridge)
 
-    message, tool_data = bridge._build_directions_message(
+    message, tool_data = bridge._build_directions_message(  # type: ignore[attr-defined]
         {"distance_m": 1450.0, "duration_s": 840.0, "polyline": "[]"},
         {"origin_label": "測試起點 A", "dest_label": "測試目的地 B"},
     )
@@ -58,7 +31,7 @@ def test_build_directions_message_returns_human_friendly_text():
 def test_build_directions_failure_response_generates_fallback_message():
     bridge = MCPAgentBridge.__new__(MCPAgentBridge)
 
-    result = bridge._build_directions_failure_response(
+    result = bridge._build_directions_failure_response(  # type: ignore[attr-defined]
         {
             "origin_lat": 25.045,
             "origin_lon": 121.516,
@@ -79,42 +52,76 @@ def test_build_directions_failure_response_generates_fallback_message():
     assert "地圖" in message
 
 
-def test_call_mcp_tool_returns_fallback_when_directions_fails():
-    bridge = MCPAgentBridge.__new__(MCPAgentBridge)
+def test_tool_coordinator_navigation_flow_uses_env_context():
+    async def _run():
+        async def env_provider(user_id):
+            return {"lat": 25.0, "lon": 121.5, "label": "目前位置"}
 
-    async def fake_enrich(tool_name, arguments, user_id):
-        return arguments
+        bridge = MCPAgentBridge(env_provider=env_provider)
+        bridge._tool_coordinator.register(ToolMetadata(name='directions', enable_reformat=False))  # type: ignore[attr-defined]
 
-    async def fake_handler(_arguments):
-        raise ExecutionError("OpenRouteService 無法提供路線")
+        async def forward_handler(arguments):
+            return {
+                "success": True,
+                "content": "定位完成",
+                "data": {"best_match": {"lat": 24.9, "lon": 121.3, "label": arguments.get("query")}},
+            }
 
-    async def fake_resolve(_self, _lat, _lon):
-        return "測試地點"
+        async def directions_handler(arguments):
+            assert arguments["origin_label"] == "目前位置"
+            assert arguments["dest_label"] == "桃園火車站"
+            return {"success": True, "content": "沿著高速公路前進", "distance_m": 1000.0}
 
-    bridge._enrich_arguments_with_env = fake_enrich  # type: ignore[attr-defined]
-    bridge._resolve_coordinate_label = fake_resolve.__get__(bridge, MCPAgentBridge)  # type: ignore[attr-defined]
+        bridge.mcp_server.tools = {
+            "forward_geocode": types.SimpleNamespace(handler=forward_handler),
+            "directions": types.SimpleNamespace(handler=directions_handler),
+        }
 
-    bridge.mcp_server = types.SimpleNamespace(
-        tools={
-            "directions": types.SimpleNamespace(
-                handler=fake_handler,
-                description="Route",
-                metadata={"category": "地理"},
-                inputSchema={"properties": {}, "required": []},
-            )
+        intent = {
+            "type": "mcp_tool",
+            "tool_name": "forward_geocode",
+            "arguments": {"query": "桃園火車站"},
         }
-    )
 
-    result = asyncio.run(
-        bridge._call_mcp_tool(
-            "directions",
-            {"origin_lat": 25.045, "origin_lon": 121.516, "dest_lat": 24.993, "dest_lon": 121.324},
-            user_id="u123",
-            original_message="測試路線",
-        )
-    )
+        result = await bridge.process_intent(intent, user_id="u1", original_message="怎麼去桃園火車站")
+        assert result["tool_name"] == "directions"
+        assert '��離約' in result['message']
+
+    asyncio.run(_run())
+
+
+def test_directions_failure_produces_fallback_tool_result():
+    async def _run():
+        async def env_provider(user_id):
+            return {"lat": 25.0, "lon": 121.5, "label": "目前位置"}
+
+        bridge = MCPAgentBridge(env_provider=env_provider)
+        bridge._tool_coordinator.register(ToolMetadata(name='directions', enable_reformat=False))  # type: ignore[attr-defined]
+
+        async def forward_handler(arguments):
+            return {
+                "success": True,
+                "content": "定位完成",
+                "data": {"best_match": {"lat": 24.9, "lon": 121.3, "label": arguments.get("query")}},
+            }
+
+        async def directions_handler(arguments):
+            raise ExecutionError("OpenRouteService 無法提供路線")
+
+        bridge.mcp_server.tools = {
+            "forward_geocode": types.SimpleNamespace(handler=forward_handler),
+            "directions": types.SimpleNamespace(handler=directions_handler),
+        }
+
+        intent = {
+            "type": "mcp_tool",
+            "tool_name": "forward_geocode",
+            "arguments": {"query": "桃園火車站"},
+        }
+
+        result = await bridge.process_intent(intent, user_id="u1", original_message="怎麼去桃園火車站")
+        assert isinstance(result, dict)
+        assert result["tool_name"] == "directions"
+        assert result['tool_data']['fallback'] is True
 
-    assert isinstance(result, dict)
-    assert result["tool_name"] == "directions"
-    assert result["tool_data"]["fallback"] is True
-    assert "目前無法向路線服務取得詳細路線" in result["message"]
+    asyncio.run(_run())

tests/features/mcp/test_navigation_fix.py

@@ -4,6 +4,10 @@
 驗證地點查詢與導航是否正常工作
 """
 
+import pytest
+
+pytestmark = pytest.mark.skip(reason='Integration script - skipped in automated suite')
+
 import asyncio
 import logging
 from features.mcp.agent_bridge import MCPAgentBridge

tests/features/mcp/test_precise_location.py

@@ -4,6 +4,10 @@
 驗證 reverse_geocode 與 forward_geocode 是否能正確提取門牌、路口資訊
 """
 
+import pytest
+
+pytestmark = pytest.mark.skip(reason='Integration script - skipped in automated suite')
+
 import asyncio
 import logging
 from features.mcp.tools.geocode_tool import ReverseGeocodeTool

tests/features/mcp/test_tool_coordinator.py

@@ -0,0 +1,107 @@
+import asyncio
+import sys
+from pathlib import Path
+from typing import Any, Dict, Optional
+
+ROOT_DIR = Path(__file__).resolve().parents[3]
+if str(ROOT_DIR) not in sys.path:
+    sys.path.insert(0, str(ROOT_DIR))
+
+from features.mcp.coordinator import ToolCoordinator
+from features.mcp.tool_models import ToolMetadata
+
+
+def test_tool_coordinator_env_injection():
+    async def _run():
+        captured: Dict[str, Any] = {}
+
+        async def env_provider(user_id: Optional[str]) -> Dict[str, Any]:
+            return {"lat": 25.0, "lon": 121.5, "city": "Taipei"}
+
+        async def weather_handler(arguments: Dict[str, Any]) -> Dict[str, Any]:
+            nonlocal captured
+            captured = dict(arguments)
+            return {"success": True, "content": "晴時多雲", "temperature": 25}
+
+        async def formatter(name: str, message: str, payload: Dict[str, Any], original: str) -> str:
+            return message
+
+        coordinator = ToolCoordinator(
+            env_provider=env_provider,
+            tool_lookup=lambda name: weather_handler if name == "weather_query" else None,
+            formatter=formatter,
+        )
+        coordinator.register(
+            ToolMetadata(
+                name="weather_query",
+                requires_env={"lat", "lon", "city"},
+                enable_reformat=False,
+            )
+        )
+
+        result = await coordinator.invoke(
+            "weather_query",
+            {},
+            user_id="user-1",
+            original_message="台北天氣",
+        )
+
+        assert captured["lat"] == 25.0
+        assert captured["lon"] == 121.5
+        assert captured["city"] == "Taipei"
+        assert result.message == "晴時多雲"
+
+    asyncio.run(_run())
+
+
+def test_navigation_flow_auto_routes():
+    async def _run():
+        directions_calls = []
+
+        async def env_provider(user_id: Optional[str]) -> Dict[str, Any]:
+            return {"lat": 25.0, "lon": 121.5, "label": "現在位置"}
+
+        async def geocode_handler(arguments: Dict[str, Any]) -> Dict[str, Any]:
+            return {
+                "success": True,
+                "content": "找到目的地",
+                "data": {"best_match": {"lat": 24.1, "lon": 120.9, "label": arguments.get("query")}},
+            }
+
+        async def directions_handler(arguments: Dict[str, Any]) -> Dict[str, Any]:
+            directions_calls.append(arguments)
+            return {"success": True, "content": "沿著高速公路前進"}
+
+        async def formatter(name: str, message: str, payload: Dict[str, Any], original: str) -> str:
+            return message
+
+        def tool_lookup(name: str):
+            if name == "forward_geocode":
+                return geocode_handler
+            if name == "directions":
+                return directions_handler
+            return None
+
+        coordinator = ToolCoordinator(
+            env_provider=env_provider,
+            tool_lookup=tool_lookup,
+            formatter=formatter,
+        )
+        coordinator.register(ToolMetadata(name="forward_geocode", flow="navigation"))
+        coordinator.register(ToolMetadata(name="directions", enable_reformat=False))
+
+        result = await coordinator.invoke(
+            "forward_geocode",
+            {"query": "桃園火車站"},
+            user_id="tester",
+            original_message="怎麼去桃園火車站",
+        )
+
+        assert result.name == "directions"
+        assert directions_calls, "directions tool should be invoked"
+        call_args = directions_calls[0]
+        assert call_args["origin_label"] == "現在位置"
+        assert call_args["dest_label"] == "桃園火車站"
+        assert "沿著高速公路前進" in result.message
+
+    asyncio.run(_run())