Browse Source

feat(mcp): 支持streamable HTTP传输并移除搜索工具集成

- 在DBMCPServer模型中添加type字段以区分传输类型
- 实现CustomStreamableHttpClientTransport用于HTTP/JSON传输
- 移除ToolsSystemFunctionHandler及相关搜索工具集成
- 优化MCP客户端初始化,根据type选择SSE或HTTP传输
- 修复iOS端工具调用ID生成逻辑,提升兼容性
newdev_shunjiawei
liwei1dao 8 months ago
parent
commit
73ccaf8015
  1. 2
      lib/data/models/appconfig_model.dart
  2. 2
      lib/data/models/appconfig_model.g.dart
  3. 2
      local_plugins/agent_service/android/src/main/kotlin/com/yunqiinnovation/agent_service/AgentService.kt
  4. 8
      local_plugins/chat_api/android/src/main/kotlin/com/yunqiinnovation/chat_api/ChatApiService.kt
  5. 145
      local_plugins/chat_api/android/src/main/kotlin/com/yunqiinnovation/chat_api/CustomSseClientTransport.kt
  6. 32
      local_plugins/chat_api/android/src/main/kotlin/com/yunqiinnovation/chat_api/MCPClient.kt
  7. 104
      local_plugins/chat_api/android/src/main/kotlin/com/yunqiinnovation/chat_api/MCPSubClient.kt
  8. 213
      local_plugins/chat_api/android/src/main/kotlin/com/yunqiinnovation/chat_api/ToolsSystemFunctionHandler.kt
  9. 11
      local_plugins/chat_api/ios/chat_api/Sources/chat_api/ChatApiService.swift
  10. 133
      local_plugins/chat_api/ios/chat_api/Sources/chat_api/CustomSseClientTransport.swift
  11. 75
      local_plugins/chat_api/ios/chat_api/Sources/chat_api/MCPClient.swift

2
lib/data/models/appconfig_model.dart

