Skip to content

fix(server): GET SSE stream crash on Netty in Streamable HTTP #681

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
devcrocod merged 1 commit into from
Apr 10, 2026
Merged

fix(server): GET SSE stream crash on Netty in Streamable HTTP #681

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
14 changes: 14 additions & 0 deletions ...-sdk-server/src/commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/server/KtorServer.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,13 +3,17 @@
package io.modelcontextprotocol.kotlin.sdk.server

import io.github.oshai.kotlinlogging.KotlinLogging
import io.ktor.http.HttpMethod
import io.ktor.http.HttpStatusCode
import io.ktor.server.application.Application
import io.ktor.server.application.ApplicationCall
import io.ktor.server.application.ApplicationCallPipeline
import io.ktor.server.application.MissingApplicationPluginException
import io.ktor.server.application.install
import io.ktor.server.request.ApplicationRequest
import io.ktor.server.request.header
import io.ktor.server.request.httpMethod
import io.ktor.server.response.header
import io.ktor.server.response.respond
import io.ktor.server.routing.Route
import io.ktor.server.routing.RoutingContext
Expand Down Expand Up @@ -114,6 +118,16 @@ private fun Application.mcpStreamableHttp(

routing {
route(path) {
// Set Mcp-Session-Id on GET responses before Ktor's sse {} commits headers.
intercept(ApplicationCallPipeline.Plugins) {
if (context.request.httpMethod == HttpMethod.Get) {
val sessionId = context.request.header(MCP_SESSION_ID_HEADER)
Copy link

Copilot AI Apr 9, 2026

Choose a reason for hiding this comment

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

The GET interceptor reads the session ID via context.request.header(...) (single value), but the transport validation later rejects requests that contain multiple Mcp-Session-Id header values. In that case this interceptor can still echo a session ID onto a response that will be rejected. Consider using context.request.headers.getAll(MCP_SESSION_ID_HEADER) and only setting the response header when there is exactly one non-blank value (matching StreamableHttpServerTransport.validateSession).

Suggested change
val sessionId = context.request.header(MCP_SESSION_ID_HEADER)
val sessionIds = context.request.headers.getAll(MCP_SESSION_ID_HEADER)
val sessionId =
sessionIds
?.singleOrNull()
?.takeIf { it.isNotBlank() }

Copilot uses AI. Check for mistakes.
if (sessionId != null && transportManager.getTransport(sessionId) != null) {
context.response.header(MCP_SESSION_ID_HEADER, sessionId)
}
}
}

sse {
val transport = existingStreamableTransport(call, transportManager) ?: return@sse
transport.handleRequest(this, call)
Expand Down
8 changes: 4 additions & 4 deletions ...monMain/kotlin/io/modelcontextprotocol/kotlin/sdk/server/StreamableHttpServerTransport.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -453,8 +453,8 @@ public class StreamableHttpServerTransport(private val configuration: Configurat
return
}

call.appendSseHeaders()
flushSse(sseSession) // flush headers immediately
// 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)
maybeSendPrimingEvent(STANDALONE_SSE_STREAM_ID, sseSession, call.request.header(MCP_PROTOCOL_VERSION_HEADER))
sseSession.coroutineContext.job.invokeOnCompletion {
Expand Down Expand Up @@ -529,8 +529,8 @@ public class StreamableHttpServerTransport(private val configuration: Configurat
}
}

call.appendSseHeaders()
flushSse(session) // flush headers immediately
// SSE headers are already set by the framework's SSE handler.
flushSse(session)

val streamId = store.replayEventsAfter(lastEventId) { eventId, message ->
try {
Expand Down
46 changes: 46 additions & 0 deletions ...est/kotlin/io/modelcontextprotocol/kotlin/sdk/server/StreamableHttpServerTransportTest.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@ package io.modelcontextprotocol.kotlin.sdk.server

import io.kotest.matchers.collections.shouldContainAll
import io.kotest.matchers.equals.shouldBeEqual
import io.kotest.matchers.nulls.shouldNotBeNull
import io.kotest.matchers.shouldBe
import io.ktor.client.HttpClient
import io.ktor.client.call.body
Expand All @@ -10,7 +11,9 @@ import io.ktor.client.plugins.logging.Logging
import io.ktor.client.request.HttpRequestBuilder
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
Expand All @@ -21,6 +24,7 @@ import io.ktor.server.routing.post
import io.ktor.server.routing.routing
import io.ktor.server.testing.ApplicationTestBuilder
import io.ktor.server.testing.testApplication
import io.ktor.utils.io.readUTF8Line
import io.modelcontextprotocol.kotlin.sdk.types.ClientCapabilities
import io.modelcontextprotocol.kotlin.sdk.types.EmptyResult
import io.modelcontextprotocol.kotlin.sdk.types.Implementation
Expand All @@ -36,6 +40,7 @@ import io.modelcontextprotocol.kotlin.sdk.types.ListToolsResult
import io.modelcontextprotocol.kotlin.sdk.types.McpJson
import io.modelcontextprotocol.kotlin.sdk.types.Method
import io.modelcontextprotocol.kotlin.sdk.types.RequestId
import io.modelcontextprotocol.kotlin.sdk.types.ServerCapabilities
import io.modelcontextprotocol.kotlin.sdk.types.Tool
import io.modelcontextprotocol.kotlin.sdk.types.ToolSchema
import io.modelcontextprotocol.kotlin.sdk.types.toJSON
Expand Down Expand Up @@ -433,6 +438,47 @@ class StreamableHttpServerTransportTest {
}
}

@Test
fun `GET SSE stream includes Mcp-Session-Id header and stays open`() = 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 GET SSE stream with session ID
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 ->
// Verify Mcp-Session-Id is present on the SSE response
response.status shouldBe HttpStatusCode.OK
response.headers[MCP_SESSION_ID_HEADER] shouldBe sessionId

// Verify the stream is alive by reading at least one line (flush event)
val channel = response.bodyAsChannel()
val firstLine = channel.readUTF8Line()
firstLine.shouldNotBeNull()
channel.isClosedForRead shouldBe false
}
}

private fun ApplicationTestBuilder.configureTransportEndpoint(transport: StreamableHttpServerTransport) {
application {
routing {
Expand Down
Loading