企业级分布式 MCP 方案

飞书原文档链接地址:https://ik3te1knhq.feishu.cn/wiki/D8kSwC9tFi61CMkRdd8cMxNTnpg

企业级分布式 MCP 方案

[!TIP]
背景:现阶段 MCP Client 和 MCP Server 是一对一的连接方式,若当前 MCP Server 挂掉了,那么 MCP Client 便不能使用 MCP Server 提供的工具能力。工具稳定性的提供得不到保证

解决:做了一些分布式 Client 连接的探索,一个 MCP Client 端可以连接多个 MCP Server(分布式部署),目前采用的方案如下:

  1. 新建一个包含服务名和对应连接的类
  2. 另外实现监听机制,可以动态的应对 MCP Server 节点上下线,去动态调整 mcpAsyncClientList
  3. (读操作)获取 MCP Server 相关信息的,采用从 mcpAsyncClientList 列表中随机中获取一个去发起请求,比如获取工具列表信息
  4. (写操作)对应 MCP Server 需要更改的信息,由 MCP Client 端发起,需要修改所有的 MCP Server

public class LoadbalancedAsyncClient implements EventListener {
private String serviceName;
private List mcpAsyncClientList;

}
给社区贡献代码:https://github.com/alibaba/spring-ai-alibaba/pull/755

模块代码解析

yaml 文件

spring:ai:mcp:client:enabled: truename: mcp-client-webfluxversion: 1.0.0type: SYNCnacos-enabled: true # 开启nacos-client配置,启动分布式alibaba:mcp:nacos: ## nacos的基础配置信息enabled: trueserver-addr: <nacos-sever-addr>service-namespace: <nacos-namespace>  service-group: <nacos-group>client:sse:connections: // 注册在nacos的MCP Server服务,这里mcp-server1代表多节点nacos-server1: mcp-server1nacos-server2: mcp-server2

自动注入部分

NacosMcpSseClientProperties(配置类)
@ConfigurationProperties("spring.ai.alibaba.mcp.client.sse")
public class NacosMcpSseClientProperties {public static final String _CONFIG_PREFIX _= "spring.ai.alibaba.mcp.client.sse";private final Map<String, String> connections = new HashMap<>();public Map<String, String> getConnections() {return connections;}}
NacosMcpSseClientAutoConfiguration

提供 Map<String, List> 的 bean

  • 键代表服务名
  • 值为对应的后续连接的 WebFluxSseClientTransport 列表
@AutoConfiguration
@EnableConfigurationProperties({ NacosMcpSseClientProperties.class, NacosMcpRegistryProperties.class })
public class NacosMcpSseClientAutoConfiguration {private static final Logger _logger _= LoggerFactory._getLogger_(NacosMcpSseClientAutoConfiguration.class);public NacosMcpSseClientAutoConfiguration() {}@Beanpublic NamingService nacosNamingService(NacosMcpRegistryProperties nacosMcpRegistryProperties) {Properties nacosProperties = nacosMcpRegistryProperties.getNacosProperties();try {return NamingFactory._createNamingService_(nacosProperties);}catch (NacosException e) {throw new RuntimeException(e);}}@Bean(name = "server2NamedTransport")public Map<String, List<NamedClientMcpTransport>> server2NamedTransport(NacosMcpSseClientProperties nacosMcpSseClientProperties, NamingService namingService,ObjectProvider<WebClient.Builder> webClientBuilderProvider,ObjectProvider<ObjectMapper> objectMapperProvider) {Map<String, List<NamedClientMcpTransport>> server2NamedTransport = new HashMap<>();WebClient.Builder webClientBuilderTemplate = (WebClient.Builder) webClientBuilderProvider.getIfAvailable(WebClient::_builder_);ObjectMapper objectMapper = (ObjectMapper) objectMapperProvider.getIfAvailable(ObjectMapper::new);Map<String, String> connections = nacosMcpSseClientProperties.getConnections();connections.forEach((serviceKey, serviceName) -> {try {List<Instance> instances = namingService.selectInstances(serviceName, true);List<NamedClientMcpTransport> namedTransports = new ArrayList<>();for (Instance instance : instances) {String url = instance.getMetadata().getOrDefault("scheme", "http") + "://" + instance.getIp() + ":"+ instance.getPort();WebClient.Builder webClientBuilder = webClientBuilderTemplate.clone().baseUrl(url);WebFluxSseClientTransport transport = new WebFluxSseClientTransport(webClientBuilder, objectMapper);namedTransports.add(new NamedClientMcpTransport(serviceName + "-" + instance.getInstanceId(), transport));}server2NamedTransport.put(serviceName, namedTransports);}catch (NacosException e) {_logger_.error("nacos naming service: {} error", serviceName, e);}});return server2NamedTransport;}}
NacosMcpClientAutoConfiguration

提供和 MCP Server 进行交互的客户端

  • List
  • List
