From 664f51b9cc210466c02699f5146130f86c4be98b Mon Sep 17 00:00:00 2001 From: Pavel Gorgulov Date: Mon, 18 May 2026 12:17:26 +0200 Subject: [PATCH] test: remove mokksy dependency (#754) Drops`dev.mokksy:mokksy` dependency ## Motivation and Context `mokksy` is a niche, single-maintainer library with very limited external adoption, introduced by its own author. Keeping a test-only dependency on it is unnecessary supply-chain exposure given that the two patterns already used in this repo, `MockEngine` for transport-level tests and `embeddedServer` for integration-level tests, cover everything it provided. ## Breaking Changes none ## Types of changes - [ ] Bug fix (non-breaking change which fixes an issue) - [ ] New feature (non-breaking change which adds functionality) - [ ] Breaking change (fix or feature that would cause existing functionality to change) - [ ] Documentation update ## Checklist - [x] I have read the [MCP Documentation](https://modelcontextprotocol.io) - [x] My code follows the repository's style guidelines - [x] New and existing tests pass locally - [x] I have added appropriate error handling - [x] I have added or updated documentation as needed --- gradle/libs.versions.toml | 2 - ...mableHttpClientTransportIntegrationTest.kt | 270 ++++++++++++++++ kotlin-sdk-client/build.gradle.kts | 1 - .../http/StreamableHttpClientTransportTest.kt | 73 +++++ .../AbstractStreamableHttpClientTest.kt | 42 --- .../kotlin/sdk/client/MockMcp.kt | 237 -------------- .../sdk/client/StreamableHttpClientTest.kt | 293 ------------------ 7 files changed, 343 insertions(+), 575 deletions(-) create mode 100644 integration-test/src/jvmTest/kotlin/io/modelcontextprotocol/kotlin/sdk/integration/streamablehttp/StreamableHttpClientTransportIntegrationTest.kt delete mode 100644 kotlin-sdk-client/src/jvmTest/kotlin/io/modelcontextprotocol/kotlin/sdk/client/AbstractStreamableHttpClientTest.kt delete mode 100644 kotlin-sdk-client/src/jvmTest/kotlin/io/modelcontextprotocol/kotlin/sdk/client/MockMcp.kt delete mode 100644 kotlin-sdk-client/src/jvmTest/kotlin/io/modelcontextprotocol/kotlin/sdk/client/StreamableHttpClientTest.kt diff --git a/gradle/libs.versions.toml b/gradle/libs.versions.toml index a1f2cc0..5b04ca4 100644 --- a/gradle/libs.versions.toml +++ b/gradle/libs.versions.toml @@ -20,7 +20,6 @@ kotlinx-io = "0.9.0" ktor = "3.3.3" logging = "8.0.02" mockk = "1.14.9" -mokksy = "0.8.1" serialization = "1.11.0" slf4j = "2.0.17" logback = "1.5.32" @@ -65,7 +64,6 @@ kotlinx-coroutines-test = { group = "org.jetbrains.kotlinx", name = "kotlinx-cor ktor-client-mock = { group = "io.ktor", name = "ktor-client-mock", version.ref = "ktor" } ktor-server-test-host = { group = "io.ktor", name = "ktor-server-test-host", version.ref = "ktor" } mockk = { module = "io.mockk:mockk", version.ref = "mockk" } -mokksy = { group = "dev.mokksy", name = "mokksy", version.ref = "mokksy" } netty-bom = { group = "io.netty", name = "netty-bom", version.ref = "netty" } slf4j-simple = { group = "org.slf4j", name = "slf4j-simple", version.ref = "slf4j" } logback-classic = { group = "ch.qos.logback", name = "logback-classic", version.ref = "logback" } diff --git a/integration-test/src/jvmTest/kotlin/io/modelcontextprotocol/kotlin/sdk/integration/streamablehttp/StreamableHttpClientTransportIntegrationTest.kt b/integration-test/src/jvmTest/kotlin/io/modelcontextprotocol/kotlin/sdk/integration/streamablehttp/StreamableHttpClientTransportIntegrationTest.kt new file mode 100644 index 0000000..598825f --- /dev/null +++ b/integration-test/src/jvmTest/kotlin/io/modelcontextprotocol/kotlin/sdk/integration/streamablehttp/StreamableHttpClientTransportIntegrationTest.kt @@ -0,0 +1,270 @@ +package io.modelcontextprotocol.kotlin.sdk.integration.streamablehttp + +import io.kotest.matchers.collections.shouldContain +import io.kotest.matchers.shouldBe +import io.ktor.client.HttpClient +import io.ktor.client.plugins.sse.SSE +import io.ktor.http.ContentType +import io.ktor.http.HttpStatusCode +import io.ktor.server.engine.embeddedServer +import io.ktor.server.request.receiveText +import io.ktor.server.response.header +import io.ktor.server.response.respond +import io.ktor.server.response.respondText +import io.ktor.server.response.respondTextWriter +import io.ktor.server.routing.Route +import io.ktor.server.routing.delete +import io.ktor.server.routing.get +import io.ktor.server.routing.post +import io.ktor.server.routing.routing +import io.modelcontextprotocol.kotlin.sdk.client.Client +import io.modelcontextprotocol.kotlin.sdk.client.ClientOptions +import io.modelcontextprotocol.kotlin.sdk.client.StreamableHttpClientTransport +import io.modelcontextprotocol.kotlin.sdk.types.ClientCapabilities +import io.modelcontextprotocol.kotlin.sdk.types.EmptyJsonObject +import io.modelcontextprotocol.kotlin.sdk.types.Implementation +import io.modelcontextprotocol.kotlin.sdk.types.Tool +import io.modelcontextprotocol.kotlin.sdk.types.ToolSchema +import io.modelcontextprotocol.kotlin.test.utils.actualPort +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.awaitCancellation +import kotlinx.coroutines.delay +import kotlinx.coroutines.runBlocking +import kotlinx.serialization.json.Json +import kotlinx.serialization.json.buildJsonObject +import kotlinx.serialization.json.jsonObject +import kotlinx.serialization.json.jsonPrimitive +import kotlinx.serialization.json.put +import kotlinx.serialization.json.putJsonObject +import kotlin.test.Test +import io.ktor.client.engine.cio.CIO as ClientCIO +import io.ktor.server.cio.CIO as ServerCIO + +private const val SESSION_ID_HEADER = "mcp-session-id" +private const val HOST = "127.0.0.1" +private const val MCP_PATH = "/mcp" + +private val INITIALIZE_RESULT = """ + { + "capabilities": {"tools": {"listChanged": false}}, + "protocolVersion": "2025-03-26", + "serverInfo": {"name": "test-server", "version": "1.0.0"} + } +""".trimIndent() + +private val TOOLS_LIST_RESULT = """ + { + "tools": [ + { + "name": "get_weather", + "title": "Weather Information Provider", + "description": "Get current weather information for a location", + "inputSchema": { + "type": "object", + "properties": { + "location": {"type": "string", "description": "City name or zip code"} + }, + "required": ["location"] + }, + "outputSchema": { + "type": "object", + "properties": { + "temperature": {"type": "number", "description": "Temperature, Celsius"} + }, + "required": ["temperature"] + }, + "_meta": {} + } + ] + } +""".trimIndent() + +private const val PROGRESS_NOTIFICATION_1 = + """{"jsonrpc":"2.0","method":"notifications/progress","params":{"progressToken":"upload-123","progress":50,"total":100}}""" + +private const val PROGRESS_NOTIFICATION_2 = + """{"jsonrpc":"2.0","method":"notifications/progress","params":{"progressToken":"upload-123","progress":75,"total":100}}""" + +internal class StreamableHttpClientTransportIntegrationTest { + + @Test + fun `client receives progress notifications via GET SSE and lists tools`(): Unit = runBlocking(Dispatchers.IO) { + val sessionId = "sid-progress" + runWithServer({ + mcpPostRoute(sessionId, extraResults = mapOf("tools/list" to TOOLS_LIST_RESULT)) + get(MCP_PATH) { + call.response.header(SESSION_ID_HEADER, sessionId) + call.respondTextWriter(contentType = ContentType.Text.EventStream) { + write(sseEvent(id = "1", data = PROGRESS_NOTIFICATION_1)) + flush() + delay(200) + write(sseEvent(id = "2", data = PROGRESS_NOTIFICATION_2)) + flush() + awaitCancellation() + } + } + }) { url -> + val client = newClient() + client.connect(StreamableHttpClientTransport(client = newHttpClient(), url = url)) + + val listToolsResult = client.listTools() + + listToolsResult.tools shouldContain Tool( + name = "get_weather", + title = "Weather Information Provider", + description = "Get current weather information for a location", + inputSchema = ToolSchema( + properties = buildJsonObject { + putJsonObject("location") { + put("type", "string") + put("description", "City name or zip code") + } + }, + required = listOf("location"), + ), + outputSchema = ToolSchema( + properties = buildJsonObject { + putJsonObject("temperature") { + put("type", "number") + put("description", "Temperature, Celsius") + } + }, + required = listOf("temperature"), + ), + annotations = null, + meta = EmptyJsonObject, + ) + + client.close() + } + } + + @Test + fun `terminateSession sends DELETE request and clears sessionId`(): Unit = runBlocking(Dispatchers.IO) { + val sessionId = "sid-terminate" + runWithServer({ + mcpPostRoute(sessionId) + get(MCP_PATH) { + call.response.header(SESSION_ID_HEADER, sessionId) + call.respondTextWriter(contentType = ContentType.Text.EventStream) { + awaitCancellation() + } + } + delete(MCP_PATH) { + call.respond(HttpStatusCode.OK) + } + }) { url -> + val client = newClient() + val transport = StreamableHttpClientTransport(client = newHttpClient(), url = url) + client.connect(transport) + + transport.sessionId shouldBe sessionId + transport.terminateSession() + transport.sessionId shouldBe null + + client.close() + } + } + + @Test + fun `client survives 405 on GET SSE`(): Unit = runBlocking(Dispatchers.IO) { + val sessionId = "sid-405" + runWithServer({ + mcpPostRoute(sessionId, extraResults = mapOf("ping" to "{}")) + get(MCP_PATH) { + call.respondText( + text = "", + contentType = ContentType.Text.EventStream, + status = HttpStatusCode.MethodNotAllowed, + ) + } + }) { url -> + val client = newClient() + client.connect(StreamableHttpClientTransport(client = newHttpClient(), url = url)) + + client.ping() + + client.close() + } + } + + @Test + fun `client survives non streaming JSON response on GET SSE`(): Unit = runBlocking(Dispatchers.IO) { + val sessionId = "sid-json" + runWithServer({ + mcpPostRoute(sessionId, extraResults = mapOf("ping" to "{}")) + get(MCP_PATH) { + call.response.header(SESSION_ID_HEADER, sessionId) + call.respondText( + text = "", + contentType = ContentType.Application.Json, + status = HttpStatusCode.OK, + ) + } + }) { url -> + val client = newClient() + client.connect(StreamableHttpClientTransport(client = newHttpClient(), url = url)) + + client.ping() + + client.close() + } + } + + private fun Route.mcpPostRoute(sessionId: String, extraResults: Map = emptyMap()) { + post(MCP_PATH) { + val body = call.receiveText() + val parsed = Json.parseToJsonElement(body).jsonObject + val method = parsed["method"]?.jsonPrimitive?.content + val idJson = parsed["id"]?.toString() ?: "null" + call.response.header(SESSION_ID_HEADER, sessionId) + when { + method == "initialize" -> call.respondText( + text = jsonRpcResult(idJson, INITIALIZE_RESULT), + contentType = ContentType.Application.Json, + status = HttpStatusCode.OK, + ) + + method == "notifications/initialized" -> call.respond(HttpStatusCode.Accepted) + + method != null && method in extraResults -> call.respondText( + text = jsonRpcResult(idJson, extraResults.getValue(method)), + contentType = ContentType.Application.Json, + status = HttpStatusCode.OK, + ) + + else -> call.respond(HttpStatusCode.NotFound) + } + } + } + + private suspend fun runWithServer(routes: Route.() -> Unit, block: suspend (url: String) -> Unit) { + val server = embeddedServer(ServerCIO, host = HOST, port = 0) { + routing { routes() } + }.startSuspend(wait = false) + try { + val port = server.actualPort() + block("http://$HOST:$port$MCP_PATH") + } finally { + server.stopSuspend(1000, 2000) + } + } + + private fun newClient(): Client = Client( + clientInfo = Implementation(name = "client1", version = "1.0.0"), + options = ClientOptions(capabilities = ClientCapabilities()), + ) + + private fun newHttpClient(): HttpClient = HttpClient(ClientCIO) { + install(SSE) + } + + private fun jsonRpcResult(idJson: String, resultJson: String): String = + """{"jsonrpc":"2.0","id":$idJson,"result":$resultJson}""" + + private fun sseEvent(id: String, data: String): String = buildString { + append("event: message\n") + append("id: $id\n") + append("data: $data\n\n") + } +} diff --git a/kotlin-sdk-client/build.gradle.kts b/kotlin-sdk-client/build.gradle.kts index ae7a314..fbb1300 100644 --- a/kotlin-sdk-client/build.gradle.kts +++ b/kotlin-sdk-client/build.gradle.kts @@ -55,7 +55,6 @@ kotlin { implementation(libs.ktor.client.apache5) implementation(libs.mockk) implementation(libs.junit.jupiter.params) - implementation(libs.mokksy) implementation(dependencies.platform(libs.netty.bom)) runtimeOnly(libs.slf4j.simple) } 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 8810552..b380f64 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 @@ -478,6 +478,79 @@ class StreamableHttpClientTransportTest { } } + @Test + fun testSkipEmptySSE() = runTest { + val transport = createTransport { request -> + if (request.method == HttpMethod.Post) { + val sseContent = buildString { + // Event 1: empty data — must be skipped without error + appendLine("id: a") + appendLine("data:") + appendLine() + // Event 2: whitespace-only data — must be skipped without error + appendLine("id: b") + appendLine("data: \t ") + appendLine() + // Event 3: multi-line data lines that together form a valid JSON-RPC response + appendLine("id: c") + appendLine("event: message") + appendLine("data: {") + appendLine("""data: "jsonrpc":"2.0",""") + appendLine("""data: "id":"test-1",""") + appendLine("""data: "result":{""") + appendLine("""data: "protocolVersion":"2025-06-18",""") + appendLine("""data: "capabilities":{},""") + appendLine("""data: "serverInfo":{"name":"server","version":"1.0.0"}""") + appendLine("data: }") + appendLine("data: }") + appendLine() + } + + respond( + content = ByteReadChannel(sseContent), + status = HttpStatusCode.OK, + headers = headersOf( + HttpHeaders.ContentType, + ContentType.Text.EventStream.toString(), + ), + ) + } else { + respond("", HttpStatusCode.OK) + } + } + + val receivedMessages = mutableListOf() + val messageReceived = CompletableDeferred() + + transport.onMessage { message -> + receivedMessages.add(message) + if (!messageReceived.isCompleted) { + messageReceived.complete(Unit) + } + } + + transport.start() + + transport.send( + JSONRPCRequest( + id = "test-1", + method = "initialize", + params = buildJsonObject { }, + ), + ) + + eventually { + messageReceived.await() + } + + receivedMessages shouldHaveSize 1 + val response = receivedMessages[0] + response.shouldBeInstanceOf() + response.id shouldBe RequestId.StringId("test-1") + + transport.close() + } + @Test fun testInlineSSEInResponse() = runTest { val transport = createTransport { request -> diff --git a/kotlin-sdk-client/src/jvmTest/kotlin/io/modelcontextprotocol/kotlin/sdk/client/AbstractStreamableHttpClientTest.kt b/kotlin-sdk-client/src/jvmTest/kotlin/io/modelcontextprotocol/kotlin/sdk/client/AbstractStreamableHttpClientTest.kt deleted file mode 100644 index 4a04e81..0000000 --- a/kotlin-sdk-client/src/jvmTest/kotlin/io/modelcontextprotocol/kotlin/sdk/client/AbstractStreamableHttpClientTest.kt +++ /dev/null @@ -1,42 +0,0 @@ -package io.modelcontextprotocol.kotlin.sdk.client - -import io.ktor.client.HttpClient -import io.ktor.client.engine.apache5.Apache5 -import io.ktor.client.plugins.logging.LogLevel -import io.ktor.client.plugins.logging.Logging -import io.ktor.client.plugins.sse.SSE -import org.junit.jupiter.api.AfterAll -import org.junit.jupiter.api.AfterEach -import org.junit.jupiter.api.TestInstance - -@TestInstance(TestInstance.Lifecycle.PER_CLASS) -internal abstract class AbstractStreamableHttpClientTest { - - // start mokksy on random port - protected val mockMcp: MockMcp = MockMcp(verbose = true) - - @AfterEach - fun afterEach() { - mockMcp.verifyNoUnexpectedRequests() - } - - @AfterAll - fun afterAll() { - mockMcp.verifyNoUnmatchedStubs() - mockMcp.close() - } - - protected suspend fun connect(client: Client) { - client.connect( - StreamableHttpClientTransport( - url = mockMcp.url, - client = HttpClient(Apache5) { - install(SSE) - install(Logging) { - level = LogLevel.ALL - } - }, - ), - ) - } -} 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 deleted file mode 100644 index de9d2b3..0000000 --- a/kotlin-sdk-client/src/jvmTest/kotlin/io/modelcontextprotocol/kotlin/sdk/client/MockMcp.kt +++ /dev/null @@ -1,237 +0,0 @@ -package io.modelcontextprotocol.kotlin.sdk.client - -import dev.mokksy.mokksy.BuildingStep -import dev.mokksy.mokksy.Mokksy -import dev.mokksy.mokksy.StubConfiguration -import dev.mokksy.mokksy.shutdown -import dev.mokksy.mokksy.start -import io.ktor.http.ContentType -import io.ktor.http.HttpMethod -import io.ktor.http.HttpStatusCode -import io.ktor.sse.ServerSentEvent -import io.modelcontextprotocol.kotlin.sdk.types.JSONRPCRequest -import io.modelcontextprotocol.kotlin.sdk.types.RequestId -import kotlinx.coroutines.flow.Flow -import kotlinx.serialization.json.Json -import kotlinx.serialization.json.JsonObject -import kotlinx.serialization.json.JsonPrimitive -import kotlinx.serialization.json.buildJsonObject -import kotlinx.serialization.json.contentOrNull -import kotlinx.serialization.json.jsonObject -import kotlinx.serialization.json.jsonPrimitive -import kotlinx.serialization.json.putJsonObject - -const val MCP_SESSION_ID_HEADER = "MCP-Session-Id" - -internal class MockMcp(verbose: Boolean = false) : AutoCloseable { - - private val mokksy: Mokksy = Mokksy(verbose = verbose).apply { - start() - } - - fun verifyNoUnexpectedRequests() { - mokksy.verifyNoUnexpectedRequests() - } - - fun verifyNoUnmatchedStubs() { - mokksy.verifyNoUnmatchedStubs() - } - - val url = "${mokksy.baseUrl()}/mcp" - - fun onInitialize( - clientName: String? = null, - sessionId: String, - protocolVersion: String = "2025-03-26", - serverName: String = "Mock MCP Server", - serverVersion: String = "1.0.0", - capabilities: JsonObject = buildJsonObject { - putJsonObject("tools") { - put("listChanged", JsonPrimitive(false)) - } - }, - ) { - val predicates = if (clientName != null) { - arrayOf<(JSONRPCRequest?) -> Boolean>({ - it?.params?.jsonObject - ?.get("clientInfo")?.jsonObject - ?.get("name")?.jsonPrimitive - ?.contentOrNull == clientName - }) - } else { - emptyArray() - } - - handleWithResult( - jsonRpcMethod = "initialize", - sessionId = sessionId, - bodyPredicates = predicates, - // language=json - result = """ - { - "capabilities": $capabilities, - "protocolVersion": "$protocolVersion", - "serverInfo": { - "name": "$serverName", - "version": "$serverVersion" - }, - "_meta": { - "foo": "bar" - } - } - """.trimIndent(), - ) - } - - fun onJSONRPCRequest( - httpMethod: HttpMethod = HttpMethod.Post, - jsonRpcMethod: String, - expectedSessionId: String? = null, - vararg bodyPredicates: (JSONRPCRequest) -> Boolean, - ): BuildingStep = mokksy.method( - configuration = StubConfiguration(removeAfterMatch = true), - httpMethod = httpMethod, - requestType = JSONRPCRequest::class, - ) { - path("/mcp") - expectedSessionId?.let { - containsHeader(MCP_SESSION_ID_HEADER, it) - } - bodyMatchesPredicate( - description = "JSON-RPC version is '2.0'", - predicate = - { - it!!.jsonrpc == "2.0" - }, - ) - bodyMatchesPredicate( - description = "JSON-RPC Method should be '$jsonRpcMethod'", - predicate = - { - it!!.method == jsonRpcMethod - }, - ) - bodyPredicates.forEach { predicate -> - bodyMatchesPredicate(predicate = { predicate.invoke(it!!) }) - } - } - - fun handleWithResult( - httpMethod: HttpMethod = HttpMethod.Post, - jsonRpcMethod: String, - expectedSessionId: String? = null, - sessionId: String, - contentType: ContentType = ContentType.Application.Json, - statusCode: HttpStatusCode = HttpStatusCode.OK, - vararg bodyPredicates: (JSONRPCRequest) -> Boolean, - result: () -> JsonObject, - ) { - onJSONRPCRequest( - httpMethod = httpMethod, - jsonRpcMethod = jsonRpcMethod, - expectedSessionId = expectedSessionId, - bodyPredicates = bodyPredicates, - ) respondsWith { - val requestId = when (val id = request.body().id) { - is RequestId.NumberId -> id.value.toString() - is RequestId.StringId -> "\"${id.value}\"" - } - val resultObject = result.invoke() - // language=json - body = """ - { - "jsonrpc": "2.0", - "id": $requestId, - "result": $resultObject - } - """.trimIndent() - this.contentType = contentType - headers += MCP_SESSION_ID_HEADER to sessionId - httpStatus = statusCode - } - } - - fun handleWithResult( - httpMethod: HttpMethod = HttpMethod.Post, - jsonRpcMethod: String, - expectedSessionId: String? = null, - sessionId: String, - contentType: ContentType = ContentType.Application.Json, - statusCode: HttpStatusCode = HttpStatusCode.OK, - vararg bodyPredicates: (JSONRPCRequest) -> Boolean, - result: String, - ) { - handleWithResult( - httpMethod = httpMethod, - jsonRpcMethod = jsonRpcMethod, - expectedSessionId = expectedSessionId, - sessionId = sessionId, - contentType = contentType, - statusCode = statusCode, - bodyPredicates = bodyPredicates, - result = { - Json.parseToJsonElement(result).jsonObject - }, - ) - } - - fun handleJSONRPCRequest( - httpMethod: HttpMethod = HttpMethod.Post, - jsonRpcMethod: String, - expectedSessionId: String? = null, - sessionId: String, - contentType: ContentType = ContentType.Application.Json, - statusCode: HttpStatusCode = HttpStatusCode.OK, - vararg bodyPredicates: (JSONRPCRequest?) -> Boolean, - bodyBuilder: () -> String = { "" }, - ) { - onJSONRPCRequest( - httpMethod = httpMethod, - jsonRpcMethod = jsonRpcMethod, - expectedSessionId = expectedSessionId, - bodyPredicates = bodyPredicates, - ) respondsWith { - body = bodyBuilder.invoke() - this.contentType = contentType - headers += MCP_SESSION_ID_HEADER to sessionId - httpStatus = statusCode - } - } - - fun onSubscribe(httpMethod: HttpMethod = HttpMethod.Post, sessionId: String): BuildingStep = mokksy.method( - httpMethod = httpMethod, - name = "MCP GETs", - requestType = Any::class, - ) { - path("/mcp") - containsHeader(MCP_SESSION_ID_HEADER, sessionId) - containsHeader("Accept", "application/json,text/event-stream") - containsHeader("Cache-Control", "no-store") - } - - fun handleSubscribeWithGet(sessionId: String, block: () -> Flow) { - onSubscribe( - httpMethod = HttpMethod.Get, - sessionId = sessionId, - ) respondsWithSseStream { - headers += MCP_SESSION_ID_HEADER to sessionId - this.flow = block.invoke() - } - } - - fun mockUnsubscribeRequest(sessionId: String) { - mokksy.delete( - configuration = StubConfiguration(removeAfterMatch = true), - requestType = JSONRPCRequest::class, - ) { - path("/mcp") - containsHeader(MCP_SESSION_ID_HEADER, sessionId) - } respondsWith { - body = null - } - } - - override fun close() { - mokksy.shutdown() - } -} diff --git a/kotlin-sdk-client/src/jvmTest/kotlin/io/modelcontextprotocol/kotlin/sdk/client/StreamableHttpClientTest.kt b/kotlin-sdk-client/src/jvmTest/kotlin/io/modelcontextprotocol/kotlin/sdk/client/StreamableHttpClientTest.kt deleted file mode 100644 index 43be8c9..0000000 --- a/kotlin-sdk-client/src/jvmTest/kotlin/io/modelcontextprotocol/kotlin/sdk/client/StreamableHttpClientTest.kt +++ /dev/null @@ -1,293 +0,0 @@ -package io.modelcontextprotocol.kotlin.sdk.client - -import io.kotest.matchers.collections.shouldContain -import io.kotest.matchers.shouldBe -import io.ktor.http.ContentType -import io.ktor.http.HttpMethod -import io.ktor.http.HttpStatusCode -import io.ktor.sse.ServerSentEvent -import io.modelcontextprotocol.kotlin.sdk.types.ClientCapabilities -import io.modelcontextprotocol.kotlin.sdk.types.EmptyJsonObject -import io.modelcontextprotocol.kotlin.sdk.types.Implementation -import io.modelcontextprotocol.kotlin.sdk.types.Tool -import io.modelcontextprotocol.kotlin.sdk.types.ToolSchema -import kotlinx.coroutines.delay -import kotlinx.coroutines.flow.emptyFlow -import kotlinx.coroutines.flow.flow -import kotlinx.coroutines.flow.flowOf -import kotlinx.coroutines.runBlocking -import kotlinx.serialization.json.buildJsonObject -import kotlinx.serialization.json.put -import kotlinx.serialization.json.putJsonObject -import org.junit.jupiter.api.TestInstance -import kotlin.test.Test -import kotlin.time.Duration.Companion.milliseconds -import kotlin.time.Duration.Companion.seconds -import kotlin.uuid.ExperimentalUuidApi -import kotlin.uuid.Uuid - -@OptIn(ExperimentalUuidApi::class) -@TestInstance(TestInstance.Lifecycle.PER_CLASS) -internal class StreamableHttpClientTest : AbstractStreamableHttpClientTest() { - - @Test - fun `Should skip empty SSE`(): Unit = runBlocking { - val client = Client( - clientInfo = Implementation( - name = "client1", - version = "1.0.0", - ), - options = ClientOptions( - capabilities = ClientCapabilities(), - ), - ) - val sessionId = Uuid.random().toString() - - mockMcp.onJSONRPCRequest( - httpMethod = HttpMethod.Post, - jsonRpcMethod = "initialize", - ).respondsWithStream { - headers += MCP_SESSION_ID_HEADER to sessionId - flow = flowOf( - "id: ${Uuid.random()}\n", - "data:\n", // empty data - "\n", - "id: ${Uuid.random()}\n", - "data: \t \n", // tabs and spaces - "\n", - "id: ${Uuid.random()}\n", - "event: message\n", - // multiline data - "data: {\n", - "data: \"result\":{\n" + - "data: \"protocolVersion\":\"2025-06-18\",\n" + - "data: \"capabilities\":{},\n" + - "data: \"serverInfo\":{\"name\":\"simple-streamable-http-server\",\"version\":\"1.0.0\"}\n" + - "data: },\n" + - "data: \"jsonrpc\":\"2.0\",\n" + - "data: \"id\":\"7ce065b0678f49e5b04ce5a0fcc7d518\"\n" + - "data: }\n", - "\n", - ) - } - - mockMcp.handleJSONRPCRequest( - jsonRpcMethod = "notifications/initialized", - expectedSessionId = sessionId, - sessionId = sessionId, - statusCode = HttpStatusCode.Accepted, - ) - - mockMcp.handleSubscribeWithGet(sessionId) { - emptyFlow() - } - - connect(client) - } - - @Test - fun `test streamableHttpClient`() = runBlocking { - val client = Client( - clientInfo = Implementation( - name = "client1", - version = "1.0.0", - ), - options = ClientOptions( - capabilities = ClientCapabilities(), - ), - ) - - val sessionId = Uuid.random().toString() - - mockMcp.onInitialize( - clientName = "client1", - sessionId = sessionId, - ) - - mockMcp.handleJSONRPCRequest( - jsonRpcMethod = "notifications/initialized", - expectedSessionId = sessionId, - sessionId = sessionId, - statusCode = HttpStatusCode.Accepted, - ) - - mockMcp.handleSubscribeWithGet(sessionId) { - flow { - delay(500.milliseconds) - emit( - ServerSentEvent( - event = "message", - id = "1", - //language=json - data = - """{"jsonrpc":"2.0","method":"notifications/progress","params":{"progressToken":"upload-123","progress":50,"total":100}}""", - ), - ) - delay(200.milliseconds) - emit( - ServerSentEvent( - //language=json - data = - """{"jsonrpc":"2.0","method":"notifications/progress","params":{"progressToken":"upload-123","progress":50,"total":100}}""", - ), - ) - } - } - - // TODO: how to get notifications via Client API? - - mockMcp.handleWithResult( - jsonRpcMethod = "tools/list", - sessionId = sessionId, - // language=json - result = """ - { - "tools": [ - { - "name": "get_weather", - "title": "Weather Information Provider", - "description": "Get current weather information for a location", - "inputSchema": { - "type": "object", - "properties": { - "location": { - "type": "string", - "description": "City name or zip code" - } - }, - "required": ["location"] - }, - "outputSchema": { - "type": "object", - "properties": { - "temperature": { - "type": "number", - "description": "Temperature, Celsius" - } - }, - "required": ["temperature"] - }, - "_meta": {} - } - ] - } - """.trimIndent(), - ) - - connect(client) - - val listToolsResult = client.listTools() - - listToolsResult.tools shouldContain Tool( - name = "get_weather", - title = "Weather Information Provider", - description = "Get current weather information for a location", - inputSchema = ToolSchema( - properties = buildJsonObject { - putJsonObject("location") { - put("type", "string") - put("description", "City name or zip code") - } - }, - required = listOf("location"), - ), - outputSchema = ToolSchema( - properties = buildJsonObject { - putJsonObject("temperature") { - put("type", "number") - put("description", "Temperature, Celsius") - } - }, - required = listOf("temperature"), - ), - annotations = null, - meta = EmptyJsonObject, - ) - - client.close() - } - - @Test - fun `terminateSession sends DELETE request`() = runBlocking { - val client = Client( - clientInfo = Implementation(name = "client1", version = "1.0.0"), - options = ClientOptions(capabilities = ClientCapabilities()), - ) - val sessionId = Uuid.random().toString() - - mockMcp.onInitialize(clientName = "client1", sessionId = sessionId) - mockMcp.handleJSONRPCRequest( - jsonRpcMethod = "notifications/initialized", - expectedSessionId = sessionId, - sessionId = sessionId, - statusCode = HttpStatusCode.Accepted, - ) - mockMcp.handleSubscribeWithGet(sessionId) { emptyFlow() } - - connect(client) - - mockMcp.mockUnsubscribeRequest(sessionId = sessionId) - (client.transport as StreamableHttpClientTransport).terminateSession() - (client.transport as StreamableHttpClientTransport).sessionId shouldBe null - - client.close() - } - - @Test - fun `handle MethodNotAllowed`() = runBlocking { - checkSupportNonStreamingResponse( - ContentType.Text.EventStream, - HttpStatusCode.MethodNotAllowed, - ) - } - - @Test - fun `handle non-streaming response`() = runBlocking { - checkSupportNonStreamingResponse( - ContentType.Application.Json, - HttpStatusCode.OK, - ) - } - - private suspend fun checkSupportNonStreamingResponse(contentType: ContentType, statusCode: HttpStatusCode) { - val sessionId = "SID_${Uuid.random().toHexString()}" - val clientName = "client-${Uuid.random().toHexString()}" - val client = Client( - clientInfo = Implementation(name = clientName, version = "1.0.0"), - options = ClientOptions( - capabilities = ClientCapabilities(), - ), - ) - - mockMcp.onInitialize(clientName = clientName, sessionId = sessionId) - - mockMcp.handleJSONRPCRequest( - jsonRpcMethod = "notifications/initialized", - expectedSessionId = sessionId, - sessionId = sessionId, - statusCode = HttpStatusCode.Accepted, - ) - - mockMcp.onSubscribe( - httpMethod = HttpMethod.Get, - sessionId = sessionId, - ) respondsWith { - headers += MCP_SESSION_ID_HEADER to sessionId - body = null - httpStatus = statusCode - this.contentType = contentType - } - - mockMcp.handleWithResult(jsonRpcMethod = "ping", sessionId = sessionId) { - buildJsonObject {} - } - - connect(client) - - delay(1.seconds) - - client.ping() // connection is still alive - - client.close() - } -}