Question: Do sse clients need to stay connected? If so, would it be better to specify it in the specification?

[noear]

Currently, based on keepAlive, many http clients will experience read timeouts if the server does not issue sse (e.g. okhttp).

Implementations can have a convention, if defined by a specification.

For example:

Keep the server alive with an sse heartbeat, or set the read timeout to 0 on the client, and so on

[xionghuyi]

I also want to know the answer

[lambochen]

Keeping the SSE heartbeat does not completely solve the problem, as the mcp client cannot actively reconnect after the connection is unexpectedly disconnected (such as when the mcp server restarts).
When implementing it within the enterprise, I implemented an automatic reconnection mechanism (a temporary solution) to address this issue.

Additionally, Streamable HTTP can solve this problem in some scenarios.

[noear]

@lambochen That's not necessarily... This is how solon-ai-mcp implements disconnection reconnections (e.g., server restarts). It has been tested and used in production.

https://github.com/opensolon/solon-ai/blob/dev/solon-ai-mcp/src/main/java/org/noear/solon/ai/mcp/client/McpClientToolProvider.java

import io.modelcontextprotocol.client.McpClient;
import io.modelcontextprotocol.client.McpSyncClient;
import io.modelcontextprotocol.client.transport.ServerParameters;
import io.modelcontextprotocol.client.transport.StdioClientTransport;
import io.modelcontextprotocol.client.transport.WebRxSseClientTransport;
import io.modelcontextprotocol.spec.McpClientTransport;
import io.modelcontextprotocol.spec.McpSchema;
import org.noear.snack.ONode;
import org.noear.solon.Utils;
import org.noear.solon.ai.chat.tool.FunctionTool;
import org.noear.solon.ai.chat.tool.RefererFunctionTool;
import org.noear.solon.ai.chat.tool.ToolProvider;
import org.noear.solon.ai.image.Image;
import org.noear.solon.ai.mcp.McpChannel;
import org.noear.solon.ai.mcp.exception.McpException;
import org.noear.solon.core.Props;
import org.noear.solon.core.util.Assert;
import org.noear.solon.core.util.ResourceUtil;
import org.noear.solon.core.util.RunUtil;
import org.noear.solon.net.http.HttpTimeout;
import org.noear.solon.net.http.HttpUtilsBuilder;

import java.io.Closeable;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.Proxy;
import java.net.URI;
import java.net.URL;
import java.time.Duration;
import java.util.*;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.ReentrantLock;

/**
 * Mcp 连接工具提供者
 *
 * @author noear
 * @since 3.1
 */
public class McpClientToolProvider implements ToolProvider, Closeable {
    private final ReentrantLock LOCKER = new ReentrantLock();

    private final AtomicBoolean isClosed = new AtomicBoolean(false);
    private final AtomicBoolean isStarted = new AtomicBoolean(false);
    private final McpClientProperties clientProps;
    private ScheduledExecutorService heartbeatExecutor;
    private McpSyncClient client;

    /**
     * 用于支持注入
     */
    public McpClientToolProvider(Properties clientProps) {
        this(Props.from(clientProps).bindTo(new McpClientProperties()));
    }

    /**
     * 用于简单构建
     */
    public McpClientToolProvider(String apiUrl) {
        this(new McpClientProperties(apiUrl));
    }

    public McpClientToolProvider(McpClientProperties clientProps) {
        if (clientProps.getHeartbeatInterval() != null) {
            if (clientProps.getHeartbeatInterval().getSeconds() < 10L) {
                throw new IllegalArgumentException("HeartbeatInterval cannot be less than 10s!");
            }
        }

        if (McpChannel.STDIO.equals(clientProps.getChannel())) {
            //stdio 通道
            if (clientProps.getServerParameters() == null) {
                throw new IllegalArgumentException("ServerParameters is null!");
            }
        } else {
            //sse 通道
            if (Utils.isEmpty(clientProps.getApiUrl())) {
                throw new IllegalArgumentException("ApiUrl is empty!");
            }
        }

        this.clientProps = clientProps;

        //开始心跳
        this.heartbeatHandle();
    }

