前言

官方示例的代码中，mcp一般是配置到yml中或者json文件中，使用自动装配的方式注入服务，这种方式不方便在程序启动后添加新的服务，这里参考cherry studio的方式动态添加mcp服务

1.确定方案

• mcp服务的维护放到mysql业务数据库维护，前端通过json来添加mcp服务，为什么是json？因为开源的mcp服务都提供json示例，拿来即用

• 后端轮询mcp服务确保服务可用状态

• 前端动态切换模型时，根据模型是否支持工具调和是否启用mcp来控制是否使用工具

• 一个mcp服务下可能有一系列的工具，提供一个查看mcp服务工具的页面

2. pom依赖

 <!-- Spring AI MCP 核心包 (包含ToolCallbackProvider) --><dependency><groupId>org.springframework.ai</groupId><artifactId>spring-ai-starter-mcp-client</artifactId></dependency><!-- Spring AI Model (ToolCallback接口等) --><dependency><groupId>org.springframework.ai</groupId><artifactId>spring-ai-model</artifactId><version>${spring-ai.version}</version></dependency>

其他的依赖见前面几篇文章

3. 新增mcp表

CREATE TABLE `ai_mcp` (`id` bigint NOT NULL AUTO_INCREMENT,`content_type` varchar(64) CHARACTER SET utf8mb4 COLLATE utf8mb4_german2_ci NOT NULL COMMENT '连接类型，sse,stdio',`name` varchar(128) CHARACTER SET utf8mb4 COLLATE utf8mb4_german2_ci NOT NULL COMMENT 'mcp名称',`description` varchar(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_german2_ci DEFAULT NULL COMMENT '功能描述',`config_json` json NOT NULL COMMENT '配置参数，json类型',`status` tinyint(1) DEFAULT NULL COMMENT '可用状态，1可用，0不可用',`update_time` timestamp NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=11 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_german2_ci;

4.补充增删改查

使用的mybatis-plus，省略mapper层

import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.lanyu.springainovel.entity.AiMcp;
import org.lanyu.springainovel.entity.ServerConfig;
import org.lanyu.springainovel.mapper.AiMcpMapper;
import org.lanyu.springainovel.util.ConversionUtil;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;import java.sql.Timestamp;
import java.util.List;@Service
public class AiMcpService {public AiMcpService() {super();}@Autowiredprivate AiMcpMapper aiMcpMapper;public List<AiMcp> getAllEnabledMcp() {QueryWrapper<AiMcp> query = new QueryWrapper<>();query.eq("status", 1);return aiMcpMapper.selectList(query);}public void addMcp(AiMcp mcp, DynamicMcpClientManager dynamicMcpClientManager) {mcp.setUpdateTime(new Timestamp(System.currentTimeMillis()));// 从configJson中提取mcpServers作为mcp名称和类型try {ObjectMapper objectMapper = new ObjectMapper();JsonNode root = objectMapper.readTree(mcp.getConfigJson());JsonNode serversNode = root.path("mcpServers");if (serversNode.isObject() && serversNode.size() > 0) {// 获取第一个服务器名称作为mcp名称String serverName = serversNode.fieldNames().next();mcp.setName(serverName);// 获取服务器类型JsonNode serverNode = serversNode.path(serverName);String type = serverNode.path("type").asText("");if ("sse".equalsIgnoreCase(type) || "stdio".equalsIgnoreCase(type)) {mcp.setContentType(type);}}} catch (Exception e) {// 日志输出}aiMcpMapper.insert(mcp);if (mcp.getStatus() != null && mcp.getStatus() == 1) {// 解析 configJson，提取所有 servertry {ObjectMapper objectMapper = new ObjectMapper();JsonNode root = objectMapper.readTree(mcp.getConfigJson());JsonNode serversNode = root.path("mcpServers");if (serversNode.isObject()) {serversNode.fields().forEachRemaining(entry -> {String serverName = entry.getKey();JsonNode serverNode = entry.getValue();String type = serverNode.path("type").asText("");if ("sse".equalsIgnoreCase(type) || "stdio".equalsIgnoreCase(type)) {ServerConfig config = ConversionUtil.parseServerConfigFromJson(serverName, serverNode, type);if (config != null) {dynamicMcpClientManager.addOrUpdateServerConfig(serverName, config);}}});}} catch (Exception e) {// 日志输出}}}public void updateMcp(AiMcp mcp) {mcp.setUpdateTime(new Timestamp(System.currentTimeMillis()));// 从configJson中提取mcpServers作为mcp名称try {ObjectMapper objectMapper = new ObjectMapper();JsonNode root = objectMapper.readTree(mcp.getConfigJson());JsonNode serversNode = root.path("mcpServers");if (serversNode.isObject() && serversNode.size() > 0) {// 获取第一个服务器名称作为mcp名称String serverName = serversNode.fieldNames().next();mcp.setName(serverName);// 获取服务器类型JsonNode serverNode = serversNode.path(serverName);String type = serverNode.path("type").asText("");if ("sse".equalsIgnoreCase(type) || "stdio".equalsIgnoreCase(type)) {mcp.setContentType(type);}}} catch (Exception e) {// 日志输出}aiMcpMapper.updateById(mcp);}public void deleteMcp(Long id, DynamicMcpClientManager dynamicMcpClientManager) {AiMcp mcp = aiMcpMapper.selectById(id);if (mcp != null) {// 解析 configJson，提取所有 servertry {ObjectMapper objectMapper = new ObjectMapper();JsonNode root = objectMapper.readTree(mcp.getConfigJson());JsonNode serversNode = root.path("mcpServers");if (serversNode.isObject()) {serversNode.fields().forEachRemaining(entry -> {String serverName = entry.getKey();dynamicMcpClientManager.removeServerConfig(serverName);});}} catch (Exception e) {// 日志输出}aiMcpMapper.deleteById(id);}}public AiMcp getById(Long id) {return aiMcpMapper.selectById(id);}public AiMcp getByName(String name) {QueryWrapper<AiMcp> query = new QueryWrapper<>();query.eq("name", name);return aiMcpMapper.selectOne(query);}public Page<AiMcp> pageQuery(String name, int page, int size) {QueryWrapper<AiMcp> query = new QueryWrapper<>();if (name != null && !name.isEmpty()) {query.like("name", name);}query.orderByDesc("update_time");return aiMcpMapper.selectPage(new Page<>(page, size), query);}
}

5.创建mcp管理类（核心）

/*** 动态MCP客户端管理器，支持热加载和配置变更。** <p>此服务提供以下功能：* <ul>* <li>动态读取MCP服务器配置</li>* <li>自动连接和重连MCP服务器</li>* <li>实时发现新工具</li>* <li>配置变更时自动更新连接</li>* <li>健康检查和故障恢复</li>* </ul>** @author Spring AI Team* @since 1.1.0*/
@Service
public class DynamicMcpClientManager implements DisposableBean {private static final Logger logger = LoggerFactory.getLogger(DynamicMcpClientManager.class);/*** 活跃的MCP客户端，key为服务器名，value为McpSyncClient实例*/private final Map<String, McpSyncClient> activeClients = new ConcurrentHashMap<>();/*** 当前连接状态的配置，key为服务器名，value为ServerConfig*/private final Map<String, ServerConfig> currentConfigs = new ConcurrentHashMap<>();/*** 定时任务调度器*/private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(2);/*** AiMcpService用于数据库操作*/private final AiMcpService aiMcpService;/*** 缓存的工具回调列表，定期刷新*/private volatile List<SyncMcpToolCallback> cachedToolCallbacks = new ArrayList<>();/*** 上次工具回调刷新的时间戳*/private volatile long lastToolRefresh = 0;/*** 工具回调缓存的有效期（毫秒）*/private static final long TOOL_CACHE_TTL = 30000; // 30秒缓存/*** 最大重连次数，所有重连逻辑统一引用该常量*/private static final int MAX_RECONNECT_ATTEMPTS = 3;public DynamicMcpClientManager(AiMcpService aiMcpService) {this.aiMcpService = aiMcpService;}@PostConstructpublic void init() {loadSpringAiMcpConfiguration();
//        scheduler.scheduleWithFixedDelay(this::loadSpringAiMcpConfiguration, 30, 30, TimeUnit.SECONDS);scheduler.scheduleWithFixedDelay(this::performHealthCheck, 30, 30, TimeUnit.SECONDS);logger.info("动态MCP客户端管理器已启动");}/*** 从数据库加载MCP配置，只有配置实际变更时才调用updateServerConfigs*/public void loadSpringAiMcpConfiguration() {logger.info("从数据库加载MCP配置");Map<String, ServerConfig> newConfigs = new ConcurrentHashMap<>();try {for (AiMcp mcp : aiMcpService.getAllEnabledMcp()) {if (mcp.getConfigJson() == null || mcp.getConfigJson().isEmpty()) {logger.warn("MCP配置[{}]缺少configJson，跳过", mcp.getName());continue;}try {JsonNode root = new ObjectMapper().readTree(mcp.getConfigJson());JsonNode serversNode = root.path("mcpServers");if (serversNode.isMissingNode() || !serversNode.isObject()) {logger.warn("MCP配置[{}]的mcpServers字段缺失或格式错误，跳过", mcp.getName());continue;}serversNode.fields().forEachRemaining(entry -> {String serverName = entry.getKey();JsonNode serverNode = entry.getValue();String type = serverNode.path("type").asText("");ServerConfig config = ConversionUtil.parseServerConfigFromJson(serverName, serverNode, type);if (config != null) {newConfigs.put(serverName, config);}});} catch (Exception e) {logger.warn("MCP配置[{}]解析configJson失败: {}，跳过", mcp.getName(), e.getMessage());}}} catch (Exception e) {logger.error("从数据库加载MCP配置失败", e);}// 只有配置实际变更时才更新if (!newConfigs.equals(currentConfigs)) {updateServerConfigs(newConfigs);} else {logger.debug("MCP配置无变更，无需更新");}}/*** 更新服务器配置并重新连接。重连次数由MAX_RECONNECT_ATTEMPTS控制。*/public synchronized void updateServerConfigs(Map<String, ServerConfig> newConfigs) {logger.info("更新MCP服务器配置，共 {} 个服务器", newConfigs.size());// 移除不再存在的服务器currentConfigs.keySet().removeIf(serverName -> {if (!newConfigs.containsKey(serverName)) {disconnectServer(serverName);return true;}return false;});// 添加或更新服务器配置for (Map.Entry<String, ServerConfig> entry : newConfigs.entrySet()) {String serverName = entry.getKey();ServerConfig newConfig = entry.getValue();ServerConfig oldConfig = currentConfigs.get(serverName);// 如果配置发生变化或新加，尝试连接if (oldConfig == null || !configEquals(oldConfig, newConfig)) {currentConfigs.put(serverName, newConfig);if (newConfig.isEnabled()) {boolean connected = false;int attempts = 0;while (!connected && attempts < MAX_RECONNECT_ATTEMPTS) {attempts++;logger.info("[重连策略] 连接/重连MCP服务器[{}]，第{}次尝试...", serverName, attempts);connectServer(serverName, newConfig);if (activeClients.containsKey(serverName)) {logger.info("[重连策略] 连接/重连MCP服务器[{}]成功", serverName);connected = true;} else {logger.warn("[重连策略] 连接/重连MCP服务器[{}]失败，等待2分钟后重试...", serverName);try {Thread.sleep(120_000);} catch (InterruptedException e) {Thread.currentThread().interrupt();}}}if (!connected) {logger.error("[重连策略] MCP服务器[{}]重连已达最大次数({})，将自动禁用该服务", serverName, MAX_RECONNECT_ATTEMPTS);disableMcpInDatabase(serverName);}} else {disconnectServer(serverName);}}}// 清除工具缓存，强制重新加载invalidateToolCache();}/*** 连接到MCP服务器（重连和首次连接参数完全一致，重连前彻底释放资源）*/private void connectServer(String serverName, ServerConfig config) {logger.info("连接到MCP服务器: {} ({})", serverName, config.getUrl());// 1. 彻底释放旧资源disconnectServer(serverName);try {// 2. 构建全新 McpSyncClient，参数与首次一致io.modelcontextprotocol.client.McpSyncClient client;if (config.getUrl().startsWith("stdio://")) {logger.info("使用STDIO传输连接服务器: {}", serverName);// TODO: STDIO传输实现logger.warn("STDIO传输暂未实现，服务器: {}", serverName);return;} else {logger.info("使用SSE传输连接服务器: {} -> {} (endpoint: {})", serverName, config.getUrl(), config.getSseEndpoint());HttpClientSseClientTransport transport = HttpClientSseClientTransport.builder(config.getUrl()).clientBuilder(HttpClient.newBuilder()).sseEndpoint(config.getSseEndpoint()).build();client = McpClient.sync(transport).requestTimeout(config.getTimeout()).build();}// 3. 初始化客户端client.initialize();activeClients.put(serverName, client);logger.info("成功连接到MCP服务器: {}", serverName);} catch (Exception e) {logger.error("连接MCP服务器失败: {} - {} (url: {}, sseEndpoint: {})", serverName, e.getMessage(), config.getUrl(), config.getSseEndpoint(), e);}}/*** 彻底断开MCP服务器连接，释放所有资源*/private void disconnectServer(String serverName) {io.modelcontextprotocol.client.McpSyncClient client = activeClients.remove(serverName);if (client != null) {try {client.close();logger.info("已断开MCP服务器连接: {}", serverName);} catch (Exception e) {logger.warn("断开MCP服务器连接时出错: {} - {}", serverName, e.getMessage());}}// 清理相关缓存// cachedToolCallbacks = new ArrayList<>(); // 若有必要可清理}/*** 获取所有活跃的客户端*/public List<McpSyncClient> getActiveClients() {return new ArrayList<>(activeClients.values());}/*** 获取所有可用的工具回调（带缓存）*/public List<ToolCallback> getAvailableToolCallbacks() {long now = System.currentTimeMillis();if (now - lastToolRefresh > TOOL_CACHE_TTL || cachedToolCallbacks.isEmpty()) {refreshToolCallbacks();lastToolRefresh = now;}return new ArrayList<>(cachedToolCallbacks);}/*** 强制刷新工具回调，重新从所有活跃客户端获取工具*/public synchronized void refreshToolCallbacks() {logger.info("刷新MCP工具回调");List<McpSyncClient> clients = getActiveClients();if (clients.isEmpty()) {cachedToolCallbacks = new ArrayList<>();return;}try {logger.info("准备调用createToolCallbacks");List<SyncMcpToolCallback> newCallbacks = SimpleMcpToolCallbackProvider.createToolCallbacks(clients);logger.info("createToolCallbacks调用工具结果：{}", newCallbacks.size());cachedToolCallbacks = newCallbacks;logger.info("刷新{}个工具", newCallbacks.size());} catch (Exception e) {logger.error("刷新工具回调失败", e);}}/*** 获取指定Mcp服务的工具列表*/public List<SyncMcpToolCallback> getToolByMcpName(String mcpName) {McpSyncClient client = activeClients.get(mcpName);if (client == null) {return new ArrayList<>();}try {return SimpleMcpToolCallbackProvider.getToolCallbacksByClientName(client);} catch (Exception e) {return new ArrayList<>();}}/*** 清除工具缓存*/public void invalidateToolCache() {lastToolRefresh = 0;logger.debug("工具缓存已清除");}/*** 执行健康检查*/private void performHealthCheck() {logger.info("执行MCP客户端健康检查");List<String> unhealthyServers = new ArrayList<>();for (Map.Entry<String, McpSyncClient> entry : activeClients.entrySet()) {String serverName = entry.getKey();McpSyncClient client = entry.getValue();try {List<McpSchema.Tool> tools = client.listTools().tools();//logger.info("检查工具：{}", tools.toString());} catch (Exception e) {logger.warn("MCP服务器 {} 健康检查失败: {}，将尝试重连", serverName, e.getMessage());unhealthyServers.add(serverName);}}// 只负责发现断开，重连交给 reconnectWithLimitfor (String serverName : unhealthyServers) {ServerConfig config = currentConfigs.get(serverName);if (config != null && config.isEnabled()) {logger.info("MCP服务器 {} 已断开，将尝试重连", serverName);reconnectWithLimit(serverName, config);}}}/*** 比较两个配置是否相等*/private boolean configEquals(ServerConfig config1, ServerConfig config2) {if (config1 == config2) return true;if (config1 == null || config2 == null) return false;return java.util.Objects.equals(config1.getUrl(), config2.getUrl()) &&java.util.Objects.equals(config1.getSseEndpoint(), config2.getSseEndpoint()) &&java.util.Objects.equals(config1.getHeaders(), config2.getHeaders()) &&java.util.Objects.equals(config1.getTimeout(), config2.getTimeout()) &&config1.isEnabled() == config2.isEnabled();}private void reconnectWithLimit(String serverName, ServerConfig config) {int attempts = 0;boolean connected = false;while (attempts < MAX_RECONNECT_ATTEMPTS && !connected) {attempts++;logger.warn("重连MCP服务器[{}]，第{}次尝试...，最大{}次", serverName, attempts, MAX_RECONNECT_ATTEMPTS);try {connectServer(serverName, config);if (activeClients.containsKey(serverName)) {logger.info("重连MCP服务器[{}]成功", serverName);connected = true;}} catch (Exception e) {logger.error("重连MCP服务器[{}]失败: {}", serverName, e.getMessage());}}if (!connected) {logger.error("MCP服务器[{}]重连已达最大次数({})，将自动禁用该服务", serverName, MAX_RECONNECT_ATTEMPTS);disableMcpInDatabase(serverName);}}private void disableMcpInDatabase(String serverName) {try {org.lanyu.springainovel.entity.AiMcp mcp = aiMcpService.getByName(serverName);if (mcp != null && mcp.getStatus() != null && mcp.getStatus() != 0) {mcp.setStatus(0);aiMcpService.updateMcp(mcp);logger.warn("已将MCP服务[{}]在数据库中禁用(status=0)", serverName);}// 同步禁用缓存ServerConfigServerConfig config = currentConfigs.get(serverName);if (config != null && config.isEnabled()) {config.setEnabled(false);logger.warn("已将MCP服务[{}]在缓存中禁用(enable=false)", serverName);}// 同步禁用McpConfigurationService的lastKnownConfigstry {// 反射获取AiMcpService中的McpConfigurationServicejava.lang.reflect.Field field = aiMcpService.getClass().getDeclaredField("mcpConfigurationService");field.setAccessible(true);Object mcpConfigServiceObj = field.get(aiMcpService);if (mcpConfigServiceObj != null) {java.lang.reflect.Field lastKnownConfigsField = mcpConfigServiceObj.getClass().getDeclaredField("lastKnownConfigs");lastKnownConfigsField.setAccessible(true);@SuppressWarnings("unchecked")Map<String, ServerConfig> lastKnownConfigs = (Map<String, ServerConfig>) lastKnownConfigsField.get(mcpConfigServiceObj);ServerConfig lastConfig = lastKnownConfigs.get(serverName);if (lastConfig != null && lastConfig.isEnabled()) {lastConfig.setEnabled(false);logger.warn("已将MCP服务[{}]在lastKnownConfigs中禁用(enable=false)", serverName);}}} catch (Exception e) {logger.error("同步禁用lastKnownConfigs缓存失败: {}", e.getMessage());}} catch (Exception e) {logger.error("禁用MCP服务[{}]数据库操作失败: {}", serverName, e.getMessage());}}/*** 新增或更新单个MCP服务器配置*/public synchronized void addOrUpdateServerConfig(String serverName, ServerConfig config) {Map<String, ServerConfig> newConfigs = new ConcurrentHashMap<>(currentConfigs);newConfigs.put(serverName, config);updateServerConfigs(newConfigs);}/*** 移除单个MCP服务器配置*/public synchronized void removeServerConfig(String serverName) {Map<String, ServerConfig> newConfigs = new ConcurrentHashMap<>(currentConfigs);newConfigs.remove(serverName);updateServerConfigs(newConfigs);}@Overridepublic void destroy() throws Exception {logger.info("关闭动态MCP客户端管理器");// 关闭调度器scheduler.shutdown();try {if (!scheduler.awaitTermination(5, TimeUnit.SECONDS)) {scheduler.shutdownNow();}} catch (InterruptedException e) {scheduler.shutdownNow();Thread.currentThread().interrupt();}// 关闭所有客户端连接for (String serverName : new ArrayList<>(activeClients.keySet())) {disconnectServer(serverName);}}
}

/*** 简化的MCP工具回调提供者，用于演示目的。** <p>此类提供了从MCP客户端创建工具回调的基本功能。** @author Spring AI Team* @since 1.1.0*/
public class SimpleMcpToolCallbackProvider {private static final Logger logger = LoggerFactory.getLogger(SimpleMcpToolCallbackProvider.class);private static final ObjectMapper objectMapper = new ObjectMapper();/*** 从MCP客户端列表创建工具回调列表*/public static List<SyncMcpToolCallback> createToolCallbacks(List<McpSyncClient> clients) {List<SyncMcpToolCallback> callbacks = new ArrayList<>();for (McpSyncClient client : clients) {try {// 获取工具列表McpSchema.ListToolsResult toolsResult = client.listTools();if (toolsResult != null && toolsResult.tools() != null) {for (McpSchema.Tool tool : toolsResult.tools()) {callbacks.add(new SyncMcpToolCallback(client, tool));}}} catch (Exception e) {logger.error("获取客户端工具失败", e);}}return callbacks;}/*** 从MCP客户端列表创建工具回调列表*/public static List<SyncMcpToolCallback> getToolCallbacksByClientName(McpSyncClient client) {List<SyncMcpToolCallback> callbacks = new ArrayList<>();try {// 获取工具列表McpSchema.ListToolsResult toolsResult = client.listTools();if (toolsResult != null && toolsResult.tools() != null) {for (McpSchema.Tool tool : toolsResult.tools()) {callbacks.add(new SyncMcpToolCallback(client, tool));}}} catch (Exception e) {logger.error("获取客户端工具失败", e);}return callbacks;}/*** 简化的MCP工具回调实现*/public static class SimpleMcpToolCallback implements ToolCallback {private final McpSyncClient client;private final McpSchema.Tool tool;private final ToolDefinition toolDefinition;public SimpleMcpToolCallback(McpSyncClient client, McpSchema.Tool tool) {this.client = client;this.tool = tool;this.toolDefinition = ToolDefinition.builder().name(tool.name()).description(tool.description()).inputSchema(tool.inputSchema() != null ? tool.inputSchema().toString() : "{}").build();}@Overridepublic ToolDefinition getToolDefinition() {return toolDefinition;}@Overridepublic String call(String arguments) {try {// 解析参数@SuppressWarnings("unchecked")Map<String, Object> args = objectMapper.readValue(arguments, Map.class);// 调用MCP工具McpSchema.CallToolRequest request = new McpSchema.CallToolRequest(tool.name(), args);McpSchema.CallToolResult result = client.callTool(request);if (result != null && result.content() != null && !result.content().isEmpty()) {// 提取文本内容StringBuilder response = new StringBuilder();for (Object content : result.content()) {if (content instanceof McpSchema.TextContent) {McpSchema.TextContent textContent = (McpSchema.TextContent) content;response.append(textContent.text());} else {response.append(content.toString());}}return response.toString();} else {return "工具执行完成，无返回内容";}} catch (Exception e) {logger.error("调用MCP工具失败: {}", tool.name(), e);return "错误：" + e.getMessage();}}}
}

/*** 服务器配置类*/
public class ServerConfig {public ServerConfig() { super(); }private String name;private String url;private String sseEndpoint; // SSE端点，默认为/sseprivate Map<String, String> headers;private Duration timeout = Duration.ofSeconds(30);private boolean enabled = true;// Getters and Setterspublic String getName() { return name; }public void setName(String name) { this.name = name; }public String getUrl() { return url; }public void setUrl(String url) { this.url = url; }public String getSseEndpoint() { return sseEndpoint; }public void setSseEndpoint(String sseEndpoint) { this.sseEndpoint = sseEndpoint; }public Map<String, String> getHeaders() { return headers; }public void setHeaders(Map<String, String> headers) { this.headers = headers; }public Duration getTimeout() { return timeout; }public void setTimeout(Duration timeout) { this.timeout = timeout; }public boolean isEnabled() { return enabled; }public void setEnabled(boolean enabled) { this.enabled = enabled; }/*** 获取完整的SSE URL（基础URL + SSE端点）*/public String getFullSseUrl() {if (url == null) return null;if (url.startsWith("stdio://")) return url;String baseUrl = url.endsWith("/") ? url.substring(0, url.length() - 1) : url;String endpoint = sseEndpoint.startsWith("/") ? sseEndpoint : "/" + sseEndpoint;return baseUrl + endpoint;}
}

6.Mcp的相关接口

/*** MCP工具控制器*/
@Tag(name = "mcp工具", description = "供统一的REST API来调用MCP工具")
@RestController
@RequestMapping("/mcp")
public class McpToolController {private static final Logger logger = LoggerFactory.getLogger(McpToolController.class);/*** 动态MCP客户端管理器，负责动态连接和管理MCP服务器*/@Autowired(required = false)private DynamicMcpClientManager dynamicClientManager;/*** AiMcpService，负责MCP工具的数据库操作*/@Autowiredprivate AiMcpService aiMcpService;public McpToolController() {super(); // 默认无参构造，调用父类构造函数}/*** 获取当前有效的工具回调数组，优先使用动态管理器** @return ToolCallback[] 当前可用的工具回调*/private ToolCallback[] getCurrentToolCallbacks() {List<ToolCallback> dynamicTools = dynamicClientManager.getAvailableToolCallbacks();if (!dynamicTools.isEmpty()) {return dynamicTools.toArray(new ToolCallback[0]);}return new ToolCallback[0];}/*** 获取MCP配置状态** @return 配置状态信息*/@Operation(summary = "获取MCP配置状态", description = "返回当前MCP工具的配置模式、工具数量、可用工具列表等状态信息。")@GetMapping("/status")public Map<String, Object> getStatus() {Map<String, Object> status = new HashMap<>();ToolCallback[] currentCallbacks = getCurrentToolCallbacks();status.put("toolCount", currentCallbacks.length);status.put("hasDynamicManager", dynamicClientManager != null);// 添加工具列表List<Map<String, String>> toolList = new ArrayList<>();for (ToolCallback callback : currentCallbacks) {Map<String, String> tool = new HashMap<>();tool.put("name", callback.getToolDefinition().name());tool.put("description", callback.getToolDefinition().description());toolList.add(tool);}status.put("tools", toolList);logger.debug("📊 状态查询 - 工具数: {}", currentCallbacks.length);return status;}/*** 获取可用工具列表** @return 工具列表*/@Operation(summary = "获取可用MCP工具列表", description = "返回当前可用的MCP工具及其描述、输入参数等信息。")@GetMapping("/tools")public Map<String, Object> getTools() {Map<String, Object> result = new HashMap<>();try {ToolCallback[] currentCallbacks = getCurrentToolCallbacks();List<Map<String, Object>> toolList = new ArrayList<>();for (ToolCallback callback : currentCallbacks) {Map<String, Object> tool = new HashMap<>();tool.put("name", callback.getToolDefinition().name());tool.put("description", callback.getToolDefinition().description());tool.put("inputSchema", callback.getToolDefinition().inputSchema());toolList.add(tool);}result.put("success", true);result.put("toolCount", currentCallbacks.length);result.put("tools", toolList);logger.debug("🔧 工具列表查询 - 返回 {} 个工具", currentCallbacks.length);} catch (Exception e) {logger.error("❌ 获取工具列表失败", e);result.put("success", false);result.put("message", "获取工具列表失败: " + e.getMessage());result.put("toolCount", 0);result.put("tools", new ArrayList<>());}return result;}/*** 获取可用工具列表** @return 工具列表*/@Operation(summary = "获取指定MCP工具列表", description = "返回当前MCP服务的工具及其描述、输入参数等信息。")@GetMapping("/toolsByName")public Map<String, Object> toolsByName(@RequestParam String mcpName) {Map<String, Object> result = new HashMap<>();List<SyncMcpToolCallback> tools = dynamicClientManager.getToolByMcpName(mcpName);List<Map<String, Object>> toolList = new ArrayList<>();for (ToolCallback callback : tools) {Map<String, Object> tool = new HashMap<>();tool.put("name", callback.getToolDefinition().name());tool.put("description", callback.getToolDefinition().description());tool.put("inputSchema", callback.getToolDefinition().inputSchema());toolList.add(tool);}result.put("success", true);result.put("toolCount", tools.size());result.put("tools", toolList);logger.debug("🔧 工具列表查询 - 返回 {} 个工具", tools.size());return result;}/*** 调用MCP工具** @param toolName  工具名称* @param arguments 工具参数（JSON格式）* @return 调用结果*/@Operation(summary = "调用指定MCP工具", description = "根据工具名称和参数调用MCP工具，返回调用结果。参数为JSON字符串。")@PostMapping("/call/{toolName}")public Map<String, Object> callTool(@PathVariable String toolName, @RequestBody String arguments) {Map<String, Object> result = new HashMap<>();try {ToolCallback[] currentCallbacks = getCurrentToolCallbacks();if (currentCallbacks.length == 0) {result.put("success", false);result.put("message", "当前没有可用的MCP工具");return result;}// 查找指定的工具ToolCallback targetTool = null;for (ToolCallback callback : currentCallbacks) {if (callback.getToolDefinition().name().equals(toolName)) {targetTool = callback;break;}}if (targetTool == null) {result.put("success", false);result.put("message", "工具不存在: " + toolName);result.put("availableTools", Arrays.stream(currentCallbacks).map(cb -> cb.getToolDefinition().name()).toArray());return result;}logger.debug("🔧 调用工具: {}, 参数: {}", toolName, arguments);// 调用工具String toolResult = targetTool.call(arguments);result.put("success", true);result.put("toolName", toolName);result.put("result", toolResult);logger.info("✅ 工具调用成功: {} -> {}", toolName, toolResult);} catch (Exception e) {logger.error("❌ 工具调用失败: {}", toolName, e);result.put("success", false);result.put("message", "调用失败: " + e.getMessage());result.put("toolName", toolName);}return result;}/*** 刷新工具列表（重新从MCP服务器获取）** @return 刷新结果*/@Operation(summary = "刷新MCP工具列表", description = "重新从MCP服务器获取工具列表并刷新缓存。")@PostMapping("/refresh")public Map<String, Object> refreshTools() {Map<String, Object> result = new HashMap<>();try {if (dynamicClientManager != null) {logger.info("🔄 刷新MCP工具列表");dynamicClientManager.refreshToolCallbacks();List<ToolCallback> tools = dynamicClientManager.getAvailableToolCallbacks();result.put("success", true);result.put("message", "工具列表已刷新");result.put("toolCount", tools.size());logger.info("✅ 工具列表刷新完成，发现 {} 个工具", tools.size());} else {result.put("success", false);result.put("message", "动态管理器不可用");}} catch (Exception e) {logger.error("❌ 刷新工具列表失败", e);result.put("success", false);result.put("message", "刷新失败: " + e.getMessage());}return result;}/*** 新增MCP工具*/@Operation(summary = "新增MCP工具", description = "新增一条MCP工具记录")@PostMapping("/entity")public RestVO<String> addMcp(@RequestBody AiMcp mcp) {try {aiMcpService.addMcp(mcp, dynamicClientManager);return RestVO.success("新增成功");} catch (Exception e) {return RestVO.fail("新增失败: " + e.getMessage());}}/*** 修改MCP工具*/@Operation(summary = "修改MCP工具", description = "根据ID修改MCP工具记录")@PutMapping("/entity/{id}")public RestVO<String> updateMcp(@PathVariable Long id, @RequestBody AiMcp mcp) {try {mcp.setId(id);aiMcpService.updateMcp(mcp);return RestVO.success("修改成功");} catch (Exception e) {return RestVO.fail("修改失败: " + e.getMessage());}}/*** 删除MCP工具*/@Operation(summary = "删除MCP工具", description = "根据ID删除MCP工具记录")@DeleteMapping("/entity/{id}")public RestVO<String> deleteMcp(@PathVariable Long id) {try {aiMcpService.deleteMcp(id, dynamicClientManager);return RestVO.success("删除成功");} catch (Exception e) {return RestVO.fail("删除失败: " + e.getMessage());}}/*** 分页查询MCP工具，支持名称模糊搜索，按添加日期倒序*/@Operation(summary = "分页查询MCP工具", description = "分页查询MCP工具，支持名称模糊搜索，按添加日期倒序排列")@GetMapping("/entity/page")public RestVO<Map<String, Object>> pageMcp(@RequestParam(defaultValue = "") String name,@RequestParam(defaultValue = "1") int page,@RequestParam(defaultValue = "10") int size) {try {Page<AiMcp> resultPage = aiMcpService.pageQuery(name, page, size);Map<String, Object> result = new HashMap<>();result.put("total", resultPage.getTotal());result.put("pages", resultPage.getPages());result.put("current", resultPage.getCurrent());result.put("size", resultPage.getSize());result.put("records", resultPage.getRecords());return RestVO.success(result);} catch (Exception e) {return RestVO.fail("分页查询失败: " + e.getMessage());}}
}

7.测试mcp工具

添加一个高德地图的mcp ，key从高德开放平台获取

{"mcpServers": {"amap-amap-sse": {"url": "https://mcp.amap.com","type": "sse","sseEndpoint": "/sse?key=xxxxxxxxxxxxxxxxxx"}}
}

查询可用工具

8.ChatClient接入toolCallbacks

private Flux<FluxVO> getFluxVOFlux(List<Message> messageList, AiModel myModel, QuestionVO body) {Prompt prompt = new Prompt(messageList);AtomicBoolean inThinking = new AtomicBoolean(false);StringBuffer outputText = body.getMemory() ? new StringBuffer() : null;ChatClient chatModel = myModel.getChatClient();// 1. 先构造 Publisher<ChatResponse>Flux<ChatResponse> publisher;//判断是否需要启用mcp工具if (body.getUseTools()) {List<ToolCallback> toolCallbacks = dynamicMcpClientManager.getAvailableToolCallbacks();publisher = chatModel.prompt(prompt).toolCallbacks(toolCallbacks).stream().chatResponse();} else {publisher = chatModel.prompt(prompt).stream().chatResponse();}// 主动推送一条“处理中”消息Flux<FluxVO> proactiveMsg = Flux.just(FluxVO.builder().text("").status("before").build());Flux<FluxVO> resp = Flux.from(publisher).doFirst(() -> {System.out.println("-------------开始输出");if (body.getMemory()) {chatMemoryService.saveMessage(body);}}).doFinally(signalType -> {System.out.println("-------------流式处理结束");if (body.getMemory() && outputText != null) {chatMemoryService.saveMessage(body.getSessionId(), "ASSISTANT", outputText.toString(), body.getModel());}});return Flux.concat(proactiveMsg, resp);

getFluxVOFlux 之前的代码可以参考同系列前几章节

9.测试最终效果