From fe97b442f6a146fbb6b508178087571af76427cf Mon Sep 17 00:00:00 2001 From: Anastasiia Zarechneva <49490937+aozherelyeva@users.noreply.github.com> Date: Fri, 20 Feb 2026 16:48:06 +0100 Subject: [PATCH] Some more Streamable HTTP client/server transport tests (#551) --- .../client/StreamableHttpClientTransport.kt | 4 + .../http/StreamableHttpClientTransportTest.kt | 255 ++++++++++++++++++ .../StreamableHttpServerTransportTest.kt | 192 ++++++++++++- 3 files changed, 440 insertions(+), 11 deletions(-) diff --git a/kotlin-sdk-client/src/commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/client/StreamableHttpClientTransport.kt b/kotlin-sdk-client/src/commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/client/StreamableHttpClientTransport.kt index efa5832..a3b4190 100644 --- a/kotlin-sdk-client/src/commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/client/StreamableHttpClientTransport.kt +++ b/kotlin-sdk-client/src/commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/client/StreamableHttpClientTransport.kt @@ -332,6 +332,10 @@ public class StreamableHttpClientTransport( throw it } } + if (eventName == "error") { + _onError(StreamableHttpError(null, data)) + return + } } @Suppress("LoopWithTooManyJumpStatements") diff --git a/kotlin-sdk-client/src/commonTest/kotlin/io/modelcontextprotocol/kotlin/sdk/client/streamable/http/StreamableHttpClientTransportTest.kt b/kotlin-sdk-client/src/commonTest/kotlin/io/modelcontextprotocol/kotlin/sdk/client/streamable/http/StreamableHttpClientTransportTest.kt index 4696184..d509813 100644 --- a/kotlin-sdk-client/src/commonTest/kotlin/io/modelcontextprotocol/kotlin/sdk/client/streamable/http/StreamableHttpClientTransportTest.kt +++ b/kotlin-sdk-client/src/commonTest/kotlin/io/modelcontextprotocol/kotlin/sdk/client/streamable/http/StreamableHttpClientTransportTest.kt @@ -1,6 +1,9 @@ package io.modelcontextprotocol.kotlin.sdk.client.streamable.http +import io.kotest.assertions.withClue +import io.kotest.matchers.collections.shouldHaveSize import io.kotest.matchers.shouldBe +import io.kotest.matchers.types.shouldBeInstanceOf import io.ktor.client.HttpClient import io.ktor.client.engine.mock.MockEngine import io.ktor.client.engine.mock.MockRequestHandler @@ -22,6 +25,7 @@ import io.modelcontextprotocol.kotlin.sdk.types.JSONRPCRequest import io.modelcontextprotocol.kotlin.sdk.types.McpException import io.modelcontextprotocol.kotlin.sdk.types.McpJson import io.modelcontextprotocol.kotlin.sdk.types.RPCError +import kotlinx.coroutines.CompletableDeferred import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.delay import kotlinx.coroutines.test.runTest @@ -51,6 +55,14 @@ class StreamableHttpClientTransportTest { return StreamableHttpClientTransport(httpClient, url = "http://localhost:8080/mcp") } + private fun buildSseMessage(id: String, method: String, params: String): String = buildString { + appendLine("event: message") + appendLine("id: $id") + val data = """{"jsonrpc":"2.0","method":"$method","params":$params}""" + appendLine("data: $data") + appendLine() + } + @Test fun testSendJsonRpcMessage() = runTest { val message = JSONRPCRequest( @@ -79,6 +91,42 @@ class StreamableHttpClientTransportTest { transport.close() } + @Test + fun testStartTransportTwoTimesThrowsException() = runTest { + val transport = createTransport { _ -> + respond( + content = "", + status = HttpStatusCode.Accepted, + ) + } + + transport.start() + try { + transport.start() + } catch (e: IllegalStateException) { + e shouldBe IllegalStateException("Can't change state: expected transport state New, but found Operational.") + } + transport.close() + } + + @Test + fun testCloseTransportTwoTimesIsPossible() = runTest { + val transport = createTransport { _ -> + respond( + content = "", + status = HttpStatusCode.Accepted, + ) + } + + transport.start() + transport.close() + try { + transport.close() + } catch (e: Exception) { + withClue("We expect no exceptions when closing already closed transport") { e shouldBe null } + } + } + @Test fun testStoreSessionId() = runTest { val initMessage = JSONRPCRequest( @@ -428,4 +476,211 @@ class StreamableHttpClientTransportTest { transport.close() } } + + @Test + fun testInlineSSEInResponse() = runTest { + val transport = createTransport { request -> + if (request.method == HttpMethod.Post) { + val sseContent = buildString { + append( + buildSseMessage( + id = "1", + method = "notifications/progress", + params = """{"progressToken":"task-1","progress":50}""", + ), + ) + append( + buildSseMessage( + id = "2", + method = "notifications/tools/list_changed", + params = "{}", + ), + ) + } + + respond( + content = ByteReadChannel(sseContent), + status = HttpStatusCode.OK, + headers = headersOf( + HttpHeaders.ContentType, + ContentType.Text.EventStream.toString(), + ), + ) + } else { + respond("", HttpStatusCode.OK) + } + } + + val receivedMessages = mutableListOf() + val twoMessagesReceived = CompletableDeferred() + + transport.onMessage { message -> + receivedMessages.add(message) + if (receivedMessages.size >= 2 && !twoMessagesReceived.isCompleted) { + twoMessagesReceived.complete(Unit) + } + } + + transport.start() + + transport.send( + JSONRPCRequest( + id = "test-1", + method = "test", + params = buildJsonObject { }, + ), + ) + + withTimeout(2.seconds) { + twoMessagesReceived.await() + } + + receivedMessages.size shouldBe 2 + + val firstNotification = receivedMessages[0] as JSONRPCNotification + firstNotification.method shouldBe "notifications/progress" + + val secondNotification = receivedMessages[1] as JSONRPCNotification + secondNotification.method shouldBe "notifications/tools/list_changed" + + transport.close() + } + + @Test + fun testErrorInSSEResponse() = runTest { + val errorMessage = "Something went wrong on the server" + val transport = createTransport { request -> + if (request.method == HttpMethod.Post) { + val sseContent = buildString { + appendLine("event: error") + appendLine("data: $errorMessage") + appendLine() + } + + respond( + content = ByteReadChannel(sseContent), + status = HttpStatusCode.OK, + headers = headersOf( + HttpHeaders.ContentType, + ContentType.Text.EventStream.toString(), + ), + ) + } else { + respond("", HttpStatusCode.OK) + } + } + + val receivedErrors = mutableListOf() + val errorDeferred = CompletableDeferred() + + transport.onError { error -> + receivedErrors.add(error) + if (!errorDeferred.isCompleted) { + errorDeferred.complete(error) + } + } + + transport.start() + + transport.send( + JSONRPCRequest( + id = "test-error", + method = "test", + params = buildJsonObject { }, + ), + ) + + val error = withTimeout(2.seconds) { errorDeferred.await() } + + receivedErrors.size shouldBe 1 + error.message shouldBe "Streamable HTTP error: $errorMessage" + + transport.close() + } + + @Test + fun testCloseTransportDuringEventStream() = runTest { + val transport = createTransport { request -> + if (request.method == HttpMethod.Post) { + val sseContent = buildString { + append( + buildSseMessage( + id = "1", + method = "notifications/progress", + params = """{"progressToken":"task-1","progress":25}""", + ), + ) + append( + buildSseMessage( + id = "2", + method = "notifications/progress", + params = """{"progressToken":"task-1","progress":50}""", + ), + ) + append( + buildSseMessage( + id = "3", + method = "notifications/progress", + params = """{"progressToken":"task-1","progress":75}""", + ), + ) + } + respond( + content = ByteReadChannel(sseContent), + status = HttpStatusCode.OK, + headers = headersOf( + HttpHeaders.ContentType, + ContentType.Text.EventStream.toString(), + ), + ) + } else { + respond("", HttpStatusCode.OK) + } + } + + val (receivedMessages, receivedErrors) = setupTransportAndCollectMessages(transport) + + withTimeout(2.seconds) { + while (receivedMessages.isEmpty()) { + delay(10) + } + } + + transport.close() + + assertTrue(receivedMessages.isNotEmpty()) + + receivedMessages.forEach { message -> + message.shouldBeInstanceOf() + message.method shouldBe "notifications/progress" + } + receivedErrors shouldHaveSize 0 + } + + private suspend fun setupTransportAndCollectMessages( + transport: StreamableHttpClientTransport, + ): Pair, MutableList> { + val receivedMessages = mutableListOf() + val receivedErrors = mutableListOf() + + transport.onMessage { message -> + receivedMessages.add(message) + } + + transport.onError { error -> + receivedErrors.add(error) + } + + transport.start() + + transport.send( + JSONRPCRequest( + id = "test-close", + method = "test", + params = buildJsonObject { }, + ), + ) + + return receivedMessages to receivedErrors + } } 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 968c057..c6e539a 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 @@ -1,6 +1,6 @@ package io.modelcontextprotocol.kotlin.sdk.server -import io.kotest.matchers.collections.shouldContain +import io.kotest.matchers.collections.shouldContainAll import io.kotest.matchers.equals.shouldBeEqual import io.kotest.matchers.shouldBe import io.ktor.client.HttpClient @@ -42,6 +42,8 @@ import io.modelcontextprotocol.kotlin.sdk.types.toJSON import kotlinx.serialization.builtins.ListSerializer import kotlinx.serialization.json.buildJsonObject import kotlinx.serialization.json.put +import org.junit.jupiter.params.ParameterizedTest +import org.junit.jupiter.params.provider.MethodSource import java.util.concurrent.atomic.AtomicBoolean import kotlin.test.Test import kotlin.test.assertEquals @@ -51,6 +53,18 @@ import io.ktor.client.plugins.contentnegotiation.ContentNegotiation as ClientCon import io.ktor.server.plugins.contentnegotiation.ContentNegotiation as ServerContentNegotiation class StreamableHttpServerTransportTest { + companion object { + @JvmStatic + @MethodSource + fun invalidPayloads() = listOf( + "", + " ", + " \n \t ", + null, + "lolol", + ) + } + private val path = "/transport" @Test @@ -114,7 +128,86 @@ class StreamableHttpServerTransportTest { } @Test - fun `notifications only payload responds with 202 Accepted`() = testApplication { + fun `second initialization request returns an HTTP error`() = testApplication { + configTestServer() + + val client = createTestClient() + + val transport = StreamableHttpServerTransport(enableJsonResponse = true) + transport.onMessage { message -> + if (message is JSONRPCRequest) { + transport.send(JSONRPCResponse(message.id, EmptyResult())) + } + } + + configureTransportEndpoint(transport) + + val payload = buildInitializeRequestPayload() + + val firstResponse = client.post(path) { + addStreamableHeaders() + setBody(payload) + } + + firstResponse.status shouldBe HttpStatusCode.OK + + val secondResponse = client.post(path) { + addStreamableHeaders() + header("mcp-session-id", firstResponse.headers[MCP_SESSION_ID_HEADER]) + setBody(payload) + } + + secondResponse.status shouldBe HttpStatusCode.BadRequest + } + + @Test + fun `request with unsupported protocol version returns an HTTP error`() = testApplication { + configTestServer() + + val client = createTestClient() + + val transport = StreamableHttpServerTransport(enableJsonResponse = true) + transport.onMessage { message -> + if (message is JSONRPCRequest) { + transport.send(JSONRPCResponse(message.id, EmptyResult())) + } + } + + configureTransportEndpoint(transport) + + val initPayload = buildInitializeRequestPayload() + val initResponse = client.post(path) { + addStreamableHeaders() + setBody(initPayload) + } + + initResponse.status shouldBe HttpStatusCode.OK + val sessionId = initResponse.headers[MCP_SESSION_ID_HEADER] + assertNotNull(sessionId) + + // TODO When https://github.com/modelcontextprotocol/kotlin-sdk/issues/547 is fixed, + // check the incompatible mcp-protocol-version in the InitializeRequest and delete the part below + val response = client.post(path) { + addStreamableHeaders() + header("mcp-session-id", sessionId) + header("mcp-protocol-version", "1900-01-01") + setBody( + encodeMessages( + listOf( + JSONRPCRequest( + id = RequestId("test-1"), + method = Method.Defined.ToolsList.value, + ), + ), + ), + ) + } + + response.status shouldBe HttpStatusCode.BadRequest + } + + @Test + fun `notifications only payload is accepted`() = testApplication { configTestServer() val client = createTestClient() @@ -202,15 +295,92 @@ class StreamableHttpServerTransportTest { assertEquals(HttpStatusCode.OK, response.status) - val responses = response.body>().map { it.result } - responses shouldContain (firstResult) - responses shouldContain (secondResult) - // TODO(check order) -// assertEquals(listOf(firstRequest.id, secondRequest.id), responses.map { it.id }) -// val firstMeta = (responses[0].result as EmptyResult).meta -// val secondMeta = (responses[1].result as EmptyResult).meta -// assertEquals("first", firstMeta?.get("label")?.jsonPrimitive?.content) -// assertEquals("second", secondMeta?.get("label")?.jsonPrimitive?.content) + val responses = response.body>() + val results = responses.map { it.result } + results.shouldContainAll(firstResult, secondResult) + + // Check responses' order + + // TODO Uncomment when fixed https://github.com/modelcontextprotocol/kotlin-sdk/issues/548 + /*assertEquals(listOf(firstRequest.id, secondRequest.id), responses.map { it.id }) + val firstMeta = (responses[0] as ListToolsResult).meta + val secondMeta = (responses[1] as ListResourcesResult).meta + assertEquals("first", firstMeta?.get("label")?.jsonPrimitive?.content) + assertEquals("second", secondMeta?.get("label")?.jsonPrimitive?.content)*/ + } + + @ParameterizedTest + @MethodSource("invalidPayloads") + fun `POST with a null or empty body returns an error`(payload: String?) = testApplication { + configTestServer() + + val client = createTestClient() + + val transport = StreamableHttpServerTransport(enableJsonResponse = true) + transport.onMessage { message -> + if (message is JSONRPCRequest) { + transport.send(JSONRPCResponse(message.id, EmptyResult())) + } + } + + configureTransportEndpoint(transport) + + val response = client.post(path) { + addStreamableHeaders() + setBody(payload) + } + + response.status shouldBe HttpStatusCode.BadRequest + } + + @Test + fun `POST with payload at max size is accepted`() = testApplication { + configTestServer() + + val client = createTestClient() + + val transport = StreamableHttpServerTransport(enableJsonResponse = true) + transport.onMessage { message -> + if (message is JSONRPCRequest) { + transport.send(JSONRPCResponse(message.id, EmptyResult())) + } + } + + configureTransportEndpoint(transport) + + val maxSizePayload = "x".repeat(4 * 1024 * 1024) + + val response = client.post(path) { + addStreamableHeaders() + setBody(maxSizePayload) + } + + response.status shouldBe HttpStatusCode.BadRequest + } + + @Test + fun `POST with oversized body returns an error`() = testApplication { + configTestServer() + + val client = createTestClient() + + val transport = StreamableHttpServerTransport(enableJsonResponse = true) + transport.onMessage { message -> + if (message is JSONRPCRequest) { + transport.send(JSONRPCResponse(message.id, EmptyResult())) + } + } + + configureTransportEndpoint(transport) + + val oversizedPayload = "x".repeat(4 * 1024 * 1024 + 1) + + val response = client.post(path) { + addStreamableHeaders() + setBody(oversizedPayload) + } + + response.status shouldBe HttpStatusCode.PayloadTooLarge } private fun ApplicationTestBuilder.configureTransportEndpoint(transport: StreamableHttpServerTransport) {