    /**
     * 构建同步客户端
     */
    private McpSyncClient buildClient() {
        McpClientTransport clientTransport;

        if (McpChannel.STDIO.equals(clientProps.getChannel())) {
            //stdio 通道
            clientTransport = new StdioClientTransport(ServerParameters.builder(clientProps.getServerParameters().getCommand())
                    .args(clientProps.getServerParameters().getArgs())
                    .env(clientProps.getServerParameters().getEnv())
                    .build());
        } else {
            //sse 通道
            URI url = URI.create(clientProps.getApiUrl());
            String baseUri = url.getScheme() + "://" + url.getAuthority();
            String sseEndpoint = null;
            if (Utils.isEmpty(url.getRawQuery())) {
                sseEndpoint = url.getRawPath();
            } else {
                sseEndpoint = url.getRawPath() + "?" + url.getRawQuery();
            }


            if (Utils.isEmpty(sseEndpoint)) {
                throw new IllegalArgumentException("SseEndpoint is empty!");
            }

            //超时
            HttpUtilsBuilder webBuilder = new HttpUtilsBuilder();
            webBuilder.baseUri(baseUri);

            if (Utils.isNotEmpty(clientProps.getApiKey())) {
                webBuilder.headerSet("Authorization", "Bearer " + clientProps.getApiKey());
            }

            clientProps.getHeaders().forEach((k, v) -> {
                webBuilder.headerSet(k, v);
            });

            if (clientProps.getHttpTimeout() != null) {
                webBuilder.timeout(clientProps.getHttpTimeout());
            }

            if (clientProps.getHttpProxy() != null) {
                webBuilder.proxy(clientProps.getHttpProxy());
            }

            clientTransport = WebRxSseClientTransport.builder(webBuilder)
                    .sseEndpoint(sseEndpoint)
                    .build();
        }

        return McpClient.sync(clientTransport)
                .clientInfo(new McpSchema.Implementation(clientProps.getName(), clientProps.getVersion()))
                .requestTimeout(clientProps.getRequestTimeout())
                .initializationTimeout(clientProps.getInitializationTimeout())
                .build();
    }


    /**
     * 获取客户端
     */
    public McpSyncClient getClient() {
        LOCKER.lock();

        try {
            if (isClosed.get()) {
                //如果已关闭
                throw new IllegalStateException("The current status has been closed.");
            }

            //设为开始
            isStarted.set(true);

            if (client == null) {
                client = buildClient();
            }

            if (client.isInitialized() == false) {
                client.initialize();
            }

            return client;
        } finally {
            LOCKER.unlock();
        }
    }

    private void heartbeatHandle() {
        if (heartbeatExecutor == null) {
            heartbeatExecutor = Executors.newSingleThreadScheduledExecutor();
        }

        heartbeatHandleDo();
    }

    /**
     * 心跳处理
     */
    private void heartbeatHandleDo() {
        if (heartbeatExecutor == null) {
            return;
        }

        if (clientProps.getHeartbeatInterval() == null) {
            return;
        }

        heartbeatExecutor.schedule(() -> {
            if (Thread.currentThread().isInterrupted()) {
                //如果中断
                return;
            }


            if (isClosed.get() == false) {
                //如果未关闭，尝试心跳
                if (isStarted.get()) {
                    //如果已开始，则心跳发送
                    RunUtil.runAndTry(() -> {
                        try {
                            getClient().ping();
                        } catch (Throwable ex) {
                            //如果失败，重置（下次会尝试重连）
                            this.reset();
                        }
                    });
                }

                heartbeatHandleDo();
            }
        }, this.clientProps.getHeartbeatInterval().toMillis(), TimeUnit.MILLISECONDS);
    }

    /**
     * 关闭
     */
    @Override
    public void close() {
        LOCKER.lock();
        try {
            if (isClosed.get() == false) {
                //如果未关闭
                isClosed.set(true);
                isStarted.set(false);

                if (heartbeatExecutor != null) {
                    heartbeatExecutor.shutdownNow();
                    heartbeatExecutor = null;
                }

                this.reset();
            }
        } finally {
            LOCKER.unlock();
        }
    }

    /**
     * 重新打开
     */
    public void reopen() {
        LOCKER.lock();
        try {
            if (isClosed.get()) {
                //如果已关闭
                isClosed.set(false);
                getClient();
                heartbeatHandle();
            }
        } finally {
            LOCKER.unlock();
        }
    }

