modelcontextprotocol/kotlin-sdk
Watch
1
0
Fork
You've already forked kotlin-sdk
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

fix(server): evict stale GET SSE stream mapping on reconnect (#716)

Browse source 
## Motivation and Context

When a GET SSE stream disconnects and the client reconnects on the same
session, the new GET is silently rejected because the previous stream's
`STANDALONE_SSE_STREAM_ID` entry in `streamsMapping` hasn't been cleaned
up yet. Cleanup via `invokeOnCompletion` is asynchronous — there is a
race window where the stale entry persists and blocks reconnection.

Additionally, Ktor's `sse {}` handler commits 200 OK headers before
`handleGetRequest` runs, so the intended
`call.reject(HttpStatusCode.Conflict, ...)` cannot change the
already-committed HTTP status. On all tested engines (test host, CIO,
Netty), clients receive a 200 OK with an immediately-closed empty SSE
stream rather than a 409 Conflict. Server-side `CallLogging` reports 409
(from the response object, not the wire status), which is misleading.

### What this PR does

1. **Explicitly closes the old SSE session** before establishing a new
stream. If the old session is already dead, `close()` is a no-op. This
is the synchronous cleanup that the async `invokeOnCompletion` couldn't
provide.

2. **Replaces `invokeOnCompletion` with `try/finally`** around
`awaitCancellation()` — cleanup runs during cancellation propagation
(earlier in the lifecycle) rather than after job completion. An identity
check (`=== newContext`) ensures only this stream's entry is removed,
not a replacement's.

3. **Adds identity-based removal in `emitOnStream`** — a failed `send()`
on an old stream can no longer accidentally evict the new stream's
mapping.

4. **Rethrows `CancellationException`** instead of swallowing it,
preserving structured concurrency.

The TypeScript reference implementation avoids this race via synchronous
`ReadableStream.cancel()` cleanup. The Kotlin coroutine equivalent
(`invokeOnCompletion`) is inherently async. The Go SDK uses `defer
stream.release()` under a mutex for the same reason. The Java SDK uses
unconditional `AtomicReference.set()` replacement with
`compareAndExchange` on close.

Fixes #715

## How Has This Been Tested?

All tests run against latest `main` (`bf8dc6d`). ktlint and detekt pass
clean.

- 2 new unit tests (Ktor test host): reconnect after disconnect,
concurrent stream takeover
- 2 new integration tests (real CIO embedded server): same scenarios
over real HTTP
- Manual verification against a live Ktor+Netty MCP server (curl):
confirmed 200 OK on the wire with immediately-closed stream on reconnect
(pre-fix), live stream on reconnect (post-fix)
- All existing server unit tests pass (76 total)
- All existing integration tests pass (306 total)

## Breaking Changes

The server no longer returns 409 Conflict for a second GET SSE stream on
the same session. Instead, the old stream is explicitly closed and the
new stream takes over. The 409 was already unreachable as an HTTP status
— Ktor's `sse {}` handler commits 200 OK before the transport handler
runs on all tested engines (test host, CIO, Netty), so clients never
received 409 on the wire.

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

## Additional context

This bug likely became reachable after #681 (April 10) fixed the Netty
`appendSseHeaders` crash that previously prevented GET SSE from working
at all on Netty.

Cross-SDK comparison of how each SDK handles a second GET SSE on the
same session:

| SDK | Behavior | Cleanup mechanism | Stale mapping bug? |
|-----|----------|-------------------|-------------------|
| TypeScript (reference) | 409 Conflict | Synchronous
`ReadableStream.cancel()` | No |
| Go | 409 Conflict | `defer stream.release()` under mutex | No |
| Java | Replaces (no 409) | `AtomicReference.set()` +
`compareAndExchange` on close | No |
| C# | 400 Bad Request | Boolean flag, never reset (no reconnect
support) | No |
| Kotlin (before fix) | 409 (unreachable on wire) | Async
`invokeOnCompletion` | **Yes** |
| Kotlin (this PR) | Closes old + replaces | `try/finally` +
identity-based removal | No |

---------

Co-authored-by: devcrocod <devcrocod@gmail.com>
This commit is contained in:
Amr Youssef 2026-04-29 03:09:05 -07:00 • committed by GitHub
commit a238fc0a2e
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
3 changed files with 309 additions and 17 deletions
Download patch file Download diff file Expand all files Collapse all files

171
integration-test/src/jvmTest/kotlin/io/modelcontextprotocol/kotlin/sdk/integration/streamablehttp/StreamableHttpSseReconnectTest.kt Normal file
View file

