#355 Handle coroutine cancellation in Server Transports (#375)

This commit is contained in:
Konstantin Pavlov 2025-11-07 12:04:56 +02:00 • committed by GitHub
commit 07f661a94c
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 12 additions and 4 deletions

View file

@ -19,6 +19,7 @@ import kotlinx.atomicfu.atomic
import kotlinx.atomicfu.update
import kotlinx.collections.immutable.PersistentMap
import kotlinx.collections.immutable.toPersistentMap
import kotlinx.coroutines.awaitCancellation
private val logger = KotlinLogging.logger {}
@ -91,6 +92,8 @@ internal suspend fun ServerSSESession.mcpSseEndpoint(
server.createSession(transport)
logger.debug { "Server connected to transport for sessionId: ${transport.sessionId}" }
awaitCancellation()
}
internal fun ServerSSESession.mcpSseTransport(

View file

@ -11,9 +11,11 @@ import io.ktor.server.sse.ServerSSESession
import io.modelcontextprotocol.kotlin.sdk.JSONRPCMessage
import io.modelcontextprotocol.kotlin.sdk.shared.AbstractTransport
import io.modelcontextprotocol.kotlin.sdk.shared.McpJson
import kotlinx.coroutines.InternalCoroutinesApi
import kotlinx.coroutines.job
import kotlin.concurrent.atomics.AtomicBoolean
import kotlin.concurrent.atomics.ExperimentalAtomicApi
import kotlin.coroutines.cancellation.CancellationException
import kotlin.uuid.ExperimentalUuidApi
import kotlin.uuid.Uuid
@ -53,10 +55,13 @@ public class SseServerTransport(private val endpoint: String, private val sessio
data = "${endpoint.encodeURLPath()}?$SESSION_ID_PARAM=$sessionId",
)
try {
session.coroutineContext.job.join()
} finally {
_onClose.invoke()
@OptIn(InternalCoroutinesApi::class)
session.coroutineContext.job.invokeOnCompletion {
if (it != null && it !is CancellationException) {
_onError.invoke(it)
} else {
_onClose.invoke()
}
}
}