317 lines
11 KiB
Python
317 lines
11 KiB
Python
"""
|
||
甄养堂 · 企微客服助手 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")
|