@ -0,0 +1,171 @@
package io.modelcontextprotocol.kotlin.sdk.integration.streamablehttp
import io.kotest.matchers.nulls.shouldNotBeNull
import io.kotest.matchers.shouldBe
import io.ktor.client.HttpClient
import io.ktor.client.plugins.sse.SSE
import io.ktor.client.request.header
import io.ktor.client.request.post
import io.ktor.client.request.prepareGet
import io.ktor.client.request.setBody
import io.ktor.client.statement.bodyAsChannel
import io.ktor.http.ContentType
import io.ktor.http.HttpHeaders
import io.ktor.http.HttpStatusCode
import io.ktor.http.contentType
import io.ktor.utils.io.readUTF8Line
import io.modelcontextprotocol.kotlin.sdk.types.ClientCapabilities
import io.modelcontextprotocol.kotlin.sdk.types.Implementation
import io.modelcontextprotocol.kotlin.sdk.types.InitializeRequest
import io.modelcontextprotocol.kotlin.sdk.types.InitializeRequestParams
import io.modelcontextprotocol.kotlin.sdk.types.JSONRPCRequest
import io.modelcontextprotocol.kotlin.sdk.types.LATEST_PROTOCOL_VERSION
import io.modelcontextprotocol.kotlin.sdk.types.toJSON
import io.modelcontextprotocol.kotlin.test.utils.actualPort
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.runBlocking
import kotlinx.serialization.json.Json
import kotlin.test.Test
import kotlin.test.assertNotNull
import io.ktor.client.engine.cio.CIO as ClientCIO
private const val SESSION_ID_HEADER = "mcp-session-id"
private const val PROTOCOL_VERSION_HEADER = "mcp-protocol-version"
/**
* Integration tests for GET SSE stream reconnection using a real embedded CIO server.
*
* Verifies that the transport correctly evicts stale STANDALONE_SSE_STREAM_ID
* entries when a client reconnects after a disconnect, rather than silently
* rejecting the new stream.
*/
class StreamableHttpSseReconnectTest : AbstractStreamableHttpIntegrationTest() {
/**
* Verifies that after a GET SSE stream disconnects and the client
* immediately reconnects, the server evicts the stale stream mapping
* and allows the new stream to succeed.
*/
@Test
fun `GET SSE reconnect after disconnect should succeed`(): Unit = runBlocking(Dispatchers.IO) {
var server: StreamableHttpTestServer? = null
var httpClient: HttpClient? = null
try {
server = initTestServer("reconnect-test")
val port = server.ktorServer.actualPort()
val mcpUrl = "http://$URL:$port/mcp"
httpClient = HttpClient(ClientCIO) { install(SSE) }
// Step 1: Initialize session via POST
val initResponse = httpClient.post(mcpUrl) {
contentType(ContentType.Application.Json)
header(
HttpHeaders.Accept,
"${ContentType.Application.Json}, ${ContentType.Text.EventStream}",
)
setBody(Json.encodeToString(buildInitPayload()))
}
initResponse.status shouldBe HttpStatusCode.OK
val sessionId = assertNotNull(initResponse.headers[SESSION_ID_HEADER])
// Step 2: Open GET SSE stream, consume the flush event, then disconnect
httpClient.prepareGet(mcpUrl) {
header(HttpHeaders.Accept, ContentType.Text.EventStream.toString())
header(SESSION_ID_HEADER, sessionId)
header(PROTOCOL_VERSION_HEADER, LATEST_PROTOCOL_VERSION)
}.execute { response ->
response.status shouldBe HttpStatusCode.OK
response.bodyAsChannel().readUTF8Line()
}
// Step 3: Immediately reconnect. The transport detects that the
// previous stream's coroutine is no longer active and evicts the
// stale mapping, allowing the new stream to succeed.
httpClient.prepareGet(mcpUrl) {
header(HttpHeaders.Accept, ContentType.Text.EventStream.toString())
header(SESSION_ID_HEADER, sessionId)
header(PROTOCOL_VERSION_HEADER, LATEST_PROTOCOL_VERSION)
}.execute { response ->
response.status shouldBe HttpStatusCode.OK
response.headers[SESSION_ID_HEADER] shouldBe sessionId
val channel = response.bodyAsChannel()
val firstLine = channel.readUTF8Line()
firstLine.shouldNotBeNull()
channel.isClosedForRead shouldBe false
}
} finally {
httpClient?.close()
server?.ktorServer?.stopSuspend(1000, 2000)
}
}
/**
* Verifies that a second concurrent GET SSE stream on the same session
* closes the old stream and takes over. The new stream should be live.
*/
@Test
fun `concurrent GET SSE stream closes old stream and takes over`(): Unit = runBlocking(Dispatchers.IO) {
var server: StreamableHttpTestServer? = null
var httpClient: HttpClient? = null
try {
server = initTestServer("takeover-test")
val port = server.ktorServer.actualPort()
val mcpUrl = "http://$URL:$port/mcp"
httpClient = HttpClient(ClientCIO) { install(SSE) }
// Step 1: Initialize session via POST
val initResponse = httpClient.post(mcpUrl) {
contentType(ContentType.Application.Json)
header(
HttpHeaders.Accept,
"${ContentType.Application.Json}, ${ContentType.Text.EventStream}",
)
setBody(Json.encodeToString(buildInitPayload()))
}
initResponse.status shouldBe HttpStatusCode.OK
val sessionId = assertNotNull(initResponse.headers[SESSION_ID_HEADER])
// Step 2: Open first GET SSE stream and keep it open
httpClient.prepareGet(mcpUrl) {
header(HttpHeaders.Accept, ContentType.Text.EventStream.toString())
header(SESSION_ID_HEADER, sessionId)
header(PROTOCOL_VERSION_HEADER, LATEST_PROTOCOL_VERSION)
}.execute { firstResponse ->
firstResponse.status shouldBe HttpStatusCode.OK
firstResponse.bodyAsChannel().readUTF8Line()
// Step 3: Open a second GET — closes old stream, new one takes over
httpClient.prepareGet(mcpUrl) {
header(HttpHeaders.Accept, ContentType.Text.EventStream.toString())
header(SESSION_ID_HEADER, sessionId)
header(PROTOCOL_VERSION_HEADER, LATEST_PROTOCOL_VERSION)
}.execute { secondResponse ->
secondResponse.status shouldBe HttpStatusCode.OK
secondResponse.headers[SESSION_ID_HEADER] shouldBe sessionId
// New stream is alive
val secondChannel = secondResponse.bodyAsChannel()
val firstLine = secondChannel.readUTF8Line()
firstLine.shouldNotBeNull()
secondChannel.isClosedForRead shouldBe false
}
}
} finally {
httpClient?.close()
server?.ktorServer?.stopSuspend(1000, 2000)
}
}
private fun buildInitPayload(): JSONRPCRequest = InitializeRequest(
InitializeRequestParams(
protocolVersion = LATEST_PROTOCOL_VERSION,
capabilities = ClientCapabilities(),
clientInfo = Implementation(name = "reconnect-test-client", version = "1.0.0"),
),
).toJSON()
}

