diff --git a/local_plugins/open_ai_service/android/src/main/kotlin/com/yunqiinnovation/open_ai_service/MCPClient.kt b/local_plugins/open_ai_service/android/src/main/kotlin/com/yunqiinnovation/open_ai_service/MCPClient.kt index c94ebb3d4..b186b6a64 100644 --- a/local_plugins/open_ai_service/android/src/main/kotlin/com/yunqiinnovation/open_ai_service/MCPClient.kt +++ b/local_plugins/open_ai_service/android/src/main/kotlin/com/yunqiinnovation/open_ai_service/MCPClient.kt @@ -33,6 +33,14 @@ import okhttp3.Interceptor import io.ktor.client.request.* import io.ktor.http.* import com.yunqiinnovation.open_ai_service.mcp.CustomSseClientTransport +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.Job +import kotlinx.coroutines.delay +import kotlinx.coroutines.isActive +import kotlinx.coroutines.launch +import kotlinx.coroutines.sync.Mutex +import kotlinx.coroutines.sync.withLock import okio.Buffer /** @@ -68,11 +76,17 @@ class MCPSubClient( // 保存服务器URL private val serverUrl: String = serverUrl - - init { - Log.d(TAG, "MCPSubClient初始化，URL: $serverUrl") - } - + + // 重连配置 + private val maxRetryAttempts = 3 + private val initialReconnectDelay = 1000L // 初始重连延迟1秒 + private val maxReconnectDelay = 30000L // 最大重连延迟30秒 + private var currentReconnectDelay = initialReconnectDelay + private var retryCount = 0 + + // 连接状态锁 + private val connectionLock = Mutex() + // 初始化MCP客户端 private val mcp: Client = Client( clientInfo = Implementation( @@ -95,7 +109,18 @@ class MCPSubClient( // 工具名称到工具的映射，用于快速查找 private val toolMapping = mutableMapOf() - + + // 心跳检测 + private var heartbeatJob: Job? = null + private val heartbeatInterval = 30000L // 30秒 + + // 网络状态监听 + private var networkStateListener: NetworkStateListener? = null + + init { + Log.d(TAG, "MCPSubClient初始化，URL: $serverUrl") + } + /** * 将JsonObject转换为JsonValue */ @@ -110,6 +135,10 @@ class MCPSubClient( * @return 是否连接成功 */ suspend fun connect(): Boolean { + if (isConnected) { + Log.d(TAG, "已经连接到服务器，跳过重复连接") + return true + } try { Log.d(TAG, "开始连接服务器: $serverUrl") @@ -130,6 +159,11 @@ class MCPSubClient( processToolsList(toolsResult) isConnected = true + retryCount = 0 + currentReconnectDelay = initialReconnectDelay + // 启动心跳检测 + startHeartbeat() + Log.d(TAG, "连接成功，可用工具: ${tools.size}个") return true } catch (e: Exception) { @@ -368,9 +402,11 @@ class MCPSubClient( * 调用工具 */ suspend fun callTool(name: String, arguments: Map): JSONObject? { - if (!isConnected) { - Log.e(TAG, "未连接到服务器") - return JSONObject().apply { put("context","未连接到MCP服务器") } + + // 先检查连接状态 + if (!checkConnection()) { + Log.e(TAG, "无法连接到服务器，工具调用失败") + return JSONObject().apply { put("context", "无法连接到MCP服务器") } } if (!containsTool(name)) { @@ -395,32 +431,123 @@ class MCPSubClient( } } catch (e: Exception) { Log.e(TAG, "调用工具失败: ${e.message}") + isConnected = false JSONObject().apply { put("context", "调用工具失败: ${e.message}") } } } - + + /** + * 设置网络状态监听器 + */ + fun setNetworkStateListener(listener: NetworkStateListener) { + this.networkStateListener = listener + } + + /** + * 检查连接状态并自动重连 + */ + suspend fun checkConnection(): Boolean { + if (!isConnected) { + Log.d(TAG, "当前未连接，尝试重新连接...") + return connect() + } + + try { + // 简单ping测试连接状态 + mcp.ping() + return true + } catch (e: Exception) { + Log.e(TAG, "连接检查失败: ${e.message}") + isConnected = false + return false + } + } /** * 检查是否已连接 */ fun isConnected(): Boolean { return isConnected } - + /** + * 处理连接错误并尝试重连 + */ + private suspend fun handleConnectionError(e: Exception) { + isConnected = false + stopHeartbeat() + + if (retryCount < maxRetryAttempts) { + retryCount++ + currentReconnectDelay = minOf(currentReconnectDelay * 2, maxReconnectDelay) + + Log.w(TAG, "连接失败，将在 ${currentReconnectDelay}ms 后尝试重连 (尝试 $retryCount/$maxRetryAttempts)") + + delay(currentReconnectDelay) + connect() + } else { + Log.e(TAG, "已达到最大重试次数($maxRetryAttempts)，停止重连") + // 可以在这里添加通知或回调，告知上层连接彻底失败 + } + } + + /** + * 启动心跳检测 + */ + private fun startHeartbeat() { + heartbeatJob?.cancel() + heartbeatJob = CoroutineScope(Dispatchers.IO).launch { + while (isActive && isConnected) { + try { + delay(heartbeatInterval) + + // 发送心跳请求 + val startTime = System.currentTimeMillis() + val response = mcp.ping() + val latency = System.currentTimeMillis() - startTime + + Log.d(TAG, "心跳检测成功，延迟: ${latency}ms") + } catch (e: Exception) { + Log.e(TAG, "心跳检测失败: ${e.message}") + handleConnectionError(e) + break + } + } + } + } + + /** + * 停止心跳检测 + */ + private fun stopHeartbeat() { + heartbeatJob?.cancel() + heartbeatJob = null + } + /** * 关闭连接 */ override fun close() { runBlocking { - try { - mcp.close() - sseTransport?.close() - isConnected = false - Log.d(TAG, "已关闭MCP服务器连接") - } catch (e: Exception) { - Log.e(TAG, "关闭MCP服务器连接失败: ${e.message}", e) + connectionLock.withLock { + try { + isConnected = false + stopHeartbeat() + mcp.close() + sseTransport?.close() + Log.d(TAG, "已关闭MCP服务器连接") + } catch (e: Exception) { + Log.e(TAG, "关闭MCP服务器连接失败: ${e.message}", e) + } } } } + + /** + * 网络状态监听接口 + */ + interface NetworkStateListener { + fun onNetworkAvailable() + fun onNetworkLost() + } } /** diff --git a/local_plugins/open_ai_service/android/src/main/kotlin/com/yunqiinnovation/open_ai_service/mcp/CustomSseClientTransport.kt b/local_plugins/open_ai_service/android/src/main/kotlin/com/yunqiinnovation/open_ai_service/mcp/CustomSseClientTransport.kt index 6d6a5eb27..b198298a2 100644 --- a/local_plugins/open_ai_service/android/src/main/kotlin/com/yunqiinnovation/open_ai_service/mcp/CustomSseClientTransport.kt +++ b/local_plugins/open_ai_service/android/src/main/kotlin/com/yunqiinnovation/open_ai_service/mcp/CustomSseClientTransport.kt @@ -19,41 +19,8 @@ import kotlin.properties.Delegates import kotlin.time.Duration import java.util.concurrent.atomic.AtomicBoolean import org.json.JSONObject -import kotlinx.serialization.Serializable -import kotlinx.serialization.json.JsonElement -import kotlinx.serialization.json.JsonPrimitive -//@Serializable -//data class JSONRPCRequest( -// val jsonrpc: String = "2.0", -// val method: String, -// val params: JsonElement? = null, -// val id: JsonPrimitive? = null -//) : JSONRPCMessage, Parcelable { -// -// constructor(parcel: Parcel) : this( -// parcel.readString() ?: "2.0", -// parcel.readString() ?: "", -// parcel.readString()?.let { Json.decodeFromString(it) }, -// parcel.readString()?.let { Json.decodeFromString(it) } -// ) -// -// override fun writeToParcel(parcel: Parcel, flags: Int) { -// parcel.writeString(jsonrpc) -// parcel.writeString(method) -// parcel.writeString(params?.let { Json.encodeToString(it) }) -// parcel.writeString(id?.let { Json.encodeToString(it) }) -// } -// -// override fun describeContents(): Int = 0 -// -// companion object CREATOR : Parcelable.Creator { -// override fun createFromParcel(parcel: Parcel): JSONRPCRequest = JSONRPCRequest(parcel) -// override fun newArray(size: Int): Array = arrayOfNulls(size) -// } -//} - /** * 自定义SSE客户端传输层，修复原始SseClientTransport中的URL拼接问题 * 解决URL查询参数与路径拼接错误的问题，确保消息端点URL格式正确 @@ -77,11 +44,6 @@ class CustomSseClientTransport( private var job: Job? = null - - private val heartbeatIntervalMs = 30_000L // 30秒 - private var heartbeatJob: Job? = null - - // 创建JSON解析器，增强灵活性设置 private val json = Json { ignoreUnknownKeys = true // 忽略未知字段 @@ -341,9 +303,6 @@ class CustomSseClientTransport( // 收集SSE事件 collectEvents() - // 启动心跳任务 - startHeartbeat() - // 等待endpoint就绪 endpoint.await() Log.d(TAG, "传输层启动完成，消息端点已就绪") @@ -391,7 +350,6 @@ class CustomSseClientTransport( Log.e(TAG, "关闭失败: 传输层未初始化") error("CustomSseClientTransport is not initialized!") } - heartbeatJob?.cancel() session.cancel() _onClose() job?.cancelAndJoin() @@ -404,29 +362,6 @@ class CustomSseClientTransport( fun isInitialized(): Boolean { return initialized.get() } - - - private fun startHeartbeat() { - heartbeatJob = scope.launch { - while (isActive) { - delay(heartbeatIntervalMs) - try { - sendHeartbeat() - } catch (e: Exception) { - Log.e(TAG, "心跳发送失败: ${e.message}", e) - _onError(e) - } - } - } - } - - private suspend fun sendHeartbeat() { - val pingMessage = JSONRPCRequest( - method = "ping", - jsonrpc = "2.0" - ) - send(pingMessage) - } } \ No newline at end of file