fix(server): notifications not delivered over Streamable HTTP (#587)

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.
 

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
 

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")))
     }
 

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

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