diff --git a/kotlin-sdk-server/src/commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/server/KtorServer.kt b/kotlin-sdk-server/src/commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/server/KtorServer.kt index 5fc3d22..8220a8b 100644 --- a/kotlin-sdk-server/src/commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/server/KtorServer.kt +++ b/kotlin-sdk-server/src/commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/server/KtorServer.kt @@ -1,6 +1,7 @@ package io.modelcontextprotocol.kotlin.sdk.server import io.github.oshai.kotlinlogging.KotlinLogging +import io.ktor.http.HttpHeaders import io.ktor.http.HttpMethod import io.ktor.http.HttpStatusCode import io.ktor.server.application.Application @@ -282,18 +283,10 @@ private fun Application.mcpStatelessStreamableHttp( ) } get { - call.reject( - HttpStatusCode.MethodNotAllowed, - RPCError.ErrorCode.CONNECTION_CLOSED, - "Method not allowed.", - ) + call.rejectUnsupportedMethod() } delete { - call.reject( - HttpStatusCode.MethodNotAllowed, - RPCError.ErrorCode.CONNECTION_CLOSED, - "Method not allowed.", - ) + call.rejectUnsupportedMethod() } } } @@ -419,6 +412,16 @@ private suspend fun RoutingContext.mcpPostEndpoint(transportManager: TransportMa logger.trace { "Message handled for sessionId: $sessionId" } } +/** A stateless endpoint serves POST only, offering neither an SSE stream to open nor a session to delete. */ +private suspend fun ApplicationCall.rejectUnsupportedMethod() { + response.header(HttpHeaders.Allow, HttpMethod.Post.value) + reject( + HttpStatusCode.MethodNotAllowed, + RPCError.ErrorCode.CONNECTION_CLOSED, + "Method not allowed.", + ) +} + private fun ApplicationRequest.sessionId(): String? = header(MCP_SESSION_ID_HEADER) private suspend fun existingStreamableTransport( diff --git a/kotlin-sdk-server/src/commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/server/StreamableHttpServerTransport.kt b/kotlin-sdk-server/src/commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/server/StreamableHttpServerTransport.kt index 3b02835..15d0b4c 100644 --- a/kotlin-sdk-server/src/commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/server/StreamableHttpServerTransport.kt +++ b/kotlin-sdk-server/src/commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/server/StreamableHttpServerTransport.kt @@ -10,7 +10,7 @@ import io.ktor.server.request.header import io.ktor.server.request.httpMethod import io.ktor.server.response.header import io.ktor.server.response.respond -import io.ktor.server.response.respondNullable +import io.ktor.server.response.respondText import io.ktor.server.sse.ServerSSESession import io.ktor.util.collections.ConcurrentMap import io.modelcontextprotocol.kotlin.sdk.shared.AbstractTransport @@ -474,12 +474,13 @@ public class StreamableHttpServerTransport(private val configuration: Configurat ) return } - if (!validateProtocolVersion(call)) return + if (!validateProtocolVersion(call, initializationRequest.id)) return if (messages.size > 1) { call.reject( HttpStatusCode.BadRequest, RPCError.ErrorCode.INVALID_REQUEST, "Invalid Request: Only one initialization request is allowed", + initializationRequest.id, ) return } @@ -491,12 +492,13 @@ public class StreamableHttpServerTransport(private val configuration: Configurat sessionId?.let { onSessionInitialized?.invoke(it) } } } else { - if (!validateSession(call) || !validateProtocolVersion(call)) return + val requestId = messages.filterIsInstance().firstOrNull()?.id + if (!validateSession(call, requestId) || !validateProtocolVersion(call, requestId)) return } val hasRequest = messages.any { it is JSONRPCRequest } if (!hasRequest) { - call.respondNullable(status = HttpStatusCode.Accepted, message = null) + call.respond(HttpStatusCode.Accepted) 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. @@ -552,7 +554,7 @@ public class StreamableHttpServerTransport(private val configuration: Configurat 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) + call.respond(HttpStatusCode.Accepted) } else { call.response.header(HttpHeaders.ContentType, ContentType.Application.Json.toString()) val payload = if (responses.size == 1) responses.first() else responses @@ -645,7 +647,7 @@ public class StreamableHttpServerTransport(private val configuration: Configurat if (!validateSession(call) || !validateProtocolVersion(call)) return sessionId?.let { onSessionClosed?.invoke(it) } close() - call.respondNullable(status = HttpStatusCode.OK, message = null) + call.respond(HttpStatusCode.OK) } /** @@ -734,7 +736,7 @@ public class StreamableHttpServerTransport(private val configuration: Configurat } } - private suspend fun validateSession(call: ApplicationCall): Boolean { + private suspend fun validateSession(call: ApplicationCall, id: RequestId? = null): Boolean { if (sessionIdGenerator == null) return true if (!initialized.load()) { @@ -742,6 +744,7 @@ public class StreamableHttpServerTransport(private val configuration: Configurat HttpStatusCode.BadRequest, RPCError.ErrorCode.CONNECTION_CLOSED, "Bad Request: Server not initialized", + id, ) return false } @@ -753,6 +756,7 @@ public class StreamableHttpServerTransport(private val configuration: Configurat HttpStatusCode.BadRequest, RPCError.ErrorCode.CONNECTION_CLOSED, "Bad Request: Mcp-Session-Id header is required", + id, ) return false } @@ -765,13 +769,14 @@ public class StreamableHttpServerTransport(private val configuration: Configurat HttpStatusCode.NotFound, REQUEST_TIMEOUT, "Session not found", + id, ) false } } } - private suspend fun validateProtocolVersion(call: ApplicationCall): Boolean { + private suspend fun validateProtocolVersion(call: ApplicationCall, id: RequestId? = null): Boolean { val version = call.request.headers[MCP_PROTOCOL_VERSION_HEADER] ?: return true return when (version) { @@ -784,6 +789,7 @@ public class StreamableHttpServerTransport(private val configuration: Configurat ", ", ) })", + id, ) false } @@ -914,12 +920,13 @@ public class StreamableHttpServerTransport(private val configuration: Configurat } } +/** Writes an already serialized body so a client accepting only `text/event-stream` still receives [status]. */ internal suspend fun ApplicationCall.reject( status: HttpStatusCode, code: Int, message: String, id: RequestId? = null, ) { - this.response.status(status) - this.respond(JSONRPCError(id = id, error = RPCError(code = code, message = message))) + val error = JSONRPCError(id = id, error = RPCError(code = code, message = message)) + respondText(McpJson.encodeToString(error), ContentType.Application.Json, status) } diff --git a/kotlin-sdk-server/src/jvmTest/kotlin/io/modelcontextprotocol/kotlin/sdk/server/StreamableHttpResponseDeliveryTest.kt b/kotlin-sdk-server/src/jvmTest/kotlin/io/modelcontextprotocol/kotlin/sdk/server/StreamableHttpResponseDeliveryTest.kt new file mode 100644 index 0000000..fc076eb --- /dev/null +++ b/kotlin-sdk-server/src/jvmTest/kotlin/io/modelcontextprotocol/kotlin/sdk/server/StreamableHttpResponseDeliveryTest.kt @@ -0,0 +1,99 @@ +package io.modelcontextprotocol.kotlin.sdk.server + +import io.kotest.matchers.shouldBe +import io.ktor.client.request.delete +import io.ktor.client.request.get +import io.ktor.client.request.header +import io.ktor.client.request.post +import io.ktor.client.request.setBody +import io.ktor.client.statement.HttpResponse +import io.ktor.client.statement.bodyAsText +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.testing.testApplication +import io.modelcontextprotocol.kotlin.sdk.types.Implementation +import io.modelcontextprotocol.kotlin.sdk.types.JSONRPCError +import io.modelcontextprotocol.kotlin.sdk.types.McpJson +import io.modelcontextprotocol.kotlin.sdk.types.RPCError +import io.modelcontextprotocol.kotlin.sdk.types.ServerCapabilities +import kotlin.test.Test + +/** + * A client opening the standalone stream sends `Accept: text/event-stream`, which matches no JSON + * converter. Responses rendered through ContentNegotiation used to reach such a client as an empty + * 406 with the intended status discarded. + */ +class StreamableHttpResponseDeliveryTest { + + private val eventStream = ContentType.Text.EventStream.toString() + + private fun testServer(): Server = Server( + Implementation("test-server", "1.0"), + ServerOptions(capabilities = ServerCapabilities()), + ) + + @Test + fun `stateless GET is rejected with 405 and advertises POST`() = testApplication { + application { mcpStatelessStreamableHttp { testServer() } } + + val response = client.get("/mcp") { + header(HttpHeaders.Host, "localhost") + header(HttpHeaders.Accept, eventStream) + } + + response.assertMethodNotAllowed() + } + + @Test + fun `stateless DELETE is rejected with 405 and advertises POST`() = testApplication { + application { mcpStatelessStreamableHttp { testServer() } } + + val response = client.delete("/mcp") { + header(HttpHeaders.Host, "localhost") + header(HttpHeaders.Accept, eventStream) + } + + response.assertMethodNotAllowed() + } + + @Test + fun `stateful DELETE for an unknown session returns 404`() = testApplication { + application { mcpStreamableHttp { testServer() } } + + val response = client.delete("/mcp") { + header(HttpHeaders.Host, "localhost") + header(HttpHeaders.Accept, eventStream) + header(MCP_SESSION_ID_HEADER, "unknown-session") + } + + response.status shouldBe HttpStatusCode.NotFound + response.decodeError().error.message shouldBe "Session not found" + } + + @Test + fun `POST carrying no request is acknowledged with a bodiless 202`() = testApplication { + application { mcpStatelessStreamableHttp { testServer() } } + + val response = client.post("/mcp") { + header(HttpHeaders.Host, "localhost") + header(HttpHeaders.Accept, "${ContentType.Application.Json}, $eventStream") + contentType(ContentType.Application.Json) + setBody("""{"jsonrpc":"2.0","method":"notifications/initialized"}""") + } + + response.status shouldBe HttpStatusCode.Accepted + response.bodyAsText() shouldBe "" + } + + private suspend fun HttpResponse.assertMethodNotAllowed() { + status shouldBe HttpStatusCode.MethodNotAllowed + headers[HttpHeaders.Allow] shouldBe HttpMethod.Post.value + contentType()?.withoutParameters() shouldBe ContentType.Application.Json + decodeError().error.code shouldBe RPCError.ErrorCode.CONNECTION_CLOSED + } + + private suspend fun HttpResponse.decodeError(): JSONRPCError = McpJson.decodeFromString(bodyAsText()) +} diff --git a/kotlin-sdk-server/src/jvmTest/kotlin/io/modelcontextprotocol/kotlin/sdk/server/StreamableHttpServerTransportTest.kt b/kotlin-sdk-server/src/jvmTest/kotlin/io/modelcontextprotocol/kotlin/sdk/server/StreamableHttpServerTransportTest.kt index bb522d6..e5c51e9 100644 --- a/kotlin-sdk-server/src/jvmTest/kotlin/io/modelcontextprotocol/kotlin/sdk/server/StreamableHttpServerTransportTest.kt +++ b/kotlin-sdk-server/src/jvmTest/kotlin/io/modelcontextprotocol/kotlin/sdk/server/StreamableHttpServerTransportTest.kt @@ -216,14 +216,16 @@ class StreamableHttpServerTransportTest { configureTransportEndpoint(transport) + val payload = buildInitializeRequestPayload() val initResponse = client.post(path) { addStreamableHeaders() header("mcp-protocol-version", "1900-01-01") - setBody(buildInitializeRequestPayload()) + setBody(payload) } initResponse.status shouldBe HttpStatusCode.BadRequest initResponse.headers[MCP_SESSION_ID_HEADER] shouldBe null + initResponse.body().id shouldBe payload.id } @Test @@ -297,6 +299,65 @@ class StreamableHttpServerTransportTest { } response.status shouldBe HttpStatusCode.BadRequest + response.body().id shouldBe RequestId("test-1") + } + + @Test + fun `batch with an initialization request echoes the initialize id when rejected`() = testApplication { + configTestServer() + + val client = createTestClient() + + val transport = StreamableHttpServerTransport(enableJsonResponse = true) + transport.onMessage { } + + configureTransportEndpoint(transport) + + val initPayload = buildInitializeRequestPayload() + val response = client.post(path) { + addStreamableHeaders() + setBody( + encodeMessages( + listOf( + initPayload, + JSONRPCRequest(id = RequestId("extra"), method = Method.Defined.ToolsList.value), + ), + ), + ) + } + + response.status shouldBe HttpStatusCode.BadRequest + val error = response.body() + error.error.message shouldBe "Invalid Request: Only one initialization request is allowed" + error.id shouldBe initPayload.id + } + + @Test + fun `non-init request before initialization echoes the request id`() = testApplication { + configTestServer() + + val client = createTestClient() + + val transport = StreamableHttpServerTransport(enableJsonResponse = true) + transport.onMessage { } + + configureTransportEndpoint(transport) + + val response = client.post(path) { + addStreamableHeaders() + setBody( + encodeMessages( + listOf( + JSONRPCRequest(id = RequestId("before-init"), method = Method.Defined.ToolsList.value), + ), + ), + ) + } + + response.status shouldBe HttpStatusCode.BadRequest + val error = response.body() + error.error.message shouldBe "Bad Request: Server not initialized" + error.id shouldBe RequestId("before-init") } @Test