@ -58,10 +58,12 @@ class DBAgent {
class DBMCPServer { class DBMCPServer {
final String servername; final String servername;
final String url; final String url;
final int type;
final String tools; final String tools;
DBMCPServer({ DBMCPServer({
required this.servername, required this.servername,
required this.url, required this.url,
required this.type,
required this.tools, required this.tools,
}); });

2
lib/data/models/appconfig_model.g.dart

@ -61,6 +61,7 @@ Map<String, dynamic> _$DBAgentToJson(DBAgent instance) => <String, dynamic>{
DBMCPServer _$DBMCPServerFromJson(Map<String, dynamic> json) => DBMCPServer( DBMCPServer _$DBMCPServerFromJson(Map<String, dynamic> json) => DBMCPServer(
servername: json['servername'] as String, servername: json['servername'] as String,
url: json['url'] as String, url: json['url'] as String,
type: json['type'] as int,
tools: json['tools'] as String? ?? '', tools: json['tools'] as String? ?? '',
); );
@ -68,6 +69,7 @@ Map<String, dynamic> _$DBMCPServerToJson(DBMCPServer instance) =>
<String, dynamic>{ <String, dynamic>{
'servername': instance.servername, 'servername': instance.servername,
'url': instance.url, 'url': instance.url,
'type': instance.type,
'tools': instance.tools, 'tools': instance.tools,
}; };

2
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["openaivisionModel"]?.toString() ?: "",
config["mcpServer"]?.toString() ?: "", config["mcpServer"]?.toString() ?: "",
config["musicServiceType"]?.toString() ?: "", config["musicServiceType"]?.toString() ?: "",
serverurl,
usertoken,
) )
Log.e(TAG, "liwei -----------读取服务地址和用户token: $serverurl $usertoken") Log.e(TAG, "liwei -----------读取服务地址和用户token: $serverurl $usertoken")

8
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服务 * 初始化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 this.apiKey = apiKey
if (baseUrl.isNotEmpty()) { if (baseUrl.isNotEmpty()) {
this.baseUrl = baseUrl this.baseUrl = baseUrl
@ -257,7 +257,7 @@ class ChatApiService(private val context: android.content.Context? = null) : Cor
mcpConfigJson = mcpServer mcpConfigJson = mcpServer
// 异步初始化MCP客户端 // 异步初始化MCP客户端
launch { launch {
initializeMcpClient(mcpServer,musicType,serverhUrl,token) initializeMcpClient(mcpServer,musicType)
// initializeMcpClient("{}") // initializeMcpClient("{}")
} }
@ -941,9 +941,9 @@ class ChatApiService(private val context: android.content.Context? = null) : Cor
/** /**
* 初始化MCP客户端 * 初始化MCP客户端
*/ */
fun initializeMcpClient(serverUrl: String,musicType: String = "AUTO",searchUrl: String = "",token: String = ""): Boolean { fun initializeMcpClient(serverUrl: String,musicType: String = "AUTO"): Boolean {
if (_mcpClient == null) { if (_mcpClient == null) {
_mcpClient = MCPClient(context,musicType,searchUrl,token) _mcpClient = MCPClient(context,musicType)
} }
// 直接使用类的CoroutineScope启动协程 // 直接使用类的CoroutineScope启动协程

145
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 io.modelcontextprotocol.kotlin.sdk.shared.AbstractTransport
import kotlinx.coroutines.* import kotlinx.coroutines.*
import kotlinx.serialization.encodeToString import kotlinx.serialization.encodeToString
import kotlinx.serialization.json.Json import kotlinx.serialization.json.*
import kotlinx.serialization.decodeFromString import kotlinx.serialization.decodeFromString
import kotlin.properties.Delegates import kotlin.properties.Delegates
import kotlin.time.Duration import kotlin.time.Duration
@ -416,4 +416,145 @@ class CustomSseClientTransport(
transportScope.cancel() transportScope.cancel()
Log.d(TAG, "$serviceidString CustomSseClientTransport已关闭") Log.d(TAG, "$serviceidString CustomSseClientTransport已关闭")
} }
} }
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<JSONRPCMessage>(item.toString())
_onMessage(message)
} catch (e: Exception) {
Log.e(TAG, "$serviceidString 解析JSON-RPC批量消息失败: ${e.message}", e)
_onError(e)
}
}
}
else -> {
try {
val message = json.decodeFromString<JSONRPCMessage>(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)
}
}
}

32
local_plugins/chat_api/android/src/main/kotlin/com/yunqiinnovation/chat_api/MCPClient.kt

