Fix Streamable handling to support JSON-only responses (#309)
quick fix #308
This commit is contained in:
parent
bc28e6c03f
commit
bfaf54e726
2 changed files with 80 additions and 3 deletions
|
|
@ -235,10 +235,22 @@ public class StreamableHttpClientTransport(
|
|||
}
|
||||
logger.debug { "Client SSE session started successfully." }
|
||||
} catch (e: SSEClientException) {
|
||||
if (e.response?.status == HttpStatusCode.MethodNotAllowed) {
|
||||
val responseStatus = e.response?.status
|
||||
val responseContentType = e.response?.contentType()
|
||||
|
||||
// 405 means server doesn't support SSE at GET endpoint - this is expected and valid
|
||||
if (responseStatus == HttpStatusCode.MethodNotAllowed) {
|
||||
logger.info { "Server returned 405 for GET/SSE, stream disabled." }
|
||||
return
|
||||
}
|
||||
|
||||
// If server returns application/json, it means it doesn't support SSE for this session
|
||||
// This is valid per spec - server can choose to only use JSON responses
|
||||
if (responseContentType?.match(ContentType.Application.Json) == true) {
|
||||
logger.info { "Server returned application/json for GET/SSE, using JSON-only mode." }
|
||||
return
|
||||
}
|
||||
|
||||
_onError(e)
|
||||
throw e
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,6 +1,8 @@
|
|||
package io.modelcontextprotocol.kotlin.sdk.client
|
||||
|
||||
import io.kotest.matchers.collections.shouldContain
|
||||
import io.ktor.http.ContentType
|
||||
import io.ktor.http.HttpMethod
|
||||
import io.ktor.http.HttpStatusCode
|
||||
import io.ktor.sse.ServerSentEvent
|
||||
import io.modelcontextprotocol.kotlin.sdk.ClientCapabilities
|
||||
|
|
@ -13,9 +15,11 @@ import kotlinx.serialization.json.buildJsonObject
|
|||
import kotlinx.serialization.json.put
|
||||
import kotlinx.serialization.json.putJsonObject
|
||||
import org.junit.jupiter.api.TestInstance
|
||||
import java.util.UUID
|
||||
import kotlin.test.Test
|
||||
import kotlin.time.Duration.Companion.milliseconds
|
||||
import kotlin.time.Duration.Companion.seconds
|
||||
import kotlin.uuid.ExperimentalUuidApi
|
||||
import kotlin.uuid.Uuid
|
||||
|
||||
/**
|
||||
* Integration tests for the `StreamableHttpClientTransport` implementation
|
||||
|
|
@ -23,6 +27,7 @@ import kotlin.time.Duration.Companion.milliseconds
|
|||
* to simulate Streaming HTTP with server-sent events (SSE).
|
||||
* @author Konstantin Pavlov
|
||||
*/
|
||||
@OptIn(ExperimentalUuidApi::class)
|
||||
@TestInstance(TestInstance.Lifecycle.PER_CLASS)
|
||||
@Suppress("LongMethod")
|
||||
internal class StreamableHttpClientTest : AbstractStreamableHttpClientTest() {
|
||||
|
|
@ -39,7 +44,7 @@ internal class StreamableHttpClientTest : AbstractStreamableHttpClientTest() {
|
|||
),
|
||||
)
|
||||
|
||||
val sessionId = UUID.randomUUID().toString()
|
||||
val sessionId = Uuid.random().toString()
|
||||
|
||||
mockMcp.onInitialize(
|
||||
clientName = "client1",
|
||||
|
|
@ -148,4 +153,64 @@ internal class StreamableHttpClientTest : AbstractStreamableHttpClientTest() {
|
|||
|
||||
client.close()
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `handle MethodNotAllowed`() = runBlocking {
|
||||
checkSupportNonStreamingResponse(
|
||||
ContentType.Text.EventStream,
|
||||
HttpStatusCode.MethodNotAllowed,
|
||||
)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `handle non-streaming response`() = runBlocking {
|
||||
checkSupportNonStreamingResponse(
|
||||
ContentType.Application.Json,
|
||||
HttpStatusCode.OK,
|
||||
)
|
||||
}
|
||||
|
||||
private suspend fun checkSupportNonStreamingResponse(contentType: ContentType, statusCode: HttpStatusCode) {
|
||||
val sessionId = "SID_${Uuid.random().toHexString()}"
|
||||
val clientName = "client-${Uuid.random().toHexString()}"
|
||||
val client = Client(
|
||||
clientInfo = Implementation(name = clientName, version = "1.0.0"),
|
||||
options = ClientOptions(
|
||||
capabilities = ClientCapabilities(),
|
||||
),
|
||||
)
|
||||
|
||||
mockMcp.onInitialize(clientName = clientName, sessionId = sessionId)
|
||||
|
||||
mockMcp.handleJSONRPCRequest(
|
||||
jsonRpcMethod = "notifications/initialized",
|
||||
expectedSessionId = sessionId,
|
||||
sessionId = sessionId,
|
||||
statusCode = HttpStatusCode.Accepted,
|
||||
)
|
||||
|
||||
mockMcp.onSubscribe(
|
||||
httpMethod = HttpMethod.Get,
|
||||
sessionId = sessionId,
|
||||
) respondsWith {
|
||||
headers += MCP_SESSION_ID_HEADER to sessionId
|
||||
body = null
|
||||
httpStatus = statusCode
|
||||
this.contentType = contentType
|
||||
}
|
||||
|
||||
mockMcp.handleWithResult(jsonRpcMethod = "ping", sessionId = sessionId) {
|
||||
buildJsonObject {}
|
||||
}
|
||||
|
||||
mockMcp.mockUnsubscribeRequest(sessionId = sessionId)
|
||||
|
||||
connect(client)
|
||||
|
||||
delay(1.seconds)
|
||||
|
||||
client.ping() // connection is still alive
|
||||
|
||||
client.close()
|
||||
}
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in a new issue