fix(sse-client): Skip SSE in StreamableHttpClientTransport when data is empty (#433)

Refactor StreamableHttpClientTransport to skip SSE when the data field
is empty.


## Motivation and Context

TypeScript SSE Server started returning Server-Sent Events with only id
and empty data field (Heartbeat/Checkpoint), followed by an event with
data, e.g.:
```
id: d85c6bb0-0fa2-4828-81ad-3e951131aea5_1764226144689_7qb3iojc
data: 

event: message
id: d85c6bb0-0fa2-4828-81ad-3e951131aea5_1764226144690_sodh20lo
data: {"result":{"protocolVersion":"2025-06-18","capabilities":{"logging":{},"tools":{"listChanged":true},"prompts":{"listChanged":true},"completions":{},"resources":{"listChanged":true}},"serverInfo":{"name":"simple-streamable-http-server","version":"1.0.0"}},"jsonrpc":"2.0","id":"7ce065b0678f49e5b04ce5a0fcc7d518"}

```

Such empty events might be used by the server as **Heartbeat** or
**Checkpoint** events. The client should retain the updated
Last-Event-ID and disregard the payload. It is essential to handle such
responses to ensure compatibility with updated TypeScript SDK.

## How Has This Been Tested?

- Added integration test
- Regression tests are green now

## Breaking Changes
No

## Types of changes
<!-- What types of changes does your code introduce? Put an `x` in all
the boxes that apply: -->
- [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
<!-- Go over all the following points, and put an `x` in all the boxes
that apply. -->
- [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
- [ ] I have added appropriate error handling
- [ ] I have added or updated documentation as needed

## Additional context
<!-- Add any other context, implementation notes, or design decisions
-->

kotlin-sdk-client/src/commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/client/StreamableHttpClientTransport.kt

@@ -42,8 +42,6 @@ import kotlin.concurrent.atomics.AtomicBoolean
 import kotlin.concurrent.atomics.ExperimentalAtomicApi
 import kotlin.time.Duration
 
-private val logger = KotlinLogging.logger {}
-
 private const val MCP_SESSION_ID_HEADER = "mcp-session-id"
 private const val MCP_PROTOCOL_VERSION_HEADER = "mcp-protocol-version"
 private const val MCP_RESUMPTION_TOKEN_HEADER = "Last-Event-ID"
@@ -67,6 +65,10 @@ public class StreamableHttpClientTransport(
     private val requestBuilder: HttpRequestBuilder.() -> Unit = {},
 ) : AbstractTransport() {
 
+    private companion object {
+        private val logger = KotlinLogging.logger {}
+    }
+
     public var sessionId: String? = null
         private set
     public var protocolVersion: String? = null
@@ -316,11 +318,14 @@ public class StreamableHttpClientTransport(
         var id: String? = null
         var eventName: String? = null
 
-        suspend fun dispatch(data: String) {
+        suspend fun dispatch(id: String?, eventName: String?, data: String) {
             id?.let {
                 lastEventId = it
                 onResumptionToken?.invoke(it)
             }
+            if (data.isBlank()) {
+                return
+            }
             if (eventName == null || eventName == "message") {
                 runCatching { McpJson.decodeFromString<JSONRPCMessage>(data) }
                     .onSuccess { msg ->
@@ -335,16 +340,16 @@ public class StreamableHttpClientTransport(
                         throw it
                     }
             }
-            // reset
-            id = null
-            eventName = null
-            sb.clear()
         }
 
         while (!channel.isClosedForRead) {
             val line = channel.readUTF8Line() ?: break
             if (line.isEmpty()) {
-                dispatch(sb.toString())
+                dispatch(id = id, eventName = eventName, data = sb.toString())
+                // reset
+                id = null
+                eventName = null
+                sb.clear()
                 continue
             }
             when {

kotlin-sdk-client/src/jvmTest/kotlin/io/modelcontextprotocol/kotlin/sdk/client/AbstractStreamableHttpClientTest.kt

@@ -12,7 +12,7 @@ import org.junit.jupiter.api.TestInstance
 internal abstract class AbstractStreamableHttpClientTest {
 
     // start mokksy on random port
-    protected val mockMcp: OldSchemaMockMcp = OldSchemaMockMcp(verbose = true)
+    protected val mockMcp: MockMcp = MockMcp(verbose = true)
 
     @AfterEach
     fun afterEach() {

kotlin-sdk-client/src/jvmTest/kotlin/io/modelcontextprotocol/kotlin/sdk/client/OldSchemaMockMcp.kt

@@ -1,236 +0,0 @@
-package io.modelcontextprotocol.kotlin.sdk.client
-
-import dev.mokksy.mokksy.BuildingStep
-import dev.mokksy.mokksy.Mokksy
-import dev.mokksy.mokksy.StubConfiguration
-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.types.JSONRPCRequest
-import io.modelcontextprotocol.kotlin.sdk.types.RequestId
-import kotlinx.coroutines.flow.Flow
-import kotlinx.serialization.json.Json
-import kotlinx.serialization.json.JsonObject
-import kotlinx.serialization.json.JsonPrimitive
-import kotlinx.serialization.json.buildJsonObject
-import kotlinx.serialization.json.contentOrNull
-import kotlinx.serialization.json.jsonObject
-import kotlinx.serialization.json.jsonPrimitive
-import kotlinx.serialization.json.putJsonObject
-
-/**
- * High-level helper for simulating an MCP server over Streaming HTTP transport with Server-Sent Events (SSE),
- * built on top of an HTTP server using the [Mokksy](https://mokksy.dev) library.
- *
- * Provides test utilities to mock server behavior based on specific request conditions.
- *
- * @param verbose Whether to print detailed logs. Defaults to `false`.
- * @author Konstantin Pavlov
- */
-internal class OldSchemaMockMcp(verbose: Boolean = false) {
-
-    private val mokksy: Mokksy = Mokksy(verbose = verbose)
-
-    fun checkForUnmatchedRequests() {
-        mokksy.checkForUnmatchedRequests()
-    }
-
-    val url = "${mokksy.baseUrl()}/mcp"
-
-    @Suppress("LongParameterList")
-    fun onInitialize(
-        clientName: String? = null,
-        sessionId: String,
-        protocolVersion: String = "2025-03-26",
-        serverName: String = "Mock MCP Server",
-        serverVersion: String = "1.0.0",
-        capabilities: JsonObject = buildJsonObject {
-            putJsonObject("tools") {
-                put("listChanged", JsonPrimitive(false))
-            }
-        },
-    ) {
-        val predicates = if (clientName != null) {
-            arrayOf<(JSONRPCRequest?) -> Boolean>({
-                it?.params?.jsonObject
-                    ?.get("clientInfo")?.jsonObject
-                    ?.get("name")?.jsonPrimitive
-                    ?.contentOrNull == clientName
-            })
-        } else {
-            emptyArray()
-        }
-
-        handleWithResult(
-            jsonRpcMethod = "initialize",
-            sessionId = sessionId,
-            bodyPredicates = predicates,
-            // language=json
-            result = """
-            {
-                "capabilities": $capabilities,
-                "protocolVersion": "$protocolVersion",
-                "serverInfo": {
-                  "name": "$serverName",
-                  "version": "$serverVersion"
-                },
-                "_meta": {
-                  "foo": "bar"
-                }
-            }
-            """.trimIndent(),
-        )
-    }
-
-    fun onJSONRPCRequest(
-        httpMethod: HttpMethod = HttpMethod.Post,
-        jsonRpcMethod: String,
-        expectedSessionId: String? = null,
-        vararg bodyPredicates: (JSONRPCRequest) -> Boolean,
-    ): BuildingStep<JSONRPCRequest> = mokksy.method(
-        configuration = StubConfiguration(removeAfterMatch = true),
-        httpMethod = httpMethod,
-        requestType = JSONRPCRequest::class,
-    ) {
-        path("/mcp")
-        expectedSessionId?.let {
-            containsHeader(MCP_SESSION_ID_HEADER, it)
-        }
-        bodyMatchesPredicate(
-            description = "JSON-RPC version is '2.0'",
-            predicate =
-            {
-                it!!.jsonrpc == "2.0"
-            },
-        )
-        bodyMatchesPredicate(
-            description = "JSON-RPC Method should be '$jsonRpcMethod'",
-            predicate =
-            {
-                it!!.method == jsonRpcMethod
-            },
-        )
-        bodyPredicates.forEach { predicate ->
-            bodyMatchesPredicate(predicate = { predicate.invoke(it!!) })
-        }
-    }
-
-    @Suppress("LongParameterList")
-    fun handleWithResult(
-        httpMethod: HttpMethod = HttpMethod.Post,
-        jsonRpcMethod: String,
-        expectedSessionId: String? = null,
-        sessionId: String,
-        contentType: ContentType = ContentType.Application.Json,
-        statusCode: HttpStatusCode = HttpStatusCode.OK,
-        vararg bodyPredicates: (JSONRPCRequest) -> Boolean,
-        result: () -> JsonObject,
-    ) {
-        onJSONRPCRequest(
-            httpMethod = httpMethod,
-            jsonRpcMethod = jsonRpcMethod,
-            expectedSessionId = expectedSessionId,
-            bodyPredicates = bodyPredicates,
-        ) respondsWith {
-            val requestId = when (request.body.id) {
-                is RequestId.NumberId -> (request.body.id as RequestId.NumberId).value.toString()
-                is RequestId.StringId -> "\"${(request.body.id as RequestId.StringId).value}\""
-            }
-            val resultObject = result!!.invoke()
-            // language=json
-            body = """
-             {
-              "jsonrpc": "2.0",
-              "id": $requestId,
-              "result": $resultObject
-            }
-            """.trimIndent()
-            this.contentType = contentType
-            headers += MCP_SESSION_ID_HEADER to sessionId
-            httpStatus = statusCode
-        }
-    }
-
-    @Suppress("LongParameterList")
-    fun handleWithResult(
-        httpMethod: HttpMethod = HttpMethod.Post,
-        jsonRpcMethod: String,
-        expectedSessionId: String? = null,
-        sessionId: String,
-        contentType: ContentType = ContentType.Application.Json,
-        statusCode: HttpStatusCode = HttpStatusCode.OK,
-        vararg bodyPredicates: (JSONRPCRequest) -> Boolean,
-        result: String,
-    ) {
-        handleWithResult(
-            httpMethod = httpMethod,
-            jsonRpcMethod = jsonRpcMethod,
-            expectedSessionId = expectedSessionId,
-            sessionId = sessionId,
-            contentType = contentType,
-            statusCode = statusCode,
-            bodyPredicates = bodyPredicates,
-            result = {
-                Json.parseToJsonElement(result).jsonObject
-            },
-        )
-    }
-
-    @Suppress("LongParameterList")
-    fun handleJSONRPCRequest(
-        httpMethod: HttpMethod = HttpMethod.Post,
-        jsonRpcMethod: String,
-        expectedSessionId: String? = null,
-        sessionId: String,
-        contentType: ContentType = ContentType.Application.Json,
-        statusCode: HttpStatusCode = HttpStatusCode.OK,
-        vararg bodyPredicates: (JSONRPCRequest?) -> Boolean,
-        bodyBuilder: () -> String = { "" },
-    ) {
-        onJSONRPCRequest(
-            httpMethod = httpMethod,
-            jsonRpcMethod = jsonRpcMethod,
-            expectedSessionId = expectedSessionId,
-            bodyPredicates = bodyPredicates,
-        ) respondsWith {
-            body = bodyBuilder.invoke()
-            this.contentType = contentType
-            headers += MCP_SESSION_ID_HEADER to sessionId
-            httpStatus = statusCode
-        }
-    }
-
-    fun onSubscribe(httpMethod: HttpMethod = HttpMethod.Post, sessionId: String): BuildingStep<Any> = mokksy.method(
-        httpMethod = httpMethod,
-        name = "MCP GETs",
-        requestType = Any::class,
-    ) {
-        path("/mcp")
-        containsHeader(MCP_SESSION_ID_HEADER, sessionId)
-        containsHeader("Accept", "application/json,text/event-stream")
-        containsHeader("Cache-Control", "no-store")
-    }
-
-    fun handleSubscribeWithGet(sessionId: String, block: () -> Flow<ServerSentEvent>) {
-        onSubscribe(
-            httpMethod = HttpMethod.Get,
-            sessionId = sessionId,
-        ) respondsWithSseStream {
-            headers += MCP_SESSION_ID_HEADER to sessionId
-            this.flow = block.invoke()
-        }
-    }
-
-    fun mockUnsubscribeRequest(sessionId: String) {
-        mokksy.delete(
-            configuration = StubConfiguration(removeAfterMatch = true),
-            requestType = JSONRPCRequest::class,
-        ) {
-            path("/mcp")
-            containsHeader(MCP_SESSION_ID_HEADER, sessionId)
-        } respondsWith {
-            body = null
-        }
-    }
-}

kotlin-sdk-client/src/jvmTest/kotlin/io/modelcontextprotocol/kotlin/sdk/client/StreamableHttpClientTest.kt

@@ -11,7 +11,9 @@ import io.modelcontextprotocol.kotlin.sdk.types.Implementation
 import io.modelcontextprotocol.kotlin.sdk.types.Tool
 import io.modelcontextprotocol.kotlin.sdk.types.ToolSchema
 import kotlinx.coroutines.delay
+import kotlinx.coroutines.flow.emptyFlow
 import kotlinx.coroutines.flow.flow
+import kotlinx.coroutines.flow.flowOf
 import kotlinx.coroutines.runBlocking
 import kotlinx.serialization.json.buildJsonObject
 import kotlinx.serialization.json.put
@@ -28,6 +30,61 @@ import kotlin.uuid.Uuid
 @Suppress("LongMethod")
 internal class StreamableHttpClientTest : AbstractStreamableHttpClientTest() {
 
+    @Test
+    fun `Should skip empty SSE`(): Unit = runBlocking {
+        val client = Client(
+            clientInfo = Implementation(
+                name = "client1",
+                version = "1.0.0",
+            ),
+            options = ClientOptions(
+                capabilities = ClientCapabilities(),
+            ),
+        )
+        val sessionId = Uuid.random().toString()
+
+        mockMcp.onJSONRPCRequest(
+            httpMethod = HttpMethod.Post,
+            jsonRpcMethod = "initialize",
+        ).respondsWithStream {
+            headers += MCP_SESSION_ID_HEADER to sessionId
+            flow = flowOf(
+                "id: ${Uuid.random()}\n",
+                "data:\n", // empty data
+                "\n",
+                "id: ${Uuid.random()}\n",
+                "data:  \t \n", // tabs and spaces
+                "\n",
+                "id: ${Uuid.random()}\n",
+                "event: message\n",
+                // multiline data
+                "data: {\n",
+                "data: \"result\":{\n" +
+                    "data:    \"protocolVersion\":\"2025-06-18\",\n" +
+                    "data:    \"capabilities\":{},\n" +
+                    "data:    \"serverInfo\":{\"name\":\"simple-streamable-http-server\",\"version\":\"1.0.0\"}\n" +
+                    "data: },\n" +
+                    "data: \"jsonrpc\":\"2.0\",\n" +
+                    "data: \"id\":\"7ce065b0678f49e5b04ce5a0fcc7d518\"\n" +
+                    "data: }\n",
+                "\n",
+            )
+        }
+
+        mockMcp.handleJSONRPCRequest(
+            jsonRpcMethod = "notifications/initialized",
+            expectedSessionId = sessionId,
+            sessionId = sessionId,
+            statusCode = HttpStatusCode.Accepted,
+        )
+
+        mockMcp.handleSubscribeWithGet(sessionId) {
+            emptyFlow()
+        }
+
+        connect(client)
+    }
+
     @Test
     fun `test streamableHttpClient`() = runBlocking {
         val client = Client(