    /**
     * 重置
     */
    private void reset() {
        LOCKER.lock();
        try {
            if (client != null) {
                client.close();
                client = null;
            }
        } finally {
            LOCKER.unlock();
        }
    }

    /// /////////////////////////////

    /**
     * 调用工具并转为文本
     *
     * @param name 工具名
     * @param args 调用参数
     */
    public String callToolAsText(String name, Map<String, Object> args) {
        McpSchema.CallToolResult result = callTool(name, args);
        if (Utils.isEmpty(result.getContent())) {
            return null;
        } else {
            return ((McpSchema.TextContent) result.getContent().get(0)).getText();
        }
    }

    /**
     * 调用工具并转为图像
     *
     * @param name 工具名
     * @param args 调用参数
     */
    public Image callToolAsImage(String name, Map<String, Object> args) {
        McpSchema.CallToolResult result = callTool(name, args);
        if (Utils.isEmpty(result.getContent())) {
            return null;
        } else {
            McpSchema.ImageContent imageContent = ((McpSchema.ImageContent) result.getContent().get(0));
            return Image.ofBase64(imageContent.getData(), imageContent.getMimeType());
        }
    }

    /**
     * 调用工具
     *
     * @param name 工具名
     * @param args 调用参数
     */
    protected McpSchema.CallToolResult callTool(String name, Map<String, Object> args) {
        try {
            McpSchema.CallToolRequest callToolRequest = new McpSchema.CallToolRequest(name, args);
            McpSchema.CallToolResult response = getClient().callTool(callToolRequest);

            if (response.getIsError() == null || response.getIsError() == false) {
                return response;
            } else {
                if (Utils.isEmpty(response.getContent())) {
                    throw new McpException("Call Toll Failed");
                } else {
                    throw new McpException(response.getContent().get(0).toString());
                }
            }
        } catch (RuntimeException ex) {
            this.reset();
            throw ex;
        }
    }

    /// ///////////
    /**
     * 获取函数工具（可用于模型绑定）
     */
    @Override
    public Collection<FunctionTool> getTools() {
        return getTools(null);
    }

    /**
     * 获取函数工具（可用于模型绑定）
     *
     * @param cursor 游标
     */
    public Collection<FunctionTool> getTools(String cursor) {
        List<FunctionTool> toolList = new ArrayList<>();

        McpSchema.ListToolsResult result = null;
        if (cursor == null) {
            result = getClient().listTools();
        } else {
            result = getClient().listTools(cursor);
        }

        for (McpSchema.Tool tool : result.getTools()) {
            String name = tool.getName();
            String description = tool.getDescription();
            String inputSchema = ONode.load(tool.getInputSchema()).toJson();

            RefererFunctionTool functionRefer = new RefererFunctionTool(
                    name,
                    description,
                    inputSchema,
                    args -> callToolAsText(name, args));

            toolList.add(functionRefer);
        }

        return toolList;
    }

    /**
     * 根据 mcpServers 配置加载客户端
     *
     * @param uri 配置资源地址
     */
    public static Map<String, McpClientToolProvider> fromMcpServers(String uri) throws IOException {
        Assert.notEmpty(uri, "Uri is empty");

        URL res = ResourceUtil.findResource(uri);
        String json = ResourceUtil.getResourceAsString(res);
        ONode jsonDom = ONode.loadStr(json);

        Map<String, McpClientToolProvider> map = new HashMap<>();
        for (Map.Entry<String, ONode> kv : jsonDom.get("mcpServers").obj().entrySet()) {
            String name = kv.getKey();
            Map<String, String> env = kv.getValue().get("env").toObject(new HashMap<String, String>() {
            }.getClass());

            String type = kv.getValue().get("type").getString();

            if (Utils.isEmpty(type)) {
                //兼容没有 type 配置的情况
                if (kv.getValue().contains("url")) {
                    type = McpChannel.SSE;
                } else {
                    type = McpChannel.STDIO;
                }
            }


            Builder builder = builder().channel(type);

            if (McpChannel.STDIO.equalsIgnoreCase(type)) {
                String command = kv.getValue().get("command").getString();
                List<String> args = kv.getValue().get("args").toObject(new ArrayList<String>() {
                }.getClass());

                builder.serverParameters(McpServerParameters.builder(command).args(args).env(env).build());
            } else {
                String url = kv.getValue().get("url").getString();
                builder.apiUrl(url).headerSet(env);
            }

            map.put(name, builder.build());
        }

        return map;
    }