54
kotlin-sdk-server/src/commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/server/StreamableHttpServerTransport.kt
View file

@ -464,27 +464,40 @@ public class StreamableHttpServerTransport(private val configuration: Configurat
}
}
if (STANDALONE_SSE_STREAM_ID in streamsMapping) {
call.reject(
HttpStatusCode.Conflict,
RPCError.ErrorCode.CONNECTION_CLOSED,
"Conflict: Only one SSE stream is allowed per session",
)
return
}
// SSE headers (Content-Type, Cache-Control, Connection) are already set by the framework's SSE handler
flushSse(sseSession)
streamsMapping[STANDALONE_SSE_STREAM_ID] = SessionContext(sseSession, call)
val newContext = SessionContext(sseSession, call)
streamMutex.withLock {
streamsMapping[STANDALONE_SSE_STREAM_ID]?.let { existingContext ->
// Close the previous SSE session. If alive, this cancels the old
// coroutine (which will hit its identity-guarded finally — that finally
// won't double-remove, since we replace the mapping below).
try {
existingContext.session?.close()
} catch (e: CancellationException) {
throw e
} catch (_: Exception) {
// Ignore — the old stream may already be closed.
}
// Evict the stale mapping — the old session is closed either way.
streamsMapping.remove(STANDALONE_SSE_STREAM_ID)
}
streamsMapping[STANDALONE_SSE_STREAM_ID] = newContext
}
val clientProtocolVersion =
call.request.header(MCP_PROTOCOL_VERSION_HEADER) ?: DEFAULT_NEGOTIATED_PROTOCOL_VERSION
maybeSendPrimingEvent(STANDALONE_SSE_STREAM_ID, sseSession, clientProtocolVersion)
sseSession.coroutineContext.job.invokeOnCompletion {
streamsMapping.remove(STANDALONE_SSE_STREAM_ID)
}
// Keep the SSE connection open until the client disconnects or the transport is closed.
// Without this, the Ktor sse{} handler returns immediately, closing the stream.
awaitCancellation()
// Cleanup uses try/finally (runs during cancellation propagation) instead of
// invokeOnCompletion (runs after job completion) to minimize the window between
// disconnect and mapping removal. Identity check ensures only this stream's entry
// is removed — not a replacement that arrived in the meantime.
try {
awaitCancellation()
} finally {
// Atomic compare-and-remove — only evict our own entry, not a replacement's.
streamsMapping.remove(STANDALONE_SSE_STREAM_ID, newContext)
}
}
/** Handles an HTTP DELETE request by closing the session and the transport. */
@ -725,10 +738,17 @@ public class StreamableHttpServerTransport(private val configuration: Configurat
try {
session?.send(event = "message", id = eventId, data = McpJson.encodeToString(message))
} catch (e: CancellationException) {
streamsMapping.remove(streamId)
// Atomic compare-and-remove — only evict this stream's entry, not a replacement's.
val current = streamsMapping[streamId]
if (current != null && current.session === session) {
streamsMapping.remove(streamId, current)
}
throw e
} catch (_: Exception) {
streamsMapping.remove(streamId)
val current = streamsMapping[streamId]
if (current != null && current.session === session) {
streamsMapping.remove(streamId, current)
}
}
}