@AutoConfiguration(after = { NacosMcpSseClientAutoConfiguration.class, McpClientAutoConfiguration.class })
@ConditionalOnClass({ McpSchema.class })
@EnableConfigurationProperties({ McpClientCommonProperties.class })
@ConditionalOnProperty(prefix = "spring.ai.mcp.client", name = { "nacos-enabled" }, havingValue = "true",matchIfMissing = false)
public class NacosMcpClientAutoConfiguration {public NacosMcpClientAutoConfiguration() {}private String connectedClientName(String clientName, String serverConnectionName) {return clientName + " - " + serverConnectionName;}@Bean@ConditionalOnProperty(prefix = "spring.ai.mcp.client", name = { "type" }, havingValue = "SYNC",matchIfMissing = true)public List<LoadbalancedMcpSyncClient> loadbalancedMcpSyncClientList(ObjectProvider<McpSyncClientConfigurer> mcpSyncClientConfigurerProvider,McpClientCommonProperties commonProperties,@Qualifier("server2NamedTransport") ObjectProvider<Map<String, List<NamedClientMcpTransport>>> server2NamedTransportProvider,ObjectProvider<NamingService> namingServiceProvider) {NamingService namingService = namingServiceProvider.getObject();McpSyncClientConfigurer mcpSyncClientConfigurer = mcpSyncClientConfigurerProvider.getObject();List<LoadbalancedMcpSyncClient> loadbalancedMcpSyncClients = new ArrayList<>();Map<String, List<NamedClientMcpTransport>> server2NamedTransport = server2NamedTransportProvider.getObject();for (Map.Entry<String, List<NamedClientMcpTransport>> entry : server2NamedTransport.entrySet()) {String serviceName = entry.getKey();List<NamedClientMcpTransport> namedTransports = entry.getValue();List<McpSyncClient> mcpSyncClients = new ArrayList<>();McpSyncClient syncClient;for (NamedClientMcpTransport namedTransport : namedTransports) {McpSchema.Implementation clientInfo = new McpSchema.Implementation(this.connectedClientName(commonProperties.getName(), namedTransport.name()),commonProperties.getVersion());McpClient.SyncSpec syncSpec = McpClient._sync_(namedTransport.transport()).clientInfo(clientInfo).requestTimeout(commonProperties.getRequestTimeout());syncSpec = mcpSyncClientConfigurer.configure(namedTransport.name(), syncSpec);syncClient = syncSpec.build();if (commonProperties.isInitialized()) {syncClient.initialize();}mcpSyncClients.add(syncClient);}LoadbalancedMcpSyncClient loadbalancedMcpSyncClient = LoadbalancedMcpSyncClient._builder_().serviceName(serviceName).mcpSyncClientList(mcpSyncClients).namingService(namingService).build();loadbalancedMcpSyncClient.subscribe();loadbalancedMcpSyncClients.add(loadbalancedMcpSyncClient);}return loadbalancedMcpSyncClients;}@Bean@ConditionalOnProperty(prefix = "spring.ai.mcp.client", name = { "type" }, havingValue = "ASYNC")public List<LoadbalancedMcpAsyncClient> loadbalancedMcpAsyncClientList(ObjectProvider<McpAsyncClientConfigurer> mcpAsyncClientConfigurerProvider,McpClientCommonProperties commonProperties,@Qualifier("server2NamedTransport") ObjectProvider<Map<String, List<NamedClientMcpTransport>>> server2NamedTransportProvider,ObjectProvider<NamingService> namingServiceProvider) {NamingService namingService = namingServiceProvider.getObject();McpAsyncClientConfigurer mcpAsyncClientConfigurer = mcpAsyncClientConfigurerProvider.getObject();List<LoadbalancedMcpAsyncClient> loadbalancedMcpAsyncClients = new ArrayList<>();Map<String, List<NamedClientMcpTransport>> server2NamedTransport = server2NamedTransportProvider.getObject();for (Map.Entry<String, List<NamedClientMcpTransport>> entry : server2NamedTransport.entrySet()) {String serviceName = entry.getKey();List<NamedClientMcpTransport> namedTransports = entry.getValue();List<McpAsyncClient> mcpAsyncClients = new ArrayList<>();McpAsyncClient asyncClient;for (NamedClientMcpTransport namedTransport : namedTransports) {McpSchema.Implementation clientInfo = new McpSchema.Implementation(this.connectedClientName(commonProperties.getName(), namedTransport.name()),commonProperties.getVersion());McpClient.AsyncSpec asyncSpec = McpClient._async_(namedTransport.transport()).clientInfo(clientInfo).requestTimeout(commonProperties.getRequestTimeout());asyncSpec = mcpAsyncClientConfigurer.configure(namedTransport.name(), asyncSpec);asyncClient = asyncSpec.build();if (commonProperties.isInitialized()) {asyncClient.initialize().block();}mcpAsyncClients.add(asyncClient);}LoadbalancedMcpAsyncClient loadbalancedMcpAsyncClient = LoadbalancedMcpAsyncClient._builder_().serviceName(serviceName).mcpAsyncClientList(mcpAsyncClients).namingService(namingService).build();loadbalancedMcpAsyncClient.subscribe();loadbalancedMcpAsyncClients.add(loadbalancedMcpAsyncClient);}return loadbalancedMcpAsyncClients;}}

Client 端部分

LoadbalancedMcpAsyncClient

各字段含义:

  • String serviceName:MCP Server 注册的服务名称
  • List<McpAsyncClient> mcpAsyncClientList:对应的多节点客户端
  • NamingService namingService:Nacos 服务
  • List<Instance> instances:Nacos 中 MCP Server 的实例列表

