fix(client): invoke onClose when SSE session disconnects (#437) (#738)

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
This commit is contained in:
johny 2026-05-16 01:44:16 +09:00 • committed by GitHub
commit 44e012e4ee
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
3 changed files with 36 additions and 0 deletions

View file

@ -136,6 +136,7 @@ public class SseClientTransport(
throw e
} finally {
closeResources()
invokeOnCloseCallback()
}
}

View file

@ -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 {

View file

@ -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)