LangChain4j核心接口使用（四）Tool和MCP（3）MCP Client

一、手动解析json法mcp-server见知识体系二MCP-CSDN博客中的demo这里主要是补充client。新增controller/** * 5、mcp tool */ PostMapping(/mcpToolGenerate) public ResponseEntity? mcpToolGenerate(Valid RequestBody ChatRequestDTO request) { try { request.setProvider(siliconflow); request.setModelName(Qwen/Qwen2.5-72B-Instruct); ChatResponseDTO response gatewayService.route(request.getProvider(), request.getModelName()).mcpToolGenerate(request); return ResponseEntity.ok(response); } catch (Exception e) { log.error(Invalid request: {}, e); } return null; }新增入参// 工具参数JSON 字符串用于手动调用工具 private String toolName; private String toolArguments;SiliconFlowProvider逻辑/* 手动调用mcp tool 方案1使用McpClient依赖langchain4j-mcp但在正式版本中不存在存在于beta版本 方案2使用 ProcessBuilder 和 JSON-RPC 手动实现 MCP 协议 */ Override public ChatResponseDTO mcpToolGenerate(ChatRequestDTO request) { Process mcpProcess null; BufferedWriter writer null; BufferedReader reader null; try { // 处理会话ID String sessionId request.getSessionId() ! null ? request.getSessionId() : default; if (Boolean.TRUE.equals(request.getClearHistory())) { memoryService.clearMemory(sessionId); log.info(Cleared memory for session: {}, sessionId); } ChatMemory memory memoryService.getOrCreateMemory(sessionId); // 启动 MCP server 进程 String jarPath C:\\mydemo\\ai-demo-mcp-server\\target\\ai-demo-mcp-server-1.0.0.jar; // 构建 Java 命令 ListString command new ArrayList(); command.add(java); command.add(-jar); command.add(jarPath); command.add(--stdio); ProcessBuilder processBuilder new ProcessBuilder(command); mcpProcess processBuilder.start(); // 获取输入输出流 writer new BufferedWriter(new OutputStreamWriter(mcpProcess.getOutputStream(), UTF-8)); reader new BufferedReader(new InputStreamReader(mcpProcess.getInputStream(), UTF-8)); // 1. 发送 initialize 请求 MapString, Object initRequest new HashMap(); initRequest.put(jsonrpc, 2.0); initRequest.put(id, 1); initRequest.put(method, initialize); MapString, Object initParams new HashMap(); initParams.put(protocolVersion, 2024-11-05); initParams.put(capabilities, new HashMap()); MapString, String clientInfo new HashMap(); clientInfo.put(name, ai-demo); clientInfo.put(version, 1.0.0); initParams.put(clientInfo, clientInfo); initRequest.put(params, initParams); String initJson objectMapper.writeValueAsString(initRequest); writer.write(initJson); writer.newLine(); writer.flush(); log.debug(Sent initialize request: {}, initJson); // 读取 initialize 响应跳过非 JSON 行如 banner 等 String initResponse readJsonResponse(reader); log.debug(MCP initialize response: {}, initResponse); if (initResponse null) { throw new RuntimeException(Failed to receive initialize response from MCP server); } // 验证 initialize 响应是否成功 try { JsonNode initResponseNode objectMapper.readTree(initResponse); if (initResponseNode.has(error)) { JsonNode error initResponseNode.path(error); String errorMessage error.path(message).asText(Unknown error); throw new RuntimeException(MCP initialize error: errorMessage); } log.info(MCP server initialized successfully); } catch (Exception e) { log.warn(Failed to parse initialize response, continuing anyway: {}, e.getMessage()); } // 2. 发送 initialized 通知 // 注意根据 JSON-RPC 规范通知notification不应该有响应 // 但某些 MCP server 实现可能会返回错误响应我们需要读取并忽略它 try { MapString, Object initialized new HashMap(); initialized.put(jsonrpc, 2.0); initialized.put(method, notifications/initialized); // 某些实现可能需要空的 params initialized.put(params, new HashMap()); writer.write(objectMapper.writeValueAsString(initialized)); writer.newLine(); writer.flush(); log.debug(Sent initialized notification); // 某些服务器可能会返回错误响应虽然不符合规范我们需要读取并忽略 // 使用短暂的等待来检查是否有响应 try { // 设置一个短暂的超时来检查是否有响应 // 由于 BufferedReader.readLine() 是阻塞的我们直接尝试读取 // 如果服务器不支持可能会立即返回错误 String notificationResponse readJsonResponse(reader); if (notificationResponse ! null) { JsonNode responseNode objectMapper.readTree(notificationResponse); if (responseNode.has(error)) { JsonNode error responseNode.path(error); String errorMessage error.path(message).asText(); // 如果错误是 Method not found说明服务器不支持这个通知这是可以接受的 if (errorMessage.contains(not found) || errorMessage.contains(Method not found)) { log.debug(Server does not support notifications/initialized notification (this is OK)); } else { log.warn(Server returned error for initialized notification: {}, errorMessage); } } else { log.debug(Server responded to initialized notification (unexpected but OK)); } } } catch (Exception e) { // 如果没有响应或读取失败这是正常的通知不应该有响应 log.debug(No response to initialized notification (this is expected): {}, e.getMessage()); } } catch (Exception e) { log.warn(Error sending initialized notification: {}, e.getMessage()); // 继续执行某些服务器可能不需要这个通知 } // 3. 获取工具名称和参数 String toolName request.getToolName(); if (toolName null || toolName.isEmpty()) { // 先列出可用工具 MapString, Object listToolsRequest new HashMap(); listToolsRequest.put(jsonrpc, 2.0); listToolsRequest.put(id, 2); listToolsRequest.put(method, tools/list); listToolsRequest.put(params, new HashMap()); writer.write(objectMapper.writeValueAsString(listToolsRequest)); writer.newLine(); writer.flush(); String toolsResponse readJsonResponse(reader); log.debug(MCP tools list response: {}, toolsResponse); // 解析工具列表可选如果需要自动选择工具 if (toolsResponse ! null) { JsonNode toolsNode objectMapper.readTree(toolsResponse); if (toolsNode.has(result) toolsNode.path(result).has(tools)) { JsonNode tools toolsNode.path(result).path(tools); if (tools.isArray() tools.size() 0) { // 使用第一个工具作为默认工具 toolName tools.get(0).path(name).asText(); log.info(Auto-selected tool: {}, toolName); } } } } if (toolName null || toolName.isEmpty()) { throw new RuntimeException(Tool name is required but not provided); } // 4. 获取工具参数 String toolArgs request.getToolArguments(); if (toolArgs null || toolArgs.isEmpty()) { toolArgs {}; } // 验证 JSON 格式 JsonNode argsNode; try { argsNode objectMapper.readTree(toolArgs); } catch (Exception e) { log.warn(Invalid JSON format for tool arguments, using empty object: {}, e.getMessage()); argsNode objectMapper.readTree({}); } // 5. 构建工具调用请求 MapString, Object callRequest new HashMap(); callRequest.put(jsonrpc, 2.0); callRequest.put(id, 3); callRequest.put(method, tools/call); MapString, Object callParams new HashMap(); callParams.put(name, toolName); callParams.put(arguments, argsNode); callRequest.put(params, callParams); // 发送工具调用请求 String callJson objectMapper.writeValueAsString(callRequest); writer.write(callJson); writer.newLine(); writer.flush(); log.debug(Sent tool call request: {}, callJson); // 读取工具执行结果跳过非 JSON 行 String toolResponse readJsonResponse(reader); log.debug(MCP tool call response: {}, toolResponse); if (toolResponse null) { throw new RuntimeException(Failed to receive tool call response from MCP server); } // 6. 解析结果 JsonNode responseNode objectMapper.readTree(toolResponse); String toolResult ; if (responseNode.has(error)) { JsonNode error responseNode.path(error); String errorMessage error.path(message).asText(Unknown error); throw new RuntimeException(MCP tool execution error: errorMessage); } if (responseNode.has(result)) { JsonNode result responseNode.path(result); if (result.has(content)) { JsonNode content result.path(content); if (content.isArray() content.size() 0) { // 提取第一个内容项的文本 JsonNode firstContent content.get(0); if (firstContent.has(text)) { toolResult firstContent.path(text).asText(); } else if (firstContent.has(type) text.equals(firstContent.path(type).asText())) { toolResult firstContent.toString(); } } else if (content.isTextual()) { toolResult content.asText(); } else if (content.isObject()) { // 尝试提取文本字段 if (content.has(text)) { toolResult content.path(text).asText(); } else { toolResult content.toString(); } } } else { // 如果没有 content 字段直接使用 result 的字符串表示 toolResult result.toString(); } } log.info(MCP tool execution result: {}, toolResult); // 7. 创建聊天模型将工具结果整合到对话中 ChatLanguageModel chatModel OpenAiChatModel.builder() .apiKey(apiKey) .baseUrl(baseUrl) .modelName(request.getModelName()) .timeout(Duration.ofSeconds(60)) .build(); // 构建包含工具结果的用户消息 String enhancedMessage request.getMessage(); if (toolResult ! null !toolResult.isEmpty()) { enhancedMessage String.format( 用户问题%s\n\n工具执行结果%s\n\n请基于工具执行结果回答用户问题。, request.getMessage(), toolResult ); } // 添加用户消息到 Memory memory.add(UserMessage.from(enhancedMessage)); // 生成回复 ResponseAiMessage response chatModel.generate(memory.messages()); String responseText response.content().text(); // 更新记忆添加 AI 回复 memory.add(response.content()); log.debug(MCP tool chat completed for session: {}, response length: {}, sessionId, responseText.length()); return ChatResponseDTO.builder() .content(responseText) .build(); } catch (Exception e) { log.error(Error in mcpToolGenerate: {}, e.getMessage(), e); return ChatResponseDTO.builder() .content(调用 MCP tool 时发生错误: e.getMessage()) .build(); } finally { // 清理资源 try { if (writer ! null) { writer.close(); } if (reader ! null) { reader.close(); } if (mcpProcess ! null) { mcpProcess.destroy(); if (!mcpProcess.waitFor(5, TimeUnit.SECONDS)) { log.warn(MCP process did not terminate gracefully, forcing destroy); mcpProcess.destroyForcibly(); } } } catch (Exception e) { log.warn(Error closing MCP process: {}, e.getMessage()); } } } /** * 从 BufferedReader 读取有效的 JSON 响应跳过所有非 JSON 行如 banner、日志等 * param reader BufferedReader 实例 * return 有效的 JSON 字符串如果超时或流关闭则返回 null */ private String readJsonResponse(BufferedReader reader) { try { String line; int maxAttempts 100; // 最多尝试读取 100 行 int attempts 0; StringBuilder jsonBuilder null; // 用于构建多行 JSON while (attempts maxAttempts) { line reader.readLine(); if (line null) { // 流已关闭 log.warn(Stream closed while reading JSON response); return jsonBuilder ! null ? jsonBuilder.toString() : null; } // 去除首尾空白 String trimmedLine line.trim(); // 跳过空行 if (trimmedLine.isEmpty()) { // 如果正在构建 JSON空行可能表示 JSON 结束 if (jsonBuilder ! null) { try { String jsonStr jsonBuilder.toString(); objectMapper.readTree(jsonStr); return jsonStr; } catch (Exception e) { // JSON 还不完整继续读取 jsonBuilder.append(\n); } } attempts; continue; } // 检查是否是有效的 JSON 开始以 { 或 [ 开头 if (trimmedLine.startsWith({) || trimmedLine.startsWith([)) { // 如果之前没有开始构建开始构建 if (jsonBuilder null) { jsonBuilder new StringBuilder(trimmedLine); } else { jsonBuilder.append(trimmedLine); } // 尝试解析 JSON 以验证是否完整 try { String jsonStr jsonBuilder.toString(); objectMapper.readTree(jsonStr); // 如果解析成功返回完整的 JSON return jsonStr; } catch (Exception e) { // 如果解析失败可能是多行 JSON继续读取 log.debug(JSON not complete yet, continuing to read...); jsonBuilder.append(\n); } } else { // 不是 JSON 开始可能是 // 1. banner 或日志跳过 // 2. 多行 JSON 的中间部分如果已开始构建 JSON则追加 if (jsonBuilder ! null) { // 正在构建 JSON追加这一行 jsonBuilder.append(trimmedLine); // 尝试解析 try { String jsonStr jsonBuilder.toString(); objectMapper.readTree(jsonStr); return jsonStr; } catch (Exception e) { // 还不完整继续 jsonBuilder.append(\n); } } else { // 不是 JSON记录并跳过可能是 banner、日志等 log.debug(Skipping non-JSON line: {}, trimmedLine.length() 100 ? trimmedLine.substring(0, 100) ... : trimmedLine); } } attempts; } // 如果最后还有未完成的 JSON尝试返回 if (jsonBuilder ! null) { try { String jsonStr jsonBuilder.toString(); objectMapper.readTree(jsonStr); return jsonStr; } catch (Exception e) { String jsonStr jsonBuilder.toString(); log.warn(Incomplete JSON after max attempts: {}, jsonStr.substring(0, Math.min(100, jsonStr.length()))); } } log.error(Failed to read valid JSON response after {} attempts, maxAttempts); return null; } catch (Exception e) { log.error(Error reading JSON response: {}, e.getMessage(), e); return null; } }测试1searchLink2searchEmployee输入错误的输入正确的三、McpClient法mcpClient McpAdapter.initMcpServer(serverDefinition.getConfig()); Constructor? constructor ToolExecutionRequest.Builder.class.getDeclaredConstructor(); constructor.setAccessible(true); ToolExecutionRequest.Builder reqBuilder (ToolExecutionRequest.Builder) constructor.newInstance(); ToolExecutionRequest request reqBuilder.name(toolName).arguments(args.toJSONString()).build(); return mcpClient.executeTool(request);