From c45f47faa000d682bd13ab668cf1f61578f4f657 Mon Sep 17 00:00:00 2001 From: hrz <1710360675@qq.com> Date: Mon, 7 Jul 2025 15:04:32 +0800 Subject: [PATCH] =?UTF-8?q?Update:=E4=BC=98=E5=8C=96mcp=E5=B7=A5=E5=85=B7?= =?UTF-8?q?=E8=AF=BB=E5=8F=96=E9=A1=BA=E5=BA=8F=20(#1775)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * update:发布0.6.3版本 * update:升级sherpa_onnx版本 * update:补充自动更新教程 * update:优化mcp工具读取顺序 * update:优化 --- .gitignore | 1 + .../impl/AgentMcpAccessPointServiceImpl.java | 99 +++++++++++++------ .../sys/utils/WebSocketClientManager.java | 42 ++++++++ 3 files changed, 112 insertions(+), 30 deletions(-) diff --git a/.gitignore b/.gitignore index 9469baac..b43a8de3 100644 --- a/.gitignore +++ b/.gitignore @@ -174,3 +174,4 @@ main/xiaozhi-server/mysql uploadfile *.json .vscode +.cursor diff --git a/main/manager-api/src/main/java/xiaozhi/modules/agent/service/impl/AgentMcpAccessPointServiceImpl.java b/main/manager-api/src/main/java/xiaozhi/modules/agent/service/impl/AgentMcpAccessPointServiceImpl.java index e6c7d4e4..839f99ff 100644 --- a/main/manager-api/src/main/java/xiaozhi/modules/agent/service/impl/AgentMcpAccessPointServiceImpl.java +++ b/main/manager-api/src/main/java/xiaozhi/modules/agent/service/impl/AgentMcpAccessPointServiceImpl.java @@ -61,14 +61,15 @@ public class AgentMcpAccessPointServiceImpl implements AgentMcpAccessPointServic wsUrl = wsUrl.replace("/mcp/", "/call/"); try { - // 创建 WebSocket 连接 + // 创建 WebSocket 连接，增加超时时间到15秒 try (WebSocketClientManager client = WebSocketClientManager.build( new WebSocketClientManager.Builder() .uri(wsUrl) - .connectTimeout(5, TimeUnit.SECONDS) - .maxSessionDuration(9, TimeUnit.SECONDS))) { + .connectTimeout(8, TimeUnit.SECONDS) + .maxSessionDuration(10, TimeUnit.SECONDS))) { - // 发送初始化消息 + // 步骤1: 发送初始化消息并等待响应 + log.info("发送MCP初始化消息，智能体ID: {}", id); McpJsonRpcRequest initializeRequest = new McpJsonRpcRequest("initialize", Map.of( "protocolVersion", "2024-11-05", @@ -81,37 +82,70 @@ public class AgentMcpAccessPointServiceImpl implements AgentMcpAccessPointServic 1); client.sendJson(initializeRequest); - // 等待初始化响应 - Thread.sleep(200); - - // 发送初始化完成通知 - // 对于通知类型的消息，手动构建JSON以避免包含null字段 - String notificationJson = "{\"jsonrpc\":\"2.0\",\"method\":\"notifications/initialized\"}"; - client.sendText(notificationJson); - - // 等待 0.2 秒 - Thread.sleep(200); - - // 发送工具列表请求 - McpJsonRpcRequest toolsRequest = new McpJsonRpcRequest("tools/list", null, 2); - client.sendJson(toolsRequest); - - // 监听响应，直到收到包含 id=2 的响应（tools/list响应） - List responses = client.listener(response -> { + // 等待初始化响应 (id=1) - 移除固定延迟，改为响应驱动 + List initResponses = client.listenerWithoutClose(response -> { try { - // 先尝试解析为通用JSON对象来获取id Map jsonMap = JsonUtils.parseObject(response, Map.class); - return jsonMap != null && Integer.valueOf(2).equals(jsonMap.get("id")); + if (jsonMap != null && Integer.valueOf(1).equals(jsonMap.get("id"))) { + // 检查是否有result字段，表示初始化成功 + return jsonMap.containsKey("result") && !jsonMap.containsKey("error"); + } + return false; } catch (Exception e) { - log.warn("解析响应失败: {}", response, e); + log.warn("解析初始化响应失败: {}", response, e); return false; } }); - // 处理响应 - for (String response : responses) { + // 验证初始化响应 + boolean initSucceeded = false; + for (String response : initResponses) { + try { + Map jsonMap = JsonUtils.parseObject(response, Map.class); + if (jsonMap != null && Integer.valueOf(1).equals(jsonMap.get("id"))) { + if (jsonMap.containsKey("result")) { + log.info("MCP初始化成功，智能体ID: {}", id); + initSucceeded = true; + break; + } else if (jsonMap.containsKey("error")) { + log.error("MCP初始化失败，智能体ID: {}, 错误: {}", id, jsonMap.get("error")); + return List.of(); + } + } + } catch (Exception e) { + log.warn("处理初始化响应失败: {}", response, e); + } + } + + if (!initSucceeded) { + log.error("未收到有效的MCP初始化响应，智能体ID: {}", id); + return List.of(); + } + + // 步骤2: 发送初始化完成通知 - 只有在收到initialize响应后才发送 + log.info("发送MCP初始化完成通知，智能体ID: {}", id); + String notificationJson = "{\"jsonrpc\":\"2.0\",\"method\":\"notifications/initialized\"}"; + client.sendText(notificationJson); + + // 步骤3: 发送工具列表请求 - 立即发送，无需额外延迟 + log.info("发送MCP工具列表请求，智能体ID: {}", id); + McpJsonRpcRequest toolsRequest = new McpJsonRpcRequest("tools/list", null, 2); + client.sendJson(toolsRequest); + + // 等待工具列表响应 (id=2) + List toolsResponses = client.listener(response -> { + try { + Map jsonMap = JsonUtils.parseObject(response, Map.class); + return jsonMap != null && Integer.valueOf(2).equals(jsonMap.get("id")); + } catch (Exception e) { + log.warn("解析工具列表响应失败: {}", response, e); + return false; + } + }); + + // 处理工具列表响应 + for (String response : toolsResponses) { try { - // 先解析为通用JSON对象 Map jsonMap = JsonUtils.parseObject(response, Map.class); if (jsonMap != null && Integer.valueOf(2).equals(jsonMap.get("id"))) { // 检查是否有result字段 @@ -122,11 +156,16 @@ public class AgentMcpAccessPointServiceImpl implements AgentMcpAccessPointServic if (toolsObj instanceof List) { List> toolsList = (List>) toolsObj; // 提取工具名称列表 - return toolsList.stream() + List result = toolsList.stream() .map(tool -> (String) tool.get("name")) .filter(name -> name != null) .collect(Collectors.toList()); + log.info("成功获取MCP工具列表，智能体ID: {}, 工具数量: {}", id, result.size()); + return result; } + } else if (jsonMap.containsKey("error")) { + log.error("获取工具列表失败，智能体ID: {}, 错误: {}", id, jsonMap.get("error")); + return List.of(); } } } catch (Exception e) { @@ -134,7 +173,7 @@ public class AgentMcpAccessPointServiceImpl implements AgentMcpAccessPointServic } } - log.warn("未找到有效的工具列表响应"); + log.warn("未找到有效的工具列表响应，智能体ID: {}", id); return List.of(); } @@ -204,4 +243,4 @@ public class AgentMcpAccessPointServiceImpl implements AgentMcpAccessPointServic // 加密后成token值 return AESUtils.encrypt(key, json); } -} +} \ No newline at end of file diff --git a/main/manager-api/src/main/java/xiaozhi/modules/sys/utils/WebSocketClientManager.java b/main/manager-api/src/main/java/xiaozhi/modules/sys/utils/WebSocketClientManager.java index 45ac488e..d7171ae7 100644 --- a/main/manager-api/src/main/java/xiaozhi/modules/sys/utils/WebSocketClientManager.java +++ b/main/manager-api/src/main/java/xiaozhi/modules/sys/utils/WebSocketClientManager.java @@ -137,6 +137,37 @@ public class WebSocketClientManager implements Closeable { return collected; } + private List listenerCustomWithoutClose( + BlockingQueue queue, + Predicate predicate) + throws InterruptedException, TimeoutException, ExecutionException { + List collected = new ArrayList<>(); + long deadline = System.currentTimeMillis() + maxSessionDurationUnit.toMillis(maxSessionDuration); + + while (true) { + if (errorFuture.isDone()) { + errorFuture.get(); + } + + long remaining = deadline - System.currentTimeMillis(); + if (remaining <= 0) { + throw new TimeoutException("等待批量消息超时"); + } + + T msg = queue.poll(remaining, TimeUnit.MILLISECONDS); + if (msg == null) { + throw new TimeoutException("等待批量消息超时"); + } + + collected.add(msg); + if (predicate.test(msg)) { + break; + } + } + // 不调用 close()，保持连接开放 + return collected; + } + /** * 同步接收多条消息，直到 predicate 为 true 或超时抛异常； * @@ -147,6 +178,17 @@ public class WebSocketClientManager implements Closeable { return listenerCustom(textMessageQueue, predicate); } + /** + * 同步接收多条消息，直到 predicate 为 true 或超时抛异常； + * 不自动关闭连接，适用于需要在同一连接上发送多个消息的场景 + * + * @return 返回监听期间的所有消息列表 + */ + public List listenerWithoutClose(Predicate predicate) + throws InterruptedException, TimeoutException, ExecutionException { + return listenerCustomWithoutClose(textMessageQueue, predicate); + } + public List listenerBinary(Predicate predicate) throws InterruptedException, TimeoutException, ExecutionException { return listenerCustom(binaryMessageQueue, predicate);