"""階段 10-B:Cooper 全域 ClaudeSDKClient 管理器(暖機 + 常駐 + 序列化)。 設計重點: - uvicorn 啟動時 connect 一次(10s),user 看不到 - 每個 LINE 訊息用 session_id 區分群組(line:) - asyncio.Lock 序列化 query — 同一個 client 一次只跑一個,避免狀態錯亂 - 工具 wrapper 用 ContextVar 取得 requester_id(取代 closure) - 9 個 Cooper MCP 工具在 client connect 時就註冊好(不再每次 LINE 訊息重建) 使用方式: # uvicorn lifespan startup await cooper_client_manager.startup() # LINE 訊息進來 response_text, rate_info, tools_called = await cooper_client_manager.respond( prompt_text="...", session_id="line:", requester_id=user_id, ) # uvicorn lifespan shutdown await cooper_client_manager.shutdown() """ from __future__ import annotations import asyncio import contextvars import json import logging import os from typing import Any from uuid import UUID from app.config import settings logger = logging.getLogger("coba.cooper_client") # ============================================================ # 工具呼叫的執行 context — 全域變數(被 asyncio.Lock 保護) # # 為什麼不用 ContextVar:ClaudeSDKClient 的 MCP server tool handler 跑在 # client 內部的 anyio task group(lifespan 啟動的),不是 LINE handler 那個 # task group。Python ContextVar 不會跨 task group 傳遞,handler 內讀不到 set。 # # 改用全域 dict + module-level reference + asyncio.Lock 序列化(同時間 # 只有一個 query 在跑,不會混)。 # ============================================================ _current_requester_id: UUID | None = None _current_tool_calls: list[dict] | None = None def _set_request_context(requester_id: UUID, tool_calls_bucket: list[dict]): global _current_requester_id, _current_tool_calls _current_requester_id = requester_id _current_tool_calls = tool_calls_bucket def _clear_request_context(): global _current_requester_id, _current_tool_calls _current_requester_id = None _current_tool_calls = None def _get_requester_id() -> UUID: if _current_requester_id is None: raise RuntimeError("Cooper tool 在沒有 requester context 時被呼叫") return _current_requester_id def _record_tool_call(name: str, args: dict, result: dict) -> None: if _current_tool_calls is not None: _current_tool_calls.append({ "name": name, "input": args, "result_keys": list(result.keys()), }) # ============================================================ # 全域 client manager # ============================================================ class CooperClientManager: """單例 — 包住 ClaudeSDKClient + lock + 暖機/關閉邏輯。""" def __init__(self): self._client: Any = None # ClaudeSDKClient,延遲 import 避免啟動時牽連 self._lock = asyncio.Lock() self._started = False self._mcp_server: Any = None # 暖機 event:startup 完成才 set。respond 進來會 await(避免 race) self._ready_event = asyncio.Event() # ---------------------------------------------------------- # MCP 工具定義(在 client connect 時固定註冊一次) # ---------------------------------------------------------- def _build_tools(self): from claude_agent_sdk import tool from app.services.coba_tools import ( tool_create_task, tool_list_tasks, tool_update_task_status, tool_update_task, tool_reassign_task, tool_delete_task, tool_share_dashboard_link, tool_get_team_members, tool_list_assignment_status, tool_get_team_completion_rate, tool_save_reflection, tool_list_recent_reflections, tool_delete_reflection, tool_update_reflection, tool_convert_reflection_to_task, tool_add_team_member, tool_record_meeting, tool_list_meetings, tool_get_meeting_detail, tool_delete_meeting, ) # 工具名稱 → 影響的 entity(完成後 broadcast,讓前端自動 refresh,不用 F5) # 列表 / 查詢類不用 broadcast(讀不會變);只有 mutation 工具才推 TOOL_BROADCAST_ENTITY = { "create_task": "tasks", "update_task": "tasks", "update_task_status": "tasks", "reassign_task": "tasks", "delete_task": "tasks", "save_reflection": "reflections", "delete_reflection": "reflections", "update_reflection": "reflections", "convert_reflection_to_task": "reflections", # 同時改 task,broadcast 兩個 "record_meeting": "meetings", "delete_meeting": "meetings", "add_team_member": "users", } def _wrap(tool_fn, name: str, description: str, schema: dict): @tool(name, description, schema) async def _wrapped(args: dict) -> dict: requester_id = _get_requester_id() try: result = await tool_fn(args or {}, requester_id) except Exception as e: logger.exception("tool %s failed", name) result = {"error": f"{type(e).__name__}: {e}"} _record_tool_call(name, args, result) # 成功改動 → broadcast WS 事件給前端,自動 refresh,不用 F5 if "error" not in result: entity = TOOL_BROADCAST_ENTITY.get(name) if entity: from app.services.coba_tools import _broadcast_data_changed try: await _broadcast_data_changed(entity, tool=name) # 特殊:reflection 升級成 task 一次改兩種 if name == "convert_reflection_to_task": await _broadcast_data_changed("tasks", tool=name) except Exception: logger.exception("broadcast after tool=%s failed", name) return { "content": [{"type": "text", "text": json.dumps(result, ensure_ascii=False)}], "is_error": "error" in result, } return _wrapped return [ _wrap(tool_create_task, "create_task", "建立一個新任務在團隊的行程規畫系統。當使用者明確命令時用。", { "type": "object", "properties": { "title": {"type": "string", "description": "任務標題"}, "assignee_name": {"type": "string", "description": "負責人 display_name"}, "due_date": {"type": "string", "description": "YYYY-MM-DD"}, "priority": {"type": "string", "enum": ["low", "medium", "high", "urgent"]}, "description": {"type": "string"}, }, "required": ["title"], }, ), _wrap(tool_list_tasks, "list_tasks", "查任務(可帶 status / assignee_name / keyword 條件)。", { "type": "object", "properties": { "status": {"type": "string", "enum": ["pending", "in_progress", "completed", "cancelled", "all"]}, "assignee_name": {"type": "string"}, "keyword": {"type": "string"}, "limit": {"type": "integer"}, }, }, ), _wrap(tool_update_task_status, "update_task_status", "改任務狀態。用 task_id_prefix(前 8 碼)或 title_keyword 找任務。", { "type": "object", "properties": { "task_id_prefix": {"type": "string", "description": "task UUID 前 8 字元"}, "title_keyword": {"type": "string"}, "new_status": {"type": "string", "enum": ["todo", "in_progress", "review", "done", "cancelled"]}, }, "required": ["new_status"], }, ), _wrap(tool_update_task, "update_task", "全功能改任務:可改 title / description / due_date / priority / assignee。" "不改 status(用 update_task_status)。" "用 task_id_prefix(前 8 碼)或 title_keyword 鎖定任務。" "把 new_due_date 設成 'null' 可清除截止日;new_assignee_name 設成 'null' 取消指派。", { "type": "object", "properties": { "task_id_prefix": {"type": "string"}, "title_keyword": {"type": "string"}, "new_title": {"type": "string"}, "new_description": {"type": "string"}, "new_due_date": {"type": "string", "description": "YYYY-MM-DD,或 'null' 清除"}, "new_priority": {"type": "string", "enum": ["low", "medium", "high", "urgent"]}, "new_assignee_name": {"type": "string", "description": "新負責人 display_name,或 'null' 取消指派"}, }, }, ), _wrap(tool_reassign_task, "reassign_task", "把任務轉指派給另一個成員。是 update_task 的 sugar 版,只動 assignee。", { "type": "object", "properties": { "task_id_prefix": {"type": "string"}, "title_keyword": {"type": "string"}, "new_assignee_name": {"type": "string"}, }, "required": ["new_assignee_name"], }, ), _wrap(tool_delete_task, "delete_task", "軟刪任務(set deleted_at)。", { "type": "object", "properties": { "task_id_prefix": {"type": "string"}, "title_keyword": {"type": "string"}, }, }, ), _wrap(tool_share_dashboard_link, "share_dashboard_link", "給對方一鍵登入 dashboard 的 magic link(只在 1 對 1 用)。", {"type": "object", "properties": {"target_user_name": {"type": "string"}}}, ), _wrap(tool_get_team_members, "get_team_members", "取得團隊成員清單(display_name + role)。", {"type": "object", "properties": {}}, ), _wrap(tool_list_assignment_status, "list_assignment_status", "列最近自動建任務的認領狀態(誰接 / 誰沒接)。", {"type": "object", "properties": {"days": {"type": "integer"}}}, ), _wrap(tool_get_team_completion_rate, "get_team_completion_rate", "列每人本週完成率 KPI。", {"type": "object", "properties": {}}, ), _wrap(tool_save_reflection, "save_reflection", "把使用者的覆盤 / 體悟 / 反省內容存進體悟紀錄(不建任務)。", { "type": "object", "properties": { "content": {"type": "string", "description": "體悟內容原文"}, "tag": {"type": "string", "description": "可選:類別標籤"}, }, "required": ["content"], }, ), _wrap(tool_list_recent_reflections, "list_recent_reflections", "列最近群組分享的覆盤 / 體悟記錄。回傳含 id_short 讓你後續能 update / delete / 轉任務。", {"type": "object", "properties": {"hours": {"type": "integer", "description": "預設 168(一週)"}}}, ), _wrap(tool_delete_reflection, "delete_reflection", "軟刪一筆覆盤。" "用 reflection_id_prefix(前 8 碼,從 list_recent_reflections 拿)或 content_keyword 鎖定。", { "type": "object", "properties": { "reflection_id_prefix": {"type": "string"}, "content_keyword": {"type": "string", "description": "覆盤內容關鍵字"}, }, }, ), _wrap(tool_update_reflection, "update_reflection", "改一筆覆盤的內容 / 摘要 / 標籤。至少要給 new_content / new_summary / new_tag 其中一個。", { "type": "object", "properties": { "reflection_id_prefix": {"type": "string"}, "content_keyword": {"type": "string"}, "new_content": {"type": "string", "description": "新覆盤完整內容"}, "new_summary": {"type": "string"}, "new_tag": {"type": "string"}, }, }, ), _wrap(tool_convert_reflection_to_task, "convert_reflection_to_task", "誤分類修正:某筆覆盤其實是任務 → 升級成任務,自動軟刪原覆盤。" "必填 title。可選 assignee_name / due_date / priority / description / mark_done(已完成的事直接 done)。", { "type": "object", "properties": { "reflection_id_prefix": {"type": "string"}, "content_keyword": {"type": "string"}, "title": {"type": "string", "description": "新任務標題,動詞開頭"}, "assignee_name": {"type": "string"}, "due_date": {"type": "string", "description": "YYYY-MM-DD"}, "priority": {"type": "string", "enum": ["low", "medium", "high", "urgent"]}, "description": {"type": "string"}, "mark_done": {"type": "boolean", "description": "true=立刻標 done(用於『我已經完成 X』類報告)"}, }, "required": ["title"], }, ), _wrap(tool_record_meeting, "record_meeting", "從文字直接記錄一場會議。" "使用者說「庫柏 記下這次會議:標題 / 結論 / action items」就用這個。" "必填 title。可選:executive_summary、key_topics、decisions(list[str|dict])、" "action_items(list[str|dict{description,assignee_name,due_date}])。" "建好後 status='completed',不走錄音流程。", { "type": "object", "properties": { "title": {"type": "string"}, "executive_summary": {"type": "string", "description": "2-4 句濃縮摘要"}, "key_topics": {"type": "array", "items": {"type": "string"}}, "decisions": { "type": "array", "items": {"type": "string"}, "description": "重要決策清單", }, "action_items": { "type": "array", "items": {"type": "string"}, "description": "待辦事項清單(可帶 - 阿綠週五前 X 之類自然語言)", }, }, "required": ["title"], }, ), _wrap(tool_list_meetings, "list_meetings", "列最近會議。回傳含 id_short 給後續查詳細用。", { "type": "object", "properties": {"limit": {"type": "integer", "description": "預設 10"}}, }, ), _wrap(tool_get_meeting_detail, "get_meeting_detail", "看一場會議完整內容(摘要 + 決策 + action items)。" "用 meeting_id_prefix(8 碼)或 title_keyword 找。", { "type": "object", "properties": { "meeting_id_prefix": {"type": "string"}, "title_keyword": {"type": "string"}, }, }, ), _wrap(tool_delete_meeting, "delete_meeting", "軟刪會議。", { "type": "object", "properties": { "meeting_id_prefix": {"type": "string"}, "title_keyword": {"type": "string"}, }, }, ), _wrap(tool_add_team_member, "add_team_member", "新增團隊成員到行程規畫系統。" "必填 display_name;可選 role(owner/manager/member,預設 member)、email、line_user_id、job_title。" "成功會回 user_id_short + 預設密碼提示給對方第一次登入用。", { "type": "object", "properties": { "display_name": {"type": "string", "description": "對方在系統內的顯示名稱"}, "role": {"type": "string", "enum": ["owner", "manager", "member"]}, "email": {"type": "string"}, "line_user_id": {"type": "string", "description": "LINE 個人 ID,Uxxxxx 格式"}, "job_title": {"type": "string"}, }, "required": ["display_name"], }, ), ] # ---------------------------------------------------------- # lifespan startup — 暖機 # ---------------------------------------------------------- async def startup(self): if self._started: return if not settings.CLAUDE_CODE_OAUTH_TOKEN: logger.warning("CLAUDE_CODE_OAUTH_TOKEN 沒設,Cooper SDK client 不啟動") return from claude_agent_sdk import ( ClaudeSDKClient, ClaudeAgentOptions, create_sdk_mcp_server, ) # MCP server 用我們塞的工具 self._mcp_server = create_sdk_mcp_server( name="cooper", version="1.0.0", tools=self._build_tools(), ) allowed = [ "mcp__cooper__create_task", "mcp__cooper__list_tasks", "mcp__cooper__update_task_status", "mcp__cooper__update_task", "mcp__cooper__reassign_task", "mcp__cooper__delete_task", "mcp__cooper__share_dashboard_link", "mcp__cooper__get_team_members", "mcp__cooper__list_assignment_status", "mcp__cooper__get_team_completion_rate", "mcp__cooper__save_reflection", "mcp__cooper__list_recent_reflections", "mcp__cooper__delete_reflection", "mcp__cooper__update_reflection", "mcp__cooper__convert_reflection_to_task", "mcp__cooper__add_team_member", "mcp__cooper__record_meeting", "mcp__cooper__list_meetings", "mcp__cooper__get_meeting_detail", "mcp__cooper__delete_meeting", ] # System prompt 用「靜態角色定義」— 動態 context(日期 / 成員清單 / 近期訊息) # 由 respond() 每次塞進 user prompt 開頭。這樣 prompt cache 才能命中。 from app.services.coba_chat import COBA_BASE_PROMPT options = ClaudeAgentOptions( model="claude-sonnet-4-5", system_prompt=COBA_BASE_PROMPT, tools=[], allowed_tools=allowed, mcp_servers={"cooper": self._mcp_server}, strict_mcp_config=True, max_turns=6, permission_mode="bypassPermissions", setting_sources=[], cwd="/tmp", env={ "ANTHROPIC_API_KEY": "", "CLAUDE_CODE_OAUTH_TOKEN": settings.CLAUDE_CODE_OAUTH_TOKEN, }, ) # SDK 第一次 spawn claude CLI 在 uvicorn process 內常 30s initialize timeout # (NAS file cache 沒暖,binary 慢)。retry 一次,第二次通常秒回。 last_err: Exception | None = None for attempt in range(1, 4): print(f"[COOPER_CLIENT_STARTUP] connecting attempt={attempt}...", flush=True) self._client = ClaudeSDKClient(options=options) try: await self._client.connect() self._started = True self._ready_event.set() print(f"[COOPER_CLIENT_STARTUP] ✓ connected (attempt={attempt})", flush=True) return except Exception as e: last_err = e err_msg = str(e)[:200] print(f"[COOPER_CLIENT_STARTUP_RETRY] attempt={attempt} fail: {type(e).__name__}: {err_msg}", flush=True) # 保險清掉破掉的 client try: if self._client: await self._client.disconnect() except Exception: pass self._client = None if attempt < 3: await asyncio.sleep(5) # 讓 NAS file cache 暖 # 三次都 fail if last_err: raise last_err # ---------------------------------------------------------- # lifespan shutdown # ---------------------------------------------------------- async def shutdown(self): if self._client and self._started: try: print("[COOPER_CLIENT_SHUTDOWN] disconnecting...", flush=True) await self._client.disconnect() print("[COOPER_CLIENT_SHUTDOWN] ✓ disconnected", flush=True) except Exception: logger.exception("Cooper client disconnect 失敗") self._started = False self._client = None # ---------------------------------------------------------- # 主 API — LINE 訊息進來呼叫這個 # ---------------------------------------------------------- async def respond( self, prompt_text: str, session_id: str, requester_id: UUID, image_blocks: list[dict] | None = None, ) -> dict: """跑一次 Cooper 對話。回 dict {response_text, rate_limit_type, total_cost_usd, tool_calls}。 並發行為:同時間多個 LINE 訊息會被 Lock 序列化(一個 client 一次只跑一個)。 image_blocks(階段 10-H):若提供,把 text + 圖片合成 multimodal content blocks 送給 Claude 看圖。每個 block 結構: {"type": "image", "source": {"type": "base64", "media_type": "image/jpeg", "data": ""}} """ # 暖機可能還在跑(uvicorn 剛啟動時),等 ready event(最多 60s) if not self._ready_event.is_set(): print("[COOPER_QUERY_WAIT_READY] 等暖機完成...", flush=True) try: await asyncio.wait_for(self._ready_event.wait(), timeout=60) except asyncio.TimeoutError: raise RuntimeError("Cooper client 暖機 60s 超時") from None if not self._started or not self._client: raise RuntimeError("Cooper client 未啟動") from claude_agent_sdk import ( AssistantMessage, TextBlock, ResultMessage, RateLimitEvent, ) tool_calls_bucket: list[dict] = [] try: async with self._lock: # 設 module-level context(在 lock 內,序列化保護) _set_request_context(requester_id, tool_calls_bucket) # 階段 10-H:若有圖片 blocks,改走 streaming mode(AsyncIterable[dict])送 multimodal # 沒圖時繼續走 string mode(較簡單,跟之前一致) if image_blocks: print( f"[COOPER_QUERY] session={session_id} prompt_len={len(prompt_text)} " f"images={len(image_blocks)} (multimodal)", flush=True, ) content_blocks = [{"type": "text", "text": prompt_text}] + image_blocks async def _msg_gen(): yield { "type": "user", "message": {"role": "user", "content": content_blocks}, "parent_tool_use_id": None, "session_id": session_id, } await self._client.query(_msg_gen(), session_id=session_id) else: print( f"[COOPER_QUERY] session={session_id} prompt_len={len(prompt_text)}", flush=True, ) await self._client.query(prompt_text, session_id=session_id) text_pieces: list[str] = [] observed_rate_type: str | None = None observed_rate_status: str | None = None total_cost_usd = 0.0 # 收訊息直到 ResultMessage(該輪結尾) async for msg in self._client.receive_messages(): if isinstance(msg, AssistantMessage): for block in msg.content: if isinstance(block, TextBlock): text_pieces.append(block.text) elif isinstance(msg, ResultMessage): total_cost_usd = msg.total_cost_usd or 0.0 print( f"[COOPER_RESULT] duration_ms={msg.duration_ms} " f"cost={total_cost_usd:.5f} stop={msg.stop_reason} " f"tools={len(tool_calls_bucket)}", flush=True, ) break elif isinstance(msg, RateLimitEvent): info = msg.rate_limit_info observed_rate_type = info.rate_limit_type observed_rate_status = info.status print( f"[COOPER_RATE_LIMIT] type={info.rate_limit_type} " f"status={info.status} overage={info.overage_status}", flush=True, ) response_text = "\n".join(p.strip() for p in text_pieces if p.strip()).strip() \ or "(庫柏沒說話)" return { "response_text": response_text, "rate_limit_type": observed_rate_type, "rate_limit_status": observed_rate_status, "total_cost_usd": total_cost_usd, "tool_calls": tool_calls_bucket, } finally: _clear_request_context() # 全域單例 cooper_client_manager = CooperClientManager()