Browse Source

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

- 在DBMCPServer模型中添加type字段以区分传输类型
- 实现CustomStreamableHttpClientTransport用于HTTP/JSON传输
- 移除ToolsSystemFunctionHandler及相关搜索工具集成
- 优化MCP客户端初始化,根据type选择SSE或HTTP传输
- 修复iOS端工具调用ID生成逻辑,提升兼容性
weicu
liwei1dao 8 months ago
parent
commit
61197c4aa9
  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 {
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,
});

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(
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<String, dynamic> _$DBMCPServerToJson(DBMCPServer instance) =>
<String, dynamic>{
'servername': instance.servername,
'url': instance.url,
'type': instance.type,
'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["mcpServer"]?.toString() ?: "",
config["musicServiceType"]?.toString() ?: "",
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服务
*/
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启动协程

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 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已关闭")
}
}
}
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客户端
* 与 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
}
}
}

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 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<Tool>()
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()
}
}
}

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 = ""
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)")
// 创建函数调用字典

133
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<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 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

Loading…
Cancel
Save