From 736efe9dc08ae083b568c7049aca511cafe6e7b8 Mon Sep 17 00:00:00 2001 From: Konstantin Pavlov <1517853+kpavlov@users.noreply.github.com> Date: Thu, 12 Mar 2026 16:52:27 +0200 Subject: [PATCH] fix(server): notifications not delivered over Streamable HTTP (#587) (#599) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Changes - fix(server): keep SSE connection open until explicitly cancelled - conformance: add logging to test_tool_with_logging and update troubleshooting documentation - chore(conformance): remove 5 passing server tests from conformance baselines: - tools-call-with-logging - tools-call-with-progress - tools-call-sampling - tools-call-elicitation - elicitation-sep1034-defaults ## Motivation and Context Two bugs in StreamableHttpServerTransport caused server-to-client notifications (e.g. notifications/message during tool calls) to be silently dropped. **Bug 1 — GET SSE stream closed immediately** `handleGetRequest` rejected GET requests with 405 when `enableJsonResponse = true` (which `mcpStreamableHttp` always sets). Since the Ktor `sse {}` handler commits response headers before the body runs, the reject failed silently and the function returned, causing Ktor to close the SSE stream. There was also no `awaitCancellation()` to keep the connection alive. **Bug 2 — Notifications discarded in JSON mode** In `send()`, notifications with a relatedRequestId entered the POST-stream path but hit `if (!isTerminated) return` without being forwarded anywhere. **Fix** - Remove the enableJsonResponse guard from handleGetRequest. The GET SSE stream is an always-required notification channel, orthogonal to how POST responses are delivered. - Add `awaitCancellation()` to keep the GET SSE connection open until the client disconnects or the transport closes. - In `send()`, route notifications with a relatedRequestId to the standalone GET SSE stream when `enableJsonResponse = true`. ## How Has This Been Tested? Verified against the MCP conformance test suite — the ToolsCallWithLogging scenario now passes. ngrep confirms the `GET /mcp` stream stays open and `notifications/message` events arrive before the `tools/call` response. ## Breaking Changes ## 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 - [ ] New and existing tests pass locally - [ ] I have added appropriate error handling - [ ] I have added or updated documentation as needed ## Additional context Fixes #587 Requires #585 to be merged first. --------- Co-authored-by: devcrocod --- conformance-test/README.md | 15 ++++------- conformance-test/conformance-baseline.yml | 5 ---- .../sdk/conformance/ConformanceTools.kt | 7 ++++++ .../main/resources/simplelogger.properties | 9 +++++++ .../server/StreamableHttpServerTransport.kt | 25 ++++++++++++------- 5 files changed, 37 insertions(+), 24 deletions(-) create mode 100644 conformance-test/src/main/resources/simplelogger.properties diff --git a/conformance-test/README.md b/conformance-test/README.md index d14fcfe..3f1e366 100644 --- a/conformance-test/README.md +++ b/conformance-test/README.md @@ -114,16 +114,11 @@ Tests the conformance server against all server scenarios: 8 scenarios are expected to fail due to current SDK limitations (tracked in [ `conformance-baseline.yml`](conformance-baseline.yml). -| Scenario | Suite | Root Cause | -|---------------------------------------|--------|--------------------------------------------------------------------------------------------------------------------------------------------------------| -| `tools-call-with-logging` | server | Notifications from tool handlers have no `relatedRequestId`; transport routes them to the standalone SSE stream instead of the request-specific stream | -| `tools-call-with-progress` | server | *(same as above)* | -| `tools-call-sampling` | server | *(same as above)* | -| `tools-call-elicitation` | server | *(same as above)* | -| `elicitation-sep1034-defaults` | server | *(same as above)* | -| `elicitation-sep1330-enums` | server | *(same as above)* | -| `resources-templates-read` | server | SDK does not implement `addResourceTemplate()` with URI pattern matching; resources are looked up by exact URI | -| `elicitation-sep1034-client-defaults` | client | SDK does not fill in `default` values from the elicitation request schema before sending the response | +| Scenario | Suite | Root Cause | +|---------------------------------------|--------|----------------------------------------------------------------------------------------------------------------| +| `elicitation-sep1330-enums` | server | *(same as above)* | +| `resources-templates-read` | server | SDK does not implement `addResourceTemplate()` with URI pattern matching; resources are looked up by exact URI | +| `elicitation-sep1034-client-defaults` | client | SDK does not fill in `default` values from the elicitation request schema before sending the response | These failures reveal SDK gaps and are intentionally not fixed in this module. diff --git a/conformance-test/conformance-baseline.yml b/conformance-test/conformance-baseline.yml index cc06a38..c15041f 100644 --- a/conformance-test/conformance-baseline.yml +++ b/conformance-test/conformance-baseline.yml @@ -1,11 +1,6 @@ # Conformance test baseline - expected failures # Add entries here as tests are identified as known SDK limitations server: - - tools-call-with-logging - - tools-call-with-progress - - tools-call-sampling - - tools-call-elicitation - - elicitation-sep1034-defaults - elicitation-sep1330-enums - resources-templates-read diff --git a/conformance-test/src/main/kotlin/io/modelcontextprotocol/kotlin/sdk/conformance/ConformanceTools.kt b/conformance-test/src/main/kotlin/io/modelcontextprotocol/kotlin/sdk/conformance/ConformanceTools.kt index ebe7a57..def59f8 100644 --- a/conformance-test/src/main/kotlin/io/modelcontextprotocol/kotlin/sdk/conformance/ConformanceTools.kt +++ b/conformance-test/src/main/kotlin/io/modelcontextprotocol/kotlin/sdk/conformance/ConformanceTools.kt @@ -1,5 +1,6 @@ package io.modelcontextprotocol.kotlin.sdk.conformance +import io.github.oshai.kotlinlogging.KotlinLogging import io.modelcontextprotocol.kotlin.sdk.server.Server import io.modelcontextprotocol.kotlin.sdk.types.AudioContent import io.modelcontextprotocol.kotlin.sdk.types.CallToolResult @@ -33,6 +34,8 @@ internal const val PNG_BASE64 = // Minimal WAV (base64) internal const val WAV_BASE64 = "UklGRiYAAABXQVZFZm10IBAAAAABAAEAQB8AAAB9AAACABAAZGF0YQIAAAA=" +private val logger = KotlinLogging.logger {} + @Suppress("LongMethod") fun Server.registerConformanceTools() { // 1. Simple text @@ -429,6 +432,7 @@ fun Server.registerConformanceTools() { name = "test_tool_with_logging", description = "test_tool_with_logging", ) { + logger.debug { "[test_tool_with_logging] Sending message 1" } sendLoggingMessage( LoggingMessageNotification( LoggingMessageNotificationParams( @@ -439,6 +443,7 @@ fun Server.registerConformanceTools() { ), ) delay(50.milliseconds) + logger.debug { "[test_tool_with_logging] Sending message #2" } sendLoggingMessage( LoggingMessageNotification( LoggingMessageNotificationParams( @@ -449,6 +454,7 @@ fun Server.registerConformanceTools() { ), ) delay(50.milliseconds) + logger.debug { "[test_tool_with_logging] Sending message 3" } sendLoggingMessage( LoggingMessageNotification( LoggingMessageNotificationParams( @@ -458,6 +464,7 @@ fun Server.registerConformanceTools() { ), ), ) + CallToolResult(listOf(TextContent("Simple text content"))) } diff --git a/conformance-test/src/main/resources/simplelogger.properties b/conformance-test/src/main/resources/simplelogger.properties new file mode 100644 index 0000000..16141d3 --- /dev/null +++ b/conformance-test/src/main/resources/simplelogger.properties @@ -0,0 +1,9 @@ +# Level of logging for the ROOT logger: ERROR, WARN, INFO, DEBUG, TRACE (default is INFO) +org.slf4j.simpleLogger.defaultLogLevel=INFO +org.slf4j.simpleLogger.showThreadName=true +org.slf4j.simpleLogger.showDateTime=false + +# Log level for specific packages or classes +org.slf4j.simpleLogger.log.io.ktor.server=DEBUG +org.slf4j.simpleLogger.log.io.modelcontextprotocol=DEBUG +org.slf4j.simpleLogger.log.io.modelcontextprotocol.kotlin.sdk.conformance=INFO diff --git a/kotlin-sdk-server/src/commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/server/StreamableHttpServerTransport.kt b/kotlin-sdk-server/src/commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/server/StreamableHttpServerTransport.kt index 9074c4c..d0bbcfb 100644 --- a/kotlin-sdk-server/src/commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/server/StreamableHttpServerTransport.kt +++ b/kotlin-sdk-server/src/commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/server/StreamableHttpServerTransport.kt @@ -28,6 +28,7 @@ import io.modelcontextprotocol.kotlin.sdk.types.RPCError import io.modelcontextprotocol.kotlin.sdk.types.RPCError.ErrorCode.REQUEST_TIMEOUT import io.modelcontextprotocol.kotlin.sdk.types.RequestId import io.modelcontextprotocol.kotlin.sdk.types.SUPPORTED_PROTOCOL_VERSIONS +import kotlinx.coroutines.awaitCancellation import kotlinx.coroutines.job import kotlinx.coroutines.sync.Mutex import kotlinx.coroutines.sync.withLock @@ -242,7 +243,15 @@ public class StreamableHttpServerTransport(private val configuration: Configurat } val isTerminated = message is JSONRPCResponse || message is JSONRPCError - if (!isTerminated) return + if (!isTerminated) { + if (configuration.enableJsonResponse) { + // In JSON response mode there is no per-request SSE stream, so route notifications + // that are logically associated with a request to the standalone GET SSE stream. + val standaloneStream = streamsMapping[STANDALONE_SSE_STREAM_ID] + standaloneStream?.let { emitOnStream(STANDALONE_SSE_STREAM_ID, it.session, message) } + } + return + } requestToResponseMapping[responseRequestId!!] = message val relatedIds = requestToStreamMapping.filterValues { it == streamId }.keys @@ -411,14 +420,9 @@ public class StreamableHttpServerTransport(private val configuration: Configurat @Suppress("ReturnCount") public suspend fun handleGetRequest(session: ServerSSESession?, call: ApplicationCall) { - if (configuration.enableJsonResponse) { - call.reject( - HttpStatusCode.MethodNotAllowed, - RPCError.ErrorCode.CONNECTION_CLOSED, - "Method not allowed.", - ) - return - } + // NOTE: enableJsonResponse only controls how POST responses are delivered (JSON vs. SSE). + // The standalone GET SSE stream is always supported — it is the only channel available + // for server-to-client notifications when enableJsonResponse = true. val sseSession = session ?: error("Server session can't be null for streaming GET requests") val acceptHeader = call.request.header(HttpHeaders.Accept) @@ -456,6 +460,9 @@ public class StreamableHttpServerTransport(private val configuration: Configurat 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() } public suspend fun handleDeleteRequest(call: ApplicationCall) {