101
kotlin-sdk-server/src/jvmTest/kotlin/io/modelcontextprotocol/kotlin/sdk/server/StreamableHttpServerTransportTest.kt
View file

@ -460,6 +460,107 @@ class StreamableHttpServerTransportTest {
}
}
@Test
fun `second concurrent GET SSE closes old stream and takes over`() = testApplication {
val mcpPath = "/mcp"
application {
mcpStreamableHttp(mcpPath) {
Server(
Implementation("test-server", "1.0.0"),
ServerOptions(capabilities = ServerCapabilities()),
)
}
}
val client = createTestClient()
// Step 1: Initialize session via POST
val initResponse = client.post(mcpPath) {
addStreamableHeaders()
setBody(buildInitializeRequestPayload())
}
initResponse.status shouldBe HttpStatusCode.OK
val sessionId = assertNotNull(initResponse.headers[MCP_SESSION_ID_HEADER])
// Step 2: Open first GET SSE stream
client.prepareGet(mcpPath) {
header(HttpHeaders.Accept, ContentType.Text.EventStream.toString())
header(MCP_SESSION_ID_HEADER, sessionId)
header("mcp-protocol-version", LATEST_PROTOCOL_VERSION)
}.execute { firstResponse ->
firstResponse.status shouldBe HttpStatusCode.OK
firstResponse.bodyAsChannel().readUTF8Line()
// Step 3: Open a second GET — the transport closes the old session
// and the new stream takes over.
client.prepareGet(mcpPath) {
header(HttpHeaders.Accept, ContentType.Text.EventStream.toString())
header(MCP_SESSION_ID_HEADER, sessionId)
header("mcp-protocol-version", LATEST_PROTOCOL_VERSION)
}.execute { secondResponse ->
secondResponse.status shouldBe HttpStatusCode.OK
secondResponse.headers[MCP_SESSION_ID_HEADER] shouldBe sessionId
// New stream is alive
val secondChannel = secondResponse.bodyAsChannel()
val firstLine = secondChannel.readUTF8Line()
firstLine.shouldNotBeNull()
secondChannel.isClosedForRead shouldBe false
}
}
}
@Test
fun `GET SSE reconnect after previous stream disconnects should succeed`() = testApplication {
val mcpPath = "/mcp"
application {
mcpStreamableHttp(mcpPath) {
Server(
Implementation("test-server", "1.0.0"),
ServerOptions(capabilities = ServerCapabilities()),
)
}
}
val client = createTestClient()
// Step 1: Initialize session via POST
val initResponse = client.post(mcpPath) {
addStreamableHeaders()
setBody(buildInitializeRequestPayload())
}
initResponse.status shouldBe HttpStatusCode.OK
val sessionId = assertNotNull(initResponse.headers[MCP_SESSION_ID_HEADER])
// Step 2: Open and then close a GET SSE stream
client.prepareGet(mcpPath) {
header(HttpHeaders.Accept, ContentType.Text.EventStream.toString())
header(MCP_SESSION_ID_HEADER, sessionId)
header("mcp-protocol-version", LATEST_PROTOCOL_VERSION)
}.execute { response ->
response.status shouldBe HttpStatusCode.OK
response.bodyAsChannel().readUTF8Line()
}
// Step 3: Immediately reconnect — the transport should close the stale
// stream and allow the new one.
client.prepareGet(mcpPath) {
header(HttpHeaders.Accept, ContentType.Text.EventStream.toString())
header(MCP_SESSION_ID_HEADER, sessionId)
header("mcp-protocol-version", LATEST_PROTOCOL_VERSION)
}.execute { response ->
response.status shouldBe HttpStatusCode.OK
response.headers[MCP_SESSION_ID_HEADER] shouldBe sessionId
val channel = response.bodyAsChannel()
val firstLine = channel.readUTF8Line()
firstLine.shouldNotBeNull()
channel.isClosedForRead shouldBe false
}
}
@Test
fun `GET SSE stream includes Mcp-Session-Id header and stays open`() = testApplication {
val mcpPath = "/mcp"