From 747a9ab5581aa07e80ef31d806fce575c9ed224f Mon Sep 17 00:00:00 2001 From: wanghanlin <1533525126@qq.com> Date: Tue, 7 Jul 2026 17:25:19 +0800 Subject: [PATCH 1/8] =?UTF-8?q?=E6=94=AF=E6=8C=81mcp=E6=9C=8D=E5=8A=A1?= =?UTF-8?q?=E5=8A=A0=E5=85=A5=E5=AF=B9=E8=AF=9D=E9=93=BE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../client/McpSyncClient.class | Bin 0 -> 9800 bytes pom.xml | 6 + .../com/wok/supportbot/app/AssistantApp.java | 223 +++++- .../supportbot/config/ChatModelFactory.java | 16 + .../supportbot/config/DatabaseInitConfig.java | 70 +- .../supportbot/config/McpClientManager.java | 660 ++++++++++++++++++ .../config/McpRequestInterceptor.java | 24 + .../supportbot/config/McpWebMvcConfig.java | 24 + .../supportbot/controller/AiController.java | 16 +- .../CustomerServiceRoleController.java | 29 + .../controller/McpServerConfigController.java | 348 +++++++++ .../converter/MessageConverter.java | 118 +++- .../supportbot/dao/McpServerConfigMapper.java | 12 + .../supportbot/entity/McpServerConfig.java | 104 +++ .../wok/supportbot/mcp/McpToolCallback.java | 265 +++++++ .../mcp/McpToolCallbackAdapter.java | 111 +++ .../wok/supportbot/mcp/SseEventBuilder.java | 73 ++ .../service/CustomerServiceRoleService.java | 76 +- .../service/McpServerConfigService.java | 320 +++++++++ src/main/resources/init-database.sql | 4 +- .../resources/static/components/ChatPanel.js | 41 +- .../static/components/McpServerManager.js | 524 ++++++++++++++ src/main/resources/static/css/main.css | 9 + src/main/resources/static/js/api.js | 58 ++ src/main/resources/static/js/app.js | 7 + src/main/resources/static/js/store.js | 4 +- src/main/resources/static/js/utils.js | 61 ++ 27 files changed, 3161 insertions(+), 42 deletions(-) create mode 100644 io/modelcontextprotocol/client/McpSyncClient.class create mode 100644 src/main/java/com/wok/supportbot/config/McpClientManager.java create mode 100644 src/main/java/com/wok/supportbot/config/McpRequestInterceptor.java create mode 100644 src/main/java/com/wok/supportbot/config/McpWebMvcConfig.java create mode 100644 src/main/java/com/wok/supportbot/controller/McpServerConfigController.java create mode 100644 src/main/java/com/wok/supportbot/dao/McpServerConfigMapper.java create mode 100644 src/main/java/com/wok/supportbot/entity/McpServerConfig.java create mode 100644 src/main/java/com/wok/supportbot/mcp/McpToolCallback.java create mode 100644 src/main/java/com/wok/supportbot/mcp/McpToolCallbackAdapter.java create mode 100644 src/main/java/com/wok/supportbot/mcp/SseEventBuilder.java create mode 100644 src/main/java/com/wok/supportbot/service/McpServerConfigService.java create mode 100644 src/main/resources/static/components/McpServerManager.js diff --git a/io/modelcontextprotocol/client/McpSyncClient.class b/io/modelcontextprotocol/client/McpSyncClient.class new file mode 100644 index 0000000000000000000000000000000000000000..b42dca95fb03ffa58121ddd1ecfd0f9688abee8d GIT binary patch literal 9800 zcmb_i3wRt=75;CUcG6_>X!@cL=mLeNDRg-i3MrKm(h}ITwMko^*2!j)OquMYv$KKn zRFRj0FI3dx3sDd%@+<{k#TSU8qN0eRsGuk+D!xVJzcaHtv$L~w7R>i`&(7UB|2^kF z=bn3JZXbT>{x1SJUZfgPhe-%e;8s2B2uxXFZx48G#k-~CIzaBX0!+#Xjthni#KPjoTH!T zjN6uDWi2yZa11lODxcSF&Y=)sEtLfpb!4I@bcIIx9oxu_bS>UA6>XTNpk2ju%n&%F z+B8|y&~r|@FFW49J(ulvuLK&3)(fOFdskhRFX?tQVismA*hj@2%oRAaCT)S4Bf8VC z+Y`Fot&MAghG{s4o)zZrBnyHn~Z?W8bwDP(fIYPye zI7*<^9a&>*S$&xKxLsfYS+I32>sWR=YuS2wyfA1Q`BB|Y_gOisYckgn{}#11VmXdg zaGZ)RtPp5-xps^(J$+Kac6njmS;KvXNp*qQ9jEy5D7%N2uf*#VoSpM9Cso?e&<5Tskm*QTH01qV zubjzpiV+o~Fi3!{k69DCH_PEMobDa!92KT?tA3mnTVm%_SQzIb$ZDoJU|Fp8n-kmL z?dxpNw-xmK-ds_@v|W}kCINIYB94lJG<_;psv|AUD>zr+=!AeWgsxT)98d#J^bJxEk7;OJMYs($uR2&aA<LG`te$l852@H8nQoFyueYtSafkeVSYUY#{Q7zUs~?r9E~_6`aRWZV#mRQXyR8;B z*A$hKrH+D6CSUg<>|!OkNyW|hH2aFY?#Qi^r7)vU=q!B$iK7tK+slHK!M|0-PJD*D zc9ztex}$e{lDm6N1=rV3%F}0SD7YnPkET0tr-Hjw+>LwK7e%-B)o$;_PD}>eYd-`I z3U5i>rk2l>nW!eaoqGuwbQ-G)j@2z&K5fwCKDy6IJ*!u3%nWSl&aCV2*)q`E*RyWp zz?Qy#0m1gBxto2nqvf1UvNI7lqI$%#rKgPWqrzf#D)rrb_+k7rMowQ_7#q~>0dI0P zRon+RX|^G+{lR+nPV5brR^8QuiGs(OF7`uxx-)7Gt=DXAj1jkqAvo*&P@fa+^r+R` z?`YX``m}L>_O2Ql{!)MBEm5K!R-K+m6Xf+Gk{sLZVh*k9Zk!@(csTf_Cqni`5^kf- z+S49#+`mC!TVex)j-Xid5|*w&QD!ztX=RMvUdLZXtb$3M4X`{T*dHX1kKCg|EeI`P zs~9YVB56r&sf3%7E!cUBT^5^)*M?i%6Z54^T3MT)l>oWKkuK`8dL#^SWgA0s8*IOk z6F59Mkk|WBeR~b|#x*~$#q{an9#YBjRx zx192Mr??N6L)EemmTSstyMk?y_fER8-+g{wZODGEz3eUTQdxmWT5m4Lm#wlzl+Ss8 zN&diE`JPq5Uj+_MR&8lc9Kb)4x5?!%Rr$)Sd_cj!ll$VmsVjfO+8#I97JWmd;3a`2 z$$Vbuy)UYlLJ=bQlT)PEVc%3S(h`iwoN`<0vL1(zyWuBuznxHo!gu^h%`ffZ4aLTQ zVXWNo715OZZB4nlUzn%Ij?N1^5ZSR5(VF}2`M0@h|soWmOOWfcz=|NfD zR>Z92c1P(A?8em2 zCHJ894vy8~3;b+HJwI3Ta}CemPDU$wAy4>bd3yK~_V9`bF22A6{63j0hC;}LEPl#~ zj-V8j4V`zRbr)vd&hOp?Q{3_OE~x6=hqi_ zUjM+I5&Ml;eV#+b>o1DVzb#gur%dtsOQQPQWA%9!oUdCV{-S(~IHZGq7woj@gCdH{XQWnLE+b-s=vXy5v5*z8;fy*6(!ZoW{iKW@^fG znoEno7qioFB}XsAOk7UzXCti>~G^DdmR2WLm=xi#ckODVmy5dR@()MMPlF25IgRKOQ()duw{ic%p7QW>h*zLRD>bpNij6K*Eg?S=mUTewu zRy=}l2j)HEnpePgeDflK&b!bPQRt^b76ySX;JZ~UJTD&R`6?DZ8Yb{j*TM_&y(-w} zyVw^+O?QCYE{j+wnyXm&7=9SU=P}p&OYx&B*jL13 z&#Hp`IDQ;pKki~*jVG#Lzs<$ICTjNqRj^~Cb~Szy&+I$np*v%ly%s<9EsQSwcM~RB zLBugo#qyAi;f4Pseir!ur0f56_<0rd_r;-~K=f5r(4WFD0`#X`^d0zR6?Q-9QgD5g z-8CWX@T!ercL#nI&+bR!p-+uv_Xa#&#lnxdFgHdmTvx@yXYlL5`)6G5Z^mz`aQmq^ zZa0(L(?aOs$UTeS2I$Ya=sWSdDwf~kT7Fy9a=8ymB~Q~s0kUzGo$Je%@G z%KuXS4|9n8f~e#5Hp)qq6O{Fo7g0{;?^7rnczqe=R9-hy%Ky1^4P`U0TPWKor%_Jl k`1O=Ccs-MHHm`4_+=th5DCeCpN6e?(kK?ydF2J0Z0oC#lD*ylh literal 0 HcmV?d00001 diff --git a/pom.xml b/pom.xml index 97bfe89..3e6f72b 100644 --- a/pom.xml +++ b/pom.xml @@ -92,6 +92,12 @@ spring-ai-openai + + + org.springframework.ai + spring-ai-starter-mcp-client + + cn.hutool diff --git a/src/main/java/com/wok/supportbot/app/AssistantApp.java b/src/main/java/com/wok/supportbot/app/AssistantApp.java index 44ede25..19204fa 100644 --- a/src/main/java/com/wok/supportbot/app/AssistantApp.java +++ b/src/main/java/com/wok/supportbot/app/AssistantApp.java @@ -10,6 +10,8 @@ import com.wok.supportbot.rag.preretrieval.MultiQueryExpanderRewriter; import com.wok.supportbot.rag.preretrieval.RewriteQueryRewriter; import com.wok.supportbot.rag.preretrieval.TranslationQueryRewriter; import com.wok.supportbot.service.FaqMatchEngine; +import com.wok.supportbot.mcp.McpToolCallback; +import com.wok.supportbot.mcp.McpToolCallbackAdapter; import jakarta.annotation.Resource; import lombok.extern.slf4j.Slf4j; import org.springframework.ai.chat.client.ChatClient; @@ -23,10 +25,12 @@ import org.springframework.ai.document.Document; import org.springframework.ai.rag.advisor.RetrievalAugmentationAdvisor; import org.springframework.ai.rag.generation.augmentation.ContextualQueryAugmenter; import org.springframework.ai.rag.retrieval.search.VectorStoreDocumentRetriever; +import org.springframework.ai.tool.ToolCallback; import org.springframework.ai.vectorstore.SearchRequest; import org.springframework.ai.vectorstore.VectorStore; import org.springframework.ai.vectorstore.filter.Filter; import org.springframework.ai.vectorstore.filter.FilterExpressionBuilder; +import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Component; import org.springframework.util.StringUtils; import reactor.core.publisher.Flux; @@ -62,6 +66,13 @@ public class AssistantApp { @Resource private FaqMatchEngine faqMatchEngine; + @Resource + private McpToolCallbackAdapter mcpToolCallbackAdapter; + + /** MCP 工具开关，默认启用，可通过 application.yml 的 chat.mcp.enabled 关闭 */ + @Value("${chat.mcp.enabled:true}") + private boolean enableMcpTools; + private final ChatModelFactory chatModelFactory; private final DatabaseChatMemory chatMemory; @@ -81,13 +92,51 @@ public class AssistantApp { } private ChatClient getChatClient(String appType) { - return chatClientCache.computeIfAbsent(appType, type -> { - ChatModel chatModel = chatModelFactory.getChatModel(type); + return getChatClient(appType, enableMcpTools ? java.util.List.of("*") : null); + } + + /** + * 获取 ChatClient（支持按角色过滤 MCP 工具） + * 缓存 key 包含工具权限信息，不同角色权限缓存不同的 ChatClient 实例。 + * + * @param appType 应用类型（CHAT / RAG_REWRITE 等） + * @param allowedMcpTools 允许的 MCP 工具名列表；null 或空 = 允许所有（默认行为）；["*"] = 全部 + */ + private ChatClient getChatClient(String appType, List allowedMcpTools) { + // 空列表降级为 ["*"]（允许所有工具），与无参数版本行为一致 + final List effectiveTools; + if (enableMcpTools && (allowedMcpTools == null || allowedMcpTools.isEmpty())) { + effectiveTools = java.util.List.of("*"); + } else { + effectiveTools = allowedMcpTools; + } + String cacheKey = appType + ":" + (effectiveTools == null ? "none" : String.join(",", effectiveTools)); + return chatClientCache.computeIfAbsent(cacheKey, type -> { + ChatModel chatModel = chatModelFactory.getChatModel(appType); var builder = ChatClient.builder(chatModel); // 仅在系统提示词非空时设置，避免 Spring AI 的 Assert.hasText 校验抛异常 if (StringUtils.hasText(SYSTEM_PROMPT)) { builder.defaultSystem(SYSTEM_PROMPT); } + // 注册 MCP 工具（按角色权限过滤） + if (enableMcpTools && effectiveTools != null && !effectiveTools.isEmpty()) { + ToolCallback[] mcpTools = mcpToolCallbackAdapter.getFilteredToolCallbacks(effectiveTools); + if (mcpTools.length > 0) { + builder.defaultToolCallbacks(mcpTools); + log.info("✅ 已注册 {} 个 MCP 工具到 ChatClient [{}]", mcpTools.length, cacheKey); + for (ToolCallback tc : mcpTools) { + log.info(" 🔧 工具: {} — {}", tc.getToolDefinition().name(), tc.getToolDefinition().description()); + } + } else { + log.warn("⚠️ MCP 工具已启用但无可注册的工具 [{}]，请检查：", cacheKey); + log.warn(" 1. 是否在「MCP 服务管理」页面添加了 MCP Server 配置？"); + log.warn(" 2. 配置是否已启用（is_active=true）？"); + log.warn(" 3. MCP Server 是否连接成功（点击「测试连接」验证）？"); + log.warn(" 4. MCP Server 是否暴露了工具（listTools 返回非空）？"); + } + } else { + log.info("ℹ️ MCP 工具未启用 [enableMcpTools={}, effectiveTools={}]", enableMcpTools, effectiveTools); + } return builder .defaultAdvisors( contentSafetyAdvisor, @@ -123,7 +172,20 @@ public class AssistantApp { * @return AI 回答 */ public String doChat(String message, String chatId, String systemPrompt) { - ChatClient.ChatClientRequestSpec spec = getChatClient("CHAT") + return doChat(message, chatId, systemPrompt, null); + } + + /** + * AI 基础对话（支持多轮对话记忆 + 角色系统提示词 + MCP 工具权限） + * + * @param message 用户消息 + * @param chatId 会话ID + * @param systemPrompt 角色人设/风格 + * @param allowedMcpTools 允许的 MCP 工具列表；null=不注册工具 + * @return AI 回答 + */ + public String doChat(String message, String chatId, String systemPrompt, List allowedMcpTools) { + ChatClient.ChatClientRequestSpec spec = getChatClient("CHAT", allowedMcpTools) .prompt() .user(message) .advisors(s -> s.param(CONVERSATION_ID, chatId)); @@ -175,6 +237,64 @@ public class AssistantApp { return spec.stream().content(); } + /** + * AI 基础对话（多轮记忆 + 角色系统提示词 + MCP 工具权限，SSE 流式传输） + * + * @param message 用户消息 + * @param chatId 会话ID + * @param systemPrompt 角色人设/风格 + * @param allowedMcpTools 允许的 MCP 工具列表；null=不注册工具 + * @return 流式回答 + */ + public Flux doChatByStream(String message, String chatId, String systemPrompt, List allowedMcpTools) { + // 重置 MCP 工具调用状态（事件收集器 + 轮次计数器） + McpToolCallback.resetEvents(); + McpToolCallback.resetCallRounds(); + + ChatClient.ChatClientRequestSpec spec = getChatClient("CHAT", allowedMcpTools) + .prompt() + .user(message) + .advisors(s -> s.param(CONVERSATION_ID, chatId)); + if (StringUtils.hasText(systemPrompt)) { + spec = spec.system(effectiveSystem(systemPrompt)); + } + return appendMcpToolEvents(spec.stream().content()); + } + + /** + * 在 SSE 文本流末尾追加 MCP 工具调用事件 + * 前端通过解析 "event: tool_call_start" / "event: tool_call_result" 行来展示工具调用过程。 + * 事件格式遵循 SSE 标准，与 readSSEStreamWithEvents 的解析逻辑匹配。 + */ + private Flux appendMcpToolEvents(Flux contentFlux) { + return contentFlux.concatMap(chunk -> Flux.just(chunk)) + .concatWith(Flux.defer(() -> { + java.util.List events = McpToolCallback.drainEvents(); + if (events.isEmpty()) { + return Flux.empty(); + } + log.info("本次对话触发了 {} 个 MCP 工具调用", events.size()); + List sseLines = new ArrayList<>(); + for (McpToolCallback.ToolCallEvent event : events) { + sseLines.add("event:tool_call_result"); + String json = String.format("{\"tool\":\"%s\",\"result\":\"%s\",\"latencyMs\":%d}", + escapeSse(event.tool()), + escapeSse(event.result()), + event.latencyMs()); + sseLines.add("data:" + json); + sseLines.add(""); // SSE 事件分隔空行 + } + return Flux.fromIterable(sseLines); + })); + } + + /** 转义 SSE 数据中的特殊字符 */ + private static String escapeSse(String s) { + if (s == null) return ""; + return s.replace("\\", "\\\\").replace("\"", "\\\"") + .replace("\n", "\\n").replace("\r", "\\r"); + } + // ==================== FAQ 优先匹配 ==================== /** @@ -278,6 +398,44 @@ public class AssistantApp { return chatResponse.getResult().getOutput().getText(); } + /** + * RAG 知识库对话（支持查询重写策略 + MCP 工具权限） + * + * @param message 用户消息 + * @param chatId 会话ID + * @param strategy 查询重写策略 + * @param categoryIds 知识库分类过滤 + * @param systemPrompt 角色人设/风格 + * @param allowedMcpTools 允许的 MCP 工具列表；null=不注册工具 + * @return AI 回答 + */ + public String doChatWithRagStrategy(String message, String chatId, String strategy, List categoryIds, + String systemPrompt, List allowedMcpTools) { + // FAQ 优先匹配 + Optional faqAnswer = tryFaqMatch(message); + if (faqAnswer.isPresent()) { + log.info("FAQ 命中，直接返回标准答案: chatId={}", chatId); + return faqAnswer.get(); + } + + if ("MULTI_QUERY".equalsIgnoreCase(strategy)) { + return doChatWithMultiQueryRag(message, chatId, categoryIds, systemPrompt, allowedMcpTools); + } + + String rewrittenMessage = rewriteQuery(message, chatId, strategy); + + ChatClient.ChatClientRequestSpec spec = getChatClient("CHAT", allowedMcpTools) + .prompt() + .user(rewrittenMessage) + .advisors(s -> s.param(CONVERSATION_ID, chatId)) + .advisors(buildRetrievalAdvisor(4, categoryIds)); + if (StringUtils.hasText(systemPrompt)) { + spec = spec.system(effectiveSystem(systemPrompt)); + } + ChatResponse chatResponse = spec.call().chatResponse(); + return chatResponse.getResult().getOutput().getText(); + } + /** * 根据策略对查询做预检索改写（MULTI_QUERY 不走此方法）。 */ @@ -344,6 +502,49 @@ public class AssistantApp { return spec.stream().content(); } + /** + * 和 RAG 知识库进行对话（支持查询重写策略 + MCP 工具权限，SSE 流式传输） + * + * @param message 用户消息 + * @param chatId 会话ID + * @param strategy 查询重写策略 + * @param categoryIds 知识库分类过滤 + * @param systemPrompt 角色人设/风格 + * @param allowedMcpTools 允许的 MCP 工具列表；null=不注册工具 + * @return 流式回答 + */ + public Flux doChatWithRagStrategyByStream(String message, String chatId, String strategy, + List categoryIds, String systemPrompt, + List allowedMcpTools) { + // 重置 MCP 工具调用状态 + McpToolCallback.resetEvents(); + McpToolCallback.resetCallRounds(); + + // FAQ 优先匹配：命中则直接以流式形式返回标准答案 + Optional faqAnswer = tryFaqMatch(message); + if (faqAnswer.isPresent()) { + log.info("FAQ 命中（流式），直接返回标准答案: chatId={}", chatId); + return Flux.just(faqAnswer.get()); + } + + // 对于 MULTI_QUERY 策略，需要先手动检索合并再流式生成 + if ("MULTI_QUERY".equalsIgnoreCase(strategy)) { + return doChatWithMultiQueryRagByStream(message, chatId, categoryIds, systemPrompt, allowedMcpTools); + } + + String rewrittenMessage = rewriteQuery(message, chatId, strategy); + + ChatClient.ChatClientRequestSpec spec = getChatClient("CHAT", allowedMcpTools) + .prompt() + .user(rewrittenMessage) + .advisors(s -> s.param(CONVERSATION_ID, chatId)) + .advisors(buildRetrievalAdvisor(4, categoryIds)); + if (StringUtils.hasText(systemPrompt)) { + spec = spec.system(effectiveSystem(systemPrompt)); + } + return appendMcpToolEvents(spec.stream().content()); + } + /** * 使用多路查询扩展的 RAG 对话 * 将原始查询扩展为多个语义不同的查询，分别检索后按文档ID去重合并， @@ -356,8 +557,13 @@ public class AssistantApp { * @return AI 回答 */ private String doChatWithMultiQueryRag(String message, String chatId, List categoryIds, String systemPrompt) { + return doChatWithMultiQueryRag(message, chatId, categoryIds, systemPrompt, null); + } + + private String doChatWithMultiQueryRag(String message, String chatId, List categoryIds, + String systemPrompt, List allowedMcpTools) { String ragSystem = buildMultiQueryRagSystem(message, categoryIds, systemPrompt); - ChatResponse chatResponse = getChatClient("CHAT") + ChatResponse chatResponse = getChatClient("CHAT", allowedMcpTools) .prompt() .system(ragSystem) .user(message) @@ -368,14 +574,19 @@ public class AssistantApp { } private Flux doChatWithMultiQueryRagByStream(String message, String chatId, List categoryIds, String systemPrompt) { + return doChatWithMultiQueryRagByStream(message, chatId, categoryIds, systemPrompt, null); + } + + private Flux doChatWithMultiQueryRagByStream(String message, String chatId, List categoryIds, + String systemPrompt, List allowedMcpTools) { String ragSystem = buildMultiQueryRagSystem(message, categoryIds, systemPrompt); - return getChatClient("CHAT") + return appendMcpToolEvents(getChatClient("CHAT", allowedMcpTools) .prompt() .system(ragSystem) .user(message) .advisors(s -> s.param(CONVERSATION_ID, chatId)) .stream() - .content(); + .content()); } /** diff --git a/src/main/java/com/wok/supportbot/config/ChatModelFactory.java b/src/main/java/com/wok/supportbot/config/ChatModelFactory.java index e6190ef..b67b06f 100644 --- a/src/main/java/com/wok/supportbot/config/ChatModelFactory.java +++ b/src/main/java/com/wok/supportbot/config/ChatModelFactory.java @@ -354,6 +354,22 @@ public class ChatModelFactory { // ==================== F6: Fallback 故障自动切换 ==================== + /** + * 判断指定提供商是否支持 Tool Calling + * DashScope: 支持（通义千问系列） + * DeepSeek: 支持 + * OpenAI: 支持 + * 其他: 默认支持（由 API 端点自行返回错误） + * + * @param provider 提供商标识（如 dashscope、deepseek、openai 等） + * @return 是否支持 Tool Calling + */ + public boolean supportsToolCalling(String provider) { + if (provider == null) return true; + // 当前所有已支持的提供商都支持 tool calling + return true; + } + /** 共享熔断器：连续 3 次失败触发，5 分钟自动恢复 */ private final SimpleCircuitBreaker circuitBreaker = new SimpleCircuitBreaker(3, 5 * 60 * 1000L, ""); diff --git a/src/main/java/com/wok/supportbot/config/DatabaseInitConfig.java b/src/main/java/com/wok/supportbot/config/DatabaseInitConfig.java index 67e832e..1e25d64 100644 --- a/src/main/java/com/wok/supportbot/config/DatabaseInitConfig.java +++ b/src/main/java/com/wok/supportbot/config/DatabaseInitConfig.java @@ -86,6 +86,8 @@ public class DatabaseInitConfig { }); safeInit("迁移 conversation_session.external_account_id 列", this::addConversationSessionExternalAccountIdColumn); + safeInit("迁移 customer_service_role.allowed_mcp_tools 列", this::addRoleAllowedMcpToolsColumn); + safeInit("同步默认客服角色", this::syncDefaultCustomerServiceRoles); safeInit("同步默认客服账号", this::syncDefaultCustomerAccounts); @@ -182,6 +184,14 @@ public class DatabaseInitConfig { } }); + // ==================== MCP Server 配置 ==================== + + safeInit("创建 MCP Server 配置表 mcp_server_config", () -> { + if (!checkTableExists("mcp_server_config")) { + createMcpServerConfigTable(); + } + }); + // P0-001: 混合检索 - 为 vector_store 添加全文检索列 safeInit("初始化 vector_store 全文检索", this::initVectorStoreFullTextSearch); @@ -220,7 +230,8 @@ public class DatabaseInitConfig { "knowledge_faq", "faq_embedding", "sys_user", "sys_role", "sys_permission", "sys_user_role", "rag_hit_log", "dashboard_snapshot", - "api_key", "webhook_config" + "api_key", "webhook_config", + "mcp_server_config" }; java.util.List missingTables = new java.util.ArrayList<>(); @@ -430,6 +441,24 @@ public class DatabaseInitConfig { log.warn("add conversation_session.external_account_id failed: {}", e.getMessage()); } } + + /** + * 为 customer_service_role 表添加 allowed_mcp_tools 列（角色 MCP 工具权限控制） + * 幂等：已有列则跳过 + */ + private void addRoleAllowedMcpToolsColumn() { + try { + String checkSql = "SELECT COUNT(*) FROM information_schema.columns WHERE table_name = 'customer_service_role' AND column_name = 'allowed_mcp_tools'"; + Integer count = jdbcTemplate.queryForObject(checkSql, Integer.class); + if (count != null && count == 0) { + log.info("添加 customer_service_role.allowed_mcp_tools 列"); + jdbcTemplate.execute("ALTER TABLE customer_service_role ADD COLUMN allowed_mcp_tools JSONB DEFAULT '[]'"); + } + } catch (Exception e) { + log.error("添加 customer_service_role.allowed_mcp_tools 列失败", e); + } + } + private void syncDefaultCustomerServiceRoles() { upsertDefaultRole("general", "客服", "用户咨询、业务办理、常见问题、问题受理与进度说明", """ @@ -1017,6 +1046,27 @@ public class DatabaseInitConfig { jdbcTemplate.execute("CREATE INDEX IF NOT EXISTS idx_webhook_enabled ON webhook_config (enabled) WHERE is_delete = false"); } + private void createMcpServerConfigTable() { + String sql = """ + CREATE TABLE IF NOT EXISTS mcp_server_config ( + id BIGINT PRIMARY KEY, + name VARCHAR(100) NOT NULL, + transport_type VARCHAR(20) NOT NULL DEFAULT 'sse', + server_url VARCHAR(500), + command VARCHAR(500), + args VARCHAR(1000), + env_vars JSONB DEFAULT '{}' NOT NULL, + description VARCHAR(500), + is_active BOOLEAN DEFAULT TRUE NOT NULL, + create_time TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, + update_time TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, + is_delete BOOLEAN NOT NULL DEFAULT FALSE + ) + """; + jdbcTemplate.execute(sql); + jdbcTemplate.execute("CREATE INDEX IF NOT EXISTS idx_mcp_server_config_active ON mcp_server_config (is_active) WHERE is_delete = FALSE"); + } + /** * 为所有自动创建的表添加注释（COMMENT ON）。 * 所有语句均为幂等操作，可安全重复执行。 @@ -1027,7 +1077,7 @@ public class DatabaseInitConfig { executeComment("TABLE chat_message", "聊天消息表（存储用户与 AI 助手的对话历史）"); executeComment("COLUMN chat_message.id", "主键（雪花算法生成）"); executeComment("COLUMN chat_message.conversation_id", "会话 ID（标识同一次对话）"); - executeComment("COLUMN chat_message.message_type", "消息类型: USER(用户消息) / ASSISTANT(AI回复) / SYSTEM(系统消息)"); + executeComment("COLUMN chat_message.message_type", "消息类型: USER(用户消息) / ASSISTANT(AI回复,可能含toolCalls) / SYSTEM(系统消息) / TOOL(MCP工具调用响应)"); executeComment("COLUMN chat_message.content", "消息内容（实际对话文本）"); executeComment("COLUMN chat_message.metadata", "元数据（JSON 格式，存储额外信息）"); executeComment("COLUMN chat_message.create_time", "创建时间"); @@ -1077,6 +1127,7 @@ public class DatabaseInitConfig { executeComment("COLUMN customer_service_role.create_time", "创建时间"); executeComment("COLUMN customer_service_role.update_time", "更新时间"); executeComment("COLUMN customer_service_role.is_delete", "逻辑删除: FALSE=正常 TRUE=已删除"); + executeComment("COLUMN customer_service_role.allowed_mcp_tools", "允许使用的 MCP 工具列表（JSONB 数组，空=不允许，[\"*\"]=全部允许）"); // ===== customer_service_role_category ===== executeComment("TABLE customer_service_role_category", "客服角色知识库关联表（角色与知识库分类的多对多关系）"); @@ -1208,6 +1259,21 @@ public class DatabaseInitConfig { // ===== webhook_config ===== executeComment("TABLE webhook_config", "Webhook 配置表（事件推送订阅）"); + // ===== mcp_server_config ===== + executeComment("TABLE mcp_server_config", "MCP Server 配置表（管理外部 MCP Server 连接配置）"); + executeComment("COLUMN mcp_server_config.id", "主键（雪花算法生成）"); + executeComment("COLUMN mcp_server_config.name", "配置名称"); + executeComment("COLUMN mcp_server_config.transport_type", "传输类型: stdio(标准输入输出) / sse(Server-Sent Events)"); + executeComment("COLUMN mcp_server_config.server_url", "SSE 模式的 MCP Server URL"); + executeComment("COLUMN mcp_server_config.command", "stdio 模式的启动命令"); + executeComment("COLUMN mcp_server_config.args", "stdio 模式的命令参数（多个用逗号分隔）"); + executeComment("COLUMN mcp_server_config.env_vars", "环境变量（JSONB 键值对）"); + executeComment("COLUMN mcp_server_config.description", "描述说明"); + executeComment("COLUMN mcp_server_config.is_active", "是否启用"); + executeComment("COLUMN mcp_server_config.create_time", "创建时间"); + executeComment("COLUMN mcp_server_config.update_time", "更新时间"); + executeComment("COLUMN mcp_server_config.is_delete", "逻辑删除: FALSE=正常 TRUE=已删除"); + // ===== ai_model_config ===== executeComment("TABLE ai_model_config", "AI 大模型配置表（管理多套模型配置，按应用类型绑定）"); executeComment("COLUMN ai_model_config.id", "主键（雪花算法生成）"); diff --git a/src/main/java/com/wok/supportbot/config/McpClientManager.java b/src/main/java/com/wok/supportbot/config/McpClientManager.java new file mode 100644 index 0000000..7041d8a --- /dev/null +++ b/src/main/java/com/wok/supportbot/config/McpClientManager.java @@ -0,0 +1,660 @@ +package com.wok.supportbot.config; + +import com.wok.supportbot.entity.McpServerConfig; +import com.wok.supportbot.service.McpServerConfigService; +import io.modelcontextprotocol.client.McpSyncClient; +import io.modelcontextprotocol.client.transport.HttpClientSseClientTransport; +import io.modelcontextprotocol.client.transport.ServerParameters; +import io.modelcontextprotocol.client.transport.StdioClientTransport; +import io.modelcontextprotocol.spec.McpSchema; +import jakarta.annotation.PostConstruct; +import jakarta.annotation.PreDestroy; +import lombok.extern.slf4j.Slf4j; +import org.springframework.ai.model.ModelOptionsUtils; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.scheduling.annotation.Scheduled; +import org.springframework.stereotype.Component; + +import java.time.Duration; +import java.time.Instant; +import java.util.*; +import java.util.concurrent.ConcurrentHashMap; + +/** + * MCP 客户端生命周期管理器 + * 负责 MCP Client 的创建、缓存、刷新和销毁。 + *

+ * 支持两种传输模式： + * - SSE（Server-Sent Events）：通过 HttpClientSseClientTransport 连接远程 MCP Server + * - stdio（标准输入输出）：通过 StdioClientTransport 启动本地 MCP Server 进程 + */ +@Component +@Slf4j +public class McpClientManager { + + @Autowired + private McpServerConfigService mcpServerConfigService; + + /** + * 客户端缓存：key = 配置ID 的字符串形式，value = MCP Client 实例 + * volatile 保证 refreshAll() 双缓冲切换时的可见性 + */ + private volatile ConcurrentHashMap clientCache = new ConcurrentHashMap<>(); + + /** + * 不可用配置集合：记录已知不存在或未启用的配置ID，避免重复查询 DB + * 配置被启用或新建时需同步移除此集合中的对应条目 + */ + private final Set unavailableConfigs = ConcurrentHashMap.newKeySet(); + + /** + * 健康状态缓存：key = 配置ID 的字符串形式，value = 健康检查结果 + */ + private final ConcurrentHashMap healthCache = new ConcurrentHashMap<>(); + + /** + * MCP Server 健康状态记录 + */ + public record HealthStatus( + /** 状态：ONLINE / OFFLINE / UNKNOWN */ + String status, + /** 响应延迟（毫秒） */ + long latencyMs, + /** 最后检查时间 */ + Instant lastCheckTime, + /** 错误信息（健康时为 null） */ + String errorMessage + ) { + public static HealthStatus online(long latencyMs) { + return new HealthStatus("ONLINE", latencyMs, Instant.now(), null); + } + + public static HealthStatus offline(long latencyMs, String errorMessage) { + return new HealthStatus("OFFLINE", latencyMs, Instant.now(), errorMessage); + } + + public static HealthStatus unknown(String errorMessage) { + return new HealthStatus("UNKNOWN", 0, Instant.now(), errorMessage); + } + } + + /** + * MCP 客户端请求超时时间，支持 Duration 格式（如 30s、PT30S、60s） + */ + @Value("${mcp.client.request-timeout:30s}") + private String requestTimeoutStr; + + /** + * MCP 客户端初始化超时时间，支持 Duration 格式（如 15s、PT15S、30s） + */ + @Value("${mcp.client.init-timeout:15s}") + private String initTimeoutStr; + + /** + * 应用启动时自动加载所有已启用的 MCP Server 配置并建立连接 + * 确保第一次对话请求时 clientCache 已就绪，MCP 工具可以被注册到 ChatClient + */ + @PostConstruct + public void init() { + log.info("McpClientManager 初始化，开始加载已启用的 MCP Server 配置..."); + refreshAll(); + } + + /** + * 刷新所有 MCP 客户端连接（双缓冲策略） + * 先在新 Map 中构建所有客户端，再原子切换引用，最后关闭旧客户端。 + * 避免清空与重建之间的请求全部失败。 + */ + public void refreshAll() { + log.info("开始刷新所有 MCP 客户端连接..."); + + // 获取所有启用的配置（一次查询代替两次按类型过滤） + List allActiveConfigs = mcpServerConfigService.listAllActiveConfigs(); + + // 第一步：在全新 Map 中构建所有客户端（不影响现有缓存） + ConcurrentHashMap newCache = new ConcurrentHashMap<>(); + for (McpServerConfig config : allActiveConfigs) { + try { + McpSyncClient client = createClientDirectly(config); + if (client != null) { + newCache.put(config.getId().toString(), client); + } + } catch (Exception e) { + log.error("创建客户端失败: id={}, name={}, transportType={}, error={}", + config.getId(), config.getName(), config.getTransportType(), e.getMessage()); + } + } + + // 第二步：原子切换引用（volatile 保证其他线程立即可见） + ConcurrentHashMap oldCache = clientCache; + clientCache = newCache; + unavailableConfigs.clear(); + + // 第三步：关闭旧缓存中的客户端（不影响新请求） + for (Map.Entry entry : oldCache.entrySet()) { + try { + McpSyncClient client = entry.getValue(); + if (client != null) { + client.close(); + log.debug("已关闭旧 MCP 客户端: configId={}", entry.getKey()); + } + } catch (Exception e) { + log.error("关闭旧 MCP 客户端失败: configId={}, error={}", entry.getKey(), e.getMessage()); + } + } + + log.info("MCP 客户端刷新完成，当前缓存数量: {}", clientCache.size()); + } + + /** + * 获取指定配置的客户端（懒加载） + * 使用 computeIfAbsent 保证同一配置只创建一次客户端，避免并发竞态。 + * 如果缓存中不存在且不在不可用集合中，从 DB 读取配置后原子创建并缓存。 + * + * @param configId 配置ID + * @return MCP Client 实例，不存在或未启用返回 null + */ + public McpSyncClient getClient(Long configId) { + String key = configId.toString(); + + // 快速路径：缓存命中 + McpSyncClient cached = clientCache.get(key); + if (cached != null) { + return cached; + } + + // 快速路径：已知不可用的配置，直接跳过 + if (unavailableConfigs.contains(key)) { + return null; + } + + // 缓存未命中，使用 computeIfAbsent 保证同一 key 只创建一次客户端 + // ConcurrentHashMap.computeIfAbsent 对同一 key 加锁，避免并发线程重复创建 + // 注意：mapping function 不能返回 null（会抛 NPE），因此不可用的情况记录到 unavailableConfigs + McpServerConfig config = mcpServerConfigService.getConfigById(configId); + if (config == null) { + log.warn("MCP 配置不存在: id={}", configId); + unavailableConfigs.add(key); + return null; + } + if (!Boolean.TRUE.equals(config.getIsActive())) { + log.warn("MCP 配置未启用，跳过创建客户端: id={}, name={}", configId, config.getName()); + unavailableConfigs.add(key); + return null; + } + + // 配置存在且启用，通过 computeIfAbsent 原子创建（防止并发重复创建） + return clientCache.computeIfAbsent(key, k -> createClientDirectly(config)); + } + + /** + * 获取所有已启用的 MCP 工具描述 + * 遍历所有已缓存的客户端，调用 listTools() 获取真实的工具列表 + * + * @return 工具描述列表，每个元素包含 config_id、name、transport_type、description、tools + */ + public List> listAvailableTools() { + List> tools = new ArrayList<>(); + + log.info("🔍 listAvailableTools: clientCache 大小 = {}", clientCache.size()); + + if (clientCache.isEmpty()) { + log.warn("⚠️ clientCache 为空！MCP Server 可能未配置或连接失败。"); + log.warn(" 请检查：1) 是否添加了 MCP Server 配置 2) 配置是否已启用 3) 启动日志是否有连接错误"); + return tools; + } + + for (Map.Entry entry : clientCache.entrySet()) { + Long configId; + try { + configId = Long.parseLong(entry.getKey()); + } catch (NumberFormatException e) { + log.warn("缓存 key 格式异常: {}", entry.getKey()); + continue; + } + + McpServerConfig config = mcpServerConfigService.getConfigById(configId); + if (config == null || !Boolean.TRUE.equals(config.getIsActive())) { + continue; + } + + Map toolInfo = new LinkedHashMap<>(); + toolInfo.put("config_id", config.getId().toString()); + toolInfo.put("name", config.getName()); + toolInfo.put("transport_type", config.getTransportType()); + toolInfo.put("description", config.getDescription()); + + // 从 MCP Client 获取真实的工具列表 + List> toolList = new ArrayList<>(); + try { + McpSyncClient client = entry.getValue(); + McpSchema.ListToolsResult listToolsResult = client.listTools(); + if (listToolsResult != null && listToolsResult.tools() != null) { + log.info(" 📦 MCP Server [{}] 暴露了 {} 个工具", config.getName(), listToolsResult.tools().size()); + for (McpSchema.Tool tool : listToolsResult.tools()) { + Map toolMeta = new LinkedHashMap<>(); + toolMeta.put("name", tool.name()); + toolMeta.put("description", tool.description()); + // JsonSchema 对象序列化为 JSON 字符串，供 McpToolCallback 构建 ToolDefinition 使用 + if (tool.inputSchema() != null) { + toolMeta.put("inputSchema", ModelOptionsUtils.toJsonString(tool.inputSchema())); + } else { + toolMeta.put("inputSchema", "{}"); + } + toolList.add(toolMeta); + } + } + log.info(" 📦 MCP Server [{}] 获取到 {} 个工具", config.getName(), toolList.size()); + if (toolList.isEmpty()) { + log.warn(" ⚠️ MCP Server [{}] 未暴露任何工具，AI 模型将无法调用", config.getName()); + } + } catch (Exception e) { + log.error("获取 MCP 工具列表失败: serverId={}, name={}, error={}", + configId, config.getName(), e.getMessage()); + } + + toolInfo.put("tools", toolList); + tools.add(toolInfo); + } + + return tools; + } + + /** + * 定时健康检查：每 5 分钟执行一次 + * 遍历所有缓存的客户端，尝试 listTools() 探测连通性，记录状态到 healthCache + */ + @Scheduled(fixedDelay = 300000) + public void healthCheck() { + log.info("开始 MCP Server 定时健康检查，当前缓存数量: {}", clientCache.size()); + for (Map.Entry entry : clientCache.entrySet()) { + String key = entry.getKey(); + checkSingleClient(key, entry.getValue()); + } + log.info("MCP Server 定时健康检查完成"); + } + + /** + * 手动触发单个配置的健康检查 + * + * @param configId 配置ID + * @return 该配置的最新健康状态 + */ + public HealthStatus checkHealthForConfig(Long configId) { + String key = configId.toString(); + McpSyncClient client = clientCache.get(key); + if (client == null) { + HealthStatus status = HealthStatus.unknown("客户端未缓存或配置不存在"); + healthCache.put(key, status); + return status; + } + return checkSingleClient(key, client); + } + + /** + * 对单个客户端执行健康探测 + * 通过调用 listTools() 判断服务端是否存活 + * + * @param key 缓存 key（配置ID 字符串） + * @param client MCP 客户端实例 + * @return 健康状态 + */ + private HealthStatus checkSingleClient(String key, McpSyncClient client) { + long startTime = System.currentTimeMillis(); + try { + McpSchema.ListToolsResult result = client.listTools(); + long latencyMs = System.currentTimeMillis() - startTime; + int toolsCount = (result != null && result.tools() != null) ? result.tools().size() : 0; + HealthStatus status = HealthStatus.online(latencyMs); + healthCache.put(key, status); + log.debug("MCP Server 健康检查通过: configId={}, latencyMs={}, toolsCount={}", key, latencyMs, toolsCount); + return status; + } catch (Exception e) { + long latencyMs = System.currentTimeMillis() - startTime; + HealthStatus status = HealthStatus.offline(latencyMs, e.getMessage()); + healthCache.put(key, status); + log.warn("MCP Server 健康检查失败: configId={}, latencyMs={}, error={}", key, latencyMs, e.getMessage()); + return status; + } + } + + /** + * 获取指定配置的健康状态 + * + * @param configId 配置ID + * @return 健康状态，未检查过返回 UNKNOWN + */ + public HealthStatus getHealthStatus(Long configId) { + HealthStatus status = healthCache.get(configId.toString()); + if (status == null) { + return HealthStatus.unknown("尚未执行健康检查"); + } + return status; + } + + /** + * 获取所有已缓存配置的健康状态 + * + * @return Map：key = 配置ID 字符串，value = 健康状态 + */ + public Map getAllHealthStatus() { + Map result = new LinkedHashMap<>(); + // 遍历所有缓存的客户端，未在 healthCache 中的标记为 UNKNOWN + for (String key : clientCache.keySet()) { + HealthStatus status = healthCache.get(key); + if (status == null) { + status = HealthStatus.unknown("尚未执行健康检查"); + } + result.put(key, status); + } + return result; + } + + /** + * 关闭所有 MCP 客户端连接 + * 在应用关闭或手动刷新时调用 + */ + @PreDestroy + public void shutdown() { + log.info("关闭所有 MCP 客户端连接，当前数量: {}", clientCache.size()); + for (Map.Entry entry : clientCache.entrySet()) { + try { + McpSyncClient client = entry.getValue(); + if (client != null) { + client.close(); + log.debug("已关闭 MCP 客户端: configId={}", entry.getKey()); + } + } catch (Exception e) { + log.error("关闭 MCP 客户端失败: configId={}, error={}", entry.getKey(), e.getMessage()); + } + } + clientCache.clear(); + unavailableConfigs.clear(); + healthCache.clear(); + log.info("所有 MCP 客户端连接已关闭"); + } + + /** + * 根据配置创建 MCP 客户端（不负责缓存，由调用方决定是否缓存） + *

+ * SSE 模式：使用 HttpClientSseClientTransport 连接远程 MCP Server + * stdio 模式：使用 StdioClientTransport 启动本地 MCP Server 进程 + * + * @param config MCP Server 配置 + * @return 创建并初始化后的客户端实例，失败返回 null + */ + private McpSyncClient createClientDirectly(McpServerConfig config) { + try { + McpSyncClient client; + + if ("sse".equals(config.getTransportType())) { + // 创建 SSE 传输层：连接远程 MCP Server + // HttpClientSseClientTransport 会自动将 /sse 作为 SSE 端点附加到 baseUri + String serverUrl = config.getServerUrl().trim(); + // 如果 URL 以 /sse 结尾，去掉后缀作为 baseUri（transport 层会自动附加） + String baseUri = serverUrl.replaceAll("/sse/?$", ""); + HttpClientSseClientTransport transport = HttpClientSseClientTransport.builder(baseUri).build(); + + client = buildSyncClient(transport); + log.info("正在初始化 SSE MCP 客户端: id={}, url={}", config.getId(), baseUri); + + } else if ("stdio".equals(config.getTransportType())) { + // 创建 stdio 传输层：启动本地 MCP Server 进程 + String command = config.getCommand().trim(); + + // 解析参数列表（逗号分隔） + List argsList = new ArrayList<>(); + if (config.getArgs() != null && !config.getArgs().trim().isEmpty()) { + argsList = Arrays.asList(config.getArgs().split("\\s*,\\s*")); + } + + // 构建 ServerParameters + ServerParameters.Builder paramsBuilder = ServerParameters.builder(command); + if (!argsList.isEmpty()) { + paramsBuilder.args(argsList); + } + + // 环境变量（Map -> Map） + if (config.getEnvVars() != null && !config.getEnvVars().isEmpty()) { + Map envStr = new HashMap<>(); + config.getEnvVars().forEach((k, v) -> envStr.put(k, v != null ? v.toString() : "")); + paramsBuilder.env(envStr); + } + + StdioClientTransport transport = new StdioClientTransport(paramsBuilder.build()); + client = buildSyncClient(transport); + log.info("正在初始化 stdio MCP 客户端: id={}, command={}", config.getId(), command); + + } else { + log.error("不支持的传输类型: {}", config.getTransportType()); + return null; + } + + // 初始化连接（握手协议） + client.initialize(); + log.info("MCP 客户端初始化成功: id={}, name={}, transportType={}", + config.getId(), config.getName(), config.getTransportType()); + + return client; + + } catch (Exception e) { + log.error("创建 MCP 客户端失败: id={}, name={}, transportType={}, error={}", + config.getId(), config.getName(), config.getTransportType(), e.getMessage(), e); + return null; + } + } + + /** + * 构建 McpSyncClient 实例 + * 统一设置客户端信息和超时参数，超时时间从配置文件读取 + * + * @param transport 传输层实例（SSE 或 stdio） + * @return 构建的 McpSyncClient + */ + private McpSyncClient buildSyncClient(io.modelcontextprotocol.spec.McpClientTransport transport) { + Duration requestTimeout = parseDuration(requestTimeoutStr, Duration.ofSeconds(30)); + Duration initTimeout = parseDuration(initTimeoutStr, Duration.ofSeconds(15)); + + return io.modelcontextprotocol.client.McpClient.sync(transport) + .clientInfo(new McpSchema.Implementation("supportbot-mcp-client", "1.0.0")) + .requestTimeout(requestTimeout) + .initializationTimeout(initTimeout) + .build(); + } + + /** + * 解析 Duration 字符串，兼容 ISO-8601 格式（PT30S）和简化格式（30s、60s） + * 解析失败时返回默认值 + * + * @param durationStr Duration 字符串 + * @param defaultValue 解析失败时的默认值 + * @return 解析后的 Duration + */ + private Duration parseDuration(String durationStr, Duration defaultValue) { + if (durationStr == null || durationStr.isBlank()) { + return defaultValue; + } + try { + // 先尝试 ISO-8601 格式（如 PT30S、PT1M30S） + return Duration.parse(durationStr.trim()); + } catch (Exception e) { + // 尝试简化格式：纯数字秒（如 "30"）或带 "s" 后缀（如 "30s"） + String trimmed = durationStr.trim().toLowerCase(); + String numStr = trimmed.replaceAll("[^0-9.]+$", ""); + if (!numStr.isEmpty()) { + try { + long seconds = Long.parseLong(numStr); + return Duration.ofSeconds(seconds); + } catch (NumberFormatException ignored) { + // 继续尝试小数 + } + } + log.warn("无法解析 Duration 配置 '{}'，使用默认值: {}", durationStr, defaultValue); + return defaultValue; + } + } + + // ==================== 增量缓存管理（供 Controller 层 CRUD 操作使用） ==================== + + /** + * 新增配置时：创建客户端并加入缓存 + * 仅对新建的单个配置创建连接，不影响其他已有连接。 + * + * @param configId 新建配置的ID + */ + public void addClient(Long configId) { + String key = configId.toString(); + // 从不可用集合中移除（新建的配置可能之前被标记为不可用） + unavailableConfigs.remove(key); + + McpServerConfig config = mcpServerConfigService.getConfigById(configId); + if (config == null || !Boolean.TRUE.equals(config.getIsActive())) { + return; + } + McpSyncClient client = createClientDirectly(config); + if (client != null) { + clientCache.put(key, client); + log.info("增量添加 MCP 客户端: id={}, name={}", configId, config.getName()); + } + } + + /** + * 更新配置时：关闭旧客户端，重建新客户端 + * 仅对更新的单个配置重建连接，不影响其他已有连接。 + * + * @param configId 更新的配置ID + */ + public void rebuildClient(Long configId) { + String key = configId.toString(); + // 先移除旧客户端 + removeClient(configId); + + McpServerConfig config = mcpServerConfigService.getConfigById(configId); + if (config == null) { + unavailableConfigs.add(key); + return; + } + if (!Boolean.TRUE.equals(config.getIsActive())) { + unavailableConfigs.add(key); + return; + } + + unavailableConfigs.remove(key); + McpSyncClient client = createClientDirectly(config); + if (client != null) { + clientCache.put(key, client); + log.info("增量重建 MCP 客户端: id={}, name={}", configId, config.getName()); + } + } + + /** + * 禁用配置时：关闭并移除客户端，标记为不可用 + * + * @param configId 禁用的配置ID + */ + public void disableClient(Long configId) { + removeClient(configId); + unavailableConfigs.add(configId.toString()); + } + + /** + * 移除指定配置的客户端缓存 + * 配置删除或禁用时调用 + * + * @param configId 配置ID + */ + public void removeClient(Long configId) { + String key = configId.toString(); + McpSyncClient removed = clientCache.remove(key); + if (removed != null) { + try { + removed.close(); + } catch (Exception e) { + log.error("关闭被移除的 MCP 客户端失败: configId={}, error={}", configId, e.getMessage()); + } + log.info("已移除并关闭 MCP 客户端缓存: configId={}", configId); + } + } + + /** + * 测试指定配置的 MCP Server 连接 + * 创建临时客户端 -> initialize -> listTools -> close，不影响正常缓存 + * + * @param configId 配置ID + * @return 测试结果 Map：success / latencyMs / toolsCount / errorMessage + */ + public Map testConnection(Long configId) { + McpServerConfig config = mcpServerConfigService.getConfigById(configId); + if (config == null) { + Map result = new LinkedHashMap<>(); + result.put("success", false); + result.put("latencyMs", 0); + result.put("toolsCount", 0); + result.put("errorMessage", "配置不存在: id=" + configId); + return result; + } + + McpSyncClient tempClient = null; + long startTime = System.currentTimeMillis(); + try { + // 创建临时客户端（不加入缓存），复用 createClientDirectly 的传输层构建逻辑 + tempClient = createClientDirectly(config); + if (tempClient == null) { + Map result = new LinkedHashMap<>(); + result.put("success", false); + result.put("latencyMs", System.currentTimeMillis() - startTime); + result.put("toolsCount", 0); + result.put("errorMessage", "客户端创建失败，请检查配置"); + return result; + } + + // 获取工具列表 + McpSchema.ListToolsResult listToolsResult = tempClient.listTools(); + int toolsCount = (listToolsResult != null && listToolsResult.tools() != null) + ? listToolsResult.tools().size() : 0; + + long latencyMs = System.currentTimeMillis() - startTime; + log.info("MCP 连接测试成功: id={}, name={}, latencyMs={}, toolsCount={}", + configId, config.getName(), latencyMs, toolsCount); + + Map result = new LinkedHashMap<>(); + result.put("success", true); + result.put("latencyMs", latencyMs); + result.put("toolsCount", toolsCount); + result.put("errorMessage", null); + return result; + + } catch (Exception e) { + long latencyMs = System.currentTimeMillis() - startTime; + log.warn("MCP 连接测试失败: id={}, name={}, latencyMs={}, error={}", + configId, config.getName(), latencyMs, e.getMessage()); + + Map result = new LinkedHashMap<>(); + result.put("success", false); + result.put("latencyMs", latencyMs); + result.put("toolsCount", 0); + result.put("errorMessage", e.getMessage()); + return result; + + } finally { + // 确保关闭临时客户端 + if (tempClient != null) { + try { + tempClient.close(); + } catch (Exception e) { + log.debug("关闭临时测试客户端失败: configId={}, error={}", configId, e.getMessage()); + } + } + } + } + + /** + * 获取当前缓存的客户端数量 + * + * @return 缓存数量 + */ + public int getCacheSize() { + return clientCache.size(); + } +} diff --git a/src/main/java/com/wok/supportbot/config/McpRequestInterceptor.java b/src/main/java/com/wok/supportbot/config/McpRequestInterceptor.java new file mode 100644 index 0000000..c482c3c --- /dev/null +++ b/src/main/java/com/wok/supportbot/config/McpRequestInterceptor.java @@ -0,0 +1,24 @@ +package com.wok.supportbot.config; + +import com.wok.supportbot.mcp.McpToolCallback; +import jakarta.servlet.http.HttpServletRequest; +import jakarta.servlet.http.HttpServletResponse; +import lombok.extern.slf4j.Slf4j; +import org.springframework.stereotype.Component; +import org.springframework.web.servlet.HandlerInterceptor; + +/** + * MCP 请求拦截器 + * 在每个 HTTP 请求开始时重置 McpToolCallback 的 ThreadLocal 调用轮次计数器， + * 防止 Tomcat 线程池复用导致计数器累积，避免内存泄漏和限流失效。 + */ +@Component +@Slf4j +public class McpRequestInterceptor implements HandlerInterceptor { + + @Override + public boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler) { + McpToolCallback.resetCallRounds(); + return true; + } +} diff --git a/src/main/java/com/wok/supportbot/config/McpWebMvcConfig.java b/src/main/java/com/wok/supportbot/config/McpWebMvcConfig.java new file mode 100644 index 0000000..95b7bfc --- /dev/null +++ b/src/main/java/com/wok/supportbot/config/McpWebMvcConfig.java @@ -0,0 +1,24 @@ +package com.wok.supportbot.config; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Configuration; +import org.springframework.web.servlet.config.annotation.InterceptorRegistry; +import org.springframework.web.servlet.config.annotation.WebMvcConfigurer; + +/** + * MCP 拦截器注册配置 + * 将 McpRequestInterceptor 注册到所有请求路径， + * 确保每次请求开始时重置 MCP 工具调用轮次计数器。 + */ +@Configuration +public class McpWebMvcConfig implements WebMvcConfigurer { + + @Autowired + private McpRequestInterceptor mcpRequestInterceptor; + + @Override + public void addInterceptors(InterceptorRegistry registry) { + registry.addInterceptor(mcpRequestInterceptor) + .addPathPatterns("/**"); + } +} diff --git a/src/main/java/com/wok/supportbot/controller/AiController.java b/src/main/java/com/wok/supportbot/controller/AiController.java index f8cbf0b..c8355b4 100644 --- a/src/main/java/com/wok/supportbot/controller/AiController.java +++ b/src/main/java/com/wok/supportbot/controller/AiController.java @@ -61,7 +61,7 @@ public class AiController { AccountRoleContext context = resolveAccountRole(accountId, roleId); bindConversation(chatId, context); RoleScope scope = customerServiceRoleService.getRoleScope(context.roleId()); - return assistantApp.doChat(message, chatId, resolveSystemPrompt(scope, systemPrompt)); + return assistantApp.doChat(message, chatId, resolveSystemPrompt(scope, systemPrompt), scope.allowedMcpTools()); } /** @@ -77,7 +77,7 @@ public class AiController { AccountRoleContext context = resolveAccountRole(accountId, roleId); bindConversation(chatId, context); RoleScope scope = customerServiceRoleService.getRoleScope(context.roleId()); - return assistantApp.doChatByStream(message, chatId, resolveSystemPrompt(scope, systemPrompt)); + return assistantApp.doChatByStream(message, chatId, resolveSystemPrompt(scope, systemPrompt), scope.allowedMcpTools()); } /** @@ -93,7 +93,7 @@ public class AiController { AccountRoleContext context = resolveAccountRole(accountId, roleId); bindConversation(chatId, context); RoleScope scope = customerServiceRoleService.getRoleScope(context.roleId()); - return assistantApp.doChatByStream(message, chatId, resolveSystemPrompt(scope, systemPrompt)) + return assistantApp.doChatByStream(message, chatId, resolveSystemPrompt(scope, systemPrompt), scope.allowedMcpTools()) .map(chunk -> ServerSentEvent.builder() .data(chunk) .build()); @@ -115,7 +115,7 @@ public class AiController { // 创建一个超时时间较长的 SseEmitter SseEmitter sseEmitter = new SseEmitter(180000L); // 3 分钟超时 // 获取 Flux 响应式数据流并且直接通过订阅推送给 SseEmitter - assistantApp.doChatByStream(message, chatId, resolveSystemPrompt(scope, systemPrompt)) + assistantApp.doChatByStream(message, chatId, resolveSystemPrompt(scope, systemPrompt), scope.allowedMcpTools()) .subscribe(chunk -> { try { sseEmitter.send(chunk); @@ -147,12 +147,12 @@ public class AiController { String sys = resolveSystemPrompt(scope, systemPrompt); // 严格隔离：未授权任何知识库的角色退回普通对话，绝不检索 KB if (isKbDenied(scope) || shouldBypassKnowledgeRetrieval(message)) { - return assistantApp.doChat(message, chatId, sys); + return assistantApp.doChat(message, chatId, sys, scope.allowedMcpTools()); } try { return assistantApp.doChatWithRagStrategy( message, chatId, normalizeStrategy(rewriteStrategy), - resolveCategoryIds(scope, categoryId, categoryIds), sys); + resolveCategoryIds(scope, categoryId, categoryIds), sys, scope.allowedMcpTools()); } catch (Exception e) { log.error("RAG 对话失败 [strategy={}, chatId={}]: {}", rewriteStrategy, chatId, e.getMessage(), e); return "抱歉，知识库检索出现异常，请稍后重试。"; @@ -179,11 +179,11 @@ public class AiController { String sys = resolveSystemPrompt(scope, systemPrompt); // 严格隔离：未授权任何知识库的角色退回普通流式对话，绝不检索 KB if (isKbDenied(scope) || shouldBypassKnowledgeRetrieval(message)) { - return assistantApp.doChatByStream(message, chatId, sys); + return assistantApp.doChatByStream(message, chatId, sys, scope.allowedMcpTools()); } return assistantApp.doChatWithRagStrategyByStream( message, chatId, normalizeStrategy(rewriteStrategy), - resolveCategoryIds(scope, categoryId, categoryIds), sys); + resolveCategoryIds(scope, categoryId, categoryIds), sys, scope.allowedMcpTools()); } /** diff --git a/src/main/java/com/wok/supportbot/controller/CustomerServiceRoleController.java b/src/main/java/com/wok/supportbot/controller/CustomerServiceRoleController.java index abc8439..28170e2 100644 --- a/src/main/java/com/wok/supportbot/controller/CustomerServiceRoleController.java +++ b/src/main/java/com/wok/supportbot/controller/CustomerServiceRoleController.java @@ -93,6 +93,35 @@ public class CustomerServiceRoleController { )); } + /** + * 更新角色的 MCP 工具权限。 + * body: { "allowedMcpTools": ["*"] } 或 { "allowedMcpTools": ["mcp_get_weather", "mcp_query_db"] } + */ + @SuppressWarnings("unchecked") + @PutMapping("/role/{id}/mcp-tools") + @PreAuthorize("hasRole('admin')") + public ResponseEntity> updateRoleMcpTools( + @PathVariable("id") Long roleId, + @RequestBody Map body) { + try { + Object raw = body.get("allowedMcpTools"); + List allowedMcpTools; + if (raw instanceof List list) { + allowedMcpTools = list.stream() + .filter(Objects::nonNull) + .map(Objects::toString) + .distinct() + .toList(); + } else { + allowedMcpTools = List.of(); + } + customerServiceRoleService.updateRoleMcpTools(roleId, allowedMcpTools); + return ResponseEntity.ok(Map.of("success", true, "message", "MCP 工具权限更新成功")); + } catch (Exception e) { + return ResponseEntity.status(500).body(Map.of("success", false, "message", "更新失败：" + e.getMessage())); + } + } + private List parseCategoryIds(Object rawCategoryIds) { if (!(rawCategoryIds instanceof List list)) { return List.of(); diff --git a/src/main/java/com/wok/supportbot/controller/McpServerConfigController.java b/src/main/java/com/wok/supportbot/controller/McpServerConfigController.java new file mode 100644 index 0000000..cb717bd --- /dev/null +++ b/src/main/java/com/wok/supportbot/controller/McpServerConfigController.java @@ -0,0 +1,348 @@ +package com.wok.supportbot.controller; + +import com.wok.supportbot.config.McpClientManager; +import com.wok.supportbot.entity.McpServerConfig; +import com.wok.supportbot.service.McpServerConfigService; +import lombok.extern.slf4j.Slf4j; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.http.ResponseEntity; +import org.springframework.security.access.prepost.PreAuthorize; +import org.springframework.web.bind.annotation.*; + +import java.util.LinkedHashMap; +import java.util.Map; + +/** + * MCP Server 配置管理控制器 + * 提供 MCP Server 配置的增删改查、启用/禁用、连接测试、缓存刷新等 API + */ +@RestController +@Slf4j +public class McpServerConfigController { + + @Autowired + private McpServerConfigService mcpServerConfigService; + + @Autowired + private McpClientManager mcpClientManager; + + // ==================== 分页列表 ==================== + + /** + * 获取 MCP Server 配置列表（分页） + * + * @param page 页码（默认1） + * @param size 每页大小（默认10） + * @return 分页配置列表 + */ + @GetMapping("/mcp-server/list") + @PreAuthorize("hasRole('admin')") + public ResponseEntity> listConfigs( + @RequestParam(defaultValue = "1") int page, + @RequestParam(defaultValue = "10") int size) { + try { + Map result = mcpServerConfigService.listConfigs(page, size); + Map data = new LinkedHashMap<>(); + data.put("success", true); + data.put("data", result.get("records")); + data.put("total", result.get("total")); + data.put("page", result.get("page")); + data.put("size", result.get("size")); + data.put("pages", result.get("pages")); + return ResponseEntity.ok(data); + } catch (Exception e) { + return ResponseEntity.status(500).body(Map.of( + "success", false, + "message", "查询失败：" + e.getMessage() + )); + } + } + + // ==================== 单条详情 ==================== + + /** + * 获取单条 MCP Server 配置详情 + * + * @param id 配置ID + * @return 配置详情 + */ + @GetMapping("/mcp-server/{id}") + @PreAuthorize("hasRole('admin')") + public ResponseEntity> getConfigDetail(@PathVariable("id") Long id) { + try { + Map config = mcpServerConfigService.getConfigDetail(id); + if (config == null) { + return ResponseEntity.status(404).body(Map.of( + "success", false, + "message", "配置不存在" + )); + } + return ResponseEntity.ok(Map.of( + "success", true, + "data", config + )); + } catch (Exception e) { + return ResponseEntity.status(500).body(Map.of( + "success", false, + "message", "查询失败：" + e.getMessage() + )); + } + } + + // ==================== 新建配置 ==================== + + /** + * 新建 MCP Server 配置 + * + * @param config 配置对象 + * @return 创建结果 + */ + @PostMapping("/mcp-server") + @PreAuthorize("hasRole('admin')") + public ResponseEntity> createConfig(@RequestBody McpServerConfig config) { + try { + Map created = mcpServerConfigService.createConfig(config); + // 增量操作：仅为新建配置创建客户端，不影响其他已有连接 + Object idObj = created.get("id"); + if (idObj != null) { + try { + mcpClientManager.addClient(Long.parseLong(idObj.toString())); + } catch (Exception e) { + log.warn("新建配置后创建客户端失败（不影响配置保存）: {}", e.getMessage()); + } + } + return ResponseEntity.ok(Map.of( + "success", true, + "data", created, + "message", "配置创建成功" + )); + } catch (IllegalArgumentException e) { + return ResponseEntity.badRequest().body(Map.of( + "success", false, + "message", e.getMessage() + )); + } catch (Exception e) { + return ResponseEntity.status(500).body(Map.of( + "success", false, + "message", "创建失败：" + e.getMessage() + )); + } + } + + // ==================== 更新配置 ==================== + + /** + * 更新 MCP Server 配置 + * + * @param id 配置ID + * @param config 更新内容 + * @return 更新结果 + */ + @PutMapping("/mcp-server/{id}") + @PreAuthorize("hasRole('admin')") + public ResponseEntity> updateConfig( + @PathVariable("id") Long id, + @RequestBody McpServerConfig config) { + try { + Map updated = mcpServerConfigService.updateConfig(id, config); + // 增量操作：仅重建更新的配置客户端，不影响其他已有连接 + mcpClientManager.rebuildClient(id); + return ResponseEntity.ok(Map.of( + "success", true, + "data", updated, + "message", "配置更新成功" + )); + } catch (RuntimeException e) { + return ResponseEntity.badRequest().body(Map.of( + "success", false, + "message", e.getMessage() + )); + } catch (Exception e) { + return ResponseEntity.status(500).body(Map.of( + "success", false, + "message", "更新失败：" + e.getMessage() + )); + } + } + + // ==================== 删除配置 ==================== + + /** + * 删除 MCP Server 配置（逻辑删除） + * + * @param id 配置ID + * @return 删除结果 + */ + @DeleteMapping("/mcp-server/{id}") + @PreAuthorize("hasRole('admin')") + public ResponseEntity> deleteConfig(@PathVariable("id") Long id) { + try { + mcpServerConfigService.deleteConfig(id); + // 增量操作：仅移除被删除的配置客户端，不影响其他已有连接 + mcpClientManager.removeClient(id); + return ResponseEntity.ok(Map.of( + "success", true, + "message", "配置删除成功" + )); + } catch (RuntimeException e) { + return ResponseEntity.badRequest().body(Map.of( + "success", false, + "message", e.getMessage() + )); + } catch (Exception e) { + return ResponseEntity.status(500).body(Map.of( + "success", false, + "message", "删除失败：" + e.getMessage() + )); + } + } + + // ==================== 启用/禁用 ==================== + + /** + * 切换配置启用/禁用状态 + * + * @param id 配置ID + * @param body 请求体：{active: true/false} + * @return 操作结果 + */ + @PutMapping("/mcp-server/{id}/toggle") + @PreAuthorize("hasRole('admin')") + public ResponseEntity> toggleActive( + @PathVariable("id") Long id, + @RequestBody Map body) { + try { + Boolean active = body.get("active"); + if (active == null) { + return ResponseEntity.badRequest().body(Map.of( + "success", false, + "message", "active 参数不能为空" + )); + } + mcpServerConfigService.toggleActive(id, active); + // 增量操作：启用时重建客户端，禁用时移除客户端 + if (active) { + mcpClientManager.rebuildClient(id); + } else { + mcpClientManager.disableClient(id); + } + return ResponseEntity.ok(Map.of( + "success", true, + "message", active ? "配置已启用" : "配置已禁用" + )); + } catch (RuntimeException e) { + return ResponseEntity.badRequest().body(Map.of( + "success", false, + "message", e.getMessage() + )); + } catch (Exception e) { + return ResponseEntity.status(500).body(Map.of( + "success", false, + "message", "操作失败：" + e.getMessage() + )); + } + } + + // ==================== 测试连接 ==================== + + /** + * 测试 MCP Server 连接 + * 创建临时客户端进行握手 + 工具列表查询，不影响正常缓存 + * + * @param id 配置ID + * @return 测试结果（success / latencyMs / toolsCount / errorMessage） + */ + @PostMapping("/mcp-server/{id}/test") + @PreAuthorize("hasRole('admin')") + public ResponseEntity> testConnection(@PathVariable("id") Long id) { + try { + Map result = mcpClientManager.testConnection(id); + return ResponseEntity.ok(result); + } catch (Exception e) { + return ResponseEntity.status(500).body(Map.of( + "success", false, + "message", "测试失败：" + e.getMessage() + )); + } + } + + // ==================== 刷新连接 ==================== + + /** + * 刷新所有 MCP 客户端连接 + * 清空现有缓存并重新建立所有已启用配置的客户端连接 + * + * @return 刷新结果 + */ + @PostMapping("/mcp-server/refresh") + @PreAuthorize("hasRole('admin')") + public ResponseEntity> refreshAll() { + try { + mcpClientManager.refreshAll(); + return ResponseEntity.ok(Map.of( + "success", true, + "message", "MCP 客户端连接已刷新" + )); + } catch (Exception e) { + return ResponseEntity.status(500).body(Map.of( + "success", false, + "message", "刷新失败：" + e.getMessage() + )); + } + } + + // ==================== 健康检查 ==================== + + /** + * 获取所有 MCP Server 的健康状态 + * 返回各配置的在线/离线状态、延迟、最后检查时间等信息 + * + * @return 所有 MCP Server 的健康状态 + */ + @GetMapping("/mcp-server/health") + @PreAuthorize("hasRole('admin')") + public ResponseEntity> getAllHealthStatus() { + try { + Map healthMap = mcpClientManager.getAllHealthStatus(); + Map data = new LinkedHashMap<>(); + data.put("success", true); + data.put("data", healthMap); + return ResponseEntity.ok(data); + } catch (Exception e) { + return ResponseEntity.status(500).body(Map.of( + "success", false, + "message", "获取健康状态失败：" + e.getMessage() + )); + } + } + + /** + * 手动触发单个 MCP Server 的健康检查 + * 对指定配置的客户端执行 listTools 探测，返回最新健康状态 + * + * @param id 配置ID + * @return 该配置的健康状态 + */ + @PostMapping("/mcp-server/{id}/health-check") + @PreAuthorize("hasRole('admin')") + public ResponseEntity> checkSingleHealth(@PathVariable("id") Long id) { + try { + McpClientManager.HealthStatus status = mcpClientManager.checkHealthForConfig(id); + return ResponseEntity.ok(Map.of( + "success", true, + "data", Map.of( + "configId", id.toString(), + "status", status.status(), + "latencyMs", status.latencyMs(), + "lastCheckTime", status.lastCheckTime().toString(), + "errorMessage", status.errorMessage() != null ? status.errorMessage() : "" + ) + )); + } catch (Exception e) { + return ResponseEntity.status(500).body(Map.of( + "success", false, + "message", "健康检查失败：" + e.getMessage() + )); + } + } +} diff --git a/src/main/java/com/wok/supportbot/converter/MessageConverter.java b/src/main/java/com/wok/supportbot/converter/MessageConverter.java index 84e3a22..d4b93b8 100644 --- a/src/main/java/com/wok/supportbot/converter/MessageConverter.java +++ b/src/main/java/com/wok/supportbot/converter/MessageConverter.java @@ -3,42 +3,136 @@ package com.wok.supportbot.converter; import com.wok.supportbot.entity.ChatMessage; import org.springframework.ai.chat.messages.*; -import java.util.List; -import java.util.Map; +import java.util.*; /** - * @Classname MessageConverter - * @Description - * @Version 1.0.0 - * @Date 2025/06/28 13:30 - * @Author lyx + * 消息转换器：Message <-> ChatMessage + * 支持 USER / ASSISTANT / SYSTEM / TOOL 四种消息类型 + * 其中 ASSISTANT 的 toolCalls 和 TOOL 的 responses 通过 metadata JSONB 字段持久化 */ public class MessageConverter { + /** metadata 中存储 toolCalls 的内部键名 */ + private static final String META_KEY_TOOL_CALLS = "_toolCalls"; + /** metadata 中存储 toolResponses 的内部键名 */ + private static final String META_KEY_TOOL_RESPONSES = "_toolResponses"; + /** - * 将 Message 转换为 ChatMessage + * 将 Spring AI Message 转换为数据库实体 ChatMessage + * 对于 ASSISTANT 消息，如果包含 toolCalls 则序列化到 metadata + * 对于 TOOL 消息，将 responses 序列化到 metadata */ public static ChatMessage toChatMessage(Message message, String conversationId) { + Map metadata = message.getMetadata() != null + ? new HashMap<>(message.getMetadata()) + : new HashMap<>(); + + // 处理 AssistantMessage 的 toolCalls + if (message instanceof AssistantMessage assistant && assistant.hasToolCalls()) { + List> toolCallsJson = new ArrayList<>(); + for (AssistantMessage.ToolCall tc : assistant.getToolCalls()) { + Map tcMap = new LinkedHashMap<>(); + tcMap.put("id", tc.id()); + tcMap.put("type", tc.type()); + tcMap.put("name", tc.name()); + tcMap.put("arguments", tc.arguments()); + toolCallsJson.add(tcMap); + } + metadata.put(META_KEY_TOOL_CALLS, toolCallsJson); + } + + // 处理 ToolResponseMessage 的 responses + if (message instanceof ToolResponseMessage trm) { + List> responsesJson = new ArrayList<>(); + for (ToolResponseMessage.ToolResponse tr : trm.getResponses()) { + Map trMap = new LinkedHashMap<>(); + trMap.put("id", tr.id()); + trMap.put("name", tr.name()); + trMap.put("responseData", tr.responseData()); + responsesJson.add(trMap); + } + metadata.put(META_KEY_TOOL_RESPONSES, responsesJson); + } + return ChatMessage.builder() .conversationId(conversationId) .messageType(message.getMessageType()) .content(message.getText()) - .metadata(message.getMetadata()) + .metadata(metadata) .build(); } /** - * 将 ChatMessage 转换为 Message + * 将数据库实体 ChatMessage 转换为 Spring AI Message + * 根据 messageType 分发到具体子类，反序列化 toolCalls / responses */ + @SuppressWarnings("unchecked") public static Message toMessage(ChatMessage chatMessage) { MessageType messageType = chatMessage.getMessageType(); String text = chatMessage.getContent(); Map metadata = chatMessage.getMetadata(); + + // 构建不含内部键的干净 metadata（供 Message 对象使用） + Map cleanMetadata = metadata != null + ? new HashMap<>(metadata) + : new HashMap<>(); + return switch (messageType) { case USER -> new UserMessage(text); - case ASSISTANT -> new AssistantMessage(text, metadata); + case ASSISTANT -> { + List toolCalls = extractToolCalls(cleanMetadata); + yield new AssistantMessage(text, cleanMetadata, toolCalls); + } case SYSTEM -> new SystemMessage(text); - case TOOL -> new ToolResponseMessage(List.of(), metadata); + case TOOL -> { + List responses = extractToolResponses(cleanMetadata); + yield new ToolResponseMessage(responses, cleanMetadata); + } }; } + + /** + * 从 metadata 中提取并移除 _toolCalls 键，反序列化为 ToolCall 列表 + */ + @SuppressWarnings("unchecked") + private static List extractToolCalls(Map metadata) { + Object raw = metadata.remove(META_KEY_TOOL_CALLS); + if (!(raw instanceof List list) || list.isEmpty()) { + return List.of(); + } + List result = new ArrayList<>(); + for (Object item : list) { + if (item instanceof Map) { + Map map = (Map) item; + String id = String.valueOf(map.getOrDefault("id", "")); + String type = String.valueOf(map.getOrDefault("type", "function")); + String name = String.valueOf(map.getOrDefault("name", "")); + String arguments = String.valueOf(map.getOrDefault("arguments", "{}")); + result.add(new AssistantMessage.ToolCall(id, type, name, arguments)); + } + } + return result; + } + + /** + * 从 metadata 中提取并移除 _toolResponses 键，反序列化为 ToolResponse 列表 + */ + @SuppressWarnings("unchecked") + private static List extractToolResponses(Map metadata) { + Object raw = metadata.remove(META_KEY_TOOL_RESPONSES); + if (!(raw instanceof List list) || list.isEmpty()) { + return List.of(); + } + List result = new ArrayList<>(); + for (Object item : list) { + if (item instanceof Map) { + Map map = (Map) item; + String id = String.valueOf(map.getOrDefault("id", "")); + String name = String.valueOf(map.getOrDefault("name", "")); + String responseData = String.valueOf(map.getOrDefault("responseData", "")); + result.add(new ToolResponseMessage.ToolResponse(id, name, responseData)); + } + } + return result; + } } diff --git a/src/main/java/com/wok/supportbot/dao/McpServerConfigMapper.java b/src/main/java/com/wok/supportbot/dao/McpServerConfigMapper.java new file mode 100644 index 0000000..da2c003 --- /dev/null +++ b/src/main/java/com/wok/supportbot/dao/McpServerConfigMapper.java @@ -0,0 +1,12 @@ +package com.wok.supportbot.dao; + +import com.baomidou.mybatisplus.core.mapper.BaseMapper; +import com.wok.supportbot.entity.McpServerConfig; +import org.apache.ibatis.annotations.Mapper; + +/** + * MCP Server 配置 Mapper - 继承 MyBatis Plus BaseMapper，自带 CRUD + */ +@Mapper +public interface McpServerConfigMapper extends BaseMapper { +} diff --git a/src/main/java/com/wok/supportbot/entity/McpServerConfig.java b/src/main/java/com/wok/supportbot/entity/McpServerConfig.java new file mode 100644 index 0000000..6abf8c9 --- /dev/null +++ b/src/main/java/com/wok/supportbot/entity/McpServerConfig.java @@ -0,0 +1,104 @@ +package com.wok.supportbot.entity; + +import com.baomidou.mybatisplus.annotation.*; +import com.fasterxml.jackson.databind.annotation.JsonSerialize; +import com.fasterxml.jackson.databind.ser.std.ToStringSerializer; +import com.wok.supportbot.handler.PostgresJsonTypeHandler; +import lombok.AllArgsConstructor; +import lombok.Builder; +import lombok.Data; +import lombok.NoArgsConstructor; + +import java.io.Serial; +import java.io.Serializable; +import java.util.Date; +import java.util.Map; + +/** + * MCP Server 配置表 - 管理外部 MCP Server 连接，支持 stdio 和 sse 两种传输模式 + */ +@Data +@Builder +@AllArgsConstructor +@NoArgsConstructor +@TableName(value = "mcp_server_config", autoResultMap = true) +public class McpServerConfig implements Serializable { + + @Serial + @TableField(exist = false) + private static final long serialVersionUID = 1L; + + /** + * 主键ID（雪花算法） + */ + @TableId(value = "id", type = IdType.ASSIGN_ID) + @JsonSerialize(using = ToStringSerializer.class) + private Long id; + + /** + * 配置名称（如"天气查询服务"） + */ + @TableField("name") + private String name; + + /** + * 传输类型: stdio（标准输入输出）/ sse（Server-Sent Events） + */ + @TableField("transport_type") + private String transportType; + + /** + * SSE 模式的 MCP Server URL + */ + @TableField("server_url") + private String serverUrl; + + /** + * stdio 模式的启动命令 + */ + @TableField("command") + private String command; + + /** + * stdio 模式的命令参数（多个用逗号分隔） + */ + @TableField("args") + private String args; + + /** + * 环境变量（JSONB，键值对形式） + */ + @TableField(value = "env_vars", typeHandler = PostgresJsonTypeHandler.class) + private Map envVars; + + /** + * 描述说明 + */ + @TableField("description") + private String description; + + /** + * 是否启用 + */ + @TableField("is_active") + private Boolean isActive; + + /** + * 创建时间 + */ + @TableField(value = "create_time", fill = FieldFill.INSERT) + private Date createTime; + + /** + * 更新时间 + */ + @TableField(value = "update_time", fill = FieldFill.INSERT_UPDATE) + private Date updateTime; + + /** + * 删除标志 - false:未删除, true:已删除(逻辑删除) + */ + @TableField("is_delete") + @TableLogic + private boolean isDelete; +} diff --git a/src/main/java/com/wok/supportbot/mcp/McpToolCallback.java b/src/main/java/com/wok/supportbot/mcp/McpToolCallback.java new file mode 100644 index 0000000..2351c40 --- /dev/null +++ b/src/main/java/com/wok/supportbot/mcp/McpToolCallback.java @@ -0,0 +1,265 @@ +package com.wok.supportbot.mcp; + +import com.fasterxml.jackson.core.type.TypeReference; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.wok.supportbot.config.McpClientManager; +import io.modelcontextprotocol.client.McpSyncClient; +import io.modelcontextprotocol.spec.McpSchema; +import lombok.extern.slf4j.Slf4j; +import org.springframework.ai.model.ModelOptionsUtils; +import org.springframework.ai.tool.ToolCallback; +import org.springframework.ai.tool.definition.DefaultToolDefinition; +import org.springframework.ai.tool.definition.ToolDefinition; + +import java.util.ArrayList; +import java.util.List; +import java.util.Map; + +/** + * MCP Tool -> Spring AI ToolCallback 适配器 + * 将单个 MCP Server 暴露的工具适配为 Spring AI 可调用的 ToolCallback。 + *

+ * 工作流程： + * 1. ChatClient 在对话时通过 getToolDefinition() 获取工具元数据（名称、描述、参数 Schema） + * 2. 当 AI 模型决定调用该工具时，Spring AI 框架自动调用 call() 方法 + * 3. call() 内部通过 McpClientManager 获取对应的 MCP Client，转发调用到远程 MCP Server + * 4. 工具调用事件通过 ThreadLocal 收集器记录，供 SSE 流式输出使用 + */ +@Slf4j +public class McpToolCallback implements ToolCallback { + + private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper(); + + // ==================== 工具调用事件收集 ==================== + + /** + * 工具调用事件记录 + */ + public record ToolCallEvent(String tool, String input, String result, long latencyMs, boolean error) {} + + /** + * 线程级事件收集器：在同一请求线程中收集所有工具调用事件 + * SSE 流式输出完成后，从这里取出事件发送给前端 + */ + private static final ThreadLocal> EVENTS = ThreadLocal.withInitial(ArrayList::new); + + /** + * 获取当前线程收集的所有工具调用事件，并清空收集器 + */ + public static List drainEvents() { + List events = EVENTS.get(); + List copy = new ArrayList<>(events); + events.clear(); + return copy; + } + + /** + * 重置事件收集器（每个新请求开始时调用） + */ + public static void resetEvents() { + EVENTS.remove(); + } + + /** + * MCP Server 配置 ID（用于路由调用到正确的 MCP Client） + */ + private final Long mcpServerConfigId; + + /** + * MCP Client Manager（用于获取实际的 MCP Client 实例） + */ + private final McpClientManager mcpClientManager; + + /** + * 工具名称（已加 mcp_ 前缀，避免与内置工具冲突） + */ + private final String toolName; + + /** + * MCP Server 上的原始工具名称（无前缀，用于发送给 MCP Server 调用） + */ + private final String originalToolName; + + /** + * 工具描述（供 AI 模型理解工具用途） + */ + private final String toolDescription; + + /** + * 工具输入参数的 JSON Schema（供 AI 模型生成合法的调用参数） + */ + private final String inputSchema; + + /** + * 单次对话请求中最大允许的调用轮次，防止无限循环调用 + */ + private final int maxCallRounds; + + /** + * 线程级调用轮次计数器，每个请求线程独立计数 + */ + private static final ThreadLocal CALL_ROUNDS = ThreadLocal.withInitial(() -> 0); + + /** + * 构造 MCP 工具适配器（含最大调用轮次限制） + * + * @param mcpServerConfigId MCP Server 配置 ID + * @param toolName 工具名称（含 mcp_ 前缀，用于 Spring AI 注册） + * @param originalToolName MCP Server 上的原始工具名称（无前缀，用于实际调用） + * @param toolDescription 工具描述 + * @param inputSchema 输入参数的 JSON Schema + * @param mcpClientManager MCP 客户端管理器 + * @param maxCallRounds 单次请求中最大允许的调用轮次 + */ + public McpToolCallback(Long mcpServerConfigId, String toolName, String originalToolName, + String toolDescription, String inputSchema, McpClientManager mcpClientManager, + int maxCallRounds) { + this.mcpServerConfigId = mcpServerConfigId; + this.toolName = toolName; + this.originalToolName = originalToolName; + this.toolDescription = toolDescription; + this.inputSchema = inputSchema; + this.mcpClientManager = mcpClientManager; + this.maxCallRounds = maxCallRounds; + } + + /** + * 构造 MCP 工具适配器（使用默认最大调用轮次 5） + */ + public McpToolCallback(Long mcpServerConfigId, String toolName, String originalToolName, + String toolDescription, String inputSchema, McpClientManager mcpClientManager) { + this(mcpServerConfigId, toolName, originalToolName, toolDescription, inputSchema, mcpClientManager, 5); + } + + /** + * 兼容旧构造方式（toolName 直接用于调用，不加前缀区分） + */ + public McpToolCallback(Long mcpServerConfigId, String toolName, String toolDescription, + String inputSchema, McpClientManager mcpClientManager) { + this(mcpServerConfigId, toolName, toolName, toolDescription, inputSchema, mcpClientManager, 5); + } + + /** + * 返回工具定义，供 Spring AI 框架注册到 ChatClient + * AI 模型通过此定义了解工具的名称、用途和参数格式 + */ + @Override + public ToolDefinition getToolDefinition() { + return DefaultToolDefinition.builder() + .name(toolName) + .description(toolDescription) + .inputSchema(inputSchema) + .build(); + } + + /** + * 执行工具调用 + * 当 AI 模型决定调用此工具时，Spring AI 框架自动调用此方法。 + * 内部通过 McpClientManager 获取对应的 MCP Client 实例， + * 将调用转发到远程 MCP Server 并返回结果。 + *

+ * 包含调用轮次限制：同一请求线程中，所有 MCP 工具的累计调用次数 + * 超过 maxCallRounds 时，返回错误 JSON 防止无限循环。 + * + * @param toolInput JSON 格式的工具输入参数 + * @return 工具执行结果（JSON 字符串形式） + */ + @Override + public String call(String toolInput) { + // 检查调用轮次是否超限 + int currentRound = CALL_ROUNDS.get(); + if (currentRound >= maxCallRounds) { + log.warn("MCP 工具调用轮次超限: tool={}, currentRound={}, maxRounds={}", + originalToolName, currentRound, maxCallRounds); + return "{\"error\": \"工具调用轮次已达上限 (" + maxCallRounds + " 次)，已终止调用以防止无限循环。请优化提示词减少工具调用次数。\"}"; + } + CALL_ROUNDS.set(currentRound + 1); + + log.info("MCP 工具调用: serverId={}, tool={}, input={}, round={}/{}", + mcpServerConfigId, originalToolName, toolInput, currentRound + 1, maxCallRounds); + long startTime = System.currentTimeMillis(); + + try { + // 获取 MCP Client 实例 + McpSyncClient client = mcpClientManager.getClient(mcpServerConfigId); + if (client == null) { + log.error("MCP 客户端不可用: serverId={}, tool={}", mcpServerConfigId, originalToolName); + return "{\"error\": \"MCP Server (id=" + mcpServerConfigId + ") 客户端未就绪，请检查配置\"}"; + } + + // 解析工具输入参数：JSON 字符串 -> Map + Map arguments = OBJECT_MAPPER.readValue(toolInput, new TypeReference<>() {}); + + // 构建 MCP 调用请求并执行 + McpSchema.CallToolRequest request = new McpSchema.CallToolRequest(originalToolName, arguments); + McpSchema.CallToolResult result = client.callTool(request); + + long latency = System.currentTimeMillis() - startTime; + log.info("MCP 工具调用完成: tool={}, latency={}ms, isError={}", + originalToolName, latency, result.isError()); + + // 收集工具调用事件（供 SSE 流式输出使用） + String resultStr = result.content() != null ? String.valueOf(result.content()) : ""; + boolean isError = result.isError() != null && result.isError(); + EVENTS.get().add(new ToolCallEvent(originalToolName, toolInput, resultStr, latency, isError)); + + // 检查是否为错误结果 + if (result.isError() != null && result.isError()) { + log.error("MCP 工具返回错误: tool={}, content={}", originalToolName, result.content()); + return "{\"error\": \"工具执行返回错误: " + + escapeJson(String.valueOf(result.content())) + "\"}"; + } + + // 将 MCP Content 列表序列化为 JSON 字符串返回给 AI 模型 + // 与官方 SyncMcpToolCallback 保持一致，使用 ModelOptionsUtils 序列化 + String resultJson = ModelOptionsUtils.toJsonString(result.content()); + log.debug("MCP 工具调用结果: tool={}, result={}", originalToolName, resultJson); + return resultJson; + + } catch (Exception e) { + long latency = System.currentTimeMillis() - startTime; + log.error("MCP 工具调用失败: tool={}, latency={}ms, error={}", originalToolName, latency, e.getMessage()); + // 记录失败事件 + EVENTS.get().add(new ToolCallEvent(originalToolName, toolInput, e.getMessage(), latency, true)); + return "{\"error\": \"" + escapeJson(e.getMessage()) + "\"}"; + } + } + + /** + * 转义字符串中的特殊字符，避免破坏 JSON 格式 + */ + private String escapeJson(String text) { + if (text == null) return "未知错误"; + return text.replace("\\", "\\\\").replace("\"", "'").replace("\n", "\\n").replace("\r", "\\r"); + } + + // ==================== Getter 方法 ==================== + + public Long getMcpServerConfigId() { + return mcpServerConfigId; + } + + public String getToolName() { + return toolName; + } + + public String getOriginalToolName() { + return originalToolName; + } + + public String getToolDescription() { + return toolDescription; + } + + public String getInputSchema() { + return inputSchema; + } + + /** + * 重置当前线程的调用轮次计数器 + * 应在每次新的对话请求开始时调用，确保轮次计数从零开始 + */ + public static void resetCallRounds() { + CALL_ROUNDS.remove(); + } +} diff --git a/src/main/java/com/wok/supportbot/mcp/McpToolCallbackAdapter.java b/src/main/java/com/wok/supportbot/mcp/McpToolCallbackAdapter.java new file mode 100644 index 0000000..510ebfa --- /dev/null +++ b/src/main/java/com/wok/supportbot/mcp/McpToolCallbackAdapter.java @@ -0,0 +1,111 @@ +package com.wok.supportbot.mcp; + +import com.wok.supportbot.config.McpClientManager; +import lombok.extern.slf4j.Slf4j; +import org.springframework.ai.tool.ToolCallback; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.stereotype.Component; + +import java.util.ArrayList; +import java.util.List; +import java.util.Map; + +/** + * MCP 工具批量适配器 + * 从 McpClientManager 获取所有已启用的 MCP 工具描述， + * 批量转换为 Spring AI ToolCallback 列表，供 ChatClient 注册。 + *

+ * 使用方式： + * 

+ *   // 注入后在 ChatClient 构建时注册所有 MCP 工具
+ *   ChatClient.builder(chatModel)
+ *       .defaultTools(mcpToolCallbackAdapter.getAllToolCallbacksAsArray())
+ *       .defaultAdvisors(...)
+ *       .build();
+ *
+ */ +@Component +@Slf4j +public class McpToolCallbackAdapter { + + private final McpClientManager mcpClientManager; + + /** + * 单次对话请求中最大允许的 MCP 工具调用轮次，防止无限循环调用 + */ + @Value("${mcp.tools.max-call-rounds:5}") + private int maxCallRounds; + + public McpToolCallbackAdapter(McpClientManager mcpClientManager) { + this.mcpClientManager = mcpClientManager; + } + + /** + * 获取所有 MCP 工具的 ToolCallback 列表 + * 遍历所有已启用的 MCP Server，将其暴露的工具转换为 ToolCallback + * + * @return ToolCallback 列表（可能为空，但不会为 null） + */ + public List getAllToolCallbacks() { + List callbacks = new ArrayList<>(); + List> tools = mcpClientManager.listAvailableTools(); + + for (Map toolInfo : tools) { + String configId = (String) toolInfo.get("config_id"); + String serverName = (String) toolInfo.get("name"); + + @SuppressWarnings("unchecked") + List> toolList = (List>) toolInfo.get("tools"); + if (toolList == null || toolList.isEmpty()) { + log.debug("MCP Server [{}] 暂无可用工具", serverName); + continue; + } + + for (Map tool : toolList) { + String toolName = (String) tool.get("name"); + String description = (String) tool.get("description"); + String inputSchema = (String) tool.get("inputSchema"); + + // 工具名加 mcp_ 前缀，避免与内置工具冲突 + String prefixedName = "mcp_" + toolName; + McpToolCallback callback = new McpToolCallback( + Long.parseLong(configId), prefixedName, toolName, description, inputSchema, + mcpClientManager, maxCallRounds); + callbacks.add(callback); + log.debug("注册 MCP 工具: server={}, tool={}", serverName, prefixedName); + } + } + + log.info("MCP 工具适配完成，共注册 {} 个工具", callbacks.size()); + return callbacks; + } + + /** + * 获取 MCP 工具的 ToolCallback 数组（供 ChatClient.defaultTools() 使用） + * ChatClient.defaultTools() 接受 ToolCallback[] 参数，此方法提供便捷转换 + * + * @return ToolCallback 数组 + */ + public ToolCallback[] getAllToolCallbacksAsArray() { + return getAllToolCallbacks().toArray(new ToolCallback[0]); + } + + /** + * 获取指定工具名列表的 ToolCallback 数组（按角色权限过滤 MCP 工具） + * + * @param allowedTools 允许的工具名列表；["*"] 表示全部允许，其他为具体工具名列表 + * @return 过滤后的 ToolCallback 数组（可能为空数组） + */ + public ToolCallback[] getFilteredToolCallbacks(List allowedTools) { + if (allowedTools == null || allowedTools.isEmpty()) { + return new ToolCallback[0]; + } + List all = getAllToolCallbacks(); + if (allowedTools.contains("*")) { + return all.toArray(new ToolCallback[0]); + } + return all.stream() + .filter(cb -> allowedTools.contains(cb.getToolDefinition().name())) + .toArray(ToolCallback[]::new); + } +} diff --git a/src/main/java/com/wok/supportbot/mcp/SseEventBuilder.java b/src/main/java/com/wok/supportbot/mcp/SseEventBuilder.java new file mode 100644 index 0000000..814c4f9 --- /dev/null +++ b/src/main/java/com/wok/supportbot/mcp/SseEventBuilder.java @@ -0,0 +1,73 @@ +package com.wok.supportbot.mcp; + +import org.springframework.http.codec.ServerSentEvent; + +/** + * SSE 事件构建器 + * 用于在 MCP 工具调用流程中构建标准 SSE 事件， + * 让前端能区分文本内容与工具调用状态。 + * + * 事件类型： + * - message: 普通文本内容 + * - tool_call_start: 工具调用开始 + * - tool_call_result: 工具调用结果返回 + * - error: 错误信息 + */ +public class SseEventBuilder { + + /** + * 构建普通文本消息事件 + */ + public static ServerSentEvent messageEvent(String data) { + return ServerSentEvent.builder() + .event("message") + .data(data) + .build(); + } + + /** + * 构建工具调用开始事件 + */ + public static ServerSentEvent toolCallStartEvent(String toolName, String input) { + String json = String.format("{\"tool\":\"%s\",\"input\":\"%s\"}", + escapeJson(toolName), escapeJson(input)); + return ServerSentEvent.builder() + .event("tool_call_start") + .data(json) + .build(); + } + + /** + * 构建工具调用结果事件 + */ + public static ServerSentEvent toolCallResultEvent(String toolName, String result, long latencyMs) { + String json = String.format("{\"tool\":\"%s\",\"result\":\"%s\",\"latencyMs\":%d}", + escapeJson(toolName), escapeJson(result), latencyMs); + return ServerSentEvent.builder() + .event("tool_call_result") + .data(json) + .build(); + } + + /** + * 构建错误事件 + */ + public static ServerSentEvent errorEvent(String message) { + return ServerSentEvent.builder() + .event("error") + .data("{\"message\":\"" + escapeJson(message) + "\"}") + .build(); + } + + /** + * JSON 字符串转义 + */ + private static String escapeJson(String s) { + if (s == null) return ""; + return s.replace("\\", "\\\\") + .replace("\"", "\\\"") + .replace("\n", "\\n") + .replace("\r", "\\r") + .replace("\t", "\\t"); + } +} diff --git a/src/main/java/com/wok/supportbot/service/CustomerServiceRoleService.java b/src/main/java/com/wok/supportbot/service/CustomerServiceRoleService.java index 9888e52..907d146 100644 --- a/src/main/java/com/wok/supportbot/service/CustomerServiceRoleService.java +++ b/src/main/java/com/wok/supportbot/service/CustomerServiceRoleService.java @@ -1,5 +1,8 @@ package com.wok.supportbot.service; +import com.fasterxml.jackson.core.type.TypeReference; +import com.fasterxml.jackson.databind.ObjectMapper; +import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.jdbc.core.JdbcTemplate; import org.springframework.stereotype.Service; @@ -12,11 +15,15 @@ import java.util.Map; import java.util.Objects; @Service +@Slf4j public class CustomerServiceRoleService { @Autowired private JdbcTemplate jdbcTemplate; + @Autowired + private ObjectMapper objectMapper; + public List> listRoles() { return listRoles(false); } @@ -25,7 +32,7 @@ public class CustomerServiceRoleService { * @param includeDisabled true=包含已停用角色（管理页用）；false=仅启用角色（对话页用） */ public List> listRoles(boolean includeDisabled) { - String roleSql = "SELECT id::text AS id, role_key, name, description, prompt, sort_order, enabled " + String roleSql = "SELECT id::text AS id, role_key, name, description, prompt, sort_order, enabled, allowed_mcp_tools " + "FROM customer_service_role WHERE is_delete = false " + (includeDisabled ? "" : "AND enabled = true ") + "ORDER BY sort_order ASC, id ASC"; @@ -137,6 +144,28 @@ public class CustomerServiceRoleService { jdbcTemplate.update("UPDATE customer_service_role_category SET is_delete = true WHERE role_id = ?", roleId); } + /** + * 更新角色的 MCP 工具权限。 + * + * @param roleId 角色ID + * @param allowedMcpTools 允许的 MCP 工具名列表；["*"]=全部允许，空/null=不允许任何工具 + */ + @Transactional(rollbackFor = Exception.class) + public void updateRoleMcpTools(Long roleId, List allowedMcpTools) { + if (roleId == null || roleId <= 0) { + throw new IllegalArgumentException("角色ID无效"); + } + String json; + try { + json = objectMapper.writeValueAsString(allowedMcpTools != null ? allowedMcpTools : List.of()); + } catch (Exception e) { + json = "[]"; + } + jdbcTemplate.update( + "UPDATE customer_service_role SET allowed_mcp_tools = ?::jsonb, update_time = CURRENT_TIMESTAMP WHERE id = ? AND is_delete = false", + json, roleId); + } + /** * 服务端解析角色的知识库范围与人设。 * 用于对话时强制约束：角色只能检索其绑定分类下的内容，由后端决定，客户端无法越权跨域。 @@ -149,7 +178,7 @@ public class CustomerServiceRoleService { return RoleScope.empty(); } List> rows = jdbcTemplate.queryForList( - "SELECT name, prompt FROM customer_service_role WHERE id = ? AND is_delete = false AND enabled = true", + "SELECT name, prompt, allowed_mcp_tools FROM customer_service_role WHERE id = ? AND is_delete = false AND enabled = true", roleId); if (rows.isEmpty()) { return RoleScope.empty(); @@ -159,22 +188,57 @@ public class CustomerServiceRoleService { List categoryIds = jdbcTemplate.queryForList( "SELECT category_id FROM customer_service_role_category WHERE role_id = ? AND is_delete = false", Long.class, roleId); - return new RoleScope(true, name, prompt, categoryIds); + // 解析 allowed_mcp_tools JSONB 字段 + List allowedMcpTools = parseAllowedMcpTools(rows.get(0).get("allowed_mcp_tools")); + return new RoleScope(true, name, prompt, categoryIds, allowedMcpTools); } /** - * 角色范围：是否命中角色、角色人设、可检索的知识库分类。 + * 解析 allowed_mcp_tools JSONB 字段为 List + * 兼容 PostgreSQL JSONB、Java List、String 等多种输入类型 */ - public record RoleScope(boolean present, String name, String prompt, List categoryIds) { + private List parseAllowedMcpTools(Object raw) { + if (raw == null) { + return List.of(); + } + try { + if (raw instanceof List list) { + return list.stream().map(Objects::toString).toList(); + } + String json = raw.toString().trim(); + if (json.isEmpty() || "[]".equals(json) || "null".equals(json)) { + return List.of(); + } + return objectMapper.readValue(json, new TypeReference<>() {}); + } catch (Exception e) { + log.warn("解析 allowed_mcp_tools 失败: {}", raw, e); + return List.of(); + } + } + + /** + * 角色范围：是否命中角色、角色人设、可检索的知识库分类、MCP 工具权限。 + */ + public record RoleScope(boolean present, String name, String prompt, List categoryIds, List allowedMcpTools) { public static RoleScope empty() { - return new RoleScope(false, "", "", List.of()); + return new RoleScope(false, "", "", List.of(), List.of()); } public boolean hasRole() { return present; } + /** 是否允许使用 MCP 工具（非空即允许） */ + public boolean isMcpToolsAllowed() { + return allowedMcpTools != null && !allowedMcpTools.isEmpty(); + } + + /** 是否允许使用所有 MCP 工具（["*"] 表示全部允许） */ + public boolean isAllMcpToolsAllowed() { + return allowedMcpTools != null && allowedMcpTools.contains("*"); + } + /** * 组合角色系统提示词：当前客服角色 + 人设；都为空时返回 null（退回基础提示词）。 */ diff --git a/src/main/java/com/wok/supportbot/service/McpServerConfigService.java b/src/main/java/com/wok/supportbot/service/McpServerConfigService.java new file mode 100644 index 0000000..cb1dd4f --- /dev/null +++ b/src/main/java/com/wok/supportbot/service/McpServerConfigService.java @@ -0,0 +1,320 @@ +package com.wok.supportbot.service; + +import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper; +import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.wok.supportbot.dao.McpServerConfigMapper; +import com.wok.supportbot.entity.McpServerConfig; +import lombok.extern.slf4j.Slf4j; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.stereotype.Service; +import org.springframework.transaction.annotation.Transactional; + +import java.util.*; + +/** + * MCP Server 配置管理服务 + * 提供 CRUD、启用/禁用、JSONB 字段显式持久化等功能 + */ +@Service +@Slf4j +public class McpServerConfigService { + + @Autowired + private McpServerConfigMapper mcpServerConfigMapper; + + @Autowired + private JdbcTemplate jdbcTemplate; + + /** 复用的 JSON 序列化器，避免每次持久化都新建实例 */ + private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper(); + + // ==================== 分页列表 ==================== + + /** + * 分页查询 MCP Server 配置列表 + * + * @param page 页码（从1开始） + * @param size 每页大小 + * @return 分页结果（records/total/page/size/pages） + */ + public Map listConfigs(int page, int size) { + // 查询总数 + LambdaQueryWrapper countWrapper = new LambdaQueryWrapper<>(); + Long total = mcpServerConfigMapper.selectCount(countWrapper); + if (total == null) total = 0L; + + // 查询列表（带排序和分页） + LambdaQueryWrapper listWrapper = new LambdaQueryWrapper<>(); + listWrapper.orderByDesc(McpServerConfig::getCreateTime); + listWrapper.last("LIMIT " + size + " OFFSET " + ((page - 1) * size)); + List records = mcpServerConfigMapper.selectList(listWrapper); + + // 格式化返回结果 + List> formattedRecords = new ArrayList<>(); + for (McpServerConfig record : records) { + formattedRecords.add(formatConfig(record)); + } + + Map result = new LinkedHashMap<>(); + result.put("records", formattedRecords); + result.put("total", total); + result.put("page", page); + result.put("size", size); + result.put("pages", (total + size - 1) / size); + return result; + } + + // ==================== 详情 ==================== + + /** + * 获取单条配置详情 + * + * @param id 配置ID + * @return 格式化的配置详情，不存在返回 null + */ + public Map getConfigDetail(Long id) { + McpServerConfig config = mcpServerConfigMapper.selectById(id); + if (config == null) { + return null; + } + return formatConfig(config); + } + + // ==================== 新建配置 ==================== + + /** + * 新建 MCP Server 配置 + * + * @param config 配置对象 + * @return 保存后的配置（格式化） + */ + @Transactional(rollbackFor = Exception.class) + public Map createConfig(McpServerConfig config) { + // 校验必填字段 + validateRequired(config); + + mcpServerConfigMapper.insert(config); + // JSONB 字段显式持久化，避免 MyBatis Plus typeHandler 不触发的问题 + persistEnvVars(config.getId(), config.getEnvVars()); + + log.info("新建 MCP Server 配置: name={}, transportType={}", config.getName(), config.getTransportType()); + return formatConfig(mcpServerConfigMapper.selectById(config.getId())); + } + + // ==================== 更新配置 ==================== + + /** + * 更新 MCP Server 配置 + * 使用 LambdaUpdateWrapper 显式设置所有字段（含 null）， + * 避免 updateById 默认跳过 null 字段导致无法清空可选字段。 + * + * @param id 配置ID + * @param config 更新内容 + * @return 更新后的配置（格式化） + */ + @Transactional(rollbackFor = Exception.class) + public Map updateConfig(Long id, McpServerConfig config) { + McpServerConfig existing = mcpServerConfigMapper.selectById(id); + if (existing == null) { + throw new RuntimeException("配置不存在: id=" + id); + } + + // 使用 LambdaUpdateWrapper 显式设置所有字段，null 也会被写入（清空语义） + LambdaUpdateWrapper updateWrapper = new LambdaUpdateWrapper<>(); + updateWrapper.eq(McpServerConfig::getId, id) + .set(McpServerConfig::getName, config.getName()) + .set(McpServerConfig::getTransportType, config.getTransportType()) + .set(McpServerConfig::getServerUrl, config.getServerUrl()) + .set(McpServerConfig::getCommand, config.getCommand()) + .set(McpServerConfig::getArgs, config.getArgs()) + .set(McpServerConfig::getDescription, config.getDescription()); + // isActive 仅在前端显式传入时更新，null 表示不修改 + if (config.getIsActive() != null) { + updateWrapper.set(McpServerConfig::getIsActive, config.getIsActive()); + } + mcpServerConfigMapper.update(null, updateWrapper); + + // JSONB 字段显式持久化（LambdaUpdateWrapper 不支持 JSONB 类型） + // null 或空 Map 均执行持久化（清空语义），与 LambdaUpdateWrapper 行为一致 + persistEnvVars(id, config.getEnvVars()); + + log.info("更新 MCP Server 配置: id={}", id); + return formatConfig(mcpServerConfigMapper.selectById(id)); + } + + // ==================== 删除配置 ==================== + + /** + * 逻辑删除 MCP Server 配置 + * + * @param id 配置ID + */ + @Transactional(rollbackFor = Exception.class) + public void deleteConfig(Long id) { + McpServerConfig config = mcpServerConfigMapper.selectById(id); + if (config == null) { + throw new RuntimeException("配置不存在: id=" + id); + } + mcpServerConfigMapper.deleteById(id); + log.info("逻辑删除 MCP Server 配置: id={}, name={}", id, config.getName()); + } + + // ==================== 启用/禁用 ==================== + + /** + * 切换配置启用状态 + * + * @param id 配置ID + * @param active 是否启用 + */ + @Transactional(rollbackFor = Exception.class) + public void toggleActive(Long id, boolean active) { + McpServerConfig config = mcpServerConfigMapper.selectById(id); + if (config == null) { + throw new RuntimeException("配置不存在: id=" + id); + } + + LambdaUpdateWrapper updateWrapper = new LambdaUpdateWrapper<>(); + updateWrapper.eq(McpServerConfig::getId, id) + .set(McpServerConfig::getIsActive, active); + mcpServerConfigMapper.update(null, updateWrapper); + + log.info("切换 MCP Server 配置状态: id={}, name={}, active={}", id, config.getName(), active); + } + + // ==================== 活跃配置查询 ==================== + + /** + * 获取所有启用的 SSE 类型配置 + * 供 McpClientManager 建立 SSE 连接使用 + * + * @return SSE 类型的活跃配置列表 + */ + public List listActiveSseConfigs() { + LambdaQueryWrapper wrapper = new LambdaQueryWrapper<>(); + wrapper.eq(McpServerConfig::getIsActive, true) + .eq(McpServerConfig::getTransportType, "sse"); + return mcpServerConfigMapper.selectList(wrapper); + } + + /** + * 获取所有启用的 stdio 类型配置 + * 供 McpClientManager 启动 stdio 进程使用 + * + * @return stdio 类型的活跃配置列表 + */ + public List listActiveStdioConfigs() { + LambdaQueryWrapper wrapper = new LambdaQueryWrapper<>(); + wrapper.eq(McpServerConfig::getIsActive, true) + .eq(McpServerConfig::getTransportType, "stdio"); + return mcpServerConfigMapper.selectList(wrapper); + } + + /** + * 根据 ID 获取配置（含完整字段，仅供内部使用） + * + * @param id 配置ID + * @return 原始配置对象，不存在返回 null + */ + public McpServerConfig getConfigById(Long id) { + return mcpServerConfigMapper.selectById(id); + } + + /** + * 获取所有启用的配置（不区分传输类型） + * 供 McpClientManager.refreshAll() 一次查询获取所有活跃配置 + * + * @return 所有启用的配置列表 + */ + public List listAllActiveConfigs() { + LambdaQueryWrapper wrapper = new LambdaQueryWrapper<>(); + wrapper.eq(McpServerConfig::getIsActive, true); + return mcpServerConfigMapper.selectList(wrapper); + } + + // ==================== 工具方法 ==================== + + /** + * 校验必填字段 + * + * @param config 配置对象 + */ + private void validateRequired(McpServerConfig config) { + if (config.getName() == null || config.getName().trim().isEmpty()) { + throw new IllegalArgumentException("配置名称不能为空"); + } + if (config.getTransportType() == null || config.getTransportType().trim().isEmpty()) { + throw new IllegalArgumentException("传输类型不能为空"); + } + // 根据传输类型校验对应必填字段 + switch (config.getTransportType()) { + case "sse": + if (config.getServerUrl() == null || config.getServerUrl().trim().isEmpty()) { + throw new IllegalArgumentException("SSE 模式下服务地址不能为空"); + } + break; + case "stdio": + if (config.getCommand() == null || config.getCommand().trim().isEmpty()) { + throw new IllegalArgumentException("stdio 模式下启动命令不能为空"); + } + break; + default: + throw new IllegalArgumentException("不支持的传输类型: " + config.getTransportType()); + } + } + + /** + * 格式化配置为前端返回格式 + * 雪花 ID 转字符串、时间格式化 + * + * @param config 原始配置 + * @return 格式化后的 Map + */ + private Map formatConfig(McpServerConfig config) { + Map formatted = new LinkedHashMap<>(); + // 雪花 ID 必须转为字符串，避免 JS 精度丢失 + formatted.put("id", config.getId().toString()); + formatted.put("name", config.getName()); + formatted.put("transport_type", config.getTransportType()); + formatted.put("server_url", config.getServerUrl()); + formatted.put("command", config.getCommand()); + formatted.put("args", config.getArgs()); + formatted.put("env_vars", config.getEnvVars()); + formatted.put("description", config.getDescription()); + formatted.put("is_active", config.getIsActive()); + formatted.put("create_time", config.getCreateTime()); + formatted.put("update_time", config.getUpdateTime()); + return formatted; + } + + /** + * 显式持久化 envVars JSONB 字段 + * MyBatis Plus 的 insert/updateById 对带 typeHandler 的 JSONB 字段可能不触发写入， + * 使用 JdbcTemplate 显式更新保证数据可靠落库。 + * + * @param id 配置ID + * @param envVars 环境变量 Map + */ + private void persistEnvVars(Long id, Map envVars) { + if (id == null) { + return; + } + try { + String json; + if (envVars == null || envVars.isEmpty()) { + json = "{}"; + } else { + json = OBJECT_MAPPER.writeValueAsString(envVars); + } + jdbcTemplate.update( + "UPDATE mcp_server_config SET env_vars = ?::jsonb WHERE id = ?", + json, id); + log.debug("持久化 envVars: id={}, envVars={}", id, json); + } catch (Exception e) { + log.error("持久化 envVars 失败: id={}, error={}", id, e.getMessage()); + throw new RuntimeException("持久化环境变量失败: " + e.getMessage(), e); + } + } +} diff --git a/src/main/resources/init-database.sql b/src/main/resources/init-database.sql index af1d7cb..5f131c3 100644 --- a/src/main/resources/init-database.sql +++ b/src/main/resources/init-database.sql @@ -141,7 +141,8 @@ CREATE TABLE IF NOT EXISTS customer_service_role ( enabled BOOLEAN NOT NULL DEFAULT TRUE, create_time TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, update_time TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, - is_delete BOOLEAN NOT NULL DEFAULT FALSE + is_delete BOOLEAN NOT NULL DEFAULT FALSE, + allowed_mcp_tools JSONB NOT NULL DEFAULT '[]' ); CREATE INDEX IF NOT EXISTS idx_customer_service_role_enabled ON customer_service_role (enabled, sort_order); @@ -157,6 +158,7 @@ COMMENT ON COLUMN customer_service_role.enabled IS '是否启用'; COMMENT ON COLUMN customer_service_role.create_time IS '创建时间'; COMMENT ON COLUMN customer_service_role.update_time IS '更新时间'; COMMENT ON COLUMN customer_service_role.is_delete IS '逻辑删除'; +COMMENT ON COLUMN customer_service_role.allowed_mcp_tools IS '允许使用的 MCP 工具列表（JSONB 数组，空=不允许，["*"]=全部允许）'; -- 默认角色种子数据 INSERT INTO customer_service_role (role_key, name, description, sort_order, enabled) diff --git a/src/main/resources/static/components/ChatPanel.js b/src/main/resources/static/components/ChatPanel.js index 0f40be7..bf9d315 100644 --- a/src/main/resources/static/components/ChatPanel.js +++ b/src/main/resources/static/components/ChatPanel.js @@ -3,7 +3,7 @@ */ import { ref, computed, nextTick, onMounted } from 'vue' import { chatSync, chatRagSync, chatSSEUrl, chatRagSSEUrl, getRoleList, getActiveModelConfig, truncateConversation, ragSources, submitFeedback as submitFeedbackApi } from '../js/api.js' -import { toast, readSSEStream, renderMarkdown } from '../js/utils.js' +import { toast, readSSEStream, readSSEStreamWithEvents, renderMarkdown } from '../js/utils.js' import { store } from '../js/store.js' import MessageSources from './MessageSources.js' @@ -108,6 +108,16 @@ export default {