fix(server): expire idle Streamable HTTP sessions and cap their count

README.md

@@ -803,6 +803,48 @@ embeddedServer(CIO, port = 3000) {
 -->
 <!--- KNIT example-server-routes-01.kt -->
 
+**Session limits:** `mcpStreamableHttp()` keeps every session in memory until the client deletes it, the server closes
+it, or it expires. A session expires after `sessionIdleTimeout` (30 minutes by default) without requests; its client
+then receives `404 Not Found` and has to initialize a new session. An open GET stream does not keep a session alive
+(expiry closes the stream too), so a client that only listens for notifications should send `ping` requests. At most
+`maxSessions` sessions (10,000 by default) are open at once; while that many are open, a request that would open another
+one receives `503 Service Unavailable`. Any request other than `initialize` that carries no `Mcp-Session-Id`, a plain
+`GET /mcp` included, receives `400 Bad Request`, so point health checks at a separate route. Size both limits to your
+traffic:
+
+<!--- CLEAR -->
+<!--- INCLUDE 
+import io.ktor.server.cio.CIO
+import io.ktor.server.engine.embeddedServer
+import io.modelcontextprotocol.kotlin.sdk.server.Server
+import io.modelcontextprotocol.kotlin.sdk.server.ServerOptions
+import io.modelcontextprotocol.kotlin.sdk.server.mcpStreamableHttp
+import io.modelcontextprotocol.kotlin.sdk.types.Implementation
+import io.modelcontextprotocol.kotlin.sdk.types.ServerCapabilities
+import kotlin.time.Duration.Companion.minutes
+
+private class MyServer :
+    Server(
+        serverInfo = Implementation(name = "ExampleServer", version = "1.0"),
+        options = ServerOptions(capabilities = ServerCapabilities()),
+    )
+
+fun main() {
+-->
+
+```kotlin
+embeddedServer(CIO, port = 3000) {
+    mcpStreamableHttp(sessionIdleTimeout = 10.minutes, maxSessions = 1_000) {
+        MyServer()
+    }
+}.start(wait = true)
+```
+
+<!--- SUFFIX 
+}
+-->
+<!--- KNIT example-server-routes-02.kt -->
+
 **CORS for browser-based clients (e.g. MCP Inspector):** if you connect from a browser-based
 client you need to install the Ktor CORS plugin so that MCP-specific headers are allowed and exposed:
 
@@ -869,7 +911,7 @@ embeddedServer(CIO, port = 3000) {
 <!--- SUFFIX 
 }
 -->
-<!--- KNIT example-server-routes-02.kt -->
+<!--- KNIT example-server-routes-03.kt -->
 
 Prefer Streamable HTTP for new projects.
 

conformance-test/run-conformance.sh

@@ -44,7 +44,7 @@ start_server() {
     echo "Waiting for server to be ready..."
     local retries=0
     local max_retries=30
-    while ! curl -sf "$SERVER_URL" > /dev/null 2>&1; do
+    while ! curl -s -o /dev/null "$SERVER_URL" 2>&1; do
         retries=$((retries + 1))
         if [ "$retries" -ge "$max_retries" ]; then
             echo "ERROR: Server failed to start after $max_retries attempts"

integration-test/src/jvmTest/kotlin/io/modelcontextprotocol/kotlin/sdk/integration/ClientDisconnectIntegrationTest.kt

@@ -0,0 +1,239 @@
+package io.modelcontextprotocol.kotlin.sdk.integration
+
+import io.kotest.assertions.nondeterministic.eventually
+import io.kotest.matchers.maps.shouldBeEmpty
+import io.kotest.matchers.maps.shouldHaveSize
+import io.kotest.matchers.nulls.shouldNotBeNull
+import io.kotest.matchers.shouldBe
+import io.kotest.matchers.string.shouldStartWith
+import io.ktor.client.HttpClient
+import io.ktor.client.request.header
+import io.ktor.client.request.post
+import io.ktor.client.request.setBody
+import io.ktor.http.ContentType
+import io.ktor.http.HttpHeaders
+import io.ktor.http.HttpMethod
+import io.ktor.http.HttpStatusCode
+import io.ktor.http.contentType
+import io.ktor.server.application.Application
+import io.ktor.server.application.createApplicationPlugin
+import io.ktor.server.application.install
+import io.ktor.server.engine.embeddedServer
+import io.ktor.server.request.httpMethod
+import io.ktor.server.routing.get
+import io.ktor.server.routing.routing
+import io.modelcontextprotocol.kotlin.sdk.server.Server
+import io.modelcontextprotocol.kotlin.sdk.server.ServerOptions
+import io.modelcontextprotocol.kotlin.sdk.server.mcp
+import io.modelcontextprotocol.kotlin.sdk.server.mcpStreamableHttp
+import io.modelcontextprotocol.kotlin.sdk.types.Implementation
+import io.modelcontextprotocol.kotlin.sdk.types.ServerCapabilities
+import kotlinx.coroutines.CancellationException
+import kotlinx.coroutines.CompletableDeferred
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.awaitCancellation
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.job
+import kotlinx.coroutines.runBlocking
+import kotlinx.coroutines.withContext
+import kotlinx.coroutines.withTimeout
+import java.net.Socket
+import java.util.concurrent.atomic.AtomicInteger
+import kotlin.test.Test
+import kotlin.time.Duration.Companion.seconds
+import io.ktor.client.engine.cio.CIO as ClientCIO
+import io.ktor.server.cio.CIO as ServerCIO
+import io.ktor.server.sse.SSE as ServerSSE
+
+/**
+ * Runs against a real CIO server, because what matters here is engine behavior: CIO never cancels a handler that is
+ * only waiting after its client disconnects, so a client that goes away without a word must not leave its session
+ * behind. The clients are raw sockets, so they can disconnect without saying anything.
+ */
+class ClientDisconnectIntegrationTest {
+
+    private fun testServer() = Server(Implementation("test-server", "1.0"), ServerOptions(ServerCapabilities()))
+
+    @Test
+    fun `an SSE session ends when its client disconnects`(): Unit = runBlocking {
+        val server = testServer()
+        withCioServer({
+            install(ServerSSE)
+            routing { mcp { server } }
+        }) { port ->
+            val sessionId = connect(port, LEGACY_SSE_REQUEST).use { socket ->
+                socket.readLineMatching { "sessionId=" in it }.substringAfter("sessionId=")
+            }
+
+            eventually(10.seconds) { server.sessions.shouldBeEmpty() }
+            HttpClient(ClientCIO).use { client ->
+                client.post("http://$HOST:$port/?sessionId=$sessionId") {
+                    contentType(ContentType.Application.Json)
+                    setBody("""{"jsonrpc":"2.0","id":1,"method":"ping"}""")
+                }.status shouldBe HttpStatusCode.NotFound
+            }
+        }
+    }
+
+    @Test
+    fun `an SSE handler ends when the server closes its session`(): Unit = runBlocking {
+        val server = testServer()
+        val finishedGets = AtomicInteger()
+        withCioServer({
+            countFinishedGets(finishedGets)
+            install(ServerSSE)
+            routing { mcp { server } }
+        }) { port ->
+            connect(port, LEGACY_SSE_REQUEST).use { socket ->
+                socket.readLineMatching { "sessionId=" in it }
+                eventually(10.seconds) { server.sessions shouldHaveSize 1 }
+
+                server.sessions.values.single().close()
+
+                // The client stays connected, so only the handler returning can finish the call.
+                eventually(10.seconds) { finishedGets.get() shouldBe 1 }
+            }
+        }
+    }
+
+    @Test
+    fun `an application route next to MCP runs to completion after its client disconnects`(): Unit = runBlocking {
+        val handlerStarted = CompletableDeferred<Unit>()
+        val handlerCompleted = CompletableDeferred<Boolean>()
+        withCioServer({
+            mcp { testServer() }
+            routing {
+                get("/slow") {
+                    handlerStarted.complete(Unit)
+                    try {
+                        delay(2.seconds)
+                        handlerCompleted.complete(true)
+                    } catch (e: CancellationException) {
+                        handlerCompleted.complete(false)
+                        throw e
+                    }
+                }
+            }
+        }) { port ->
+            connect(port, "GET /slow HTTP/1.1\r\nHost: localhost\r\n\r\n").use {
+                withTimeout(10.seconds) { handlerStarted.await() }
+            }
+
+            withTimeout(10.seconds) { handlerCompleted.await() } shouldBe true
+        }
+    }
+
+    @Test
+    fun `a GET stream ends as soon as its client disconnects`(): Unit = runBlocking {
+        val server = testServer()
+        val finishedGets = AtomicInteger()
+        withCioServer({
+            countFinishedGets(finishedGets)
+            mcpStreamableHttp { server }
+        }) { port ->
+            val sessionId = initializeSession(port)
+            val stream = openStream(port, sessionId)
+            withContext(Dispatchers.IO) { stream.close() }
+
+            // Long before the session could expire.
+            eventually(10.seconds) { finishedGets.get() shouldBe 1 }
+            server.sessions shouldHaveSize 1
+        }
+    }
+
+    @Test
+    fun `a POST whose client disconnects does not keep its session from expiring`(): Unit = runBlocking {
+        val toolStarted = CompletableDeferred<Unit>()
+        val server = Server(
+            Implementation("test-server", "1.0"),
+            ServerOptions(ServerCapabilities(tools = ServerCapabilities.Tools())),
+        ) {
+            addTool(name = "wait", description = "Never answers") {
+                toolStarted.complete(Unit)
+                awaitCancellation()
+            }
+        }
+        withCioServer({ mcpStreamableHttp(sessionIdleTimeout = 1.seconds) { server } }) { port ->
+            val sessionId = initializeSession(port)
+            val body = """{"jsonrpc":"2.0","id":2,"method":"tools/call","params":{"name":"wait","arguments":{}}}"""
+            val request = "POST /mcp HTTP/1.1\r\nHost: localhost\r\nAccept: application/json, text/event-stream\r\n" +
+                "Content-Type: application/json\r\nMcp-Session-Id: $sessionId\r\n" +
+                "Content-Length: ${body.toByteArray().size}\r\n\r\n$body"
+            connect(port, request).use {
+                withTimeout(10.seconds) { toolStarted.await() }
+            }
+
+            // The request is still unanswered, but with its client gone it no longer holds the session.
+            eventually(10.seconds) { server.sessions.shouldBeEmpty() }
+        }
+    }
+
+    private suspend fun initializeSession(port: Int): String = HttpClient(ClientCIO).use { client ->
+        val sessionId = client.post("http://$HOST:$port/mcp") {
+            header(HttpHeaders.Accept, "${ContentType.Application.Json}, ${ContentType.Text.EventStream}")
+            contentType(ContentType.Application.Json)
+            setBody(INITIALIZE_REQUEST)
+        }.headers["mcp-session-id"].shouldNotBeNull()
+        client.post("http://$HOST:$port/mcp") {
+            header(HttpHeaders.Accept, "${ContentType.Application.Json}, ${ContentType.Text.EventStream}")
+            header("mcp-session-id", sessionId)
+            contentType(ContentType.Application.Json)
+            setBody("""{"jsonrpc":"2.0","method":"notifications/initialized"}""")
+        }.status shouldBe HttpStatusCode.Accepted
+        sessionId
+    }
+
+    /** Opens the session's GET stream on a raw socket, which the caller can drop without a word. */
+    private suspend fun openStream(port: Int, sessionId: String): Socket {
+        val socket = connect(
+            port,
+            "GET /mcp HTTP/1.1\r\nHost: localhost\r\nAccept: text/event-stream\r\nMcp-Session-Id: $sessionId\r\n\r\n",
+        )
+        socket.readLineMatching { it.startsWith("HTTP/") } shouldStartWith "HTTP/1.1 200"
+        return socket
+    }
+
+    /** Opens a raw connection and sends [request] on it, so that the test can later drop it without a word. */
+    private suspend fun connect(port: Int, request: String): Socket = withContext(Dispatchers.IO) {
+        Socket(HOST, port).apply { getOutputStream().write(request.toByteArray()) }
+    }
+
+    /** Reads this connection until a line matches [predicate], and returns that line. */
+    private suspend fun Socket.readLineMatching(predicate: (String) -> Boolean): String = withContext(Dispatchers.IO) {
+        val lines = getInputStream().bufferedReader()
+        generateSequence { lines.readLine() }.first(predicate)
+    }
+
+    /**
+     * Counts the GET calls that have ended, whether their handler returned or they were cancelled. A cancelled call
+     * never finishes sending its response, so its end shows only as the completion of its job.
+     */
+    private fun Application.countFinishedGets(counter: AtomicInteger) {
+        install(
+            createApplicationPlugin("FinishedGets") {
+                onCall { call ->
+                    if (call.request.httpMethod == HttpMethod.Get) {
+                        call.coroutineContext.job.invokeOnCompletion { counter.incrementAndGet() }
+                    }
+                }
+            },
+        )
+    }
+
+    private suspend fun withCioServer(module: Application.() -> Unit, block: suspend (port: Int) -> Unit) {
+        val server = embeddedServer(ServerCIO, host = HOST, port = 0, module = module).start(wait = false)
+        try {
+            block(server.engine.resolvedConnectors().first().port)
+        } finally {
+            server.stop(0, 0)
+        }
+    }
+
+    private companion object {
+        const val HOST = "127.0.0.1"
+        const val LEGACY_SSE_REQUEST = "GET / HTTP/1.1\r\nHost: localhost\r\nAccept: text/event-stream\r\n\r\n"
+        const val INITIALIZE_REQUEST =
+            """{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2025-11-25",""" +
+                """"capabilities":{},"clientInfo":{"name":"test-client","version":"1.0"}}}"""
+    }
+}

integration-test/src/jvmTest/kotlin/io/modelcontextprotocol/kotlin/sdk/integration/streamablehttp/StreamableHttpClientTransportIntegrationTest.kt

@@ -17,14 +17,20 @@ import io.ktor.server.routing.post
 import io.ktor.server.routing.routing
 import io.modelcontextprotocol.kotlin.sdk.client.Client
 import io.modelcontextprotocol.kotlin.sdk.client.StreamableHttpClientTransport
+import io.modelcontextprotocol.kotlin.sdk.client.StreamableHttpError
+import io.modelcontextprotocol.kotlin.sdk.server.Server
+import io.modelcontextprotocol.kotlin.sdk.server.ServerOptions
+import io.modelcontextprotocol.kotlin.sdk.server.mcpStreamableHttp
 import io.modelcontextprotocol.kotlin.sdk.types.Implementation
 import io.modelcontextprotocol.kotlin.sdk.types.LoggingMessageNotification
 import io.modelcontextprotocol.kotlin.sdk.types.Method
+import io.modelcontextprotocol.kotlin.sdk.types.ServerCapabilities
 import io.modelcontextprotocol.kotlin.test.utils.actualPort
 import io.modelcontextprotocol.kotlin.test.utils.runIntegrationTest
 import kotlinx.coroutines.CompletableDeferred
 import kotlinx.coroutines.awaitCancellation
 import kotlinx.coroutines.channels.Channel
+import kotlinx.coroutines.withTimeout
 import kotlinx.serialization.json.Json
 import kotlinx.serialization.json.jsonObject
 import kotlinx.serialization.json.jsonPrimitive
@@ -96,6 +102,30 @@ class StreamableHttpClientTransportIntegrationTest {
         client.ping()
     }
 
+    @Test
+    fun `client reports an expired session through onError`() = runIntegrationTest(timeout = 20.seconds) {
+        val server = Server(Implementation("test-server", "1.0.0"), ServerOptions(ServerCapabilities()))
+        val ktorServer = embeddedServer(ServerCIO, host = HOST, port = 0) {
+            mcpStreamableHttp(sessionIdleTimeout = 1.seconds) { server }
+        }.startSuspend(wait = false)
+        val httpClient = HttpClient(ClientCIO) { install(SSE) }
+        val client = Client(Implementation("test-client", "1.0.0"))
+        try {
+            val url = "http://$HOST:${ktorServer.actualPort()}$MCP_PATH"
+            val transport = StreamableHttpClientTransport(httpClient, url)
+            val sessionNotFound = CompletableDeferred<StreamableHttpError>()
+            transport.onError { if (it is StreamableHttpError && it.code == 404) sessionNotFound.complete(it) }
+            client.connect(transport)
+
+            // The open GET stream does not keep the session alive, so its reconnect after expiry gets 404.
+            withTimeout(10.seconds) { sessionNotFound.await() }
+        } finally {
+            client.close()
+            httpClient.close()
+            ktorServer.stopSuspend(1000, 2000)
+        }
+    }
+
     private fun withConnectedClient(
         getRoute: Route.() -> Unit,
         configure: Client.() -> Unit = {},

kotlin-sdk-client/src/commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/client/StreamableHttpClientTransport.kt

@@ -90,6 +90,9 @@ private sealed interface ConnectResult {
  * Supports automatic SSE reconnection with exponential backoff, stream resumption via the
  * `Last-Event-ID` header, and explicit session termination.
  *
+ * A session the server has terminated answers `404`, reported through [onError] as a [StreamableHttpError].
+ * The transport does not start a new session; connect the client again with a new transport.
+ *
  * @param client Ktor HTTP client used for all requests
  * @param url MCP endpoint URL
  * @param reconnectionOptions reconnection backoff and retry-limit settings for the SSE stream
@@ -378,6 +381,13 @@ public class StreamableHttpClientTransport(
         val responseContentType = e.response?.contentType()
 
         return when {
+            responseStatus == HttpStatusCode.NotFound && sessionId != null -> {
+                val error = StreamableHttpError(responseStatus.value, "Session $sessionId not found, stream disabled")
+                logger.warn { error.message }
+                _onError(error)
+                true
+            }
+
             responseStatus == HttpStatusCode.NotFound || responseStatus == HttpStatusCode.MethodNotAllowed -> {
                 logger.info { "Server returned ${responseStatus.value} for GET/SSE, stream disabled." }
                 true

kotlin-sdk-server/api/kotlin-sdk-server.api

@@ -61,8 +61,8 @@ public final class io/modelcontextprotocol/kotlin/sdk/server/KtorServerKt {
 	public static final fun mcpStatelessStreamableHttp (Lio/ktor/server/application/Application;Ljava/lang/String;ZLjava/util/List;Ljava/util/List;Lkotlin/jvm/functions/Function1;)V
 	public static synthetic fun mcpStatelessStreamableHttp$default (Lio/ktor/server/application/Application;Ljava/lang/String;ZLjava/util/List;Ljava/util/List;Lio/modelcontextprotocol/kotlin/sdk/server/EventStore;Lkotlin/jvm/functions/Function1;ILjava/lang/Object;)V
 	public static synthetic fun mcpStatelessStreamableHttp$default (Lio/ktor/server/application/Application;Ljava/lang/String;ZLjava/util/List;Ljava/util/List;Lkotlin/jvm/functions/Function1;ILjava/lang/Object;)V
-	public static final fun mcpStreamableHttp (Lio/ktor/server/application/Application;Ljava/lang/String;ZLjava/util/List;Ljava/util/List;Lio/modelcontextprotocol/kotlin/sdk/server/EventStore;Lkotlin/jvm/functions/Function1;Lkotlin/jvm/functions/Function1;)V
-	public static synthetic fun mcpStreamableHttp$default (Lio/ktor/server/application/Application;Ljava/lang/String;ZLjava/util/List;Ljava/util/List;Lio/modelcontextprotocol/kotlin/sdk/server/EventStore;Lkotlin/jvm/functions/Function1;Lkotlin/jvm/functions/Function1;ILjava/lang/Object;)V
+	public static final fun mcpStreamableHttp-u4uj9nY (Lio/ktor/server/application/Application;Ljava/lang/String;ZLjava/util/List;Ljava/util/List;Lio/modelcontextprotocol/kotlin/sdk/server/EventStore;Lkotlin/jvm/functions/Function1;JILkotlin/jvm/functions/Function1;)V
+	public static synthetic fun mcpStreamableHttp-u4uj9nY$default (Lio/ktor/server/application/Application;Ljava/lang/String;ZLjava/util/List;Ljava/util/List;Lio/modelcontextprotocol/kotlin/sdk/server/EventStore;Lkotlin/jvm/functions/Function1;JILkotlin/jvm/functions/Function1;ILjava/lang/Object;)V
 }
 
 public final class io/modelcontextprotocol/kotlin/sdk/server/RegisteredPrompt : io/modelcontextprotocol/kotlin/sdk/server/Feature {