""" 甄养堂 · 企微客服助手 MCP Server ================================ 把企微 RPA 的配置、会话档案、AI 草稿能力暴露给 Cursor / Claude 等 MCP 客户端。 运行（stdio，供 Cursor 拉起）： python mcp_server.py 本地调试： mcp dev mcp_server.py """ from __future__ import annotations import json import os import sys import time from typing import Any, Optional # 保证可导入同目录模块 _ROOT = os.path.dirname(os.path.abspath(__file__)) if _ROOT not in sys.path: sys.path.insert(0, _ROOT) from mcp.server.fastmcp import FastMCP import ai_config from conversation_store import ConversationStore from ai_chat import call_ai_text mcp = FastMCP( "wechat-rpa", instructions=( "甄养堂企微客服助手 MCP。" "可查询/修改 AI 配置、查看会话档案、用当前人设草稿回复。" "不会直接操控企业微信鼠标；自动回复仍由 wechat_gui / wechat_bot 负责。" ), ) _STORE = ConversationStore(os.path.join(_ROOT, "conversations.json")) def _mask_key(key: str) -> str: if not key or len(key) < 10: return "***" return key[:6] + "…" + key[-4:] def _reload_store() -> ConversationStore: """每次读取前重新加载，避免外部进程写盘后读到旧缓存。""" global _STORE _STORE = ConversationStore(os.path.join(_ROOT, "conversations.json")) return _STORE def _session_summaries(limit: int = 50) -> list[dict[str, Any]]: store = _reload_store() items = [] for fp, entry in store._data.items(): hist = entry.get("history") or [] last = hist[-1] if hist else None items.append({ "session_id": fp, "message_count": len(hist), "updated": entry.get("updated") or 0, "updated_iso": time.strftime( "%Y-%m-%d %H:%M:%S", time.localtime(entry["updated"]), ) if entry.get("updated") else None, "last_role": (last or {}).get("role"), "last_content_preview": ((last or {}).get("content") or "")[:80], }) items.sort(key=lambda x: x["updated"] or 0, reverse=True) return items[:limit] # ────────────────────────────────────────────────────────────────────────────── # Tools # ────────────────────────────────────────────────────────────────────────────── @mcp.tool() def get_status() -> dict[str, Any]: """获取企微窗口挂载状态、AI 开关、会话档案数量等总览。""" hwnd = 0 title = "" try: import win32gui from wechat_bot import find_wx_hwnd, WX_WINDOW_CLASS hwnd = find_wx_hwnd() if hwnd: title = win32gui.GetWindowText(hwnd) cls = win32gui.GetClassName(hwnd) else: cls = "" except Exception as e: cls = f"error: {e}" sessions = _session_summaries(limit=3) return { "wechat_found": bool(hwnd), "hwnd": f"0x{hwnd:08X}" if hwnd else None, "window_title": title or None, "window_class": cls or None, "ai_enabled": bool(ai_config.AI_ENABLED), "ai_model": ai_config.AI_MODEL, "agent_name": ai_config.AI_AGENT_NAME, "hospital": ai_config.AI_HOSPITAL_NAME, "context_enabled": bool(ai_config.AI_CONTEXT_ENABLED), "counter_insult": bool(ai_config.AI_COUNTER_INSULT_ENABLED), "session_count": len(_reload_store()._data), "recent_sessions": sessions, "project_root": _ROOT, } @mcp.tool() def get_ai_config(reveal_api_key: bool = False) -> dict[str, Any]: """读取当前 AI 配置（默认脱敏 API Key；reveal_api_key=true 时显示完整 Key）。""" ai_config.load_settings() data = {k: getattr(ai_config, k) for k in ai_config.CONFIGURABLE_KEYS} if not reveal_api_key and "AI_API_KEY" in data: data["AI_API_KEY"] = _mask_key(str(data["AI_API_KEY"])) data["system_prompt_preview"] = ai_config.build_system_prompt()[:500] return data @mcp.tool() def update_ai_config(updates: dict[str, Any]) -> dict[str, Any]: """ 更新 AI 配置并写入 ai_settings.json。 只允许修改 CONFIGURABLE_KEYS 中的字段，例如： {"AI_AGENT_NAME": "高兴亮", "AI_COUNTER_INSULT_ENABLED": false, "AI_TEMPERATURE": 0.8} """ ai_config.load_settings() changed = {} ignored = [] for k, v in (updates or {}).items(): if k not in ai_config.CONFIGURABLE_KEYS: ignored.append(k) continue # 轻量类型校正 cur = getattr(ai_config, k) try: if isinstance(cur, bool): if isinstance(v, str): v = v.strip().lower() in ("1", "true", "yes", "on") else: v = bool(v) elif isinstance(cur, int) and not isinstance(cur, bool): v = int(v) elif isinstance(cur, float): v = float(v) else: v = str(v) except Exception: ignored.append(k) continue setattr(ai_config, k, v) changed[k] = v if k != "AI_API_KEY" else _mask_key(str(v)) if "AI_AGENT_NAME" in changed or "AI_HOSPITAL_NAME" in changed: ai_config.AI_SYSTEM_PROMPT = ai_config.build_system_prompt() ai_config.save_settings() return {"ok": True, "changed": changed, "ignored": ignored} @mcp.tool() def get_system_prompt() -> str: """返回当前渲染后的完整系统提示词（含对骂模式开关效果说明）。""" ai_config.load_settings() prompt = ai_config.build_system_prompt() if getattr(ai_config, "AI_COUNTER_INSULT_ENABLED", False): prompt += "\n" + getattr(ai_config, "AI_COUNTER_INSULT_PROMPT", "") return prompt @mcp.tool() def list_sessions(limit: int = 30) -> list[dict[str, Any]]: """列出已持久化的会话档案（按最近更新排序）。session_id 为头像指纹 hex。""" return _session_summaries(limit=max(1, min(limit, 200))) @mcp.tool() def get_session_history(session_id: str, max_messages: int = 40) -> dict[str, Any]: """获取指定会话的对话历史。session_id 来自 list_sessions。""" store = _reload_store() if session_id not in store._data: return {"ok": False, "error": f"session not found: {session_id}"} hist = store.history(session_id) max_messages = max(1, min(max_messages, 200)) slice_ = hist[-max_messages:] return { "ok": True, "session_id": session_id, "total": len(hist), "returned": len(slice_), "messages": [ { "role": m.get("role"), "content": m.get("content"), "ts": m.get("ts"), "time": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(m["ts"])) if m.get("ts") else None, } for m in slice_ ], } @mcp.tool() def clear_session(session_id: str) -> dict[str, Any]: """清空并删除指定会话档案。""" store = _reload_store() if session_id not in store._data: return {"ok": False, "error": f"session not found: {session_id}"} del store._data[session_id] store.save() return {"ok": True, "deleted": session_id} @mcp.tool() def draft_reply( message: str, session_id: Optional[str] = None, use_history: bool = True, ) -> dict[str, Any]: """ 用当前客服人设生成一条草稿回复（不发送到企业微信）。 可选传入 session_id，携带该会话档案历史上下文。 """ if not (message or "").strip(): return {"ok": False, "error": "message is empty"} history = None if use_history and session_id: store = _reload_store() if session_id in store._data: history = store.history(session_id) try: reply = call_ai_text(message.strip(), history=history) except Exception as e: return {"ok": False, "error": str(e)} return { "ok": True, "reply": reply, "session_id": session_id, "history_used": bool(history), "history_rounds": (len(history) // 2) if history else 0, } @mcp.tool() def append_session_message( session_id: str, role: str, content: str, ) -> dict[str, Any]: """ 向会话档案追加一条消息。role 必须是 user 或 assistant。 用于人工校正上下文，或外部系统写入对话记录。 """ if role not in ("user", "assistant"): return {"ok": False, "error": "role must be 'user' or 'assistant'"} if not content.strip(): return {"ok": False, "error": "content is empty"} if not session_id.strip(): return {"ok": False, "error": "session_id is empty"} store = _reload_store() store.append(session_id.strip(), role, content.strip()) store.save() return { "ok": True, "session_id": session_id, "message_count": len(store.history(session_id)), } # ────────────────────────────────────────────────────────────────────────────── # Resources # ────────────────────────────────────────────────────────────────────────────── @mcp.resource("wechat-rpa://status") def resource_status() -> str: """总览状态 JSON。""" return json.dumps(get_status(), ensure_ascii=False, indent=2) @mcp.resource("wechat-rpa://sessions") def resource_sessions() -> str: """会话列表 JSON。""" return json.dumps(list_sessions(50), ensure_ascii=False, indent=2) @mcp.resource("wechat-rpa://config") def resource_config() -> str: """AI 配置 JSON（Key 脱敏）。""" return json.dumps(get_ai_config(False), ensure_ascii=False, indent=2) @mcp.prompt() def customer_service_draft(customer_message: str) -> str: """按甄养堂糖尿病客服人设，为客户消息起草回复。""" return ( f"请以甄养堂互联网医院客服「{ai_config.AI_AGENT_NAME}」身份，" f"针对下面客户消息起草一条微信口语化回复（不要解释过程）：\n\n" f"{customer_message}" ) if __name__ == "__main__": mcp.run(transport="stdio")