其余方法的使用和 McpAsyncClient 保持一致,已经全面封装好了

  1. 读操作:通过 getMcpAsyncClient()方法轮询得到 McpAsyncClient 列表
  2. 写操作:对所有 List<McpAsyncClient> 进行操作

通过实现 EventListener 接口,动态增加 or 减少 McpAsyncClient

public class LoadbalancedMcpAsyncClient implements EventListener {private static final Logger _logger _= LoggerFactory._getLogger_(LoadbalancedMcpAsyncClient.class);private final String serviceName;private final List<McpAsyncClient> mcpAsyncClientList;private final AtomicInteger currentIndex = new AtomicInteger(0);private final NamingService namingService;private List<Instance> instances;public LoadbalancedMcpAsyncClient(String serviceName, List<McpAsyncClient> mcpAsyncClientList,NamingService namingService) {Assert._notNull_(serviceName, "serviceName cannot be null");Assert._notNull_(mcpAsyncClientList, "mcpAsyncClientList cannot be null");Assert._notNull_(namingService, "namingService cannot be null");this.serviceName = serviceName;this.mcpAsyncClientList = mcpAsyncClientList;try {this.namingService = namingService;this.instances = namingService.selectInstances(serviceName, true);}catch (NacosException e) {throw new RuntimeException(String._format_("Failed to get instances for service: %s", serviceName));}}public void subscribe() {try {this.namingService.subscribe(this.serviceName, this);}catch (NacosException e) {throw new RuntimeException(String._format_("Failed to subscribe to service: %s", this.serviceName));}}public String getServiceName() {return serviceName;}public List<McpAsyncClient> getMcpAsyncClientList() {return mcpAsyncClientList;}public NamingService getNamingService() {return this.namingService;}public List<Instance> getInstances() {return this.instances;}private McpAsyncClient getMcpAsyncClient() {if (mcpAsyncClientList.isEmpty()) {throw new IllegalStateException("No McpAsyncClient available");}int index = currentIndex.getAndIncrement() % mcpAsyncClientList.size();return mcpAsyncClientList.get(index);}// ------------------------------------------------------------------------------------------------------------------------------------------------public McpSchema.ServerCapabilities getServerCapabilities() {return getMcpAsyncClient().getServerCapabilities();}public McpSchema.Implementation getServerInfo() {return getMcpAsyncClient().getServerInfo();}public boolean isInitialized() {return getMcpAsyncClient().isInitialized();}public McpSchema.ClientCapabilities getClientCapabilities() {return getMcpAsyncClient().getClientCapabilities();}public McpSchema.Implementation getClientInfo() {return getMcpAsyncClient().getClientInfo();}public void close() {Iterator<McpAsyncClient> iterator = mcpAsyncClientList.iterator();while (iterator.hasNext()) {McpAsyncClient mcpAsyncClient = iterator.next();mcpAsyncClient.close();iterator.remove();_logger_.info("Closed and removed McpSyncClient: {}", mcpAsyncClient.getClientInfo().name());}}public Mono<Void> closeGracefully() {Iterator<McpAsyncClient> iterator = mcpAsyncClientList.iterator();List<Mono<Void>> closeMonos = new ArrayList<>();while (iterator.hasNext()) {McpAsyncClient mcpAsyncClient = iterator.next();Mono<Void> voidMono = mcpAsyncClient.closeGracefully().doOnSuccess(v -> {iterator.remove();_logger_.info("Closed and removed McpAsyncClient: {}", mcpAsyncClient.getClientInfo().name());});closeMonos.add(voidMono);}return Mono._when_(closeMonos);}public Mono<Object> ping() {return getMcpAsyncClient().ping();}public Mono<Void> addRoot(McpSchema.Root root) {return Mono._when_(mcpAsyncClientList.stream().map(mcpAsyncClient -> mcpAsyncClient.addRoot(root)).collect(Collectors._toList_()));}public Mono<Void> removeRoot(String rootUri) {return Mono._when_(mcpAsyncClientList.stream().map(mcpAsyncClient -> mcpAsyncClient.removeRoot(rootUri)).collect(Collectors._toList_()));}public Mono<Void> rootsListChangedNotification() {return Mono._when_(mcpAsyncClientList.stream().map(McpAsyncClient::rootsListChangedNotification).collect(Collectors._toList_()));}public Mono<McpSchema.CallToolResult> callTool(McpSchema.CallToolRequest callToolRequest) {return getMcpAsyncClient().callTool(callToolRequest);}public Mono<McpSchema.ListToolsResult> listTools() {return getMcpAsyncClient().listTools();}public Mono<McpSchema.ListToolsResult> listTools(String cursor) {return getMcpAsyncClient().listTools(cursor);}public Mono<McpSchema.ListResourcesResult> listResources() {return getMcpAsyncClient().listResources();}public Mono<McpSchema.ListResourcesResult> listResources(String cursor) {return getMcpAsyncClient().listResources(cursor);}public Mono<McpSchema.ReadResourceResult> readResource(McpSchema.Resource resource) {return getMcpAsyncClient().readResource(resource);}public Mono<McpSchema.ReadResourceResult> readResource(McpSchema.ReadResourceRequest readResourceRequest) {return getMcpAsyncClient().readResource(readResourceRequest);}public Mono<McpSchema.ListResourceTemplatesResult> listResourceTemplates() {return getMcpAsyncClient().listResourceTemplates();}public Mono<McpSchema.ListResourceTemplatesResult> listResourceTemplates(String cursor) {return getMcpAsyncClient().listResourceTemplates(cursor);}public Mono<Void> subscribeResource(McpSchema.SubscribeRequest subscribeRequest) {return Mono._when_(mcpAsyncClientList.stream().map(mcpAsyncClient -> mcpAsyncClient.subscribeResource(subscribeRequest)).collect(Collectors._toList_()));}public Mono<Void> unsubscribeResource(McpSchema.UnsubscribeRequest unsubscribeRequest) {return Mono._when_(mcpAsyncClientList.stream().map(mcpAsyncClient -> mcpAsyncClient.unsubscribeResource(unsubscribeRequest)).collect(Collectors._toList_()));}public Mono<McpSchema.ListPromptsResult> listPrompts() {return getMcpAsyncClient().listPrompts();}public Mono<McpSchema.ListPromptsResult> listPrompts(String cursor) {return getMcpAsyncClient().listPrompts(cursor);}public Mono<McpSchema.GetPromptResult> getPrompt(McpSchema.GetPromptRequest getPromptRequest) {return getMcpAsyncClient().getPrompt(getPromptRequest);}public Mono<Void> setLoggingLevel(McpSchema.LoggingLevel loggingLevel) {return Mono._when_(mcpAsyncClientList.stream().map(mcpAsyncClient -> mcpAsyncClient.setLoggingLevel(loggingLevel)).collect(Collectors._toList_()));}// ------------------------------------------------------------------------------------------------------------------------------------------------@Overridepublic void onEvent(Event event) {if (event instanceof NamingEvent namingEvent) {if (this.serviceName.equals(namingEvent.getServiceName())) {_logger_.info("Received service instance change event for service: {}", namingEvent.getServiceName());List<Instance> instances = namingEvent.getInstances();_logger_.info("Updated instances count: {}", instances.size());// 打印每个实例的详细信息instances.forEach(instance -> {_logger_.info("Instance: {}:{} (Healthy: {}, Enabled: {}, Metadata: {})", instance.getIp(),instance.getPort(), instance.isHealthy(), instance.isEnabled(),JacksonUtils._toJson_(instance.getMetadata()));});updateClientList(instances);}}}private void updateClientList(List<Instance> currentInstances) {McpClientCommonProperties commonProperties = ApplicationContextHolder._getBean_(McpClientCommonProperties.class);McpAsyncClientConfigurer mcpSyncClientConfigurer = ApplicationContextHolder._getBean_(McpAsyncClientConfigurer.class);ObjectMapper objectMapper = ApplicationContextHolder._getBean_(ObjectMapper.class);WebClient.Builder webClientBuilderTemplate = ApplicationContextHolder._getBean_(WebClient.Builder.class);// 移除的实例列表List<Instance> removeInstances = instances.stream().filter(instance -> !currentInstances.contains(instance)).collect(Collectors._toList_());// 新增的实例列表List<Instance> addInstances = currentInstances.stream().filter(instance -> !instances.contains(instance)).collect(Collectors._toList_());// 删除McpAsyncClient实例List<String> clientInfoNames = removeInstances.stream().map(instance -> connectedClientName(commonProperties.getName(),this.serviceName + "-" + instance.getInstanceId())).toList();Iterator<McpAsyncClient> iterator = mcpAsyncClientList.iterator();while (iterator.hasNext()) {McpAsyncClient mcpAsyncClient = iterator.next();McpSchema.Implementation clientInfo = mcpAsyncClient.getClientInfo();if (clientInfoNames.contains(clientInfo.name())) {_logger_.info("Removing McpAsyncClient: {}", clientInfo.name());mcpAsyncClient.closeGracefully().subscribe(v -> {iterator.remove();}, e -> _logger_.error("Failed to remove McpAsyncClient: {}", clientInfo.name(), e));}}// 新增McpAsyncClient实例McpAsyncClient asyncClient;for (Instance instance : addInstances) {String baseUrl = instance.getMetadata().getOrDefault("scheme", "http") + "://" + instance.getIp() + ":"+ instance.getPort();WebClient.Builder webClientBuilder = webClientBuilderTemplate.clone().baseUrl(baseUrl);WebFluxSseClientTransport transport = new WebFluxSseClientTransport(webClientBuilder, objectMapper);NamedClientMcpTransport namedTransport = new NamedClientMcpTransport(serviceName + "-" + instance.getInstanceId(), transport);McpSchema.Implementation clientInfo = new McpSchema.Implementation(this.connectedClientName(commonProperties.getName(), namedTransport.name()),commonProperties.getVersion());McpClient.AsyncSpec asyncSpec = McpClient._async_(namedTransport.transport()).clientInfo(clientInfo).requestTimeout(commonProperties.getRequestTimeout());asyncSpec = mcpSyncClientConfigurer.configure(namedTransport.name(), asyncSpec);asyncClient = asyncSpec.build();if (commonProperties.isInitialized()) {asyncClient.initialize().block();}_logger_.info("Added McpAsyncClient: {}", clientInfo.name());mcpAsyncClientList.add(asyncClient);}private String connectedClientName(String clientName, String serverConnectionName) {return clientName + " - " + serverConnectionName;}public static Builder builder() {return new Builder();}public static class Builder {private String serviceName;private List<McpAsyncClient> mcpAsyncClientList;private NamingService namingService;public Builder serviceName(String serviceName) {this.serviceName = serviceName;return this;}public Builder mcpAsyncClientList(List<McpAsyncClient> mcpAsyncClientList) {this.mcpAsyncClientList = mcpAsyncClientList;return this;}public Builder namingService(NamingService namingService) {this.namingService = namingService;return this;}public LoadbalancedMcpAsyncClient build() {return new LoadbalancedMcpAsyncClient(this.serviceName, this.mcpAsyncClientList, this.namingService);}}}
LoadbalancedMcpSyncClient

同上

public class LoadbalancedMcpSyncClient implements EventListener {private static final Logger _logger _= LoggerFactory._getLogger_(LoadbalancedMcpAsyncClient.class);private final String serviceName;private final List<McpSyncClient> mcpSyncClientList;private final AtomicInteger currentIndex = new AtomicInteger(0);private final NamingService namingService;private List<Instance> instances;public LoadbalancedMcpSyncClient(String serviceName, List<McpSyncClient> mcpSyncClientList,NamingService namingService) {Assert._notNull_(serviceName, "Service name must not be null");Assert._notNull_(mcpSyncClientList, "McpSyncClient list must not be null");Assert._notNull_(namingService, "NamingService must not be null");this.serviceName = serviceName;this.mcpSyncClientList = mcpSyncClientList;try {this.namingService = namingService;this.instances = namingService.selectInstances(serviceName, true);}catch (NacosException e) {throw new RuntimeException(String._format_("Failed to get instances for service: %s", serviceName));}}public void subscribe() {try {this.namingService.subscribe(this.serviceName, this);}catch (NacosException e) {throw new RuntimeException(String._format_("Failed to subscribe to service: %s", this.serviceName));}}public String getServiceName() {return this.serviceName;}public List<McpSyncClient> getMcpSyncClientList() {return this.mcpSyncClientList;}public NamingService getNamingService() {return this.namingService;}public List<Instance> getInstances() {return this.instances;}private McpSyncClient getMcpSyncClient() {if (mcpSyncClientList.isEmpty()) {throw new IllegalStateException("No McpAsyncClient available");}int index = currentIndex.getAndIncrement() % mcpSyncClientList.size();return mcpSyncClientList.get(index);}// ------------------------------------------------------------------------------------------------------------------------------------------------public McpSchema.ServerCapabilities getServerCapabilities() {return getMcpSyncClient().getServerCapabilities();}public McpSchema.Implementation getServerInfo() {return getMcpSyncClient().getServerInfo();}public McpSchema.ClientCapabilities getClientCapabilities() {return getMcpSyncClient().getClientCapabilities();}public McpSchema.Implementation getClientInfo() {return getMcpSyncClient().getClientInfo();}public void close() {Iterator<McpSyncClient> iterator = mcpSyncClientList.iterator();while (iterator.hasNext()) {McpSyncClient mcpSyncClient = iterator.next();mcpSyncClient.close();iterator.remove();_logger_.info("Closed and removed McpSyncClient: {}", mcpSyncClient.getClientInfo().name());}}public boolean closeGracefully() {List<Boolean> flagList = new ArrayList<>();Iterator<McpSyncClient> iterator = mcpSyncClientList.iterator();while (iterator.hasNext()) {McpSyncClient mcpSyncClient = iterator.next();boolean flag = mcpSyncClient.closeGracefully();flagList.add(flag);if (flag) {iterator.remove();_logger_.info("Closed and removed McpSyncClient: {}", mcpSyncClient.getClientInfo().name());}}return !flagList.stream().allMatch(flag -> flag);}public Object ping() {return getMcpSyncClient().ping();}public void addRoot(McpSchema.Root root) {for (McpSyncClient mcpSyncClient : mcpSyncClientList) {mcpSyncClient.addRoot(root);}}public void removeRoot(String rootUri) {for (McpSyncClient mcpSyncClient : mcpSyncClientList) {mcpSyncClient.removeRoot(rootUri);}}public McpSchema.CallToolResult callTool(McpSchema.CallToolRequest callToolRequest) {return getMcpSyncClient().callTool(callToolRequest);}public McpSchema.ListToolsResult listTools() {return getMcpSyncClient().listTools();}public McpSchema.ListToolsResult listTools(String cursor) {return getMcpSyncClient().listTools(cursor);}public McpSchema.ListResourcesResult listResources(String cursor) {return getMcpSyncClient().listResources(cursor);}public McpSchema.ListResourcesResult listResources() {return getMcpSyncClient().listResources();}public McpSchema.ReadResourceResult readResource(McpSchema.Resource resource) {return getMcpSyncClient().readResource(resource);}public McpSchema.ReadResourceResult readResource(McpSchema.ReadResourceRequest readResourceRequest) {return getMcpSyncClient().readResource(readResourceRequest);}public McpSchema.ListResourceTemplatesResult listResourceTemplates(String cursor) {return getMcpSyncClient().listResourceTemplates(cursor);}public McpSchema.ListResourceTemplatesResult listResourceTemplates() {return getMcpSyncClient().listResourceTemplates();}public void subscribeResource(McpSchema.SubscribeRequest subscribeRequest) {for (McpSyncClient mcpSyncClient : mcpSyncClientList) {mcpSyncClient.subscribeResource(subscribeRequest);}}public void unsubscribeResource(McpSchema.UnsubscribeRequest unsubscribeRequest) {for (McpSyncClient mcpSyncClient : mcpSyncClientList) {mcpSyncClient.unsubscribeResource(unsubscribeRequest);}}public McpSchema.ListPromptsResult listPrompts(String cursor) {return getMcpSyncClient().listPrompts(cursor);}public McpSchema.ListPromptsResult listPrompts() {return getMcpSyncClient().listPrompts();}public McpSchema.GetPromptResult getPrompt(McpSchema.GetPromptRequest getPromptRequest) {return getMcpSyncClient().getPrompt(getPromptRequest);}public void setLoggingLevel(McpSchema.LoggingLevel loggingLevel) {for (McpSyncClient mcpSyncClient : mcpSyncClientList) {mcpSyncClient.setLoggingLevel(loggingLevel);}}// ------------------------------------------------------------------------------------------------------------------------------------------------@Overridepublic void onEvent(Event event) {if (event instanceof NamingEvent namingEvent) {if (this.serviceName.equals(namingEvent.getServiceName())) {_logger_.info("Received service instance change event for service: {}", namingEvent.getServiceName());List<Instance> instances = namingEvent.getInstances();_logger_.info("Updated instances count: {}", instances.size());// 打印每个实例的详细信息instances.forEach(instance -> {_logger_.info("Instance: {}:{} (Healthy: {}, Enabled: {}, Metadata: {})", instance.getIp(),instance.getPort(), instance.isHealthy(), instance.isEnabled(),JacksonUtils._toJson_(instance.getMetadata()));});updateClientList(instances);}}}private void updateClientList(List<Instance> currentInstances) {McpClientCommonProperties commonProperties = ApplicationContextHolder._getBean_(McpClientCommonProperties.class);McpSyncClientConfigurer mcpSyncClientConfigurer = ApplicationContextHolder._getBean_(McpSyncClientConfigurer.class);ObjectMapper objectMapper = ApplicationContextHolder._getBean_(ObjectMapper.class);WebClient.Builder webClientBuilderTemplate = ApplicationContextHolder._getBean_(WebClient.Builder.class);// 移除的实例列表List<Instance> removeInstances = instances.stream().filter(instance -> !currentInstances.contains(instance)).collect(Collectors._toList_());// 新增的实例列表List<Instance> addInstances = currentInstances.stream().filter(instance -> !instances.contains(instance)).collect(Collectors._toList_());// 删除McpSyncClient实例List<String> clientInfoNames = removeInstances.stream().map(instance -> connectedClientName(commonProperties.getName(),this.serviceName + "-" + instance.getInstanceId())).toList();Iterator<McpSyncClient> iterator = mcpSyncClientList.iterator();while (iterator.hasNext()) {McpSyncClient mcpSyncClient = iterator.next();McpSchema.Implementation clientInfo = mcpSyncClient.getClientInfo();if (clientInfoNames.contains(clientInfo.name())) {_logger_.info("Removing McpsyncClient: {}", clientInfo.name());if (mcpSyncClient.closeGracefully()) {iterator.remove();}else {_logger_.warn("Failed to remove mcpSyncClient: {}", clientInfo.name());}}}// 新增McpSyncClient实例McpSyncClient syncClient;for (Instance instance : addInstances) {String baseUrl = instance.getMetadata().getOrDefault("scheme", "http") + "://" + instance.getIp() + ":"+ instance.getPort();WebClient.Builder webClientBuilder = webClientBuilderTemplate.clone().baseUrl(baseUrl);WebFluxSseClientTransport transport = new WebFluxSseClientTransport(webClientBuilder, objectMapper);NamedClientMcpTransport namedTransport = new NamedClientMcpTransport(serviceName + "-" + instance.getInstanceId(), transport);McpSchema.Implementation clientInfo = new McpSchema.Implementation(this.connectedClientName(commonProperties.getName(), namedTransport.name()),commonProperties.getVersion());McpClient.SyncSpec syncSpec = McpClient._sync_(namedTransport.transport()).clientInfo(clientInfo).requestTimeout(commonProperties.getRequestTimeout());syncSpec = mcpSyncClientConfigurer.configure(namedTransport.name(), syncSpec);syncClient = syncSpec.build();if (commonProperties.isInitialized()) {syncClient.initialize();}_logger_.info("Added McpAsyncClient: {}", clientInfo.name());mcpSyncClientList.add(syncClient);}this.instances = currentInstances;}private String connectedClientName(String clientName, String serverConnectionName) {return clientName + " - " + serverConnectionName;}public static Builder builder() {return new Builder();}public static class Builder {private String serviceName;private List<McpSyncClient> mcpSyncClientList;private NamingService namingService;public Builder serviceName(String serviceName) {this.serviceName = serviceName;return this;}public Builder mcpSyncClientList(List<McpSyncClient> mcpSyncClientList) {this.mcpSyncClientList = mcpSyncClientList;return this;}public Builder namingService(NamingService namingService) {this.namingService = namingService;return this;}public LoadbalancedMcpSyncClient build() {return new LoadbalancedMcpSyncClient(this.serviceName, this.mcpSyncClientList, this.namingService);}}}

工具类

ApplicationContextHolder
@Component
public class ApplicationContextHolder implements ApplicationContextAware {private static ApplicationContext _applicationContext_;@Overridepublic void setApplicationContext(ApplicationContext applicationContext) throws BeansException {ApplicationContextHolder._applicationContext _= applicationContext;}public static <T> T getBean(Class<T> clazz) {return _applicationContext_.getBean(clazz);}}

效果演示

我在 nacos 中,注册了 MCP Server 服务,部署两个节点

