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

Author: kpavlov

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

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