close stateless streamable http session after request completion (#872)
fixes #786 stateless streamable http endpoint never closed its per request session, so the session registry and notification subscriptions grew forever. now the session is closed in a finally block after each request, which triggers the existing cleanup. also removed the per request server.onClose logging callback that accumulated on shared servers. added tests proving sessions are removed after normal and rejected requests. jvmTest, ktlint, detekt, apiCheck all pass. no public api changes. Co-authored-by: Pavel Gorgulov <devcrocod@gmail.com>
This commit is contained in:
parent
8f104268c7
commit
dbd34ff327
2 changed files with 110 additions and 4 deletions
|
|
@ -380,11 +380,17 @@ private suspend fun RoutingContext.mcpStatelessStreamableHttpEndpoint(
|
|||
logger.info { "New stateless StreamableHttp connection established without sessionId" }
|
||||
|
||||
val server = block()
|
||||
server.onClose { logger.info { "Server connection closed without sessionId" } }
|
||||
server.createSession(transport)
|
||||
val session = server.createSession(transport)
|
||||
|
||||
transport.handleRequest(null, this.call)
|
||||
logger.debug { "Server connected to transport without sessionId" }
|
||||
try {
|
||||
transport.handleRequest(null, this.call)
|
||||
logger.debug { "Server connected to transport without sessionId" }
|
||||
} finally {
|
||||
// A stateless session serves exactly one request: close it so the server's
|
||||
// session registry and notification subscriptions do not grow unboundedly.
|
||||
session.close()
|
||||
logger.debug { "Stateless session closed after request completion" }
|
||||
}
|
||||
}
|
||||
|
||||
private suspend fun RoutingContext.mcpPostEndpoint(transportManager: TransportManager<SseServerTransport>) {
|
||||
|
|
|
|||
|
|
@ -0,0 +1,100 @@
|
|||
package io.modelcontextprotocol.kotlin.sdk.server
|
||||
|
||||
import io.kotest.assertions.nondeterministic.eventually
|
||||
import io.kotest.matchers.maps.shouldBeEmpty
|
||||
import io.kotest.matchers.shouldBe
|
||||
import io.ktor.client.request.HttpRequestBuilder
|
||||
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.HttpStatusCode
|
||||
import io.ktor.http.contentType
|
||||
import io.ktor.server.testing.testApplication
|
||||
import io.modelcontextprotocol.kotlin.sdk.types.ClientCapabilities
|
||||
import io.modelcontextprotocol.kotlin.sdk.types.Implementation
|
||||
import io.modelcontextprotocol.kotlin.sdk.types.InitializeRequest
|
||||
import io.modelcontextprotocol.kotlin.sdk.types.InitializeRequestParams
|
||||
import io.modelcontextprotocol.kotlin.sdk.types.JSONRPCMessage
|
||||
import io.modelcontextprotocol.kotlin.sdk.types.LATEST_PROTOCOL_VERSION
|
||||
import io.modelcontextprotocol.kotlin.sdk.types.McpJson
|
||||
import io.modelcontextprotocol.kotlin.sdk.types.ServerCapabilities
|
||||
import io.modelcontextprotocol.kotlin.sdk.types.toJSON
|
||||
import kotlin.test.Test
|
||||
import kotlin.time.Duration.Companion.seconds
|
||||
|
||||
/**
|
||||
* Verifies that the stateless Streamable HTTP endpoint does not leak server sessions:
|
||||
* every session created to serve a single request must be closed and removed from the
|
||||
* session registry once that request completes (https://github.com/modelcontextprotocol/kotlin-sdk/issues/786).
|
||||
*/
|
||||
class StatelessStreamableHttpSessionLifecycleTest {
|
||||
|
||||
private fun testServer(): Server = Server(
|
||||
Implementation("test-server", "1.0"),
|
||||
ServerOptions(capabilities = ServerCapabilities()),
|
||||
)
|
||||
|
||||
@Test
|
||||
fun `stateless endpoint removes session after each request`() = testApplication {
|
||||
val server = testServer()
|
||||
application {
|
||||
mcpStatelessStreamableHttp { server }
|
||||
}
|
||||
|
||||
repeat(5) {
|
||||
val response = client.post("/mcp") {
|
||||
addStreamableHeaders()
|
||||
setBody(initializeRequestBody())
|
||||
}
|
||||
response.status shouldBe HttpStatusCode.OK
|
||||
}
|
||||
|
||||
eventually(5.seconds) {
|
||||
server.sessions.shouldBeEmpty()
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `stateless endpoint removes session when the request is rejected`() = testApplication {
|
||||
val server = testServer()
|
||||
application {
|
||||
mcpStatelessStreamableHttp { server }
|
||||
}
|
||||
|
||||
// Missing "text/event-stream" in Accept makes the transport reject the request.
|
||||
val response = client.post("/mcp") {
|
||||
header(HttpHeaders.Host, "localhost")
|
||||
header(HttpHeaders.Accept, ContentType.Application.Json.toString())
|
||||
contentType(ContentType.Application.Json)
|
||||
setBody(initializeRequestBody())
|
||||
}
|
||||
response.status shouldBe HttpStatusCode.NotAcceptable
|
||||
|
||||
eventually(5.seconds) {
|
||||
server.sessions.shouldBeEmpty()
|
||||
}
|
||||
}
|
||||
|
||||
private fun HttpRequestBuilder.addStreamableHeaders() {
|
||||
header(HttpHeaders.Host, "localhost")
|
||||
header(
|
||||
HttpHeaders.Accept,
|
||||
listOf(ContentType.Application.Json, ContentType.Text.EventStream).joinToString(", "),
|
||||
)
|
||||
contentType(ContentType.Application.Json)
|
||||
}
|
||||
|
||||
private fun initializeRequestBody(): String {
|
||||
val request = InitializeRequest(
|
||||
InitializeRequestParams(
|
||||
protocolVersion = LATEST_PROTOCOL_VERSION,
|
||||
capabilities = ClientCapabilities(),
|
||||
clientInfo = Implementation(name = "test-client", version = "1.0.0"),
|
||||
),
|
||||
).toJSON()
|
||||
|
||||
return McpJson.encodeToString(JSONRPCMessage.serializer(), request)
|
||||
}
|
||||
}
|
||||
Loading…
Reference in a new issue