  • 同一台机器以不同端口号启动的 MCP Server 服务,分别是 19000、19001,注册在 Nacos 中以 mcp-server-provider 为服务名

yml 配置如下

server:port: 8080spring:application:name: mcp-client-webflux_  _ai:alibaba:mcp:nacos:enabled: trueserver-addr: 127.0.0.1:8848username: nacospassword: nacosclient:sse:connections:nacos-server1: mcp-server-providermcp:client:enabled: truename: mcp-client-webfluxversion: 0.0.1initialized: truerequest-timeout: 600snacos-enabled: true

我们能发现已经成功注入 LoadbalancedMcpSyncClient 类,其中 mcp-server-provider 有两个实例,对应的两个 McpSyncClient

我们停掉其中的 MCP Server19001 端口的服务,通过 removeInstances 获取移除的实例列表,同步在 mcpSyncClientList 移除对应的 McpSyncClient

我们再新启动 MCP Server19001 端口的服务,通过 addInstances 获取新增的实例列表,同步在 mcpSyncClientList 新增对应的 McpSyncClient

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.mzph.cn/web/78162.shtml

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

【AI提示词】奥卡姆剃刀思维模型专家

提示说明 一位专注于奥卡姆剃刀思维模型的专业人士&#xff0c;擅长将简洁性原则应用于复杂问题的分析与解决。 提示词 # Role: 奥卡姆剃刀思维模型专家## Profile - language: 中文 - description: 一位专注于奥卡姆剃刀思维模型的专业人士&#xff0c;擅长将简洁性原则应用…

2.1 行列式

引言 行列式是线性代数的核心工具&#xff0c;贯穿矩阵运算、特征值计算与微分方程求解。本文系统梳理2.1节核心考点&#xff0c;结合公式速查与典型例题&#xff0c;助你高效突破行列式难点&#xff01; 考点一&#xff1a;数值型行列式计算 1️⃣ 行列式的定义 (1) 定义方…

单词规律(简单)

思路和同构字符串那道题一样。、但是这道题要注意的地方就是&#xff0c;检查 pattern 和 s 的单词数量是否一致以及在进行字符串比较的时候应该用equals来进行比较&#xff0c;而不能用“&#xff01;”&#xff0c;“&#xff01;”比较的是对象引用而非内容。 class Soluti…

【C++】认识map和set

目录 前言&#xff1a; 一&#xff1a;认识map和set 二&#xff1a;map和set的使用 1.set的使用 2.map的使用 三&#xff1a;map的insert方法返回值 四&#xff1a;map的[ ]的使用 五&#xff1a;multiset和multimap 六&#xff1a;map和set的底层数据结构 七&#x…

Mybatis中的一级二级缓存扫盲

思维导图&#xff1a; MyBatis 提供了一级缓存和二级缓存机制&#xff0c;用于提高数据库查询的性能&#xff0c;减少对数据库的访问次数。&#xff08;本质上是减少IO次数&#xff09;。 一级缓存 1. 概念 一级缓存也称为会话缓存&#xff0c;它是基于 SqlSession 的缓存。在同…

uniapp 实现低功耗蓝牙连接并读写数据实战指南

在物联网应用场景中&#xff0c;低功耗蓝牙&#xff08;BLE&#xff09;凭借其低能耗、连接便捷的特点&#xff0c;成为设备间数据交互的重要方式。Uniapp 作为一款跨平台开发框架&#xff0c;提供了丰富的 API 支持&#xff0c;使得在多个端实现低功耗蓝牙功能变得轻松高效。本…

OpenSSL应用实践:嵌入式数据安全实战指南

文章目录 OpenSSL应用实践:嵌入式数据安全实战指南一、嵌入式安全现状与OpenSSL适配方案1.1 嵌入式安全挑战1.2 OpenSSL精简方案二、开发环境搭建2.1 交叉编译工具链2.2 OpenSSL交叉编译三、核心功能实现3.1 AES-GCM加密实践四、实战项目:安全OTA升级4.1 系统架构4.2 关键代码…

harmonyOS 手机,双折叠,平板,PC端屏幕适配

由于HarmonyOS设备的屏幕尺寸和分辨率各不相同&#xff0c;开发者需要采取适当的措施来适配不同的屏幕。 1.EntryAbility.ets文件里&#xff1a;onWindowStageCreate方法里判断设备类型&#xff0c; 如果是pad&#xff0c;需全屏展示&#xff08;按客户需求来&#xff0c;本次…

跟韩学AiOps系列之2025学MySQL系列_如何在MySQL中开启和提交事务?!

跟韩学AiOps系列之2025学MySQL系列_如何在MySQL中开启和提交事务&#xff1f;! 文章目录 一、事务的基本操作1. 开启事务2. 执行事务内操作3. 提交事务4. 回滚事务 二、验证示例&#xff08;适用于 MySQL 5.7&#xff09;步骤 1&#xff1a;准备测试表和数据步骤 2&#xff1a…

Java生成微信小程序码及小程序短链接

使用wx-java-miniapp-spring-boot-starter 生成微信小程序码及小程序短链接 在pom.xml文件中引入依赖 <dependency><groupId>com.github.binarywang</groupId><artifactId>wx-java-miniapp-spring-boot-starter</artifactId><version>4.7…

如何让通义千问大模型支持结构化输出?

之前的文章提到通义千问API无法通过with_structured_output/json schema的方式支持结构化输出&#xff0c;如果就是想使用通义千问大模型做结构化输出&#xff0c;应该怎么办呢&#xff1f;有两种办法 使用Ollama来运行通义千问大模型 从Ollama博客文章 Structured output 中…

一条 SQL 查询语句是如何执行的(MySQL)

第一讲&#xff1a;一条 SQL 查询语句是如何执行的 总览图示 MySQL 查询的执行流程可以大致分为以下步骤&#xff08;如图所示&#xff09;&#xff1a; 连接器&#xff08;Connection&#xff09;查询缓存&#xff08;Query Cache&#xff0c;MySQL 8.0 已废弃&#xff09;…

汽车OTA在线升级法规分析

摘要 本文介绍了R156法规即《关于批准车辆的软件升级和软件升级管理体系统一规定的法规》、该法规专注于汽车软件升级功能&#xff0c;并为此提出了一系列具体要求&#xff0c;旨在确保软件升级流程的安全性、可控性和合规性&#xff0c;从而顺应汽车行业智能化、联网化的发展趋…

Notepad编辑器实现换行符替换

在不同的Note编辑器中&#xff0c;批量把换行替换为空的方法有所不同&#xff0c;以下是常见编辑器的操作方法&#xff1a; Notepad 打开文件后&#xff0c;按CtrlH打开“查找和替换”对话框&#xff0c;在“查找”字段中输入\r\n&#xff0c;在“替换为”字段中输入一个空格…

Rust多线程性能优化:打破Arc+锁的瓶颈,效率提升10倍

一、引言 在 Rust 开发中&#xff0c;多线程编程是提升程序性能的重要手段。Arc&#xff08;原子引用计数&#xff09;和锁的组合是实现多线程数据共享的常见方式。然而&#xff0c;很多程序员在使用 Arc 和锁时会遇到性能瓶颈&#xff0c;导致程序运行效率低下。本文将深入剖…

【安装指南】Centos7 在 Docker 上安装 RabbitMQ4.0.x

目录 前置知识:RabbitMQ 的介绍 一、单机安装 RabbitMQ 4.0.7版本 1.1 在线拉取镜像 二、延迟插件的安装 2.1 安装延迟插件 步骤一:下载延迟插件 步骤二:将延迟插件放到插件目录 步骤三:启动延迟插件 步骤四:重启 RabbitMQ 服务 步骤五:验收成果 步骤六:手动…

【quantity】5 derive_more库 2.0 版介绍

derive_more 是一个 Rust 过程宏库&#xff0c;旨在通过派生宏自动生成常见 trait 的实现&#xff0c;减少样板代码。2.0 版本带来了多项改进和新特性。 主要特性 1. 支持的 Trait 派生 derive_more 2.0 支持派生以下 trait&#xff1a; 基本操作 trait: Display - 格式化显…

网站备份,网站数据备份的步骤

网站备份&#xff08;尤其是网站数据备份&#xff09;是保障业务连续性、防止数据丢失和应对安全威胁的关键措施。以下是系统化的备份步骤和实施建议&#xff0c;涵盖技术操作、策略规划及常见问题处理&#xff1a; 一、备份前的准备工作 明确备份范围 核心数据&#xff1a;…

OpenCV 图形API(72)图像与通道拼接函数-----根据指定的方式翻转图像(GMat)函数 flip()

操作系统&#xff1a;ubuntu22.04 OpenCV版本&#xff1a;OpenCV4.9 IDE:Visual Studio Code 编程语言&#xff1a;C11 算法描述 翻转一个2D矩阵&#xff0c;围绕垂直轴、水平轴或同时围绕两个轴。 该函数以三种不同的方式之一翻转矩阵&#xff08;行和列的索引是从0开始的&a…

医生视角下转录组学的生物信息学分析

医生视角下转录组学的生物信息学分析 转录组学的生物信息学分析是医生解决临床与科研问题的有力工具。这里罗列医学转录组学相关的几个概念&#xff0c;从使用者&#xff08;医生&#xff09;的角度看待理解相关技术&#xff0c;为后续使用该技术说明临床和科研问题奠定基础。…