    public static Builder builder() {
        return new Builder();
    }

    public static class Builder {
        private McpClientProperties props = new McpClientProperties();

        public Builder name(String name) {
            props.setName(name);
            return this;
        }

        public Builder version(String version) {
            props.setVersion(version);
            return this;
        }

        public Builder channel(String channel) {
            props.setChannel(channel);
            return this;
        }

        public Builder apiUrl(String apiUrl) {
            props.setApiUrl(apiUrl);
            return this;
        }

        public Builder apiKey(String apiKey) {
            props.setApiKey(apiKey);
            return this;
        }

        public Builder headerSet(String name, String value) {
            props.getHeaders().put(name, value);
            return this;
        }

        public Builder headerSet(Map<String, String> headers) {
            if (Utils.isNotEmpty(headers)) {
                props.getHeaders().putAll(headers);
            }
            return this;
        }

        public Builder httpTimeout(HttpTimeout httpTimeout) {
            props.setHttpTimeout(httpTimeout);
            return this;
        }

        public Builder httpProxy(Proxy httpProxy) {
            props.setHttpProxy(httpProxy);
            return this;
        }

        public Builder httpProxy(String host, int port) {
            return httpProxy(new Proxy(Proxy.Type.HTTP, new InetSocketAddress(host, port)));
        }

        public Builder requestTimeout(Duration requestTimeout) {
            props.setRequestTimeout(requestTimeout);
            return this;
        }

        public Builder initializationTimeout(Duration initializationTimeout) {
            props.setInitializationTimeout(initializationTimeout);
            return this;
        }

        public Builder heartbeatInterval(Duration heartbeatInterval) {
            props.setHeartbeatInterval(heartbeatInterval);
            return this;
        }

        public Builder serverParameters(McpServerParameters serverParameters) {
            props.setServerParameters(serverParameters);
            return this;
        }

        public McpClientToolProvider build() {
            return new McpClientToolProvider(props);
        }
    }
}

[lambochen]

@lambochen That's not necessarily... This is how solon-ai-mcp implements disconnection reconnections (e.g., server restarts). It has been tested and used in production.

https://github.com/opensolon/solon-ai/blob/dev/solon-ai-mcp/src/main/java/org/noear/solon/ai/mcp/client/McpClientToolProvider.java

import io.modelcontextprotocol.client.McpClient;
import io.modelcontextprotocol.client.McpSyncClient;
import io.modelcontextprotocol.client.transport.ServerParameters;
import io.modelcontextprotocol.client.transport.StdioClientTransport;
import io.modelcontextprotocol.client.transport.WebRxSseClientTransport;
import io.modelcontextprotocol.spec.McpClientTransport;
import io.modelcontextprotocol.spec.McpSchema;
import org.noear.snack.ONode;
import org.noear.solon.Utils;
import org.noear.solon.ai.chat.tool.FunctionTool;
import org.noear.solon.ai.chat.tool.RefererFunctionTool;
import org.noear.solon.ai.chat.tool.ToolProvider;
import org.noear.solon.ai.image.Image;
import org.noear.solon.ai.mcp.McpChannel;
import org.noear.solon.ai.mcp.exception.McpException;
import org.noear.solon.core.Props;
import org.noear.solon.core.util.Assert;
import org.noear.solon.core.util.ResourceUtil;
import org.noear.solon.core.util.RunUtil;
import org.noear.solon.net.http.HttpTimeout;
import org.noear.solon.net.http.HttpUtilsBuilder;

import java.io.Closeable;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.Proxy;
import java.net.URI;
import java.net.URL;
import java.time.Duration;
import java.util.*;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.ReentrantLock;