@ -31,7 +31,7 @@ interface FunctionHandler {
* MCP客户端 * MCP客户端
* 与 iOS 版本 MCPClient 功能对等 * 与 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 { companion object {
private const val TAG = "MCPClient" private const val TAG = "MCPClient"
@ -54,7 +54,7 @@ class MCPClient(private val context: Context? = null,musicType: String = "AUTO",
init { init {
initializeSystemFunctions() initializeSystemFunctions()
initializeToolsFunctions(searchUrl,token) // initializeToolsFunctions(searchUrl,token)
Log.d(TAG, "初始化音乐服务: $musicType") Log.d(TAG, "初始化音乐服务: $musicType")
when (musicType) { when (musicType) {
"WY" -> initializeMusiceFunctions() //网易 "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) 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 url = serverConfig.optString("url", "")
val tools = serverConfig.optString("tools", "") 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 if (url.isEmpty()) continue
Log.d(TAG, "MCP服务器配置: id=$serverId, typeRaw=${serverConfig.opt("type")}, type=$transportType, url=$url")
Log.d(TAG, "开始连接MCP服务器: $serverId") Log.d(TAG, "开始连接MCP服务器: $serverId")
val subClient = MCPSubClient(serverId, url, tools, sharedHttpClient) val subClient = MCPSubClient(serverId, url, tools, transportType, sharedHttpClient)
// 使用协程并发连接,但每个服务器都会进行重试 // 使用协程并发连接,但每个服务器都会进行重试
try { try {
@ -430,4 +432,4 @@ class MCPClient(private val context: Context? = null,musicType: String = "AUTO",
return list return list
} }
} }

104
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 serverId: String,
private val serverUrl: String, private val serverUrl: String,
private val filterTools: String, private val filterTools: String,
private val transportType: Int = 1,
private val httpClient: HttpClient? = null private val httpClient: HttpClient? = null
) : AutoCloseable { ) : AutoCloseable {
@ -44,7 +45,7 @@ class MCPSubClient(
private var mcpClient: Client? = null private var mcpClient: Client? = null
private var isConnected = false private var isConnected = false
private var availableTools = mutableListOf<Tool>() private var availableTools = mutableListOf<Tool>()
private var transport: CustomSseClientTransport? = null private var transport: AbstractTransport? = null
/** /**
* 连接到MCP服务器 * 连接到MCP服务器
@ -52,11 +53,13 @@ class MCPSubClient(
suspend fun connect(): Boolean = connectionMutex.withLock { suspend fun connect(): Boolean = connectionMutex.withLock {
if (isConnected) return true if (isConnected) return true
Log.e(TAG, "[$serverId] 开始连接mcp服务器: $serverUrl") Log.e(TAG, "[$serverId] 开始连接mcp服务器: $serverUrl")
Log.d(TAG, "[$serverId] 连接参数: type=$transportType, filterTools='${filterTools}', url=$serverUrl")
// 重试连接逻辑 val maxAttempts = if (transportType == 0) 1 else maxRetryAttempts
for (attempt in 0 until maxRetryAttempts) { // 重试连接逻辑(HTTP模式仅做一次性探测)
for (attempt in 0 until maxAttempts) {
try { try {
Log.d(TAG, "[$serverId] 连接尝试 ${attempt + 1}/$maxRetryAttempts") Log.d(TAG, "[$serverId] 连接尝试 ${attempt + 1}/$maxAttempts")
// 创建MCP客户端实例 // 创建MCP客户端实例
val client = Client( val client = Client(
@ -66,9 +69,9 @@ class MCPSubClient(
) )
) )
// 根据URL类型选择传输方式 // 根据传输类型选择传输方式
val newTransport = when { val newTransport = when (transportType) {
serverUrl.startsWith("http://") || serverUrl.startsWith("https://") -> { 1 -> {
// SSE传输 - 使用自定义的CustomSseClientTransport // SSE传输 - 使用自定义的CustomSseClientTransport
val mcpHttpClient = httpClient ?: createMcpHttpClient() val mcpHttpClient = httpClient ?: createMcpHttpClient()
CustomSseClientTransport( 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 -> { else -> {
Log.e(TAG, "[$serverId] 不支持的服务器URL格式: $serverUrl") Log.e(TAG, "[$serverId] 不支持的传输类型: $transportType (url=$serverUrl)")
return false return false
} }
} }
@ -94,14 +106,17 @@ class MCPSubClient(
// 连接到服务器 - 增加超时时间 // 连接到服务器 - 增加超时时间
try { try {
Log.d(TAG, "[$serverId] 尝试建立连接 ${transport?.urlString}") Log.d(TAG, "[$serverId] 尝试建立连接 $serverUrl (type=$transportType)")
withTimeout(30000) { // 30秒超时 withTimeout(30000) { // 30秒超时
client.connect(newTransport) client.connect(newTransport)
} }
Log.d(TAG, "[$serverId] 连接建立成功") Log.d(TAG, "[$serverId] 连接建立成功")
} catch (e: TimeoutCancellationException) { } catch (e: TimeoutCancellationException) {
Log.w(TAG, "[$serverId] 连接超时 (尝试 ${attempt + 1}/$maxRetryAttempts)") Log.w(TAG, "[$serverId] 连接超时 (尝试 ${attempt + 1}/$maxAttempts)")
if (attempt < maxRetryAttempts - 1) { if (transportType == 0) {
return false
}
if (attempt < maxAttempts - 1) {
delay(currentReconnectDelay) delay(currentReconnectDelay)
currentReconnectDelay = (currentReconnectDelay * 2).coerceAtMost(maxReconnectDelay) currentReconnectDelay = (currentReconnectDelay * 2).coerceAtMost(maxReconnectDelay)
continue // 继续下一次重试 continue // 继续下一次重试
@ -110,8 +125,11 @@ class MCPSubClient(
return false return false
} }
} catch (e: Exception) { } catch (e: Exception) {
Log.w(TAG, "[$serverId] 连接异常 (尝试 ${attempt + 1}/$maxRetryAttempts): ${e.message}") Log.w(TAG, "[$serverId] 连接异常 (尝试 ${attempt + 1}/$maxAttempts): ${e.message}")
if (attempt < maxRetryAttempts - 1) { if (transportType == 0) {
return false
}
if (attempt < maxAttempts - 1) {
delay(currentReconnectDelay) delay(currentReconnectDelay)
currentReconnectDelay = (currentReconnectDelay * 2).coerceAtMost(maxReconnectDelay) currentReconnectDelay = (currentReconnectDelay * 2).coerceAtMost(maxReconnectDelay)
continue // 继续下一次重试 continue // 继续下一次重试
@ -123,21 +141,37 @@ class MCPSubClient(
// 获取可用工具列表 // 获取可用工具列表
try { try {
Log.d(TAG, "[$serverId] 开始拉取工具列表: type=$transportType, url=$serverUrl")
val toolsResult = client.listTools() val toolsResult = client.listTools()
if (toolsResult != null) { if (toolsResult != null) {
availableTools.clear() availableTools.clear()
val allToolNames = toolsResult.tools.map { it.name } 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 -> val filtered = toolsResult.tools.filter { tool ->
filterTools.isEmpty() || filterTools.contains(tool.name) filterTools.isEmpty() || filterTools.contains(tool.name)
} }
val filteredToolNames = filtered.map { it.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) availableTools.addAll(filtered)
} else {
Log.w(TAG, "[$serverId] 工具列表拉取结果为空: toolsResult=null")
if (transportType == 0) {
runCatching { newTransport.close() }
runCatching { client.close() }
return false
}
} }
} catch (e: Exception) { } 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 mcpClient = client
@ -150,7 +184,10 @@ class MCPSubClient(
} catch (e: Exception) { } catch (e: Exception) {
Log.w(TAG, "[$serverId] 连接尝试 ${attempt + 1} 失败: ${e.message}") Log.w(TAG, "[$serverId] 连接尝试 ${attempt + 1} 失败: ${e.message}")
if (attempt < maxRetryAttempts - 1) { if (transportType == 0) {
return false
}
if (attempt < maxAttempts - 1) {
delay(currentReconnectDelay) delay(currentReconnectDelay)
currentReconnectDelay = (currentReconnectDelay * 2).coerceAtMost(maxReconnectDelay) currentReconnectDelay = (currentReconnectDelay * 2).coerceAtMost(maxReconnectDelay)
} }
@ -368,7 +405,10 @@ class MCPSubClient(
val toolsResult = client.listTools() val toolsResult = client.listTools()
if (toolsResult != null) { if (toolsResult != null) {
availableTools.clear() availableTools.clear()
availableTools.addAll(toolsResult.tools) val filtered = toolsResult.tools.filter { tool ->
filterTools.isEmpty() || filterTools.contains(tool.name)
}
availableTools.addAll(filtered)
true true
} else { } else {
false false
@ -383,13 +423,25 @@ class MCPSubClient(
* 检查连接状态并自动重连 * 检查连接状态并自动重连
*/ */
suspend fun checkConnection(): Boolean { 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) { if (!isConnected) {
Log.d(TAG, "[$serverId] 当前未连接,尝试重新连接...") Log.d(TAG, "[$serverId] 当前未连接,尝试重新连接...")
return connect() 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) { if (!transportActive) {
Log.w(TAG, "[$serverId] 传输层连接已断开") Log.w(TAG, "[$serverId] 传输层连接已断开")
isConnected = false isConnected = false
@ -427,7 +479,12 @@ class MCPSubClient(
* 获取连接状态 * 获取连接状态
*/ */
fun getConnectionStatus(): Boolean { 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 isConnected = false
stopHeartbeat() stopHeartbeat()
if (transportType == 0) {
return
}
if (retryCount < maxRetryAttempts) { if (retryCount < maxRetryAttempts) {
retryCount++ retryCount++
currentReconnectDelay = minOf(currentReconnectDelay * 2, maxReconnectDelay) currentReconnectDelay = minOf(currentReconnectDelay * 2, maxReconnectDelay)
@ -504,4 +564,4 @@ class MCPSubClient(
} }
scope.cancel() scope.cancel()
} }
} }

213
local_plugins/chat_api/android/src/main/kotlin/com/yunqiinnovation/chat_api/ToolsSystemFunctionHandler.kt

@ -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<List<Map<String, Any?>>>) -> 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<Map<String, Any?>> {
val results = mutableListOf<Map<String, Any?>>()
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, Any>): 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<String, Any?>
// 添加调试日志
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}\"}"
}
}
}

11
local_plugins/chat_api/ios/chat_api/Sources/chat_api/ChatApiService.swift

@ -32,7 +32,7 @@ private class ToolCallInfo {
var arguments: String = "" var arguments: String = ""
func isValid() -> Bool { 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) 个工具调用") print("[Session: \(sessionId)] 收集到 \(toolCalls.count) 个工具调用")
for (index, toolCall) in toolCalls { 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! 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)") print("[Session: \(sessionId)] 处理工具调用: \(firstToolCall.name),参数: \(firstToolCall.arguments)")
// 创建函数调用字典 // 创建函数调用字典

133
local_plugins/chat_api/ios/chat_api/Sources/chat_api/CustomSseClientTransport.swift

@ -6,6 +6,10 @@ import MCP
import EventSource import EventSource
#endif #endif
public protocol ConnectionActiveCheckable: Transport {
func isConnectionActive() -> Bool
}
/// 自定义SSE客户端传输层 /// 自定义SSE客户端传输层
public actor CustomSseClientTransport: Transport { public actor CustomSseClientTransport: Transport {
@ -477,3 +481,132 @@ public actor CustomSseClientTransport: Transport {
eventSource = nil 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<Data, Swift.Error>
private let messageContinuation: AsyncThrowingStream<Data, Swift.Error>.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<Data, Swift.Error>.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<Data, Swift.Error> {
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)")
}
}
}

75
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 serverId: String
private let serverUrl: String private let serverUrl: String
private let filterTools: String private let filterTools: String
private let transportType: Int
// 使用官方MCP Swift SDK // 使用官方MCP Swift SDK
private var mcpClient: Client? private var mcpClient: Client?
private var transport: CustomSseClientTransport? private var transport: ConnectionActiveCheckable?
private var tools: [Tool] = [] private var tools: [Tool] = []
private var toolMaps: [[String: Any]] = [] private var toolMaps: [[String: Any]] = []
private var isConnectedFlag = false private var isConnectedFlag = false
@ -61,10 +62,11 @@ public class MCPSubClient {
// 连接状态锁 // 连接状态锁
private let connectionLock = NSLock() 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.serverId = serverId
self.serverUrl = serverUrl self.serverUrl = serverUrl
self.filterTools = filterTools self.filterTools = filterTools
self.transportType = transportType
self.currentReconnectDelay = initialReconnectDelay self.currentReconnectDelay = initialReconnectDelay
} }
@ -83,18 +85,26 @@ public class MCPSubClient {
// 1. 创建MCP客户端 // 1. 创建MCP客户端
self.mcpClient = Client(name: "deep_voice_\(serverId)", version: "1.0.0") self.mcpClient = Client(name: "deep_voice_\(serverId)", version: "1.0.0")
// 2. 创建自定义SSE传输层,添加断线回调 // 2. 根据类型创建传输层
let swiftLogger = Logger(label: "mcp.transport.custom_sse.\(serverId)") let swiftLogger = Logger(label: "mcp.transport.\(serverId)")
let transport = CustomSseClientTransport( let transport: ConnectionActiveCheckable
urlString: serverUrl, if transportType == 0 {
reconnectionTime: 1.0, transport = CustomStreamableHttpClientTransport(
logger: swiftLogger, urlString: serverUrl,
onConnectionLost: { [weak self] in logger: swiftLogger
Task { )
await self?.handleConnectionLost() } else {
transport = CustomSseClientTransport(
urlString: serverUrl,
reconnectionTime: 1.0,
logger: swiftLogger,
onConnectionLost: { [weak self] in
Task {
await self?.handleConnectionLost()
}
} }
} )
) }
self.transport = transport self.transport = transport
// 3. 连接到服务器(添加超时) // 3. 连接到服务器(添加超时)
@ -104,8 +114,9 @@ public class MCPSubClient {
// 4. 获取工具列表 // 4. 获取工具列表
let (toolList, _) = try await self.mcpClient!.listTools() let (toolList, _) = try await self.mcpClient!.listTools()
let toolNames = toolList.map { $0.name }
print("[MCPSubClient] \(serverId):工具列表 \(toolList)") let toolNamesPreview = toolNames.count > 50 ? Array(toolNames.prefix(50)) : toolNames
print("[MCPSubClient] \(serverId) listTools 成功: total=\(toolList.count), names=\(toolNamesPreview), type=\(transportType), filterTools='\(filterTools)'")
// 5. 处理工具列表 // 5. 处理工具列表
processTools(toolList) processTools(toolList)
@ -113,7 +124,9 @@ public class MCPSubClient {
isConnectedFlag = true isConnectedFlag = true
retryCount = 0 // 重置重试计数 retryCount = 0 // 重置重试计数
currentReconnectDelay = initialReconnectDelay // 重置延迟 currentReconnectDelay = initialReconnectDelay // 重置延迟
startKeepAlive() if transportType != 0 {
startKeepAlive()
}
return true return true
@ -129,7 +142,7 @@ public class MCPSubClient {
tools.removeAll() tools.removeAll()
toolMaps.removeAll() toolMaps.removeAll()
var acceptedNames: [String] = []
for tool in toolList { for tool in toolList {
if !filterTools.isEmpty && !filterTools.contains(tool.name) { if !filterTools.isEmpty && !filterTools.contains(tool.name) {
continue continue
@ -142,11 +155,15 @@ public class MCPSubClient {
"parameters": parameters "parameters": parameters
] ]
tools.append(tool) tools.append(tool)
acceptedNames.append(tool.name)
toolMaps.append([ toolMaps.append([
"type": "function", "type": "function",
"function": functionMap "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] { private func convertInputSchema(_ inputSchema: Any?) -> [String: Any] {
@ -390,13 +407,18 @@ public class MCPSubClient {
isConnectedFlag = false isConnectedFlag = false
keepAliveTask?.cancel() keepAliveTask?.cancel()
keepAliveTask = nil keepAliveTask = nil
startReconnection() if transportType != 0 {
startReconnection()
}
} }
} }
} }
/// 启动重连 /// 启动重连
private func startReconnection() { private func startReconnection() {
if transportType == 0 {
return
}
// 取消之前的重连任务 // 取消之前的重连任务
if reconnectTask != nil { if reconnectTask != nil {
return return
@ -438,6 +460,9 @@ public class MCPSubClient {
isConnectedFlag = false isConnectedFlag = false
keepAliveTask?.cancel() keepAliveTask?.cancel()
keepAliveTask = nil keepAliveTask = nil
if transportType == 0 {
return
}
// 如果是MCP特定错误,可以进行特殊处理 // 如果是MCP特定错误,可以进行特殊处理
if let mcpError = error as? MCPError { if let mcpError = error as? MCPError {
@ -582,7 +607,19 @@ public class MCPClient {
continue continue
} }
let filtertools = configDict["tools"] as? String ?? "" 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() { if await subClient.connect() {
subClients[serverId] = subClient subClients[serverId] = subClient

Loading…
Cancel
Save