Add StreamableHttpClientTest with test infrastructure setup and dependencies (#316)
Add `StreamableHttpClientTest` with test infrastructure setup and dependencies <!-- Provide a brief summary of your changes --> ## Motivation and Context To test Streaming HTTP Transport ## How Has This Been Tested? Integration test added ## Breaking Changes No, only tests ## Types of changes <!-- What types of changes does your code introduce? Put an `x` in all the boxes that apply: --> - [ ] 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 - [x] Tests ## Checklist <!-- Go over all the following points, and put an `x` in all the boxes that apply. --> - [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 - [ ] I have added appropriate error handling - [ ] I have added or updated documentation as needed ## Additional context <!-- Add any other context, implementation notes, or design decisions -->
This commit is contained in:
Konstantin Pavlov
GitHub
parent
3b56429126
commit
d60e8b6fb2
5 changed files with 446 additions and 10 deletions
|
|
@ -228,7 +228,8 @@ public class StreamableHttpClientTransport(
|
|||
) {
|
||||
method = HttpMethod.Get
|
||||
applyCommonHeaders(this)
|
||||
accept(ContentType.Text.EventStream)
|
||||
// sseSession will add ContentType.Text.EventStream automatically
|
||||
accept(ContentType.Application.Json)
|
||||
(resumptionToken ?: lastEventId)?.let { headers.append(MCP_RESUMPTION_TOKEN_HEADER, it) }
|
||||
requestBuilder()
|
||||
}
|
||||
|
|
|
|||
|
|
@ -0,0 +1,89 @@
|
|||
package io.modelcontextprotocol.kotlin.sdk.client
|
||||
|
||||
import dev.mokksy.mokksy.Mokksy
|
||||
import dev.mokksy.mokksy.StubConfiguration
|
||||
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.JSONRPCRequest
|
||||
import kotlinx.coroutines.flow.Flow
|
||||
|
||||
/**
|
||||
* High-level helper for simulating an MCP server over Streaming HTTP transport with Server-Sent Events (SSE),
|
||||
* built on top of an HTTP server using the [Mokksy](https://mokksy.dev) library.
|
||||
*
|
||||
* Provides test utilities to mock server behavior based on specific request conditions.
|
||||
*
|
||||
* @param verbose Whether to print detailed logs. Defaults to `false`.
|
||||
* @author Konstantin Pavlov
|
||||
*/
|
||||
internal class MockMcp(verbose: Boolean = false) {
|
||||
|
||||
private val mokksy: Mokksy = Mokksy(verbose = verbose)
|
||||
|
||||
fun checkForUnmatchedRequests() {
|
||||
mokksy.checkForUnmatchedRequests()
|
||||
}
|
||||
|
||||
val url = mokksy.baseUrl() + "/mcp"
|
||||
|
||||
@Suppress("LongParameterList")
|
||||
fun onJSONRPCRequest(
|
||||
httpMethod: HttpMethod = HttpMethod.Post,
|
||||
jsonRpcMethod: String,
|
||||
expectedSessionId: String? = null,
|
||||
sessionId: String,
|
||||
contentType: ContentType = ContentType.Application.Json,
|
||||
statusCode: HttpStatusCode = HttpStatusCode.OK,
|
||||
bodyBuilder: () -> String,
|
||||
) {
|
||||
mokksy.method(
|
||||
configuration = StubConfiguration(removeAfterMatch = true),
|
||||
httpMethod = httpMethod,
|
||||
requestType = JSONRPCRequest::class,
|
||||
) {
|
||||
path("/mcp")
|
||||
expectedSessionId?.let {
|
||||
containsHeader("Mcp-Session-Id", it)
|
||||
}
|
||||
bodyMatchesPredicates(
|
||||
{
|
||||
it!!.method == jsonRpcMethod
|
||||
},
|
||||
{
|
||||
it!!.jsonrpc == "2.0"
|
||||
},
|
||||
)
|
||||
} respondsWith {
|
||||
body = bodyBuilder.invoke()
|
||||
this.contentType = contentType
|
||||
headers += "Mcp-Session-Id" to sessionId
|
||||
httpStatus = statusCode
|
||||
}
|
||||
}
|
||||
|
||||
fun onSubscribeWithGet(sessionId: String, block: () -> Flow<ServerSentEvent>) {
|
||||
mokksy.get(name = "MCP GETs", requestType = Any::class) {
|
||||
path("/mcp")
|
||||
containsHeader("Mcp-Session-Id", sessionId)
|
||||
containsHeader("Accept", "application/json,text/event-stream")
|
||||
containsHeader("Cache-Control", "no-store")
|
||||
} respondsWithSseStream {
|
||||
headers += "Mcp-Session-Id" 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", sessionId)
|
||||
} respondsWith {
|
||||
body = null
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,202 @@
|
|||
package io.modelcontextprotocol.kotlin.sdk.client
|
||||
|
||||
import io.kotest.matchers.collections.shouldContain
|
||||
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 io.ktor.http.HttpStatusCode
|
||||
import io.ktor.sse.ServerSentEvent
|
||||
import io.modelcontextprotocol.kotlin.sdk.ClientCapabilities
|
||||
import io.modelcontextprotocol.kotlin.sdk.Implementation
|
||||
import io.modelcontextprotocol.kotlin.sdk.Tool
|
||||
import kotlinx.coroutines.delay
|
||||
import kotlinx.coroutines.flow.flow
|
||||
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 java.util.UUID
|
||||
import kotlin.test.AfterTest
|
||||
import kotlin.test.Test
|
||||
import kotlin.time.Duration.Companion.milliseconds
|
||||
|
||||
/**
|
||||
* Integration tests for the `StreamableHttpClientTransport` implementation
|
||||
* using the [Mokksy](https://mokksy.dev) library
|
||||
* to simulate Streaming HTTP with server-sent events (SSE).
|
||||
* @author Konstantin Pavlov
|
||||
*/
|
||||
@TestInstance(TestInstance.Lifecycle.PER_CLASS)
|
||||
class StreamableHttpClientTest {
|
||||
|
||||
// start mokksy on random port
|
||||
private val mockMcp: MockMcp = MockMcp(verbose = true)
|
||||
|
||||
@AfterTest
|
||||
fun afterEach() {
|
||||
mockMcp.checkForUnmatchedRequests()
|
||||
}
|
||||
|
||||
@Test
|
||||
@Suppress("LongMethod")
|
||||
fun `test streamableHttpClient`(): Unit = runBlocking {
|
||||
val client = Client(
|
||||
clientInfo = Implementation(name = "sample-client", version = "1.0.0"),
|
||||
options = ClientOptions(
|
||||
capabilities = ClientCapabilities(),
|
||||
),
|
||||
)
|
||||
|
||||
val sessionId = UUID.randomUUID().toString()
|
||||
|
||||
mockMcp.onJSONRPCRequest(
|
||||
jsonRpcMethod = "initialize",
|
||||
sessionId = sessionId,
|
||||
) {
|
||||
// language=json
|
||||
"""
|
||||
{
|
||||
"jsonrpc": "2.0",
|
||||
"id": 1,
|
||||
"result": {
|
||||
"capabilities": {
|
||||
"tools": {
|
||||
"listChanged": false
|
||||
}
|
||||
},
|
||||
"protocolVersion": "2025-03-26",
|
||||
"serverInfo": {
|
||||
"name": "Mock MCP Server",
|
||||
"version": "1.0.0"
|
||||
},
|
||||
"_meta": {
|
||||
"foo": "bar"
|
||||
}
|
||||
}
|
||||
}
|
||||
""".trimIndent()
|
||||
}
|
||||
|
||||
mockMcp.onJSONRPCRequest(
|
||||
jsonRpcMethod = "notifications/initialized",
|
||||
expectedSessionId = sessionId,
|
||||
sessionId = sessionId,
|
||||
statusCode = HttpStatusCode.Accepted,
|
||||
) {
|
||||
""
|
||||
}
|
||||
|
||||
mockMcp.onSubscribeWithGet(sessionId) {
|
||||
flow {
|
||||
delay(500.milliseconds)
|
||||
emit(
|
||||
ServerSentEvent(
|
||||
event = "message",
|
||||
id = "1",
|
||||
data = @Suppress("MaxLineLength")
|
||||
//language=json
|
||||
"""{"jsonrpc":"2.0","method":"notifications/progress","params":{"progressToken":"upload-123","progress":50,"total":100}}""",
|
||||
),
|
||||
)
|
||||
delay(200.milliseconds)
|
||||
emit(
|
||||
ServerSentEvent(
|
||||
data = @Suppress("MaxLineLength")
|
||||
//language=json
|
||||
"""{"jsonrpc":"2.0","method":"notifications/progress","params":{"progressToken":"upload-123","progress":50,"total":100}}""",
|
||||
),
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
client.connect(
|
||||
StreamableHttpClientTransport(
|
||||
url = mockMcp.url,
|
||||
client = HttpClient(Apache5) {
|
||||
install(SSE)
|
||||
install(Logging) {
|
||||
level = LogLevel.ALL
|
||||
}
|
||||
},
|
||||
),
|
||||
)
|
||||
|
||||
// TODO: how to get notifications via Client API?
|
||||
|
||||
mockMcp.onJSONRPCRequest(
|
||||
jsonRpcMethod = "tools/list",
|
||||
sessionId = sessionId,
|
||||
) {
|
||||
// language=json
|
||||
"""
|
||||
{
|
||||
"jsonrpc": "2.0",
|
||||
"id": 3,
|
||||
"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"]
|
||||
}
|
||||
}
|
||||
]
|
||||
}
|
||||
}
|
||||
""".trimIndent()
|
||||
}
|
||||
|
||||
val listToolsResult = client.listTools()
|
||||
|
||||
listToolsResult.tools shouldContain Tool(
|
||||
name = "get_weather",
|
||||
title = "Weather Information Provider",
|
||||
description = "Get current weather information for a location",
|
||||
inputSchema = Tool.Input(
|
||||
properties = buildJsonObject {
|
||||
putJsonObject("location") {
|
||||
put("type", "string")
|
||||
put("description", "City name or zip code")
|
||||
}
|
||||
},
|
||||
required = listOf("location"),
|
||||
),
|
||||
outputSchema = Tool.Output(
|
||||
properties = buildJsonObject {
|
||||
putJsonObject("temperature") {
|
||||
put("type", "number")
|
||||
put("description", "Temperature, Celsius")
|
||||
}
|
||||
},
|
||||
required = listOf("temperature"),
|
||||
),
|
||||
annotations = null,
|
||||
)
|
||||
|
||||
mockMcp.mockUnsubscribeRequest(sessionId = sessionId)
|
||||
|
||||
client.close()
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,6 @@
|
|||
## https://docs.junit.org/5.3.0-M1/user-guide/index.html#writing-tests-parallel-execution
|
||||
junit.jupiter.execution.parallel.enabled=true
|
||||
junit.jupiter.execution.parallel.config.strategy=dynamic
|
||||
junit.jupiter.execution.parallel.mode.default=concurrent
|
||||
junit.jupiter.execution.parallel.mode.classes.default=concurrent
|
||||
junit.jupiter.execution.timeout.default=2m
|
||||
|
|
@ -0,0 +1,14 @@
|
|||
# Level of logging for the ROOT logger: ERROR, WARN, INFO, DEBUG, TRACE (default is INFO)
|
||||
org.slf4j.simpleLogger.defaultLogLevel=INFO
|
||||
|
||||
|
||||
org.slf4j.simpleLogger.showThreadName=true
|
||||
org.slf4j.simpleLogger.showDateTime=false
|
||||
|
||||
# Whether to enable stack traces for exceptions (true/false, default is true)
|
||||
org.slf4j.simpleLogger.showShortLogName=false
|
||||
|
||||
# Log level for specific packages or classes
|
||||
org.slf4j.simpleLogger.log.io.ktor.server=DEBUG
|
||||
org.slf4j.simpleLogger.log.io.modelcontextprotocol=TRACE
|
||||
org.slf4j.simpleLogger.log.dev.mokksy=DEBUG
|
||||
Loading…
Reference in a new issue