From 44e012e4eefc9ad1128a814c0be09769613367ba Mon Sep 17 00:00:00 2001 From: johny <69754601+jskjw157@users.noreply.github.com> Date: Sat, 16 May 2026 01:44:16 +0900 Subject: [PATCH] fix(client): invoke onClose when SSE session disconnects (#437) (#738) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fixes the SSE client transport so that registered `onClose` callbacks fire when the underlying SSE session ends — whether the server closes the stream gracefully or an error event terminates it. closes #437 ## Motivation and Context `SseClientTransport.collectMessages` already runs `closeResources()` in its `finally` block when `session.incoming` completes, but it never invoked the `onClose` callback chain. As a result, the MCP `Client` (and any user code calling `transport.onClose { ... }`) had no way to detect that the server-side SSE connection had dropped. The other client transports already do this correctly: - `StdioClientTransport` calls `invokeOnCloseCallback()` from its read-loop `finally` (kotlin-sdk-client/src/.../StdioClientTransport.kt:224). - The protected `AbstractTransport.invokeOnCloseCallback()` helper guards against duplicate firing with an `AtomicBoolean`, so it is safe to call from both the read-loop `finally` and the user-initiated `close()` path. The fix mirrors that established pattern in one line. ## How Has This Been Tested? Added `onClose callback fires when the SSE stream is disconnected by the server` to `SseClientTransportTest`. The test: 1. Starts the transport against a mock SSE engine. 2. Closes the SSE byte channel server-side via a new `MockSseClientEngine.disconnectSseStream()` helper. 3. Asserts the registered `onClose` callback fires (using Kotest `eventually`, since the close propagates on a real-time dispatcher). I also verified the test fails before the fix and passes after. ``` ./gradlew :kotlin-sdk-client:jvmTest ./gradlew apiCheck ktlintCheck detekt ``` ## Breaking Changes None. `invokeOnCloseCallback()` is `protected` on `AbstractTransport`; this PR only adds a call site inside `SseClientTransport`. No public API surface changes. ## Types of changes - [x] Bug fix (non-breaking change which fixes an issue) - [ ] New feature (non-breaking change which adds functionality) - [ ] Breaking change (fix or feature that would cause existing functionality to change) - [ ] Documentation update ## Checklist - [x] I have read the [MCP Documentation](https://modelcontextprotocol.io) - [x] My code follows the repository's style guidelines - [x] New and existing tests pass locally - [x] I have added appropriate error handling - [x] I have added or updated documentation as needed --- .../kotlin/sdk/client/SseClientTransport.kt | 1 + .../sdk/client/sse/MockSseClientEngine.kt | 8 ++++++ .../sdk/client/sse/SseClientTransportTest.kt | 27 +++++++++++++++++++ 3 files changed, 36 insertions(+) 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)