modelcontextprotocol/kotlin-sdk
Watch
1
0
Fork
You've already forked kotlin-sdk
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 2

[WIP]test(integration): Add Streamable HTTP transport integration tests (#486)

Browse source 
## 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>
This commit is contained in:
johny 2026-02-05 21:49:47 +09:00 • committed by GitHub
commit def0e925e2
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
11 changed files with 189 additions and 71 deletions
Download patch file Download diff file Expand all files Collapse all files

2
integration-test/build.gradle.kts
View file

@ -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 {

166
integration-test/src/jvmTest/kotlin/io/modelcontextprotocol/kotlin/sdk/integration/kotlin/KotlinTestBase.kt
View file

@ -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
}
}
}
}

7
integration-test/src/jvmTest/kotlin/io/modelcontextprotocol/kotlin/sdk/integration/kotlin/streamablehttp/PromptIntegrationTestStreamableHttp.kt Normal file
View file

@ -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
}

7
integration-test/src/jvmTest/kotlin/io/modelcontextprotocol/kotlin/sdk/integration/kotlin/streamablehttp/ResourceIntegrationTestStreamableHttp.kt Normal file
View file

@ -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
}

7
integration-test/src/jvmTest/kotlin/io/modelcontextprotocol/kotlin/sdk/integration/kotlin/streamablehttp/ToolIntegrationTestStreamableHttp.kt Normal file
View file

@ -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
}

8
integration-test/src/jvmTest/kotlin/io/modelcontextprotocol/kotlin/sdk/integration/typescript/AbstractKotlinClientTsServerTest.kt
View file

@ -15,7 +15,7 @@ abstract class AbstractKotlinClientTsServerTest : TsTestBase() {
protected abstract suspend fun <T> 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 ->

28
integration-test/src/jvmTest/kotlin/io/modelcontextprotocol/kotlin/sdk/integration/typescript/TsTestBase.kt
View file

@ -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 {

2
integration-test/src/jvmTest/kotlin/io/modelcontextprotocol/kotlin/sdk/integration/typescript/TypeScriptServer.kt
View file

@ -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"
}

15
integration-test/src/jvmTest/kotlin/io/modelcontextprotocol/kotlin/sdk/integration/typescript/stdio/KotlinClientTsServerTestStdio.kt
View file

@ -8,16 +8,7 @@ class KotlinClientTsServerTestStdio : AbstractKotlinClientTsServerTest() {
override val transportKind = TransportKind.STDIO
override suspend fun <T> 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 <T> useClient(block: suspend (Client) -> T): T =
// TsTestBase.withClientStdio already handles client.close() + process shutdown.
withClientStdio { client, _ -> block(client) }
}

8
integration-test/src/jvmTest/typescript/client/stdio-client.ts
View file

@ -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');

2
kotlin-sdk-client/src/jvmTest/kotlin/io/modelcontextprotocol/kotlin/sdk/client/MockMcp.kt
View file

@ -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) {