diff --git a/lib/data/models/appconfig_model.dart b/lib/data/models/appconfig_model.dart index e656a62c9..ec566e102 100644 --- a/lib/data/models/appconfig_model.dart +++ b/lib/data/models/appconfig_model.dart @@ -58,10 +58,12 @@ class DBAgent { class DBMCPServer { final String servername; final String url; + final int type; final String tools; DBMCPServer({ required this.servername, required this.url, + required this.type, required this.tools, }); diff --git a/lib/data/models/appconfig_model.g.dart b/lib/data/models/appconfig_model.g.dart index fc97723ed..4c8be2f92 100644 --- a/lib/data/models/appconfig_model.g.dart +++ b/lib/data/models/appconfig_model.g.dart @@ -61,6 +61,7 @@ Map _$DBAgentToJson(DBAgent instance) => { DBMCPServer _$DBMCPServerFromJson(Map json) => DBMCPServer( servername: json['servername'] as String, url: json['url'] as String, + type: json['type'] as int, tools: json['tools'] as String? ?? '', ); @@ -68,6 +69,7 @@ Map _$DBMCPServerToJson(DBMCPServer instance) => { 'servername': instance.servername, 'url': instance.url, + 'type': instance.type, 'tools': instance.tools, }; diff --git a/local_plugins/agent_service/android/src/main/kotlin/com/yunqiinnovation/agent_service/AgentService.kt b/local_plugins/agent_service/android/src/main/kotlin/com/yunqiinnovation/agent_service/AgentService.kt index 37adabf4f..3a1dddbd6 100644 --- a/local_plugins/agent_service/android/src/main/kotlin/com/yunqiinnovation/agent_service/AgentService.kt +++ b/local_plugins/agent_service/android/src/main/kotlin/com/yunqiinnovation/agent_service/AgentService.kt @@ -295,8 +295,6 @@ object AgentService : CoroutineScope { config["openaivisionModel"]?.toString() ?: "", config["mcpServer"]?.toString() ?: "", config["musicServiceType"]?.toString() ?: "", - serverurl, - usertoken, ) Log.e(TAG, "liwei -----------读取服务地址和用户token: $serverurl $usertoken") diff --git a/local_plugins/chat_api/android/src/main/kotlin/com/yunqiinnovation/chat_api/ChatApiService.kt b/local_plugins/chat_api/android/src/main/kotlin/com/yunqiinnovation/chat_api/ChatApiService.kt index d9d2423dd..7be026f4c 100644 --- a/local_plugins/chat_api/android/src/main/kotlin/com/yunqiinnovation/chat_api/ChatApiService.kt +++ b/local_plugins/chat_api/android/src/main/kotlin/com/yunqiinnovation/chat_api/ChatApiService.kt @@ -219,7 +219,7 @@ class ChatApiService(private val context: android.content.Context? = null) : Cor /** * 初始化ChatAPI服务 */ - fun initialize(apiKey: String, baseUrl: String, model: String, visionModel: String, mcpServer: String,musicType: String = "AUTO",serverhUrl: String = "",token: String = ""): Boolean { + fun initialize(apiKey: String, baseUrl: String, model: String, visionModel: String, mcpServer: String,musicType: String = "AUTO"): Boolean { this.apiKey = apiKey if (baseUrl.isNotEmpty()) { this.baseUrl = baseUrl @@ -257,7 +257,7 @@ class ChatApiService(private val context: android.content.Context? = null) : Cor mcpConfigJson = mcpServer // 异步初始化MCP客户端 launch { - initializeMcpClient(mcpServer,musicType,serverhUrl,token) + initializeMcpClient(mcpServer,musicType) // initializeMcpClient("{}") } @@ -941,9 +941,9 @@ class ChatApiService(private val context: android.content.Context? = null) : Cor /** * 初始化MCP客户端 */ - fun initializeMcpClient(serverUrl: String,musicType: String = "AUTO",searchUrl: String = "",token: String = ""): Boolean { + fun initializeMcpClient(serverUrl: String,musicType: String = "AUTO"): Boolean { if (_mcpClient == null) { - _mcpClient = MCPClient(context,musicType,searchUrl,token) + _mcpClient = MCPClient(context,musicType) } // 直接使用类的CoroutineScope启动协程 diff --git a/local_plugins/chat_api/android/src/main/kotlin/com/yunqiinnovation/chat_api/CustomSseClientTransport.kt b/local_plugins/chat_api/android/src/main/kotlin/com/yunqiinnovation/chat_api/CustomSseClientTransport.kt index da85cbaeb..80cfe5e40 100644 --- a/local_plugins/chat_api/android/src/main/kotlin/com/yunqiinnovation/chat_api/CustomSseClientTransport.kt +++ b/local_plugins/chat_api/android/src/main/kotlin/com/yunqiinnovation/chat_api/CustomSseClientTransport.kt @@ -10,7 +10,7 @@ import io.modelcontextprotocol.kotlin.sdk.JSONRPCMessage import io.modelcontextprotocol.kotlin.sdk.shared.AbstractTransport import kotlinx.coroutines.* import kotlinx.serialization.encodeToString -import kotlinx.serialization.json.Json +import kotlinx.serialization.json.* import kotlinx.serialization.decodeFromString import kotlin.properties.Delegates import kotlin.time.Duration @@ -416,4 +416,145 @@ class CustomSseClientTransport( transportScope.cancel() Log.d(TAG, "$serviceidString CustomSseClientTransport已关闭") } -} \ No newline at end of file +} + +class CustomStreamableHttpClientTransport( + private val client: HttpClient, + private val serviceidString: String?, + public val urlString: String?, + private val requestBuilder: HttpRequestBuilder.() -> Unit = {}, +) : AbstractTransport() { + private val TAG = "CustomStreamableHttpClientTransport" + + private val transportScope = CoroutineScope(Dispatchers.IO + SupervisorJob()) + private val initialized = AtomicBoolean(false) + private val closed = AtomicBoolean(false) + + private val json = Json { + ignoreUnknownKeys = true + isLenient = true + coerceInputValues = true + encodeDefaults = true + explicitNulls = false + } + + override suspend fun start() { + if (!initialized.compareAndSet(false, true)) { + Log.e(TAG, "$serviceidString 传输层已经启动，不能重复启动") + error("CustomStreamableHttpClientTransport already started!") + } + if (urlString.isNullOrBlank()) { + Log.e(TAG, "$serviceidString 启动失败: urlString为空") + throw IllegalArgumentException("urlString is required") + } + closed.set(false) + Log.d(TAG, "$serviceidString CustomStreamableHttpClientTransport启动完成: $urlString") + } + + override suspend fun send(message: JSONRPCMessage) { + if (!initialized.get() || closed.get()) { + Log.e(TAG, "$serviceidString 发送失败: 传输层未启动或已关闭") + error("Transport not started or already closed") + } + if (urlString.isNullOrBlank()) { + Log.e(TAG, "$serviceidString 发送失败: urlString为空") + error("Invalid urlString") + } + + try { + val jsonString = json.encodeToString(message) + val response = withTimeout(15000) { + client.post(urlString) { + headers.append(HttpHeaders.ContentType, ContentType.Application.Json.toString()) + headers.append(HttpHeaders.Accept, "${ContentType.Application.Json}, ${ContentType.Text.EventStream}") + setBody(jsonString) + requestBuilder() + } + } + + when { + response.status == HttpStatusCode.Accepted || response.status == HttpStatusCode.NoContent -> { + Log.d(TAG, "$serviceidString 消息已被接受: HTTP ${response.status}") + return + } + response.status.isSuccess() -> { + val contentType = response.headers[HttpHeaders.ContentType].orEmpty() + val bodyText = response.bodyAsText() + val snippet = if (bodyText.length > 500) bodyText.take(500) + "..." else bodyText + Log.d( + TAG, + "$serviceidString HTTP响应: status=${response.status.value}, contentType=$contentType, bodyLen=${bodyText.length}, bodySnippet=${snippet.replace("\n", "\\n")}" + ) + if (bodyText.isBlank()) { + return + } + if (!contentType.contains("application/json", ignoreCase = true)) { + Log.w(TAG, "$serviceidString 收到非JSON响应: Content-Type=$contentType") + return + } + + emitJsonRpcFromBody(bodyText) + } + else -> { + val text = response.bodyAsText() + val snippet = if (text.length > 500) text.take(500) + "..." else text + Log.w( + TAG, + "$serviceidString HTTP错误响应: status=${response.status.value}, bodyLen=${text.length}, bodySnippet=${snippet.replace("\n", "\\n")}" + ) + throw Exception("HTTP ${response.status}: $text") + } + } + } catch (e: Exception) { + Log.e(TAG, "$serviceidString 发送消息异常: ${e.message}", e) + _onError(e) + throw e + } + } + + fun isConnectionActive(): Boolean { + return initialized.get() && !closed.get() && transportScope.isActive + } + + override suspend fun close() { + if (!initialized.get()) { + Log.e(TAG, "$serviceidString 关闭失败: 传输层未初始化") + error("CustomStreamableHttpClientTransport is not initialized!") + } + closed.set(true) + transportScope.cancel() + _onClose() + Log.d(TAG, "$serviceidString CustomStreamableHttpClientTransport已关闭") + } + + private suspend fun emitJsonRpcFromBody(bodyText: String) { + try { + val element = json.parseToJsonElement(bodyText) + when (element) { + is JsonArray -> { + element.forEach { item -> + try { + val message = json.decodeFromString(item.toString()) + _onMessage(message) + } catch (e: Exception) { + Log.e(TAG, "$serviceidString 解析JSON-RPC批量消息失败: ${e.message}", e) + _onError(e) + } + } + } + else -> { + try { + val message = json.decodeFromString(element.toString()) + _onMessage(message) + } catch (e: Exception) { + Log.e(TAG, "$serviceidString 解析JSON-RPC消息失败: ${e.message}", e) + _onError(e) + } + } + } + } catch (e: Exception) { + Log.e(TAG, "$serviceidString 解析HTTP响应体失败: ${e.message}", e) + _onError(e) + } + } +} diff --git a/local_plugins/chat_api/android/src/main/kotlin/com/yunqiinnovation/chat_api/MCPClient.kt b/local_plugins/chat_api/android/src/main/kotlin/com/yunqiinnovation/chat_api/MCPClient.kt index 08f05d3f1..b8013e6dd 100644 --- a/local_plugins/chat_api/android/src/main/kotlin/com/yunqiinnovation/chat_api/MCPClient.kt +++ b/local_plugins/chat_api/android/src/main/kotlin/com/yunqiinnovation/chat_api/MCPClient.kt @@ -31,7 +31,7 @@ interface FunctionHandler { * MCP客户端 * 与 iOS 版本 MCPClient 功能对等 */ -class MCPClient(private val context: Context? = null,musicType: String = "AUTO",searchUrl: String = "",token: String = "") : AutoCloseable { +class MCPClient(private val context: Context? = null,musicType: String = "AUTO") : AutoCloseable { companion object { private const val TAG = "MCPClient" @@ -54,7 +54,7 @@ class MCPClient(private val context: Context? = null,musicType: String = "AUTO", init { initializeSystemFunctions() - initializeToolsFunctions(searchUrl,token) + // initializeToolsFunctions(searchUrl,token) Log.d(TAG, "初始化音乐服务: $musicType") when (musicType) { "WY" -> initializeMusiceFunctions() //网易 @@ -82,17 +82,6 @@ class MCPClient(private val context: Context? = null,musicType: String = "AUTO", Log.w(TAG, "Failed to initialize system functions", e) } } - /** - * 初始化系统函数 - */ - private fun initializeToolsFunctions(searchUrl: String = "",token: String = "") { - try { - val handler = ToolsSystemFunctionHandler(context,searchUrl,token) - handler.registerAllFunctions(this) - } catch (e: Exception) { - Log.w(TAG, "Failed to initialize system functions", e) - } - } /** * 初始化音乐函数 @@ -141,10 +130,23 @@ class MCPClient(private val context: Context? = null,musicType: String = "AUTO", val url = serverConfig.optString("url", "") val tools = serverConfig.optString("tools", "") + val transportType = run { + if (!serverConfig.has("type")) { + 1 + } else { + val rawType = serverConfig.opt("type") + when (rawType) { + is Number -> rawType.toInt() + is String -> rawType.toIntOrNull() ?: 1 + else -> 1 + } + } + }.coerceIn(0, 1) if (url.isEmpty()) continue + Log.d(TAG, "MCP服务器配置: id=$serverId, typeRaw=${serverConfig.opt("type")}, type=$transportType, url=$url") Log.d(TAG, "开始连接MCP服务器: $serverId") - val subClient = MCPSubClient(serverId, url, tools, sharedHttpClient) + val subClient = MCPSubClient(serverId, url, tools, transportType, sharedHttpClient) // 使用协程并发连接，但每个服务器都会进行重试 try { @@ -430,4 +432,4 @@ class MCPClient(private val context: Context? = null,musicType: String = "AUTO", return list } -} \ No newline at end of file +} diff --git a/local_plugins/chat_api/android/src/main/kotlin/com/yunqiinnovation/chat_api/MCPSubClient.kt b/local_plugins/chat_api/android/src/main/kotlin/com/yunqiinnovation/chat_api/MCPSubClient.kt index f83f812e6..1b74ae39e 100644 --- a/local_plugins/chat_api/android/src/main/kotlin/com/yunqiinnovation/chat_api/MCPSubClient.kt +++ b/local_plugins/chat_api/android/src/main/kotlin/com/yunqiinnovation/chat_api/MCPSubClient.kt @@ -23,6 +23,7 @@ class MCPSubClient( private val serverId: String, private val serverUrl: String, private val filterTools: String, + private val transportType: Int = 1, private val httpClient: HttpClient? = null ) : AutoCloseable { @@ -44,7 +45,7 @@ class MCPSubClient( private var mcpClient: Client? = null private var isConnected = false private var availableTools = mutableListOf() - private var transport: CustomSseClientTransport? = null + private var transport: AbstractTransport? = null /** * 连接到MCP服务器 @@ -52,11 +53,13 @@ class MCPSubClient( suspend fun connect(): Boolean = connectionMutex.withLock { if (isConnected) return true Log.e(TAG, "[$serverId] 开始连接mcp服务器: $serverUrl") + Log.d(TAG, "[$serverId] 连接参数: type=$transportType, filterTools='${filterTools}', url=$serverUrl") - // 重试连接逻辑 - for (attempt in 0 until maxRetryAttempts) { + val maxAttempts = if (transportType == 0) 1 else maxRetryAttempts + // 重试连接逻辑（HTTP模式仅做一次性探测） + for (attempt in 0 until maxAttempts) { try { - Log.d(TAG, "[$serverId] 连接尝试 ${attempt + 1}/$maxRetryAttempts") + Log.d(TAG, "[$serverId] 连接尝试 ${attempt + 1}/$maxAttempts") // 创建MCP客户端实例 val client = Client( @@ -66,9 +69,9 @@ class MCPSubClient( ) ) - // 根据URL类型选择传输方式 - val newTransport = when { - serverUrl.startsWith("http://") || serverUrl.startsWith("https://") -> { + // 根据传输类型选择传输方式 + val newTransport = when (transportType) { + 1 -> { // SSE传输 - 使用自定义的CustomSseClientTransport val mcpHttpClient = httpClient ?: createMcpHttpClient() CustomSseClientTransport( @@ -84,8 +87,17 @@ class MCPSubClient( } ) } + 0 -> { + // HTTPS(Streamable HTTP)传输 - 单端点POST/JSON响应 + val mcpHttpClient = httpClient ?: createMcpHttpClient() + CustomStreamableHttpClientTransport( + client = mcpHttpClient, + serviceidString = serverId, + urlString = serverUrl + ) + } else -> { - Log.e(TAG, "[$serverId] 不支持的服务器URL格式: $serverUrl") + Log.e(TAG, "[$serverId] 不支持的传输类型: $transportType (url=$serverUrl)") return false } } @@ -94,14 +106,17 @@ class MCPSubClient( // 连接到服务器 - 增加超时时间 try { - Log.d(TAG, "[$serverId] 尝试建立连接 ${transport?.urlString}") + Log.d(TAG, "[$serverId] 尝试建立连接 $serverUrl (type=$transportType)") withTimeout(30000) { // 30秒超时 client.connect(newTransport) } Log.d(TAG, "[$serverId] 连接建立成功") } catch (e: TimeoutCancellationException) { - Log.w(TAG, "[$serverId] 连接超时 (尝试 ${attempt + 1}/$maxRetryAttempts)") - if (attempt < maxRetryAttempts - 1) { + Log.w(TAG, "[$serverId] 连接超时 (尝试 ${attempt + 1}/$maxAttempts)") + if (transportType == 0) { + return false + } + if (attempt < maxAttempts - 1) { delay(currentReconnectDelay) currentReconnectDelay = (currentReconnectDelay * 2).coerceAtMost(maxReconnectDelay) continue // 继续下一次重试 @@ -110,8 +125,11 @@ class MCPSubClient( return false } } catch (e: Exception) { - Log.w(TAG, "[$serverId] 连接异常 (尝试 ${attempt + 1}/$maxRetryAttempts): ${e.message}") - if (attempt < maxRetryAttempts - 1) { + Log.w(TAG, "[$serverId] 连接异常 (尝试 ${attempt + 1}/$maxAttempts): ${e.message}") + if (transportType == 0) { + return false + } + if (attempt < maxAttempts - 1) { delay(currentReconnectDelay) currentReconnectDelay = (currentReconnectDelay * 2).coerceAtMost(maxReconnectDelay) continue // 继续下一次重试 @@ -123,21 +141,37 @@ class MCPSubClient( // 获取可用工具列表 try { + Log.d(TAG, "[$serverId] 开始拉取工具列表: type=$transportType, url=$serverUrl") val toolsResult = client.listTools() if (toolsResult != null) { availableTools.clear() val allToolNames = toolsResult.tools.map { it.name } - Log.d(TAG, "$serverId:所有工具名称列表: $allToolNames") + Log.d(TAG, "[$serverId] 工具列表拉取成功: total=${toolsResult.tools.size}, names=$allToolNames") val filtered = toolsResult.tools.filter { tool -> filterTools.isEmpty() || filterTools.contains(tool.name) } val filteredToolNames = filtered.map { it.name } - Log.d(TAG, "$serverId: 过滤后的工具: $filteredToolNames") + Log.d(TAG, "[$serverId] 工具列表过滤完成: filtered=${filtered.size}, names=$filteredToolNames, filterTools='${filterTools}'") + if (toolsResult.tools.isNotEmpty() && filtered.isEmpty() && filterTools.isNotEmpty()) { + Log.w(TAG, "[$serverId] 过滤后工具为空，请检查 tools 配置是否包含正确的工具名") + } availableTools.addAll(filtered) + } else { + Log.w(TAG, "[$serverId] 工具列表拉取结果为空: toolsResult=null") + if (transportType == 0) { + runCatching { newTransport.close() } + runCatching { client.close() } + return false + } } } catch (e: Exception) { - Log.w(TAG, "[$serverId] 获取工具列表失败: ${e.message}") - // 即使获取工具失败，连接也可能是成功的 + Log.w(TAG, "[$serverId] 获取工具列表失败: ${e.message}, type=$transportType, url=$serverUrl", e) + if (transportType == 0) { + runCatching { newTransport.close() } + runCatching { client.close() } + return false + } + // SSE模式下即使获取工具失败，连接也可能是成功的 } mcpClient = client @@ -150,7 +184,10 @@ class MCPSubClient( } catch (e: Exception) { Log.w(TAG, "[$serverId] 连接尝试 ${attempt + 1} 失败: ${e.message}") - if (attempt < maxRetryAttempts - 1) { + if (transportType == 0) { + return false + } + if (attempt < maxAttempts - 1) { delay(currentReconnectDelay) currentReconnectDelay = (currentReconnectDelay * 2).coerceAtMost(maxReconnectDelay) } @@ -368,7 +405,10 @@ class MCPSubClient( val toolsResult = client.listTools() if (toolsResult != null) { availableTools.clear() - availableTools.addAll(toolsResult.tools) + val filtered = toolsResult.tools.filter { tool -> + filterTools.isEmpty() || filterTools.contains(tool.name) + } + availableTools.addAll(filtered) true } else { false @@ -383,13 +423,25 @@ class MCPSubClient( * 检查连接状态并自动重连 */ suspend fun checkConnection(): Boolean { + if (transportType == 0) { + val transportActive = when (val currentTransport = transport) { + is CustomSseClientTransport -> currentTransport.isConnectionActive() + is CustomStreamableHttpClientTransport -> currentTransport.isConnectionActive() + else -> false + } + return isConnected && transportActive + } if (!isConnected) { Log.d(TAG, "[$serverId] 当前未连接，尝试重新连接...") return connect() } // 检查传输层连接状态 - val transportActive = transport?.isConnectionActive() ?: false + val transportActive = when (val currentTransport = transport) { + is CustomSseClientTransport -> currentTransport.isConnectionActive() + is CustomStreamableHttpClientTransport -> currentTransport.isConnectionActive() + else -> false + } if (!transportActive) { Log.w(TAG, "[$serverId] 传输层连接已断开") isConnected = false @@ -427,7 +479,12 @@ class MCPSubClient( * 获取连接状态 */ fun getConnectionStatus(): Boolean { - return isConnected && (transport?.isConnectionActive() ?: false) + val transportActive = when (val currentTransport = transport) { + is CustomSseClientTransport -> currentTransport.isConnectionActive() + is CustomStreamableHttpClientTransport -> currentTransport.isConnectionActive() + else -> false + } + return isConnected && transportActive } /** * 停止心跳检测 @@ -468,6 +525,9 @@ class MCPSubClient( isConnected = false stopHeartbeat() + if (transportType == 0) { + return + } if (retryCount < maxRetryAttempts) { retryCount++ currentReconnectDelay = minOf(currentReconnectDelay * 2, maxReconnectDelay) @@ -504,4 +564,4 @@ class MCPSubClient( } scope.cancel() } -} \ No newline at end of file +} diff --git a/local_plugins/chat_api/android/src/main/kotlin/com/yunqiinnovation/chat_api/ToolsSystemFunctionHandler.kt b/local_plugins/chat_api/android/src/main/kotlin/com/yunqiinnovation/chat_api/ToolsSystemFunctionHandler.kt deleted file mode 100644 index 462ae5c63..000000000 --- a/local_plugins/chat_api/android/src/main/kotlin/com/yunqiinnovation/chat_api/ToolsSystemFunctionHandler.kt +++ /dev/null @@ -1,213 +0,0 @@ -package com.yunqiinnovation.chat_api - -import android.content.Context -import android.content.Intent -import android.util.Log -import com.yunqiinnovation.music_service.MusicSingleton -import org.json.JSONObject -import kotlin.coroutines.resume -import kotlin.coroutines.suspendCoroutine -import androidx.core.content.ContextCompat -import kotlinx.coroutines.launch -import kotlinx.coroutines.CoroutineScope -import kotlinx.coroutines.Dispatchers -import kotlinx.coroutines.SupervisorJob -import okhttp3.Call -import okhttp3.Callback -import okhttp3.MediaType.Companion.toMediaType -import okhttp3.OkHttpClient -import okhttp3.Request -import okhttp3.RequestBody.Companion.toRequestBody -import okhttp3.Response -import org.json.JSONArray -import java.io.IOException -import java.util.concurrent.TimeUnit - -/** - * QQ音乐系统功能处理器 - * 负责注册QQ音乐相关的MCP函数 - */ -class ToolsSystemFunctionHandler(private val context: Context? = null,private val searchUrl:String,private val token:String) { - companion object { - const val TAG = "ToolsSystem" - } - - fun registerAllFunctions(client: MCPClient) { - try { - client.registerLocalFunction( - "web_search", - "通过搜索网络搜索回复用户相关问题，例如今日新闻、百科、医疗、股票、汇率等等", - mapOf( - "type" to "object", - "properties" to mapOf( - "keyword" to mapOf( - "type" to "string", - "description" to "搜索关键词，例如今日新闻、今日人民币汇率等等" - ), - "count" to mapOf( - "type" to "integer", - "description" to "返回的搜索结果数量，默认值为5", - "default" to 5 - ) - ), - "required" to listOf("keyword") - ), - SearchSearchHandler(context,searchUrl,token) - ) - Log.d(TAG, "音乐所有功能注册完成") - } catch (e: Exception) { - Log.e(TAG, "注册音乐功能失败: ${e.message}", e) - } - } -} - -// 搜索系统函数处理类 -private class SearchSearchHandler(private val context: Context?,private val searchUrl:String,private val token:String) : FunctionHandler { - companion object { - const val TAG = "SearchSearchHandler" - } - private val coroutineScope = CoroutineScope(Dispatchers.IO + SupervisorJob()) - /** - * 搜索网络内容 - */ - fun webSearch( - keyword: String, - count: Int = 0, - callback: (Result>>) -> Unit - ) { - // 使用协程异步处理，避免阻塞主线程 - coroutineScope.launch { - try { - // 配置带超时的 OkHttpClient - val client = OkHttpClient.Builder() - .connectTimeout(10, TimeUnit.SECONDS) - .writeTimeout(10, TimeUnit.SECONDS) - .readTimeout(30, TimeUnit.SECONDS) - .build() - - // 构建搜索请求的 JSON 数据 - val jsonObject = JSONObject().apply { - put("keywords", keyword) - put("count", count) - } - - val json = jsonObject.toString() - val mediaType = "application/json".toMediaType() - val body = json.toRequestBody(mediaType) - - // 构建请求 - val request = Request.Builder() - .url("$searchUrl/api/home/aitools_bochasearch") - .addHeader("Authorization", token) - .addHeader("Content-Type", "application/json") - .post(body) - .build() - - Log.d("SearchSearchHandler", "发送搜索请求: $json") - - // 使用异步调用 - client.newCall(request).enqueue(object : Callback { - override fun onFailure(call: Call, e: IOException) { - Log.e(TAG, "搜索歌曲请求失败: ${e.message}", e) - callback(Result.failure(e)) - } - - override fun onResponse(call: Call, response: Response) { - response.use { - try { - if (response.isSuccessful) { - val responseBody = response.body?.string() - Log.d(TAG, "搜索响应: $responseBody") - - if (responseBody != null) { - val jsonResponse = JSONObject(responseBody) - val results = parseResultsFromResponse(jsonResponse) - callback(Result.success(results)) - } else { - callback(Result.failure(Exception("响应体为空"))) - } - } else { - val errorMsg = "搜索失败，HTTP状态码: ${response.code}" - Log.w(TAG, errorMsg) - callback(Result.failure(Exception(errorMsg))) - } - } catch (e: Exception) { - Log.e(TAG, "解析搜索响应时发生异常: ${e.message}", e) - callback(Result.failure(e)) - } - } - } - }) - - } catch (e: Exception) { - Log.e(TAG, "搜索歌曲时发生异常: ${e.message}", e) - callback(Result.failure(e)) - } - } - } - - /** - * 解析服务器响应中的歌曲列表 - */ - private fun parseResultsFromResponse(jsonResponse: JSONObject): List> { - val results = mutableListOf>() - try { - val data = jsonResponse.optJSONObject("data") - val resultsArray = data?.optJSONArray("results") - - resultsArray?.let { array -> - for (i in 0 until array.length()) { - val songObj = array.getJSONObject(i) - results.add(mapOf( - "role" to songObj.optLong("role"), - "type" to songObj.optString("type"), - "content_type" to songObj.optString("content_type"), - "content" to songObj.optString("content"), - )) - } - } - } catch (e: Exception) { - Log.e(TAG, "Failed to parse songs from response", e) - } - return results - } - - - override suspend fun handle(arguments: Map): String { - return try { - val keyword = arguments["keyword"] as? String - ?: return "{\"success\": false, \"message\": \"缺少搜索关键词\"}" - val count = arguments["count"] as? Int ?: 5 // 默认搜索歌曲 - // 使用 suspendCoroutine 将回调转换为协程 - suspendCoroutine { continuation -> - webSearch(keyword, count) { searchResult -> - val result = if (searchResult.isSuccess) { - val datas = searchResult.getOrNull() ?: emptyList() - - val songsJson = datas.joinToString(",") { song -> - val resultMap = song as Map - // 添加调试日志 - Log.d("SearchSearchHandler", "原始数据: $resultMap") - // 修正字段名映射 - val role = resultMap["role"] ?: 0 - val type = resultMap["type"] ?: "unknown" - val content_type = resultMap["content_type"] ?: "unknown" - val content = resultMap["content"] ?: "unknown" - "{\"role\": $role, \"type\": \"$type\", \"content_type\": \"$content_type\", \"content\": \"$content\"}" - } - "{\"success\": true, \"iscard\": true, \"broadcast\": true, \"message\": \"搜索成功,共找到数据:${datas.size}\", \"data\": [$songsJson]}" - } else { - val error = searchResult.exceptionOrNull() - "{\"success\": false, \"message\": \"搜索失败: ${error?.message}\"}" - } - // 恢复协程执行 - continuation.resume(result) - } - } - } catch (e: Exception) { - Log.e("QQMusicSearchHandler", "QQ音乐搜索失败", e) - "{\"success\": false, \"message\": \"搜索异常: ${e.message}\"}" - } - } -} - diff --git a/local_plugins/chat_api/ios/chat_api/Sources/chat_api/ChatApiService.swift b/local_plugins/chat_api/ios/chat_api/Sources/chat_api/ChatApiService.swift index dece8d55f..db7ae2e49 100644 --- a/local_plugins/chat_api/ios/chat_api/Sources/chat_api/ChatApiService.swift +++ b/local_plugins/chat_api/ios/chat_api/Sources/chat_api/ChatApiService.swift @@ -32,7 +32,7 @@ private class ToolCallInfo { var arguments: String = "" func isValid() -> Bool { - return !id.isEmpty && !name.isEmpty + return !name.isEmpty } } @@ -461,7 +461,7 @@ private func processToolCalls(_ sessionId:String,vipexptime: Int) async -> Bool // 输出所有收集到的工具调用信息（用于调试） print("[Session: \(sessionId)] 收集到 \(toolCalls.count) 个工具调用") for (index, toolCall) in toolCalls { - print("[Session: \(sessionId)] 工具调用[\(index)]: name=\(toolCall.name), args.length=\(toolCall.arguments.count), args=\(toolCall.arguments.prefix(100))...") + print("[Session: \(sessionId)] 工具调用[\(index)]: id=\(toolCall.id), name=\(toolCall.name), args.length=\(toolCall.arguments.count), args=\(toolCall.arguments.prefix(100))...") } // 只处理有效的工具调用 @@ -473,6 +473,13 @@ private func processToolCalls(_ sessionId:String,vipexptime: Int) async -> Bool // 处理第一个有效的工具调用 let firstToolCall = validToolCalls.first! + if firstToolCall.id.isEmpty { + if let (index, _) = toolCalls.first(where: { $0.value === firstToolCall }) { + firstToolCall.id = "toolcall-\(index)" + } else { + firstToolCall.id = "toolcall-0" + } + } print("[Session: \(sessionId)] 处理工具调用: \(firstToolCall.name)，参数: \(firstToolCall.arguments)") // 创建函数调用字典 diff --git a/local_plugins/chat_api/ios/chat_api/Sources/chat_api/CustomSseClientTransport.swift b/local_plugins/chat_api/ios/chat_api/Sources/chat_api/CustomSseClientTransport.swift index e5da6b689..bb6931d0d 100644 --- a/local_plugins/chat_api/ios/chat_api/Sources/chat_api/CustomSseClientTransport.swift +++ b/local_plugins/chat_api/ios/chat_api/Sources/chat_api/CustomSseClientTransport.swift @@ -6,6 +6,10 @@ import MCP import EventSource #endif +public protocol ConnectionActiveCheckable: Transport { + func isConnectionActive() -> Bool +} + /// 自定义SSE客户端传输层 public actor CustomSseClientTransport: Transport { @@ -477,3 +481,132 @@ public actor CustomSseClientTransport: Transport { eventSource = nil } } + +extension CustomSseClientTransport: ConnectionActiveCheckable {} + +public actor CustomStreamableHttpClientTransport: ConnectionActiveCheckable { + public nonisolated let logger: Logger + + private let urlString: String + private let session: URLSession + + private var isConnected = false + private var isInitialized = false + + private let messageStream: AsyncThrowingStream + private let messageContinuation: AsyncThrowingStream.Continuation + + public init( + urlString: String, + session: URLSession = .shared, + logger: Logger? = nil + ) { + self.urlString = urlString + self.session = session + self.logger = logger ?? Logger( + label: "com.yunqiinnovation.chat_api.CustomStreamableHttpClientTransport", + factory: { _ in SwiftLogNoOpLogHandler() } + ) + + var continuation: AsyncThrowingStream.Continuation! + self.messageStream = AsyncThrowingStream { continuation = $0 } + self.messageContinuation = continuation + } + + public func connect() async throws { + guard !isInitialized else { + throw MCPError.internalError("CustomStreamableHttpClientTransport already started!") + } + guard URL(string: urlString) != nil else { + throw MCPError.internalError("Invalid MCP endpoint URL: \(urlString)") + } + + isInitialized = true + isConnected = true + } + + public func disconnect() async { + guard isInitialized else { + return + } + + isConnected = false + isInitialized = false + messageContinuation.finish() + } + + public func isConnectionActive() -> Bool { + return isConnected && isInitialized + } + + public func send(_ data: Data) async throws { + guard isConnected, isInitialized else { + throw MCPError.internalError("Not connected to MCP server") + } + guard let url = URL(string: urlString) else { + throw MCPError.internalError("Invalid MCP endpoint URL: \(urlString)") + } + + var request = URLRequest(url: url) + request.httpMethod = "POST" + request.setValue("application/json", forHTTPHeaderField: "Content-Type") + request.setValue("application/json, text/event-stream", forHTTPHeaderField: "Accept") + request.httpBody = data + + let (responseData, response) = try await session.data(for: request) + + guard let httpResponse = response as? HTTPURLResponse else { + throw MCPError.internalError("Invalid HTTP response") + } + + let contentType = httpResponse.value(forHTTPHeaderField: "Content-Type") ?? "" + let bodyPreview: String = { + let body = String(data: responseData, encoding: .utf8) ?? "" + if body.count > 500 { + return String(body.prefix(500)) + "..." + } + return body + }() + logger.info("HTTP响应: status=\(httpResponse.statusCode), contentType=\(contentType), bodyLen=\(responseData.count), bodySnippet=\(bodyPreview.replacingOccurrences(of: "\n", with: "\\n"))") + + if httpResponse.statusCode == 202 || httpResponse.statusCode == 204 { + return + } + + guard 200..<300 ~= httpResponse.statusCode else { + let body = String(data: responseData, encoding: .utf8) ?? "" + throw MCPError.internalError("HTTP \(httpResponse.statusCode): \(body)") + } + + if responseData.isEmpty { + return + } + + emitJsonRpcFromBody(responseData) + } + + public func receive() -> AsyncThrowingStream { + return messageStream + } + + private func emitJsonRpcFromBody(_ responseData: Data) { + do { + let jsonObject = try JSONSerialization.jsonObject(with: responseData) + if let array = jsonObject as? [Any] { + for item in array { + if let itemData = try? JSONSerialization.data(withJSONObject: item) { + messageContinuation.yield(itemData) + } + } + return + } + + if let dict = jsonObject as? [String: Any], + let dictData = try? JSONSerialization.data(withJSONObject: dict) { + messageContinuation.yield(dictData) + } + } catch { + logger.error("解析HTTP响应体失败: \(error.localizedDescription)") + } + } +} diff --git a/local_plugins/chat_api/ios/chat_api/Sources/chat_api/MCPClient.swift b/local_plugins/chat_api/ios/chat_api/Sources/chat_api/MCPClient.swift index 5d7146d1a..7df122c15 100644 --- a/local_plugins/chat_api/ios/chat_api/Sources/chat_api/MCPClient.swift +++ b/local_plugins/chat_api/ios/chat_api/Sources/chat_api/MCPClient.swift @@ -33,10 +33,11 @@ public class MCPSubClient { private let serverId: String private let serverUrl: String private let filterTools: String + private let transportType: Int // 使用官方MCP Swift SDK private var mcpClient: Client? - private var transport: CustomSseClientTransport? + private var transport: ConnectionActiveCheckable? private var tools: [Tool] = [] private var toolMaps: [[String: Any]] = [] private var isConnectedFlag = false @@ -61,10 +62,11 @@ public class MCPSubClient { // 连接状态锁 private let connectionLock = NSLock() - public init(serverId: String, serverUrl: String, filterTools: String) { + public init(serverId: String, serverUrl: String, filterTools: String, transportType: Int = 1) { self.serverId = serverId self.serverUrl = serverUrl self.filterTools = filterTools + self.transportType = transportType self.currentReconnectDelay = initialReconnectDelay } @@ -83,18 +85,26 @@ public class MCPSubClient { // 1. 创建MCP客户端 self.mcpClient = Client(name: "deep_voice_\(serverId)", version: "1.0.0") - // 2. 创建自定义SSE传输层，添加断线回调 - let swiftLogger = Logger(label: "mcp.transport.custom_sse.\(serverId)") - let transport = CustomSseClientTransport( - urlString: serverUrl, - reconnectionTime: 1.0, - logger: swiftLogger, - onConnectionLost: { [weak self] in - Task { - await self?.handleConnectionLost() + // 2. 根据类型创建传输层 + let swiftLogger = Logger(label: "mcp.transport.\(serverId)") + let transport: ConnectionActiveCheckable + if transportType == 0 { + transport = CustomStreamableHttpClientTransport( + urlString: serverUrl, + logger: swiftLogger + ) + } else { + transport = CustomSseClientTransport( + urlString: serverUrl, + reconnectionTime: 1.0, + logger: swiftLogger, + onConnectionLost: { [weak self] in + Task { + await self?.handleConnectionLost() + } } - } - ) + ) + } self.transport = transport // 3. 连接到服务器（添加超时） @@ -104,8 +114,9 @@ public class MCPSubClient { // 4. 获取工具列表 let (toolList, _) = try await self.mcpClient!.listTools() - - print("[MCPSubClient] \(serverId):工具列表 \(toolList)") + let toolNames = toolList.map { $0.name } + let toolNamesPreview = toolNames.count > 50 ? Array(toolNames.prefix(50)) : toolNames + print("[MCPSubClient] \(serverId) listTools 成功: total=\(toolList.count), names=\(toolNamesPreview), type=\(transportType), filterTools='\(filterTools)'") // 5. 处理工具列表 processTools(toolList) @@ -113,7 +124,9 @@ public class MCPSubClient { isConnectedFlag = true retryCount = 0 // 重置重试计数 currentReconnectDelay = initialReconnectDelay // 重置延迟 - startKeepAlive() + if transportType != 0 { + startKeepAlive() + } return true @@ -129,7 +142,7 @@ public class MCPSubClient { tools.removeAll() toolMaps.removeAll() - + var acceptedNames: [String] = [] for tool in toolList { if !filterTools.isEmpty && !filterTools.contains(tool.name) { continue @@ -142,11 +155,15 @@ public class MCPSubClient { "parameters": parameters ] tools.append(tool) + acceptedNames.append(tool.name) toolMaps.append([ "type": "function", "function": functionMap ]) } + + let acceptedPreview = acceptedNames.count > 50 ? Array(acceptedNames.prefix(50)) : acceptedNames + print("[MCPSubClient] \(serverId) 工具过滤完成: accepted=\(acceptedNames.count)/\(toolList.count), names=\(acceptedPreview), filterTools='\(filterTools)'") } private func convertInputSchema(_ inputSchema: Any?) -> [String: Any] { @@ -390,13 +407,18 @@ public class MCPSubClient { isConnectedFlag = false keepAliveTask?.cancel() keepAliveTask = nil - startReconnection() + if transportType != 0 { + startReconnection() + } } } } /// 启动重连 private func startReconnection() { + if transportType == 0 { + return + } // 取消之前的重连任务 if reconnectTask != nil { return @@ -438,6 +460,9 @@ public class MCPSubClient { isConnectedFlag = false keepAliveTask?.cancel() keepAliveTask = nil + if transportType == 0 { + return + } // 如果是MCP特定错误，可以进行特殊处理 if let mcpError = error as? MCPError { @@ -582,7 +607,19 @@ public class MCPClient { continue } let filtertools = configDict["tools"] as? String ?? "" - let subClient = MCPSubClient(serverId: serverId, serverUrl: url,filterTools: filtertools) + let transportType: Int = { + if let value = configDict["type"] as? Int { + return value + } + if let value = configDict["type"] as? NSNumber { + return value.intValue + } + if let value = configDict["type"] as? String { + return Int(value) ?? 1 + } + return 1 + }() + let subClient = MCPSubClient(serverId: serverId, serverUrl: url, filterTools: filtertools, transportType: transportType) if await subClient.connect() { subClients[serverId] = subClient