From def0e925e258a8d1ff833aeaf091cbb9e576d2a3 Mon Sep 17 00:00:00 2001 From: johny <69754601+jskjw157@users.noreply.github.com> Date: Thu, 5 Feb 2026 21:49:47 +0900 Subject: [PATCH] [WIP]test(integration): Add Streamable HTTP transport integration tests (#486) ## Summary Add integration tests for Prompts, Resources, and Tools over Streamable HTTP transport as outlined in #183. ## Changes - [x] Infrastructure: Enable `STREAMABLE_HTTP` transport in test base - [x] Prompts: Add functional integration tests - [x] Resources: Add functional integration tests - [x] Tools: Add functional integration tests Related to #332 --------- Co-authored-by: Konstantin Pavlov <1517853+kpavlov@users.noreply.github.com> --- integration-test/build.gradle.kts | 2 + .../sdk/integration/kotlin/KotlinTestBase.kt | 166 +++++++++++++----- .../PromptIntegrationTestStreamableHttp.kt | 7 + .../ResourceIntegrationTestStreamableHttp.kt | 7 + .../ToolIntegrationTestStreamableHttp.kt | 7 + .../AbstractKotlinClientTsServerTest.kt | 8 +- .../sdk/integration/typescript/TsTestBase.kt | 28 ++- .../typescript/TypeScriptServer.kt | 2 +- .../stdio/KotlinClientTsServerTestStdio.kt | 15 +- .../jvmTest/typescript/client/stdio-client.ts | 8 +- .../kotlin/sdk/client/MockMcp.kt | 2 +- 11 files changed, 185 insertions(+), 67 deletions(-) create mode 100644 integration-test/src/jvmTest/kotlin/io/modelcontextprotocol/kotlin/sdk/integration/kotlin/streamablehttp/PromptIntegrationTestStreamableHttp.kt create mode 100644 integration-test/src/jvmTest/kotlin/io/modelcontextprotocol/kotlin/sdk/integration/kotlin/streamablehttp/ResourceIntegrationTestStreamableHttp.kt create mode 100644 integration-test/src/jvmTest/kotlin/io/modelcontextprotocol/kotlin/sdk/integration/kotlin/streamablehttp/ToolIntegrationTestStreamableHttp.kt diff --git a/integration-test/build.gradle.kts b/integration-test/build.gradle.kts index d5b8ef8..1b763ed 100644 --- a/integration-test/build.gradle.kts +++ b/integration-test/build.gradle.kts @@ -21,6 +21,8 @@ kotlin { implementation(libs.ktor.server.cio) implementation(libs.ktor.server.websockets) implementation(libs.ktor.server.test.host) + implementation(libs.ktor.server.content.negotiation) + implementation(libs.ktor.serialization) } } jvmTest { diff --git a/integration-test/src/jvmTest/kotlin/io/modelcontextprotocol/kotlin/sdk/integration/kotlin/KotlinTestBase.kt b/integration-test/src/jvmTest/kotlin/io/modelcontextprotocol/kotlin/sdk/integration/kotlin/KotlinTestBase.kt index 43b1117..1a8ffe0 100644 --- a/integration-test/src/jvmTest/kotlin/io/modelcontextprotocol/kotlin/sdk/integration/kotlin/KotlinTestBase.kt +++ b/integration-test/src/jvmTest/kotlin/io/modelcontextprotocol/kotlin/sdk/integration/kotlin/KotlinTestBase.kt @@ -3,18 +3,27 @@ package io.modelcontextprotocol.kotlin.sdk.integration.kotlin import io.ktor.client.HttpClient import io.ktor.client.engine.cio.CIO import io.ktor.client.plugins.sse.SSE +import io.ktor.serialization.kotlinx.json.json import io.ktor.server.application.install import io.ktor.server.engine.EmbeddedServer import io.ktor.server.engine.embeddedServer +import io.ktor.server.plugins.contentnegotiation.ContentNegotiation +import io.ktor.server.routing.delete +import io.ktor.server.routing.get +import io.ktor.server.routing.post +import io.ktor.server.routing.route import io.ktor.server.routing.routing import io.modelcontextprotocol.kotlin.sdk.client.Client import io.modelcontextprotocol.kotlin.sdk.client.SseClientTransport import io.modelcontextprotocol.kotlin.sdk.client.StdioClientTransport +import io.modelcontextprotocol.kotlin.sdk.client.StreamableHttpClientTransport import io.modelcontextprotocol.kotlin.sdk.server.Server import io.modelcontextprotocol.kotlin.sdk.server.ServerOptions import io.modelcontextprotocol.kotlin.sdk.server.StdioServerTransport +import io.modelcontextprotocol.kotlin.sdk.server.StreamableHttpServerTransport import io.modelcontextprotocol.kotlin.sdk.server.mcp import io.modelcontextprotocol.kotlin.sdk.types.Implementation +import io.modelcontextprotocol.kotlin.sdk.types.McpJson import io.modelcontextprotocol.kotlin.sdk.types.ServerCapabilities import io.modelcontextprotocol.kotlin.test.utils.Retry import kotlinx.coroutines.runBlocking @@ -44,7 +53,7 @@ abstract class KotlinTestBase { protected lateinit var serverEngine: EmbeddedServer<*, *> // Transport selection - protected enum class TransportKind { SSE, STDIO } + protected enum class TransportKind { SSE, STDIO, STREAMABLE_HTTP } protected open val transportKind: TransportKind = TransportKind.STDIO // STDIO-specific fields @@ -52,13 +61,16 @@ abstract class KotlinTestBase { private var stdioClientInput: Source? = null private var stdioClientOutput: Sink? = null + // StreamableHTTP-specific fields + private var streamableHttpServerTransport: StreamableHttpServerTransport? = null + protected abstract fun configureServerCapabilities(): ServerCapabilities protected abstract fun configureServer() @BeforeEach fun setUp() { setupServer() - if (transportKind == TransportKind.SSE) { + if (transportKind == TransportKind.SSE || transportKind == TransportKind.STREAMABLE_HTTP) { await .ignoreExceptions() .until { @@ -98,6 +110,19 @@ abstract class KotlinTestBase { ) client.connect(transport) } + + TransportKind.STREAMABLE_HTTP -> { + val transport = StreamableHttpClientTransport( + client = HttpClient(CIO) { + install(SSE) + }, + url = "http://$host:$port/mcp", + ) + client = Client( + Implementation("test", "1.0"), + ) + client.connect(transport) + } } } @@ -111,35 +136,77 @@ abstract class KotlinTestBase { configureServer() - if (transportKind == TransportKind.SSE) { - serverEngine = embeddedServer(ServerCIO, host = host, port = port) { - install(ServerSSE) - routing { - mcp { server } + when (transportKind) { + TransportKind.SSE -> { + serverEngine = embeddedServer(ServerCIO, host = host, port = port) { + install(ServerSSE) + routing { + mcp { server } + } + }.start(wait = false) + } + + TransportKind.STREAMABLE_HTTP -> { + // Create StreamableHTTP server transport + // Using JSON response mode for simpler testing (no SSE session required) + val transport = StreamableHttpServerTransport( + enableJsonResponse = true, // Use JSON response mode for testing + ) + // Use stateless mode to skip session validation for simpler testing + transport.setSessionIdGenerator(null) + streamableHttpServerTransport = transport + + // IMPORTANT: Create server session BEFORE starting the HTTP server + // This ensures message handlers are set up before any requests come in + runBlocking { + server.createSession(transport) } - }.start(wait = false) - } else { - // Create in-memory stdio pipes: client->server and server->client - val clientToServerOut = PipedOutputStream() - val clientToServerIn = PipedInputStream(clientToServerOut) - val serverToClientOut = PipedOutputStream() - val serverToClientIn = PipedInputStream(serverToClientOut) + // Start embedded server with routing for StreamableHTTP + serverEngine = embeddedServer(ServerCIO, host = host, port = port) { + // Install ContentNegotiation for JSON serialization + install(ContentNegotiation) { + json(McpJson) + } + routing { + route("/mcp") { + post { + transport.handlePostRequest(null, call) + } + get { + transport.handleGetRequest(null, call) + } + delete { + transport.handleDeleteRequest(call) + } + } + } + }.start(wait = false) + } - // Server transport reads from client and writes to client - val serverTransport = StdioServerTransport( - inputStream = clientToServerIn.asSource().buffered(), - outputStream = serverToClientOut.asSink().buffered(), - ) - stdioServerTransport = serverTransport + TransportKind.STDIO -> { + // Create in-memory stdio pipes: client->server and server->client + val clientToServerOut = PipedOutputStream() + val clientToServerIn = PipedInputStream(clientToServerOut) - // Prepare client-side streams for later client initialization - stdioClientInput = serverToClientIn.asSource().buffered() - stdioClientOutput = clientToServerOut.asSink().buffered() + val serverToClientOut = PipedOutputStream() + val serverToClientIn = PipedInputStream(serverToClientOut) - // Start server transport by connecting the server - runBlocking { - server.createSession(serverTransport) + // Server transport reads from client and writes to client + val serverTransport = StdioServerTransport( + inputStream = clientToServerIn.asSource().buffered(), + outputStream = serverToClientOut.asSink().buffered(), + ) + stdioServerTransport = serverTransport + + // Prepare client-side streams for later client initialization + stdioClientInput = serverToClientIn.asSource().buffered() + stdioClientOutput = clientToServerOut.asSink().buffered() + + // Start server transport by connecting the server + runBlocking { + server.createSession(serverTransport) + } } } } @@ -160,24 +227,39 @@ abstract class KotlinTestBase { } // stop server - if (transportKind == TransportKind.SSE) { - if (::serverEngine.isInitialized) { - try { - serverEngine.stop(500, 1000) - } catch (e: Exception) { - println("Warning: Error during server stop: ${e.message}") + when (transportKind) { + TransportKind.SSE, TransportKind.STREAMABLE_HTTP -> { + if (::serverEngine.isInitialized) { + try { + serverEngine.stop(500, 1000) + } catch (e: Exception) { + println("Warning: Error during server stop: ${e.message}") + } + } + if (transportKind == TransportKind.STREAMABLE_HTTP) { + streamableHttpServerTransport?.let { + try { + runBlocking { it.close() } + } catch (e: Exception) { + println("Warning: Error during streamable http server stop: ${e.message}") + } finally { + streamableHttpServerTransport = null + } + } } } - } else { - stdioServerTransport?.let { - try { - runBlocking { it.close() } - } catch (e: Exception) { - println("Warning: Error during stdio server stop: ${e.message}") - } finally { - stdioServerTransport = null - stdioClientInput = null - stdioClientOutput = null + + TransportKind.STDIO -> { + stdioServerTransport?.let { + try { + runBlocking { it.close() } + } catch (e: Exception) { + println("Warning: Error during stdio server stop: ${e.message}") + } finally { + stdioServerTransport = null + stdioClientInput = null + stdioClientOutput = null + } } } } diff --git a/integration-test/src/jvmTest/kotlin/io/modelcontextprotocol/kotlin/sdk/integration/kotlin/streamablehttp/PromptIntegrationTestStreamableHttp.kt b/integration-test/src/jvmTest/kotlin/io/modelcontextprotocol/kotlin/sdk/integration/kotlin/streamablehttp/PromptIntegrationTestStreamableHttp.kt new file mode 100644 index 0000000..8f4e334 --- /dev/null +++ b/integration-test/src/jvmTest/kotlin/io/modelcontextprotocol/kotlin/sdk/integration/kotlin/streamablehttp/PromptIntegrationTestStreamableHttp.kt @@ -0,0 +1,7 @@ +package io.modelcontextprotocol.kotlin.sdk.integration.kotlin.streamablehttp + +import io.modelcontextprotocol.kotlin.sdk.integration.kotlin.AbstractPromptIntegrationTest + +class PromptIntegrationTestStreamableHttp : AbstractPromptIntegrationTest() { + override val transportKind: TransportKind = TransportKind.STREAMABLE_HTTP +} diff --git a/integration-test/src/jvmTest/kotlin/io/modelcontextprotocol/kotlin/sdk/integration/kotlin/streamablehttp/ResourceIntegrationTestStreamableHttp.kt b/integration-test/src/jvmTest/kotlin/io/modelcontextprotocol/kotlin/sdk/integration/kotlin/streamablehttp/ResourceIntegrationTestStreamableHttp.kt new file mode 100644 index 0000000..d72bcfd --- /dev/null +++ b/integration-test/src/jvmTest/kotlin/io/modelcontextprotocol/kotlin/sdk/integration/kotlin/streamablehttp/ResourceIntegrationTestStreamableHttp.kt @@ -0,0 +1,7 @@ +package io.modelcontextprotocol.kotlin.sdk.integration.kotlin.streamablehttp + +import io.modelcontextprotocol.kotlin.sdk.integration.kotlin.AbstractResourceIntegrationTest + +class ResourceIntegrationTestStreamableHttp : AbstractResourceIntegrationTest() { + override val transportKind: TransportKind = TransportKind.STREAMABLE_HTTP +} diff --git a/integration-test/src/jvmTest/kotlin/io/modelcontextprotocol/kotlin/sdk/integration/kotlin/streamablehttp/ToolIntegrationTestStreamableHttp.kt b/integration-test/src/jvmTest/kotlin/io/modelcontextprotocol/kotlin/sdk/integration/kotlin/streamablehttp/ToolIntegrationTestStreamableHttp.kt new file mode 100644 index 0000000..288ff07 --- /dev/null +++ b/integration-test/src/jvmTest/kotlin/io/modelcontextprotocol/kotlin/sdk/integration/kotlin/streamablehttp/ToolIntegrationTestStreamableHttp.kt @@ -0,0 +1,7 @@ +package io.modelcontextprotocol.kotlin.sdk.integration.kotlin.streamablehttp + +import io.modelcontextprotocol.kotlin.sdk.integration.kotlin.AbstractToolIntegrationTest + +class ToolIntegrationTestStreamableHttp : AbstractToolIntegrationTest() { + override val transportKind: TransportKind = TransportKind.STREAMABLE_HTTP +} diff --git a/integration-test/src/jvmTest/kotlin/io/modelcontextprotocol/kotlin/sdk/integration/typescript/AbstractKotlinClientTsServerTest.kt b/integration-test/src/jvmTest/kotlin/io/modelcontextprotocol/kotlin/sdk/integration/typescript/AbstractKotlinClientTsServerTest.kt index 1ed3cf4..9d27a1d 100644 --- a/integration-test/src/jvmTest/kotlin/io/modelcontextprotocol/kotlin/sdk/integration/typescript/AbstractKotlinClientTsServerTest.kt +++ b/integration-test/src/jvmTest/kotlin/io/modelcontextprotocol/kotlin/sdk/integration/typescript/AbstractKotlinClientTsServerTest.kt @@ -15,7 +15,7 @@ abstract class AbstractKotlinClientTsServerTest : TsTestBase() { protected abstract suspend fun useClient(block: suspend (Client) -> T): T @Test - @Timeout(20, unit = TimeUnit.SECONDS) + @Timeout(60, unit = TimeUnit.SECONDS) fun connectsAndPings() = runBlocking(Dispatchers.IO) { useClient { client -> assertNotNull(client, "Client should be initialized") @@ -28,7 +28,7 @@ abstract class AbstractKotlinClientTsServerTest : TsTestBase() { } @Test - @Timeout(20, unit = TimeUnit.SECONDS) + @Timeout(60, unit = TimeUnit.SECONDS) fun listsTools() = runBlocking(Dispatchers.IO) { useClient { client -> val result = client.listTools() @@ -42,7 +42,7 @@ abstract class AbstractKotlinClientTsServerTest : TsTestBase() { } @Test - @Timeout(20, unit = TimeUnit.SECONDS) + @Timeout(60, unit = TimeUnit.SECONDS) fun callGreet() = runBlocking(Dispatchers.IO) { useClient { client -> val testName = "TestUser" @@ -57,7 +57,7 @@ abstract class AbstractKotlinClientTsServerTest : TsTestBase() { } @Test - @Timeout(20, unit = TimeUnit.SECONDS) + @Timeout(60, unit = TimeUnit.SECONDS) fun multipleClients() = runBlocking(Dispatchers.IO) { useClient { client1 -> useClient { client2 -> diff --git a/integration-test/src/jvmTest/kotlin/io/modelcontextprotocol/kotlin/sdk/integration/typescript/TsTestBase.kt b/integration-test/src/jvmTest/kotlin/io/modelcontextprotocol/kotlin/sdk/integration/typescript/TsTestBase.kt index 49632c4..aa47a70 100644 --- a/integration-test/src/jvmTest/kotlin/io/modelcontextprotocol/kotlin/sdk/integration/typescript/TsTestBase.kt +++ b/integration-test/src/jvmTest/kotlin/io/modelcontextprotocol/kotlin/sdk/integration/typescript/TsTestBase.kt @@ -213,14 +213,19 @@ abstract class TsTestBase { serverThread.isDaemon = true serverThread.start() - // Read ONLY stderr from client for human-readable output + // Read ONLY stderr from client for human-readable output. + // Also use it as a signal that the client is done (some Node processes keep the event-loop alive). val output = StringBuilder() + val doneLatch = java.util.concurrent.CountDownLatch(1) val errReader = Thread { try { process.errorStream.bufferedReader().useLines { lines -> lines.forEach { line -> println("[TS-CLIENT-STDIO][err] $line") output.append(line).append('\n') + if (line.contains("Disconnected from server")) { + doneLatch.countDown() + } } } } catch (e: Exception) { @@ -230,11 +235,26 @@ abstract class TsTestBase { errReader.isDaemon = true errReader.start() - // Wait up to 25s for client to exit - val finished = process.waitFor(25, TimeUnit.SECONDS) + // Wait for a "done" signal first, then give the process a short grace period to exit. + // This avoids hanging under the test's outer timeout (often ~30s). + val doneTimeoutSeconds = System.getenv("MCP_TS_CLIENT_DONE_TIMEOUT_SECONDS")?.toLongOrNull() ?: 25L + val sawDone = doneLatch.await(doneTimeoutSeconds, TimeUnit.SECONDS) + if (sawDone) { + // Encourage Node to exit if it is waiting on stdin. + try { + process.outputStream.close() + } catch (_: Exception) { + } + } + + // Primary completion condition: the process exits. + // (Some error-paths may not print the disconnect line, but should still exit.) + val clientTimeoutSeconds = System.getenv("MCP_TS_CLIENT_TIMEOUT_SECONDS")?.toLongOrNull() ?: 30L + val finished = process.waitFor(clientTimeoutSeconds, TimeUnit.SECONDS) if (!finished) { - println("Stdio client did not finish in time; destroying") + println("Stdio client did not finish in time (${clientTimeoutSeconds}s); destroying") process.destroyForcibly() + process.waitFor(2, TimeUnit.SECONDS) } try { diff --git a/integration-test/src/jvmTest/kotlin/io/modelcontextprotocol/kotlin/sdk/integration/typescript/TypeScriptServer.kt b/integration-test/src/jvmTest/kotlin/io/modelcontextprotocol/kotlin/sdk/integration/typescript/TypeScriptServer.kt index d926590..579f96d 100644 --- a/integration-test/src/jvmTest/kotlin/io/modelcontextprotocol/kotlin/sdk/integration/typescript/TypeScriptServer.kt +++ b/integration-test/src/jvmTest/kotlin/io/modelcontextprotocol/kotlin/sdk/integration/typescript/TypeScriptServer.kt @@ -56,7 +56,7 @@ class TypeScriptServer(private val typescriptDir: File) { ) process = proc - check(waitForPort(port = port, timeoutSeconds = 10)) { + check(waitForPort(port = port, timeoutSeconds = 60)) { proc.destroyForcibly() "TypeScript SSE server did not become ready on localhost:$port within timeout" } diff --git a/integration-test/src/jvmTest/kotlin/io/modelcontextprotocol/kotlin/sdk/integration/typescript/stdio/KotlinClientTsServerTestStdio.kt b/integration-test/src/jvmTest/kotlin/io/modelcontextprotocol/kotlin/sdk/integration/typescript/stdio/KotlinClientTsServerTestStdio.kt index 8273724..928d591 100644 --- a/integration-test/src/jvmTest/kotlin/io/modelcontextprotocol/kotlin/sdk/integration/typescript/stdio/KotlinClientTsServerTestStdio.kt +++ b/integration-test/src/jvmTest/kotlin/io/modelcontextprotocol/kotlin/sdk/integration/typescript/stdio/KotlinClientTsServerTestStdio.kt @@ -8,16 +8,7 @@ class KotlinClientTsServerTestStdio : AbstractKotlinClientTsServerTest() { override val transportKind = TransportKind.STDIO - override suspend fun useClient(block: suspend (Client) -> T): T = withClientStdio { client, proc -> - try { - block(client) - } finally { - try { - client.close() - } catch (_: Exception) {} - try { - stopProcess(proc, name = "TypeScript stdio server") - } catch (_: Exception) {} - } - } + override suspend fun useClient(block: suspend (Client) -> T): T = + // TsTestBase.withClientStdio already handles client.close() + process shutdown. + withClientStdio { client, _ -> block(client) } } diff --git a/integration-test/src/jvmTest/typescript/client/stdio-client.ts b/integration-test/src/jvmTest/typescript/client/stdio-client.ts index c6eaca6..82c25de 100644 --- a/integration-test/src/jvmTest/typescript/client/stdio-client.ts +++ b/integration-test/src/jvmTest/typescript/client/stdio-client.ts @@ -99,8 +99,9 @@ async function main() { const tool = tools.find((t: { name: string; }) => t.name === toolName); if (!tool) { - console.error(`Tool "${toolName}" not found`); - process.exit(1); + // Don't call process.exit() here: it prevents finally{} from running, + // which breaks tests that expect a clean disconnect signal. + throw new Error(`Tool "${toolName}" not found`); } const toolArguments: any = {}; @@ -136,7 +137,8 @@ async function main() { } catch (error) { console.error('Error:', error); - process.exit(1); + // Don't hard-exit; allow finally{} to run so the client always closes cleanly. + process.exitCode = 1; } finally { await client.close(); console.error('Disconnected from server'); diff --git a/kotlin-sdk-client/src/jvmTest/kotlin/io/modelcontextprotocol/kotlin/sdk/client/MockMcp.kt b/kotlin-sdk-client/src/jvmTest/kotlin/io/modelcontextprotocol/kotlin/sdk/client/MockMcp.kt index a4447cf..ff39973 100644 --- a/kotlin-sdk-client/src/jvmTest/kotlin/io/modelcontextprotocol/kotlin/sdk/client/MockMcp.kt +++ b/kotlin-sdk-client/src/jvmTest/kotlin/io/modelcontextprotocol/kotlin/sdk/client/MockMcp.kt @@ -19,7 +19,7 @@ import kotlinx.serialization.json.jsonObject import kotlinx.serialization.json.jsonPrimitive import kotlinx.serialization.json.putJsonObject -const val MCP_SESSION_ID_HEADER = "Mcp-Session-Id" +const val MCP_SESSION_ID_HEADER = "MCP-Session-Id" internal class MockMcp(verbose: Boolean = false) {