Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 4 commits into from
Nov 27, 2025
+71 −245
Merged

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

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
926c487
fix(sse-client): Fix StreamableHttpClientTransport to ignore empty data
kpavlov Nov 27, 2025
d90c1ea
Add test to skip empty Server-Sent Events (SSE)
kpavlov Nov 27, 2025
2a5dd16
Record lastEventId even when data is empty
kpavlov Nov 27, 2025
5a839c7
Update StreamableHttpClientTest to handle various SSE data formatting…
kpavlov Nov 27, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 13 additions & 8 deletions ...monMain/kotlin/io/modelcontextprotocol/kotlin/sdk/client/StreamableHttpClientTransport.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand All @@ -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
Expand Down Expand Up @@ -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()) {
Copy link
Contributor

@skarpovdev skarpovdev Nov 27, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Should it be isEmpty() instead?

Copy link
Contributor

@devcrocod devcrocod Nov 27, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Agree
isBlank will also skip a string if it contains only whitespace. But such a value is valid

Copy link
Contributor Author

@kpavlov kpavlov Nov 27, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

There’s no need to worry about spaces when parsing, since it’s not JSON.

return
}
if (eventName == null || eventName == "message") {
runCatching { McpJson.decodeFromString<JSONRPCMessage>(data) }
.onSuccess { msg ->
Expand All @@ -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())
Copy link
Contributor

@devcrocod devcrocod Nov 27, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

In that case, you can move dispatch fun out of handleInlineSse

Copy link
Contributor Author

@kpavlov kpavlov Nov 27, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Not quite yet. The "lastEventId" is updated inside. I wanted to avoid a major refactoring for now.

// reset
id = null
eventName = null
sb.clear()
Comment on lines +349 to +352
Copy link
Contributor Author

@kpavlov kpavlov Nov 27, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

less side-effects inside dispath

continue
}
when {
Expand Down
2 changes: 1 addition & 1 deletion ...Test/kotlin/io/modelcontextprotocol/kotlin/sdk/client/AbstractStreamableHttpClientTest.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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() {
Expand Down
236 changes: 0 additions & 236 deletions ...k-client/src/jvmTest/kotlin/io/modelcontextprotocol/kotlin/sdk/client/OldSchemaMockMcp.kt
View file
Open in desktop

This file was deleted.

46 changes: 46 additions & 0 deletions .../src/jvmTest/kotlin/io/modelcontextprotocol/kotlin/sdk/client/StreamableHttpClientTest.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -28,6 +30,50 @@ import kotlin.uuid.Uuid
@Suppress("LongMethod")
internal class StreamableHttpClientTest : AbstractStreamableHttpClientTest() {

@Test
fun `Should skip empty SSE`(): Unit = runBlocking {
Copy link
Contributor

@skarpovdev skarpovdev Nov 27, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

plus the following cases?

  • should handle whitespace-only SSE data fields
  • should handle multi-line SSE data fields
flowOf(
        "event: message\n",
        "data: {\"result\":{\"protocolVersion\":\"2025-06-18\",\n",
        "data: \"capabilities\":{},\n",
        "data: \"serverInfo\":{\"name\":\"simple-streamable-http-server\",\"version\":\"1.0.0\"}},\n",
        "data: \"jsonrpc\":\"2.0\",\"id\":\"7ce065b0678f49e5b04ce5a0fcc7d518\"}\n",
        "\n",
)

Copy link
Contributor Author

@kpavlov kpavlov Nov 27, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Good point. I will update the integration test for multiline

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",
"\n",
"id: ${Uuid.random()}\n",
"event: message\n",
@Suppress("MaxLineLength")
"data: {\"result\":{\"protocolVersion\":\"2025-06-18\",\"capabilities\":{},\"serverInfo\":{\"name\":\"simple-streamable-http-server\",\"version\":\"1.0.0\"}},\"jsonrpc\":\"2.0\",\"id\":\"7ce065b0678f49e5b04ce5a0fcc7d518\"}\n",
"\n",
Comment on lines 52 to 70
Copy link
Contributor Author

@kpavlov kpavlov Nov 27, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

raw SSE stream data

)
}

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(
Expand Down
Loading