/**

• Mcp 连接工具提供者

• @author noear

• @SInCE 3.1
  */
  public class McpClientToolProvider implements ToolProvider, Closeable {
  private final ReentrantLock LOCKER = new ReentrantLock();

  private final AtomicBoolean isClosed = new AtomicBoolean(false);
  private final AtomicBoolean isStarted = new AtomicBoolean(false);
  private final McpClientProperties clientProps;
  private ScheduledExecutorService heartbeatExecutor;
  private McpSyncClient client;

  /**

  • 用于支持注入
    */
    public McpClientToolProvider(Properties clientProps) {
    this(Props.from(clientProps).bindTo(new McpClientProperties()));
    }

  /**

  • 用于简单构建
    */
    public McpClientToolProvider(String apiUrl) {
    this(new McpClientProperties(apiUrl));
    }

  public McpClientToolProvider(McpClientProperties clientProps) {
  if (clientProps.getHeartbeatInterval() != null) {
  if (clientProps.getHeartbeatInterval().getSeconds() < 10L) {
  throw new IllegalArgumentException("HeartbeatInterval cannot be less than 10s!");
  }
  }

   if (McpChannel.STDIO.equals(clientProps.getChannel())) {
       //stdio 通道
       if (clientProps.getServerParameters() == null) {
           throw new IllegalArgumentException("ServerParameters is null!");
       }
   } else {
       //sse 通道
       if (Utils.isEmpty(clientProps.getApiUrl())) {
           throw new IllegalArgumentException("ApiUrl is empty!");
       }
   }
  
   this.clientProps = clientProps;
  
   //开始心跳
   this.heartbeatHandle();
  

  }

  /**

  • 构建同步客户端
    */
    private McpSyncClient buildClient() {
    McpClientTransport clientTransport;

    if (McpChannel.STDIO.equals(clientProps.getChannel())) {
    //stdio 通道
    clientTransport = new StdioClientTransport(ServerParameters.builder(clientProps.getServerParameters().getCommand())
    .args(clientProps.getServerParameters().getArgs())
    .env(clientProps.getServerParameters().getEnv())
    .build());
    } else {
    //sse 通道
    URI url = URI.create(clientProps.getApiUrl());
    String baseUri = url.getScheme() + "://" + url.getAuthority();
    String sseEndpoint = null;
    if (Utils.isEmpty(url.getRawQuery())) {
    sseEndpoint = url.getRawPath();
    } else {
    sseEndpoint = url.getRawPath() + "?" + url.getRawQuery();
    }

     if (Utils.isEmpty(sseEndpoint)) {
         throw new IllegalArgumentException("SseEndpoint is empty!");
     }
    
     //超时
     HttpUtilsBuilder webBuilder = new HttpUtilsBuilder();
     webBuilder.baseUri(baseUri);
    
     if (Utils.isNotEmpty(clientProps.getApiKey())) {
         webBuilder.headerSet("Authorization", "Bearer " + clientProps.getApiKey());
     }
    
     clientProps.getHeaders().forEach((k, v) -> {
         webBuilder.headerSet(k, v);
     });
    
     if (clientProps.getHttpTimeout() != null) {
         webBuilder.timeout(clientProps.getHttpTimeout());
     }
    
     if (clientProps.getHttpProxy() != null) {
         webBuilder.proxy(clientProps.getHttpProxy());
     }
    
     clientTransport = WebRxSseClientTransport.builder(webBuilder)
             .sseEndpoint(sseEndpoint)
             .build();
    

    }

    return McpClient.sync(clientTransport)
    .clientInfo(new McpSchema.Implementation(clientProps.getName(), clientProps.getVersion()))
    .requestTimeout(clientProps.getRequestTimeout())
    .initializationTimeout(clientProps.getInitializationTimeout())
    .build();
    }

  /**

  • 获取客户端
    */
    public McpSyncClient getClient() {
    LOCKER.lock();

    try {
    if (isClosed.get()) {
    //如果已关闭
    throw new IllegalStateException("The current status has been closed.");
    }

     //设为开始
     isStarted.set(true);
    
     if (client == null) {
         client = buildClient();
     }
    
     if (client.isInitialized() == false) {
         client.initialize();
     }
    
     return client;
    

    } finally {
    LOCKER.unlock();
    }
    }

  private void heartbeatHandle() {
  if (heartbeatExecutor == null) {
  heartbeatExecutor = Executors.newSingleThreadScheduledExecutor();
  }

   heartbeatHandleDo();
  

  }

  /**

  • 心跳处理
    */
    private void heartbeatHandleDo() {
    if (heartbeatExecutor == null) {
    return;
    }

    if (clientProps.getHeartbeatInterval() == null) {
    return;
    }

    heartbeatExecutor.schedule(() -> {
    if (Thread.currentThread().isInterrupted()) {
    //如果中断
    return;
    }

     if (isClosed.get() == false) {
         //如果未关闭，尝试心跳
         if (isStarted.get()) {
             //如果已开始，则心跳发送
             RunUtil.runAndTry(() -> {
                 try {
                     getClient().ping();
                 } catch (Throwable ex) {
                     //如果失败，重置（下次会尝试重连）
                     this.reset();
                 }
             });
         }
    
         heartbeatHandleDo();
     }
    

    }, this.clientProps.getHeartbeatInterval().toMillis(), TimeUnit.MILLISECONDS);
    }

  /**

  • 关闭
    */
    @OverRide
    public void close() {
    LOCKER.lock();
    try {
    if (isClosed.get() == false) {
    //如果未关闭
    isClosed.set(true);
    isStarted.set(false);

         if (heartbeatExecutor != null) {
             heartbeatExecutor.shutdownNow();
             heartbeatExecutor = null;
         }
    
         this.reset();
     }
    

    } finally {
    LOCKER.unlock();
    }
    }

  /**

  • 重新打开
    */
    public void reopen() {
    LOCKER.lock();
    try {
    if (isClosed.get()) {
    //如果已关闭
    isClosed.set(false);
    getClient();
    heartbeatHandle();
    }
    } finally {
    LOCKER.unlock();
    }
    }

  /**

  • 重置
    */
    private void reset() {
    LOCKER.lock();
    try {
    if (client != null) {
    client.close();
    client = null;
    }
    } finally {
    LOCKER.unlock();
    }
    }

  /// /////////////////////////////

  /**

  • 调用工具并转为文本
  • @param name 工具名
  • @param args 调用参数
    */
    public String callToolAsText(String name, Map<String, Object> args) {
    McpSchema.CallToolResult result = callTool(name, args);
    if (Utils.isEmpty(result.getContent())) {
    return null;
    } else {
    return ((McpSchema.TextContent) result.getContent().get(0)).getText();
    }
    }

  /**

  • 调用工具并转为图像
  • @param name 工具名
  • @param args 调用参数
    */
    public Image callToolAsImage(String name, Map<String, Object> args) {
    McpSchema.CallToolResult result = callTool(name, args);
    if (Utils.isEmpty(result.getContent())) {
    return null;
    } else {
    McpSchema.ImageContent imageContent = ((McpSchema.ImageContent) result.getContent().get(0));
    return Image.ofBase64(imageContent.getData(), imageContent.getMimeType());
    }
    }

  /**

  • 调用工具

  • @param name 工具名

  • @param args 调用参数
    */
    protected McpSchema.CallToolResult callTool(String name, Map<String, Object> args) {
    try {
    McpSchema.CallToolRequest callToolRequest = new McpSchema.CallToolRequest(name, args);
    McpSchema.CallToolResult response = getClient().callTool(callToolRequest);

     if (response.getIsError() == null || response.getIsError() == false) {
         return response;
     } else {
         if (Utils.isEmpty(response.getContent())) {
             throw new McpException("Call Toll Failed");
         } else {
             throw new McpException(response.getContent().get(0).toString());
         }
     }
    

    } catch (RuntimeException ex) {
    this.reset();
    throw ex;
    }
    }

  /// ///////////
  /**

  • 获取函数工具（可用于模型绑定）
    */
    @OverRide
    public Collection getTools() {
    return getTools(null);
    }

  /**

  • 获取函数工具（可用于模型绑定）

  • @param cursor 游标
    */
    public Collection getTools(String cursor) {
    List toolList = new ArrayList<>();

    McpSchema.ListToolsResult result = null;
    if (cursor == null) {
    result = getClient().listTools();
    } else {
    result = getClient().listTools(cursor);
    }

    for (McpSchema.Tool tool : result.getTools()) {
    String name = tool.getName();
    String description = tool.getDescription();
    String inputSchema = ONode.load(tool.getInputSchema()).toJson();

     RefererFunctionTool functionRefer = new RefererFunctionTool(
             name,
             description,
             inputSchema,
             args -> callToolAsText(name, args));
    
     toolList.add(functionRefer);
    

    }

    return toolList;
    }

  /**

  • 根据 mcpServers 配置加载客户端

  • @param uri 配置资源地址
    */
    public static Map<String, McpClientToolProvider> fromMcpServers(String uri) throws IOException {
    Assert.notEmpty(uri, "Uri is empty");

    URL res = ResourceUtil.findResource(uri);
    String json = ResourceUtil.getResourceAsString(res);
    ONode jsonDom = ONode.loadStr(json);

    Map<String, McpClientToolProvider> map = new HashMap<>();
    for (Map.Entry<String, ONode> kv : jsonDom.get("mcpServers").obj().entrySet()) {
    String name = kv.getKey();
    Map<String, String> env = kv.getValue().get("env").toObject(new HashMap<String, String>() {
    }.getClass());

     String type = kv.getValue().get("type").getString();
    
     if (Utils.isEmpty(type)) {
         //兼容没有 type 配置的情况
         if (kv.getValue().contains("url")) {
             type = McpChannel.SSE;
         } else {
             type = McpChannel.STDIO;
         }
     }
    
    
     Builder builder = builder().channel(type);
    
     if (McpChannel.STDIO.equalsIgnoreCase(type)) {
         String command = kv.getValue().get("command").getString();
         List<String> args = kv.getValue().get("args").toObject(new ArrayList<String>() {
         }.getClass());
    
         builder.serverParameters(McpServerParameters.builder(command).args(args).env(env).build());
     } else {
         String url = kv.getValue().get("url").getString();
         builder.apiUrl(url).headerSet(env);
     }
    
     map.put(name, builder.build());
    

    }

    return map;
    }

  public static Builder builder() {
  return new Builder();
  }

  public static class Builder {
  private McpClientProperties props = new McpClientProperties();

   public Builder name(String name) {
       props.setName(name);
       return this;
   }
  
   public Builder version(String version) {
       props.setVersion(version);
       return this;
   }
  
   public Builder channel(String channel) {
       props.setChannel(channel);
       return this;
   }
  
   public Builder apiUrl(String apiUrl) {
       props.setApiUrl(apiUrl);
       return this;
   }
  
   public Builder apiKey(String apiKey) {
       props.setApiKey(apiKey);
       return this;
   }
  
   public Builder headerSet(String name, String value) {
       props.getHeaders().put(name, value);
       return this;
   }
  
   public Builder headerSet(Map<String, String> headers) {
       if (Utils.isNotEmpty(headers)) {
           props.getHeaders().putAll(headers);
       }
       return this;
   }
  
   public Builder httpTimeout(HttpTimeout httpTimeout) {
       props.setHttpTimeout(httpTimeout);
       return this;
   }
  
   public Builder httpProxy(Proxy httpProxy) {
       props.setHttpProxy(httpProxy);
       return this;
   }
  
   public Builder httpProxy(String host, int port) {
       return httpProxy(new Proxy(Proxy.Type.HTTP, new InetSocketAddress(host, port)));
   }
  
   public Builder requestTimeout(Duration requestTimeout) {
       props.setRequestTimeout(requestTimeout);
       return this;
   }
  
   public Builder initializationTimeout(Duration initializationTimeout) {
       props.setInitializationTimeout(initializationTimeout);
       return this;
   }
  
   public Builder heartbeatInterval(Duration heartbeatInterval) {
       props.setHeartbeatInterval(heartbeatInterval);
       return this;
   }
  
   public Builder serverParameters(McpServerParameters serverParameters) {
       props.setServerParameters(serverParameters);
       return this;
   }
  
   public McpClientToolProvider build() {
       return new McpClientToolProvider(props);
   }
  

  }
  }

👍, I have also implemented a similar reconnection mechanism in my production practice.

[noear]

@lambochen If you are based on java8, consider using solon-ai-mcp

[HarryFQG]

@lambochen Could you take a look at the specific implementation method for reconnecting after your heart rate detection fails?

[noear]

@HarryFQG if heartbeat and automatic reconnect? can see solon-ai-mcp

https://github.com/opensolon/solon-ai

[HarryFQG]

My runtime environment is JDK17. Error: The Transport exception was not passed out and could not be caught.

[oldmanpushcart]

It's a pity that when the connection is reset, you can only detect it through a heartbeat (Eg. Ping) mechanism and then actively reconnect. Be sure not to let your code permanently hold onto the McpAsyncClient / McpSyncClient instance, because reconnect will recreate them.