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
This commit is contained in:
Pavel Gorgulov
GitHub
parent
44e012e4ee
commit
664f51b9cc
7 changed files with 343 additions and 575 deletions
|
|
@ -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" }
|
||||
|
|
|
|||
|
|
@ -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<String, String> = 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")
|
||||
}
|
||||
}
|
||||
|
|
@ -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)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<JSONRPCMessage>()
|
||||
val messageReceived = CompletableDeferred<Unit>()
|
||||
|
||||
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<JSONRPCResponse>()
|
||||
response.id shouldBe RequestId.StringId("test-1")
|
||||
|
||||
transport.close()
|
||||
}
|
||||
|
||||
@Test
|
||||
fun testInlineSSEInResponse() = runTest {
|
||||
val transport = createTransport { request ->
|
||||
|
|
|
|||
|
|
@ -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
|
||||
}
|
||||
},
|
||||
),
|
||||
)
|
||||
}
|
||||
}
|
||||
|
|
@ -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<JSONRPCRequest> = 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<Any> = 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<ServerSentEvent>) {
|
||||
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()
|
||||
}
|
||||
}
|
||||
|
|
@ -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()
|
||||
}
|
||||
}
|
||||
Loading…
Reference in a new issue