diff --git a/kotlin-sdk-client/src/commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/client/SseClientTransport.kt b/kotlin-sdk-client/src/commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/client/SseClientTransport.kt index a48db60..0295eb9 100644 --- a/kotlin-sdk-client/src/commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/client/SseClientTransport.kt +++ b/kotlin-sdk-client/src/commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/client/SseClientTransport.kt @@ -136,6 +136,7 @@ public class SseClientTransport( throw e } finally { closeResources() + invokeOnCloseCallback() } } diff --git a/kotlin-sdk-client/src/commonTest/kotlin/io/modelcontextprotocol/kotlin/sdk/client/sse/MockSseClientEngine.kt b/kotlin-sdk-client/src/commonTest/kotlin/io/modelcontextprotocol/kotlin/sdk/client/sse/MockSseClientEngine.kt index 3627b6c..14af546 100644 --- a/kotlin-sdk-client/src/commonTest/kotlin/io/modelcontextprotocol/kotlin/sdk/client/sse/MockSseClientEngine.kt +++ b/kotlin-sdk-client/src/commonTest/kotlin/io/modelcontextprotocol/kotlin/sdk/client/sse/MockSseClientEngine.kt @@ -53,6 +53,14 @@ internal open class MockSseClientEngine( job.cancel() } + /** + * Closes the SSE byte channel without disposing the engine, simulating a server-side + * disconnect of the SSE stream. + */ + fun disconnectSseStream() { + channel.close() + } + private fun responseContext(): CoroutineContext = dispatcher + Job() private suspend fun respondWithSseEndpointEvent(data: HttpRequestData): HttpResponseData { diff --git a/kotlin-sdk-client/src/commonTest/kotlin/io/modelcontextprotocol/kotlin/sdk/client/sse/SseClientTransportTest.kt b/kotlin-sdk-client/src/commonTest/kotlin/io/modelcontextprotocol/kotlin/sdk/client/sse/SseClientTransportTest.kt index 6cc7f6e..319ba2a 100644 --- a/kotlin-sdk-client/src/commonTest/kotlin/io/modelcontextprotocol/kotlin/sdk/client/sse/SseClientTransportTest.kt +++ b/kotlin-sdk-client/src/commonTest/kotlin/io/modelcontextprotocol/kotlin/sdk/client/sse/SseClientTransportTest.kt @@ -1,5 +1,6 @@ package io.modelcontextprotocol.kotlin.sdk.client.sse +import io.kotest.assertions.nondeterministic.eventually import io.kotest.matchers.collections.shouldHaveSize import io.kotest.matchers.shouldBe import io.ktor.client.HttpClient @@ -9,6 +10,7 @@ import io.modelcontextprotocol.kotlin.sdk.client.SseClientTransport import io.modelcontextprotocol.kotlin.sdk.types.JSONRPCNotification import kotlinx.coroutines.test.runTest import kotlin.test.Test +import kotlin.time.Duration.Companion.seconds class SseClientTransportTest { @@ -90,6 +92,31 @@ class SseClientTransportTest { engine.close() } + @Test + fun `onClose callback fires when the SSE stream is disconnected by the server`() = runTest { + // Given + val sseUrl = "http://example.com/api/mcp/sse" + + // And + val engine = CapturingSseClientEngine(endpoint = "/messages?sessionId=abc") + val transport = sseTransport(sseUrl, engine) + var onCloseFired = false + transport.onClose { onCloseFired = true } + + // When + transport.start() + engine.disconnectSseStream() + + // Then + eventually(2.seconds) { + onCloseFired shouldBe true + } + + // Cleanup + transport.close() + engine.close() + } + private fun sseTransport(sseUrl: String, engine: CapturingSseClientEngine) = SseClientTransport(HttpClient(engine) { install(SSE) }, sseUrl)