feat(core): concurrent inbound dispatch behind an initialization gate (#884)

Make inbound message handling concurrent after the initialization
handshake, so a slow or peer-awaiting handler no longer blocks other
messages on the same connection

fixes #176, #572 

## Motivation and Context
Inbound messages were processed serially on the transport read loop. One
slow or peer-awaiting request handler (sampling, elicitation, roots)
blocked every later message on that connection, including the responses
and `notifications/cancelled` those handlers depend on which could
deadlock request/response flows.

## How Has This Been Tested?
New and old tests

## Breaking Changes
- after the handshake, inbound handlers may run concurrently (processing
stays serial during initialization). There is no opt-out flag by design,
concurrency is bounded by `maxConcurrentHandlers` /
`maxInFlightHandlers`.
- `ProtocolOptions` / `ClientOptions` / `ServerOptions` gain new
parameters with defaults
- `RequestHandlerExtra` is now a coroutine-context element and is no
longer user-constructible

## Types of changes
- [x] Bug fix (non-breaking change which fixes an issue)
- [x] New feature (non-breaking change which adds functionality)
- [x] 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
- [x] New and existing tests pass locally
- [x] I have added appropriate error handling
- [x] I have added or updated documentation as needed

kotlin-sdk-server/src/commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/server/Server.kt

@@ -53,9 +53,11 @@ import io.modelcontextprotocol.kotlin.sdk.utils.ResourceTemplateMatcher
 import io.modelcontextprotocol.kotlin.sdk.utils.ResourceTemplateMatcherFactory
 import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.Deferred
+import kotlinx.coroutines.Dispatchers
 import kotlinx.serialization.json.JsonObject
 import kotlinx.serialization.json.buildJsonObject
 import kotlinx.serialization.json.put
+import kotlin.coroutines.CoroutineContext
 import kotlin.jvm.JvmOverloads
 import kotlin.time.ExperimentalTime
 
@@ -69,12 +71,17 @@ private val logger = KotlinLogging.logger {}
  * Forwarded to [ProtocolOptions.enforceStrictCapabilities].
  * @property resourceTemplateMatcherFactory The factory used to create [ResourceTemplateMatcher] instances
  *   for matching resource URIs against registered templates. Defaults to [PathSegmentTemplateMatcher.factory].
+ * @param handlerCoroutineContext Coroutine context for inbound handlers. See [ProtocolOptions.handlerCoroutineContext].
  */
 public class ServerOptions(
     public val capabilities: ServerCapabilities,
     enforceStrictCapabilities: Boolean = true,
     public val resourceTemplateMatcherFactory: ResourceTemplateMatcherFactory = PathSegmentTemplateMatcher.factory,
-) : ProtocolOptions(enforceStrictCapabilities = enforceStrictCapabilities) {
+    handlerCoroutineContext: CoroutineContext = Dispatchers.Default,
+) : ProtocolOptions(
+    enforceStrictCapabilities = enforceStrictCapabilities,
+    handlerCoroutineContext = handlerCoroutineContext,
+) {
     @JvmOverloads
     public constructor(
         capabilities: ServerCapabilities,

kotlin-sdk-server/src/commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/server/ServerSession.kt

@@ -108,10 +108,18 @@ public open class ServerSession(
         }
     }
 
+    override fun onInitializedNotification() {
+        // Enable concurrency before the user's onInitialized callback, which runs synchronously.
+        enableConcurrentDispatch()
+    }
+
     internal val clientConnection: ClientConnection = ClientConnectionImpl(this)
 
     /**
      * Registers a callback to be invoked when the server has completed initialization.
+     *
+     * The callback must be synchronous and fast: it runs on the message-dispatch path for
+     * `notifications/initialized`, after concurrent dispatch has been enabled for the session.
      */
     public fun onInitialized(block: () -> Unit) {
         val old = _onInitialized

kotlin-sdk-server/src/commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/server/StreamableHttpServerTransport.kt

@@ -15,10 +15,12 @@ import io.ktor.server.sse.ServerSSESession
 import io.ktor.util.collections.ConcurrentMap
 import io.modelcontextprotocol.kotlin.sdk.shared.AbstractTransport
 import io.modelcontextprotocol.kotlin.sdk.shared.TransportSendOptions
+import io.modelcontextprotocol.kotlin.sdk.types.CancelledNotificationParams
 import io.modelcontextprotocol.kotlin.sdk.types.DEFAULT_NEGOTIATED_PROTOCOL_VERSION
 import io.modelcontextprotocol.kotlin.sdk.types.JSONRPCEmptyMessage
 import io.modelcontextprotocol.kotlin.sdk.types.JSONRPCError
 import io.modelcontextprotocol.kotlin.sdk.types.JSONRPCMessage
+import io.modelcontextprotocol.kotlin.sdk.types.JSONRPCNotification
 import io.modelcontextprotocol.kotlin.sdk.types.JSONRPCRequest
 import io.modelcontextprotocol.kotlin.sdk.types.JSONRPCResponse
 import io.modelcontextprotocol.kotlin.sdk.types.McpJson
@@ -27,6 +29,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.CompletableDeferred
 import kotlinx.coroutines.NonCancellable
 import kotlinx.coroutines.awaitCancellation
 import kotlinx.coroutines.job
@@ -175,6 +178,20 @@ public class StreamableHttpServerTransport(private val configuration: Configurat
     private val requestToStreamMapping: ConcurrentMap<RequestId, String> = ConcurrentMap()
     private val requestToResponseMapping: ConcurrentMap<RequestId, JSONRPCMessage> = ConcurrentMap()
 
+    /**
+     * JSON-response mode only: one entry per in-flight POST, completed by [send] with the collected
+     * responses once every request delivered by that POST has been answered. [handlePostRequest]
+     * awaits it so the HTTP exchange stays open while handlers run concurrently.
+     */
+    private val pendingJsonResponses: ConcurrentMap<String, CompletableDeferred<List<JSONRPCMessage>>> = ConcurrentMap()
+
+    /**
+     * JSON-response mode only: request ids retired by an inbound cancellation while their POST is
+     * still awaiting. Treated as settled-without-response so the POST's "a response for every id"
+     * completion condition is not blocked by a request that will never be answered.
+     */
+    private val cancelledRequestIds: ConcurrentMap<RequestId, Unit> = ConcurrentMap()
+
     private val sessionMutex = Mutex()
     private val streamMutex = Mutex()
 
@@ -241,54 +258,117 @@ public class StreamableHttpServerTransport(private val configuration: Configurat
             return
         }
 
+        val isTerminated = message is JSONRPCResponse || message is JSONRPCError
+
+        // In JSON mode request-related notifications have no per-request SSE stream. Route them to the
+        // standalone GET stream before the lookup below, so a notification for an already-retired
+        // request is delivered there instead of becoming a spurious error.
+        if (configuration.enableJsonResponse && !isTerminated) {
+            val standaloneStream = streamsMapping[STANDALONE_SSE_STREAM_ID]
+            standaloneStream?.let { emitOnStream(STANDALONE_SSE_STREAM_ID, it.session, message) }
+            return
+        }
+
         val streamId = requestToStreamMapping[routingRequestId]
-            ?: error("No connection established for request id $routingRequestId")
+        if (streamId == null) {
+            // Request already retired (completed, cancelled, or the JSON POST disconnected). In JSON
+            // mode a late terminal response is undeliverable and dropped. In SSE mode an unknown id
+            // is a real routing error.
+            if (configuration.enableJsonResponse) return
+            error("No connection established for request id $routingRequestId")
+        }
         val activeStream = streamsMapping[streamId]
 
         if (!configuration.enableJsonResponse) {
             activeStream?.let { stream ->
                 emitOnStream(streamId, stream.session, message)
             }
-        }
-
-        val isTerminated = message is JSONRPCResponse || message is JSONRPCError
-        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
+            if (!isTerminated) return
         }
 
         requestToResponseMapping[responseRequestId!!] = message
-        val relatedIds = requestToStreamMapping.filterValues { it == streamId }.keys
-
-        if (relatedIds.any { it !in requestToResponseMapping }) return
 
         streamMutex.withLock {
-            if (activeStream == null) error("No connection established for request ID: $routingRequestId")
+            // The request may have been retired (disconnect or cancellation cleanup) between recording
+            // the response above and acquiring the lock. Reclaim the stray mapping and drop it.
+            if (routingRequestId !in requestToStreamMapping) {
+                requestToResponseMapping.remove(routingRequestId)
+                return
+            }
+            // Recomputed under the lock. Handlers may respond concurrently, and the sender that
+            // observes the last settled request also retires the ids, so a losing sender sees an
+            // empty set. A cancelled request counts as settled even though it produces no response.
+            val relatedIds = requestToStreamMapping.filterValues { it == streamId }.keys
+            if (relatedIds.isEmpty() ||
+                relatedIds.any { it !in requestToResponseMapping && it !in cancelledRequestIds }
+            ) {
+                return
+            }
 
             if (configuration.enableJsonResponse) {
-                activeStream.call.response.header(HttpHeaders.ContentType, ContentType.Application.Json.toString())
-                sessionId?.let { activeStream.call.response.header(MCP_SESSION_ID_HEADER, it) }
-                val responses = relatedIds.mapNotNull { requestToResponseMapping[it] }
-                val payload = if (responses.size == 1) {
-                    responses.first()
-                } else {
-                    responses
+                // Hand the collected responses to the POST suspended in [handlePostRequest], which
+                // completes the HTTP exchange. A null pendingPost means the POST was already retired
+                // (disconnect or cancellation), so the responses are undeliverable and dropped quietly.
+                val pendingPost = pendingJsonResponses.remove(streamId)
+                if (pendingPost == null) {
+                    cleanUpRequests(relatedIds)
+                    return
                 }
-                activeStream.call.respond(payload)
+                val responses = relatedIds.mapNotNull { requestToResponseMapping[it] }
+                cleanUpRequests(relatedIds)
+                pendingPost.complete(responses)
             } else {
-                activeStream.session?.close()
+                // A null stream means the per-request SSE connection closed before the response
+                // settled, so the ids are retired rather than leaked.
+                activeStream?.session?.close()
+                cleanUpRequests(relatedIds)
             }
+        }
+    }
 
-            // Clean up
-            relatedIds.forEach { requestId ->
-                requestToResponseMapping.remove(requestId)
-                requestToStreamMapping.remove(requestId)
+    private fun cleanUpRequests(ids: Collection<RequestId>) {
+        ids.forEach { requestId ->
+            requestToResponseMapping.remove(requestId)
+            requestToStreamMapping.remove(requestId)
+            cancelledRequestIds.remove(requestId)
+        }
+    }
+
+    /**
+     * Scans a delivered POST body for `notifications/cancelled` and retires each referenced request
+     * that is awaiting on a JSON-mode POST. No-op unless [Configuration.enableJsonResponse].
+     */
+    private suspend fun retireCancelledJsonRequests(messages: List<JSONRPCMessage>) {
+        if (!configuration.enableJsonResponse) return
+        val cancelledIds = messages.asSequence()
+            .filterIsInstance<JSONRPCNotification>()
+            .filter { it.method == Method.Defined.NotificationsCancelled.value }
+            .mapNotNull { notification ->
+                val params = notification.params ?: return@mapNotNull null
+                runCatching {
+                    McpJson.decodeFromJsonElement<CancelledNotificationParams>(params).requestId
+                }.getOrNull()
             }
+            .toList()
+        cancelledIds.forEach { retireCancelledJsonRequest(it) }
+    }
+
+    /**
+     * Marks a request cancelled by the peer as settled without a response and completes its POST once
+     * every request that POST delivered is settled (answered or cancelled). Without this, a cancelled
+     * request leaves the POST awaiting a response that, by design, is never sent.
+     */
+    private suspend fun retireCancelledJsonRequest(requestId: RequestId) {
+        streamMutex.withLock {
+            val streamId = requestToStreamMapping[requestId] ?: return
+            cancelledRequestIds[requestId] = Unit
+            val relatedIds = requestToStreamMapping.filterValues { it == streamId }.keys
+            val allSettled = relatedIds.all { it in requestToResponseMapping || it in cancelledRequestIds }
+            if (!allSettled) return
+            val pendingPost = pendingJsonResponses.remove(streamId) ?: return
+            val responses = relatedIds.mapNotNull { requestToResponseMapping[it] }
+            cleanUpRequests(relatedIds)
+            pendingPost.complete(responses)
         }
     }
 
@@ -304,6 +384,10 @@ public class StreamableHttpServerTransport(private val configuration: Configurat
                 streamsMapping.clear()
                 requestToStreamMapping.clear()
                 requestToResponseMapping.clear()
+                cancelledRequestIds.clear()
+                // Release any POST calls suspended awaiting JSON responses.
+                pendingJsonResponses.values.forEach { it.cancel() }
+                pendingJsonResponses.clear()
                 invokeOnCloseCallback()
             }
         }
@@ -339,7 +423,12 @@ public class StreamableHttpServerTransport(private val configuration: Configurat
     }
 
     /**
-     * Handles POST requests containing JSON-RPC messages
+     * Handles POST requests containing JSON-RPC messages.
+     *
+     * In JSON-response mode this suspends until a response has been produced for every request in
+     * the POST body, then completes the HTTP exchange with the collected JSON (a single object for
+     * one request, an array for a batch). Bodies without requests are acknowledged with 202
+     * immediately. In SSE mode the responses are emitted onto the per-request SSE stream instead.
      */
     public suspend fun handlePostRequest(session: ServerSSESession?, call: ApplicationCall) {
         try {
@@ -407,6 +496,9 @@ public class StreamableHttpServerTransport(private val configuration: Configurat
             if (!hasRequest) {
                 call.respondNullable(status = HttpStatusCode.Accepted, message = null)
                 messages.forEach { message -> _onMessage(message) }
+                // A cancellation may target a request still awaiting delivery on a JSON-mode POST;
+                // retire it so that POST completes instead of waiting for a response that never comes.
+                retireCancelledJsonRequests(messages)
                 return
             }
 
@@ -429,13 +521,50 @@ public class StreamableHttpServerTransport(private val configuration: Configurat
                 maybeSendPrimingEvent(streamId, session, clientProtocolVersion)
             }
 
+            val pendingPost = if (configuration.enableJsonResponse) {
+                CompletableDeferred<List<JSONRPCMessage>>()
+            } else {
+                null
+            }
+
             streamMutex.withLock {
                 streamsMapping[streamId] = SessionContext(session, call)
                 messages.filterIsInstance<JSONRPCRequest>().forEach { requestToStreamMapping[it.id] = streamId }
+                pendingPost?.let { pendingJsonResponses[streamId] = it }
             }
             call.coroutineContext.job.invokeOnCompletion { streamsMapping.remove(streamId) }
 
-            messages.forEach { message -> _onMessage(message) }
+            if (pendingPost == null) {
+                // SSE mode: responses are emitted onto this call's SSE stream by send().
+                messages.forEach { message -> _onMessage(message) }
+                return
+            }
+
+            // JSON-response mode. Handlers may run concurrently, so returning from delivery no longer
+            // implies the responses were produced. Suspend until send() has collected a response for
+            // every request in this POST. A client disconnect or transport close cancels this call.
+            try {
+                messages.forEach { message -> _onMessage(message) }
+                retireCancelledJsonRequests(messages)
+                val responses = pendingPost.await()
+                sessionId?.let { call.response.header(MCP_SESSION_ID_HEADER, it) }
+                if (responses.isEmpty()) {
+                    // Every request in this POST was cancelled before producing a response; there is
+                    // nothing to return, so acknowledge with 202 rather than an empty JSON body.
+                    call.respondNullable(status = HttpStatusCode.Accepted, message = null)
+                } else {
+                    call.response.header(HttpHeaders.ContentType, ContentType.Application.Json.toString())
+                    val payload = if (responses.size == 1) responses.first() else responses
+                    call.respond(payload)
+                }
+            } finally {
+                // On cancellation or failure, retire this POST's ids so a late response is dropped
+                // quietly at the send() boundary instead of leaking mappings. Idempotent no-ops on
+                // the success path (send() or retirement already cleaned up).
+                pendingJsonResponses.remove(streamId)
+                pendingPost.cancel()
+                cleanUpRequests(messages.filterIsInstance<JSONRPCRequest>().map { it.id })
+            }
         } catch (e: CancellationException) {
             throw e
         } catch (e: Exception) {

kotlin-sdk-server/src/jvmTest/kotlin/io/modelcontextprotocol/kotlin/sdk/server/StreamableHttpServerTransportTest.kt

@@ -19,14 +19,19 @@ import io.ktor.http.HttpHeaders
 import io.ktor.http.HttpStatusCode
 import io.ktor.http.contentType
 import io.ktor.serialization.kotlinx.json.json
+import io.ktor.server.application.ApplicationCall
 import io.ktor.server.application.install
 import io.ktor.server.routing.post
 import io.ktor.server.routing.routing
+import io.ktor.server.sse.ServerSSESession
 import io.ktor.server.testing.ApplicationTestBuilder
 import io.ktor.server.testing.testApplication
+import io.ktor.sse.ServerSentEvent
 import io.ktor.utils.io.ByteReadChannel
 import io.ktor.utils.io.readLine
 import io.ktor.utils.io.readUTF8Line
+import io.modelcontextprotocol.kotlin.sdk.types.CancelledNotification
+import io.modelcontextprotocol.kotlin.sdk.types.CancelledNotificationParams
 import io.modelcontextprotocol.kotlin.sdk.types.ClientCapabilities
 import io.modelcontextprotocol.kotlin.sdk.types.EmptyResult
 import io.modelcontextprotocol.kotlin.sdk.types.Implementation
@@ -46,6 +51,16 @@ 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
+import kotlinx.coroutines.CompletableDeferred
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.SupervisorJob
+import kotlinx.coroutines.async
+import kotlinx.coroutines.cancel
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.test.runTest
+import kotlinx.coroutines.withTimeout
 import kotlinx.serialization.builtins.ListSerializer
 import kotlinx.serialization.json.buildJsonObject
 import kotlinx.serialization.json.put
@@ -53,11 +68,13 @@ import org.junit.jupiter.params.ParameterizedTest
 import org.junit.jupiter.params.provider.Arguments
 import org.junit.jupiter.params.provider.MethodSource
 import java.util.concurrent.atomic.AtomicBoolean
+import kotlin.coroutines.CoroutineContext
 import kotlin.test.Test
 import kotlin.test.assertEquals
 import kotlin.test.assertFailsWith
 import kotlin.test.assertFalse
 import kotlin.test.assertNotNull
+import kotlin.time.Duration.Companion.seconds
 import io.ktor.client.plugins.contentnegotiation.ContentNegotiation as ClientContentNegotiation
 import io.ktor.server.plugins.contentnegotiation.ContentNegotiation as ServerContentNegotiation
 
@@ -349,6 +366,306 @@ class StreamableHttpServerTransportTest {
         assertEquals("second", secondMeta?.get("label")?.jsonPrimitive?.content)*/
     }
 
+    @Test
+    fun `json response waits for a response produced after message delivery returns`() = testApplication {
+        configTestServer()
+
+        val client = createTestClient()
+
+        val transport = StreamableHttpServerTransport(
+            StreamableHttpServerTransport.Configuration(enableJsonResponse = true),
+        )
+        // Simulates concurrent dispatch: message delivery returns before the handler responds.
+        val handlerScope = CoroutineScope(SupervisorJob() + Dispatchers.Default)
+        try {
+            transport.onMessage { message ->
+                if (message is JSONRPCRequest) {
+                    if (message.method == Method.Defined.Initialize.value) {
+                        transport.send(JSONRPCResponse(message.id, EmptyResult()))
+                    } else {
+                        handlerScope.launch {
+                            delay(100)
+                            transport.send(JSONRPCResponse(message.id, EmptyResult()))
+                        }
+                    }
+                }
+            }
+
+            configureTransportEndpoint(transport)
+
+            val initResponse = client.post(path) {
+                addStreamableHeaders()
+                setBody(buildInitializeRequestPayload())
+            }
+            initResponse.status shouldBe HttpStatusCode.OK
+
+            val request = JSONRPCRequest(id = RequestId("async-1"), method = Method.Defined.ToolsList.value)
+            val response = withTimeout(10.seconds) {
+                client.post(path) {
+                    addStreamableHeaders()
+                    header(MCP_SESSION_ID_HEADER, initResponse.headers[MCP_SESSION_ID_HEADER])
+                    setBody(encodeMessages(listOf(request)))
+                }
+            }
+
+            response.status shouldBe HttpStatusCode.OK
+            response.body<JSONRPCResponse>() shouldBe JSONRPCResponse(request.id)
+        } finally {
+            handlerScope.cancel()
+        }
+    }
+
+    @Test
+    fun `json response for a batch waits until every asynchronously produced response is ready`() = testApplication {
+        configTestServer()
+
+        val client = createTestClient()
+
+        val transport = StreamableHttpServerTransport(
+            StreamableHttpServerTransport.Configuration(enableJsonResponse = true),
+        )
+        val firstRequest = JSONRPCRequest(id = RequestId("slow"), method = Method.Defined.ToolsList.value)
+        val secondRequest = JSONRPCRequest(id = RequestId("fast"), method = Method.Defined.ResourcesList.value)
+
+        // Simulates concurrent dispatch with out-of-order completion: the second request responds
+        // first; the POST must stay open until the slower first response settles too.
+        val handlerScope = CoroutineScope(SupervisorJob() + Dispatchers.Default)
+        try {
+            transport.onMessage { message ->
+                if (message is JSONRPCRequest) {
+                    if (message.method == Method.Defined.Initialize.value) {
+                        transport.send(JSONRPCResponse(message.id, EmptyResult()))
+                    } else {
+                        handlerScope.launch {
+                            delay(if (message.id == firstRequest.id) 200 else 50)
+                            transport.send(JSONRPCResponse(message.id, EmptyResult()))
+                        }
+                    }
+                }
+            }
+
+            configureTransportEndpoint(transport)
+
+            val initResponse = client.post(path) {
+                addStreamableHeaders()
+                setBody(buildInitializeRequestPayload())
+            }
+            initResponse.status shouldBe HttpStatusCode.OK
+
+            val response = withTimeout(10.seconds) {
+                client.post(path) {
+                    addStreamableHeaders()
+                    header(MCP_SESSION_ID_HEADER, initResponse.headers[MCP_SESSION_ID_HEADER])
+                    setBody(encodeMessages(listOf(firstRequest, secondRequest)))
+                }
+            }
+
+            response.status shouldBe HttpStatusCode.OK
+            val responses = response.body<List<JSONRPCResponse>>()
+            responses.map { it.id }.toSet() shouldBe setOf(firstRequest.id, secondRequest.id)
+        } finally {
+            handlerScope.cancel()
+        }
+    }
+
+    @Test
+    fun `json response POST completes with 202 when its only request is cancelled`() = testApplication {
+        configTestServer()
+
+        val client = createTestClient()
+
+        val transport = StreamableHttpServerTransport(
+            StreamableHttpServerTransport.Configuration(enableJsonResponse = true),
+        )
+        val requestDelivered = CompletableDeferred<Unit>()
+        // Concurrent dispatch: delivery returns before a response is produced. This request is never
+        // answered here, modelling a handler cancelled by notifications/cancelled.
+        transport.onMessage { message ->
+            if (message is JSONRPCRequest) {
+                if (message.method == Method.Defined.Initialize.value) {
+                    transport.send(JSONRPCResponse(message.id, EmptyResult()))
+                } else {
+                    requestDelivered.complete(Unit)
+                }
+            }
+        }
+
+        configureTransportEndpoint(transport)
+
+        val clientScope = CoroutineScope(SupervisorJob() + Dispatchers.Default)
+        try {
+            val initResponse = client.post(path) {
+                addStreamableHeaders()
+                setBody(buildInitializeRequestPayload())
+            }
+            initResponse.status shouldBe HttpStatusCode.OK
+            val sessionId = initResponse.headers[MCP_SESSION_ID_HEADER]
+
+            val request = JSONRPCRequest(id = RequestId("cancel-me"), method = Method.Defined.ToolsList.value)
+            val requestPost = clientScope.async {
+                client.post(path) {
+                    addStreamableHeaders()
+                    header(MCP_SESSION_ID_HEADER, sessionId)
+                    setBody(encodeMessages(listOf(request)))
+                }
+            }
+
+            // The request must be registered in-flight before it can be cancelled.
+            withTimeout(5.seconds) { requestDelivered.await() }
+
+            val cancellation = CancelledNotification(
+                CancelledNotificationParams(requestId = request.id, reason = "client cancelled"),
+            )
+            val cancelResponse = client.post(path) {
+                addStreamableHeaders()
+                header(MCP_SESSION_ID_HEADER, sessionId)
+                setBody(encodeMessages(listOf(cancellation.toJSON())))
+            }
+            cancelResponse.status shouldBe HttpStatusCode.Accepted
+
+            // Without retirement of the cancelled id this await never completes and the timeout fires.
+            val response = withTimeout(10.seconds) { requestPost.await() }
+            response.status shouldBe HttpStatusCode.Accepted
+        } finally {
+            clientScope.cancel()
+        }
+    }
+
+    @Test
+    fun `json response for a batch returns only the non-cancelled responses`() = testApplication {
+        configTestServer()
+
+        val client = createTestClient()
+
+        val transport = StreamableHttpServerTransport(
+            StreamableHttpServerTransport.Configuration(enableJsonResponse = true),
+        )
+        val answeredA = JSONRPCRequest(id = RequestId("a"), method = Method.Defined.ToolsList.value)
+        val answeredB = JSONRPCRequest(id = RequestId("b"), method = Method.Defined.ToolsList.value)
+        val cancelledC = JSONRPCRequest(id = RequestId("c"), method = Method.Defined.ToolsList.value)
+
+        val cancelledDelivered = CompletableDeferred<Unit>()
+        val handlerScope = CoroutineScope(SupervisorJob() + Dispatchers.Default)
+        try {
+            transport.onMessage { message ->
+                if (message is JSONRPCRequest) {
+                    when {
+                        message.method == Method.Defined.Initialize.value ->
+                            transport.send(JSONRPCResponse(message.id, EmptyResult()))
+
+                        message.id == answeredA.id || message.id == answeredB.id ->
+                            handlerScope.launch { transport.send(JSONRPCResponse(message.id, EmptyResult())) }
+
+                        message.id == cancelledC.id -> cancelledDelivered.complete(Unit)
+                    }
+                }
+            }
+
+            configureTransportEndpoint(transport)
+
+            val initResponse = client.post(path) {
+                addStreamableHeaders()
+                setBody(buildInitializeRequestPayload())
+            }
+            initResponse.status shouldBe HttpStatusCode.OK
+            val sessionId = initResponse.headers[MCP_SESSION_ID_HEADER]
+
+            val batchPost = handlerScope.async {
+                client.post(path) {
+                    addStreamableHeaders()
+                    header(MCP_SESSION_ID_HEADER, sessionId)
+                    setBody(encodeMessages(listOf(answeredA, answeredB, cancelledC)))
+                }
+            }
+
+            withTimeout(5.seconds) { cancelledDelivered.await() }
+
+            val cancellation = CancelledNotification(CancelledNotificationParams(requestId = cancelledC.id))
+            client.post(path) {
+                addStreamableHeaders()
+                header(MCP_SESSION_ID_HEADER, sessionId)
+                setBody(encodeMessages(listOf(cancellation.toJSON())))
+            }.status shouldBe HttpStatusCode.Accepted
+
+            val response = withTimeout(10.seconds) { batchPost.await() }
+            response.status shouldBe HttpStatusCode.OK
+            val responses = response.body<List<JSONRPCResponse>>()
+            responses.map { it.id }.toSet() shouldBe setOf(answeredA.id, answeredB.id)
+        } finally {
+            handlerScope.cancel()
+        }
+    }
+
+    @Test
+    fun `json response send for a retired request id is dropped without error`() = runTest {
+        val transport = StreamableHttpServerTransport(enableJsonResponse = true)
+        // No stream is mapped for this id (already retired by disconnect/cancellation cleanup);
+        // a late terminal response must be dropped quietly, not raise "No connection established".
+        transport.send(JSONRPCResponse(RequestId("retired"), EmptyResult()))
+    }
+
+    @Test
+    fun `sse response send for an unknown request id still errors`() = runTest {
+        val transport = StreamableHttpServerTransport(enableJsonResponse = false)
+        assertFailsWith<IllegalStateException> {
+            transport.send(JSONRPCResponse(RequestId("unknown"), EmptyResult()))
+        }
+    }
+
+    @Test
+    fun `sse terminal response is dropped without leaking when the per-request stream is gone`() = testApplication {
+        configTestServer()
+        val client = createTestClient()
+
+        val transport = StreamableHttpServerTransport(
+            StreamableHttpServerTransport.Configuration(enableJsonResponse = false),
+        )
+        transport.setSessionIdGenerator(null) // stateless: accept requests without an init handshake
+        val requestId = RequestId("held-request")
+        val delivered = CompletableDeferred<Unit>()
+        val release = CompletableDeferred<Unit>()
+        transport.onMessage { message ->
+            if (message is JSONRPCRequest) {
+                delivered.complete(Unit)
+                release.await() // keep the POST and its stream mapping alive until the test releases it
+            }
+        }
+
+        application {
+            routing {
+                post(path) {
+                    transport.handlePostRequest(FakeServerSSESession(call, call.coroutineContext), call)
+                }
+            }
+        }
+
+        val scope = CoroutineScope(SupervisorJob() + Dispatchers.Default)
+        try {
+            val post = scope.async {
+                client.post(path) {
+                    addStreamableHeaders()
+                    setBody(
+                        encodeMessages(
+                            listOf(JSONRPCRequest(id = requestId, method = Method.Defined.ToolsList.value)),
+                        ),
+                    )
+                }
+            }
+            withTimeout(5.seconds) { delivered.await() }
+
+            // Drop the per-request SSE stream while the request is still mapped in-flight; the handler's
+            // terminal response is now undeliverable and must be dropped quietly rather than raise
+            // "No connection established" (which would also strand the request mappings).
+            transport.closeSseStream(requestId)
+            transport.send(JSONRPCResponse(requestId, EmptyResult()))
+
+            post.cancel()
+        } finally {
+            release.complete(Unit)
+            scope.cancel()
+        }
+    }
+
     @ParameterizedTest
     @MethodSource("invalidPayloads")
     fun `POST with a null or empty body returns an error`(payload: String?) = testApplication {
@@ -694,3 +1011,12 @@ class StreamableHttpServerTransportTest {
         }
     }
 }
+
+/** No-op [ServerSSESession] double that lets a POST route drive [StreamableHttpServerTransport] in SSE mode. */
+private class FakeServerSSESession(
+    override val call: ApplicationCall,
+    override val coroutineContext: CoroutineContext,
+) : ServerSSESession {
+    override suspend fun send(event: ServerSentEvent) {}
+    override suspend fun close() {}
+}