Compare commits

..
77 changed files with 457 additions and 5769 deletions

View file

@ -27,14 +27,14 @@ jobs:
- uses: actions/checkout@v7
- name: Set up JDK ${{ env.JAVA_VERSION }}
uses: actions/setup-java@v6
uses: actions/setup-java@v5
with:
java-version: ${{ env.JAVA_VERSION }}
distribution: ${{ env.JAVA_DISTRIBUTION }}
cache: gradle
- name: Setup Gradle
uses: gradle/actions/setup-gradle@v6.3.0
uses: gradle/actions/setup-gradle@v6
- name: Generate Dokka Site
run: |-

View file

@ -27,7 +27,7 @@ jobs:
name: Mark/unmark stale PRs and issues
runs-on: ubuntu-latest
steps:
- uses: actions/stale@v11
- uses: actions/stale@v10
with:
debug-only: ${{ env.DRY_RUN }}
exempt-draft-pr: true
@ -66,7 +66,7 @@ jobs:
name: Close stale PRs and issues
runs-on: ubuntu-latest
steps:
- uses: actions/stale@v11
- uses: actions/stale@v10
with:
debug-only: ${{ env.DRY_RUN }}
only-labels: 'stale'

View file

@ -46,14 +46,14 @@ jobs:
- uses: actions/checkout@v7
- name: Set up JDK ${{ env.JAVA_VERSION }}
uses: actions/setup-java@v6
uses: actions/setup-java@v5
with:
java-version: ${{ env.JAVA_VERSION }}
distribution: ${{ env.JAVA_DISTRIBUTION }}
- name: Set up Node.js
if: matrix.os == 'ubuntu-latest'
uses: actions/setup-node@v7
uses: actions/setup-node@v6
with:
node-version: '22' # increase only after https://github.com/nodejs/node/issues/56645 will be fixed
@ -63,7 +63,7 @@ jobs:
run: npm ci
- name: Setup Gradle
uses: gradle/actions/setup-gradle@v6.3.0
uses: gradle/actions/setup-gradle@v6
with:
validate-wrappers: true
add-job-summary: 'always'
@ -129,13 +129,13 @@ jobs:
- uses: actions/checkout@v7
- name: Set up JDK ${{ env.JAVA_VERSION }}
uses: actions/setup-java@v6
uses: actions/setup-java@v5
with:
java-version: ${{ env.JAVA_VERSION }}
distribution: ${{ env.JAVA_DISTRIBUTION }}
- name: Setup Gradle
uses: gradle/actions/setup-gradle@v6.3.0
uses: gradle/actions/setup-gradle@v6
with:
validate-wrappers: true
add-job-summary: 'always'
@ -171,13 +171,13 @@ jobs:
- uses: actions/checkout@v7
- name: Set up JDK ${{ env.JAVA_VERSION }}
uses: actions/setup-java@v6
uses: actions/setup-java@v5
with:
java-version: ${{ env.JAVA_VERSION }}
distribution: ${{ env.JAVA_DISTRIBUTION }}
- name: Setup Gradle
uses: gradle/actions/setup-gradle@v6.3.0
uses: gradle/actions/setup-gradle@v6
with:
add-job-summary: 'always'
cache-read-only: true

View file

@ -28,19 +28,19 @@ jobs:
uses: actions/checkout@v7
- name: Set up JDK ${{ env.JAVA_VERSION }}
uses: actions/setup-java@v6
uses: actions/setup-java@v5
with:
distribution: ${{ env.JAVA_DISTRIBUTION }}
java-version: ${{ env.JAVA_VERSION }}
- name: Initialize CodeQL
uses: github/codeql-action/init@v4.38.0
uses: github/codeql-action/init@v4
with:
languages: java-kotlin
build-mode: manual
- name: Setup Gradle
uses: gradle/actions/setup-gradle@v6.3.0
uses: gradle/actions/setup-gradle@v6
with:
validate-wrappers: true
cache-read-only: true
@ -52,6 +52,6 @@ jobs:
--no-daemon --rerun-tasks --stacktrace
- name: Analyze
uses: github/codeql-action/analyze@v4.38.0
uses: github/codeql-action/analyze@v4
with:
category: '/language:java-kotlin'

View file

@ -28,13 +28,13 @@ jobs:
- uses: actions/checkout@v7
- name: Set up JDK
uses: actions/setup-java@v6
uses: actions/setup-java@v5
with:
java-version: ${{ env.JAVA_VERSION }}
distribution: ${{ env.JAVA_DISTRIBUTION }}
- name: Setup Gradle
uses: gradle/actions/setup-gradle@v6.3.0
uses: gradle/actions/setup-gradle@v6
with:
cache-read-only: ${{ github.ref != 'refs/heads/main' }}
@ -79,13 +79,13 @@ jobs:
- uses: actions/checkout@v7
- name: Set up JDK
uses: actions/setup-java@v6
uses: actions/setup-java@v5
with:
java-version: ${{ env.JAVA_VERSION }}
distribution: ${{ env.JAVA_DISTRIBUTION }}
- name: Setup Gradle
uses: gradle/actions/setup-gradle@v6.3.0
uses: gradle/actions/setup-gradle@v6
with:
cache-read-only: ${{ github.ref != 'refs/heads/main' }}
@ -110,13 +110,13 @@ jobs:
- uses: actions/checkout@v7
- name: Set up JDK
uses: actions/setup-java@v6
uses: actions/setup-java@v5
with:
java-version: ${{ env.JAVA_VERSION }}
distribution: ${{ env.JAVA_DISTRIBUTION }}
- name: Setup Gradle
uses: gradle/actions/setup-gradle@v6.3.0
uses: gradle/actions/setup-gradle@v6
with:
cache-read-only: ${{ github.ref != 'refs/heads/main' }}
@ -149,13 +149,13 @@ jobs:
- uses: actions/checkout@v7
- name: Set up JDK
uses: actions/setup-java@v6
uses: actions/setup-java@v5
with:
java-version: ${{ env.JAVA_VERSION }}
distribution: ${{ env.JAVA_DISTRIBUTION }}
- name: Setup Gradle
uses: gradle/actions/setup-gradle@v6.3.0
uses: gradle/actions/setup-gradle@v6
with:
cache-read-only: ${{ github.ref != 'refs/heads/main' }}

View file

@ -19,18 +19,18 @@ jobs:
steps:
- uses: actions/checkout@v7
- name: Set up JDK 21
uses: actions/setup-java@v6
uses: actions/setup-java@v5
with:
java-version: '21'
distribution: 'temurin'
- name: Set up Node.js
uses: actions/setup-node@v7
uses: actions/setup-node@v6
with:
node-version: '22' # increase only after https://github.com/nodejs/node/issues/56645 will be fixed
- name: Setup Gradle
uses: gradle/actions/setup-gradle@v6.3.0
uses: gradle/actions/setup-gradle@v6
- name: Check
run: ./gradlew check --no-configuration-cache

View file

@ -39,13 +39,13 @@ jobs:
- uses: actions/checkout@v7
- name: Set up JDK ${{ env.JAVA_VERSION }}
uses: actions/setup-java@v6
uses: actions/setup-java@v5
with:
java-version: ${{ env.JAVA_VERSION }}
distribution: ${{ env.JAVA_DISTRIBUTION }}
- name: Setup Gradle
uses: gradle/actions/setup-gradle@v6.3.0
uses: gradle/actions/setup-gradle@v6
with:
validate-wrappers: true
add-job-summary: 'always'

View file

@ -729,7 +729,7 @@ val client = Client(
)
client.setRequestHandler<CreateMessageRequest>(Method.Defined.SamplingCreateMessage) { request, _ ->
val content = request.messages.lastOrNull()?.content?.firstOrNull()
val content = request.messages.lastOrNull()?.content
val prompt = if (content is TextContent) content.text else "your topic"
CreateMessageResult(
model = "gpt-4o-mini",

View file

@ -416,6 +416,7 @@ potential-bugs:
Deprecation:
active: false
aliases: ['DEPRECATION']
excludeImportStatements: false
DontDowncastCollectionTypes:
active: false
DoubleMutabilityForCollection:

View file

@ -8,4 +8,4 @@ kotlin.daemon.jvmargs=-Xmx4G
kotlin.native.ignoreDisabledTargets=true
group=io.modelcontextprotocol
version=0.15.0
version=0.14.0

View file

@ -1,28 +1,28 @@
[versions]
# plugins version
atomicfu = "0.33.0"
binaryCompatibilityValidatorPlugin = "0.18.2"
detekt = "2.0.0-alpha.6"
binaryCompatibilityValidatorPlugin = "0.18.1"
detekt = "2.0.0-alpha.3"
dokka = "2.2.0"
kotlin = "2.4.10"
kover = "0.9.9"
kotlin = "2.3.21"
kover = "0.9.8"
ktlint = "14.2.0"
knit = "0.5.1"
mavenPublish = "0.37.0"
netty = "4.2.18.Final"
netty = "4.2.15.Final"
# libraries version
awaitility = "4.3.0"
collections-immutable = "0.5.2"
collections-immutable = "0.5.0"
coroutines = "1.11.0"
kotest = "6.2.5"
kotest = "6.2.1"
kotlinx-io = "0.9.1"
ktor = "3.5.2"
ktor = "3.4.3"
logging = "8.0.4"
mockk = "1.14.11"
serialization = "1.11.0"
slf4j = "2.0.19"
logback = "1.6.3"
slf4j = "2.0.18"
logback = "1.5.37"
junit = "5.14.4"
[libraries]

Binary file not shown.

View file

@ -1,6 +1,6 @@
distributionBase=GRADLE_USER_HOME
distributionPath=wrapper/dists
distributionUrl=https\://services.gradle.org/distributions/gradle-9.7.1-bin.zip
distributionUrl=https\://services.gradle.org/distributions/gradle-9.6.1-bin.zip
networkTimeout=10000
retries=0
retryBackOffMs=500

View file

@ -24,7 +24,6 @@ import io.modelcontextprotocol.kotlin.sdk.types.InitializeRequest
import io.modelcontextprotocol.kotlin.sdk.types.InitializeResult
import io.modelcontextprotocol.kotlin.sdk.types.IntegerSchema
import io.modelcontextprotocol.kotlin.sdk.types.JSONRPCMessage
import io.modelcontextprotocol.kotlin.sdk.types.JSONRPCNotification
import io.modelcontextprotocol.kotlin.sdk.types.JSONRPCRequest
import io.modelcontextprotocol.kotlin.sdk.types.JSONRPCResponse
import io.modelcontextprotocol.kotlin.sdk.types.LATEST_PROTOCOL_VERSION
@ -53,7 +52,6 @@ import io.modelcontextprotocol.kotlin.sdk.types.UntitledSingleSelectEnumSchema
import io.modelcontextprotocol.kotlin.sdk.types.UrlElicitationRequiredException
import kotlinx.coroutines.CompletableDeferred
import kotlinx.coroutines.TimeoutCancellationException
import kotlinx.coroutines.awaitCancellation
import kotlinx.coroutines.cancel
import kotlinx.coroutines.delay
import kotlinx.coroutines.joinAll
@ -73,7 +71,6 @@ import kotlin.test.assertIs
import kotlin.test.assertNull
import kotlin.test.assertTrue
import kotlin.test.fail
import kotlin.time.Duration.Companion.seconds
class ClientTest {
@Test
@ -546,15 +543,6 @@ class ClientTest {
val serverSession = serverSessionResult.await()
// Register the tap after createSession so it chains onto the session's subscription instead
// of pre-empting the handshake.
val cancelledOnWire = CompletableDeferred<Unit>()
serverTransport.onMessage { message ->
if (message is JSONRPCNotification && message.method == Method.Defined.NotificationsCancelled.value) {
cancelledOnWire.complete(Unit)
}
}
serverSession.setRequestHandler<ListResourcesRequest>(Method.Defined.ResourcesList) { _, _ ->
// Simulate delay
def.complete(Unit)
@ -580,7 +568,6 @@ class ClientTest {
def.await()
runCatching { job.cancel("Cancelled by test") }
defCancel.await()
cancelledOnWire.await()
defTimeOut.await()
}
@ -619,9 +606,17 @@ class ClientTest {
val serverSession = serverSessionResult.await()
serverSession.setRequestHandler<ListResourcesRequest>(Method.Defined.ResourcesList) { _, _ ->
// Never respond: the handler runs on a real dispatcher while the 1 ms timeout runs on the
// virtual test clock, so a real-time delay could race it. Parking keeps the test deterministic.
awaitCancellation()
// Simulate a delayed response
// Wait ~100ms unless canceled
try {
withTimeout(100L) {
// Just delay here, if timeout is 0 on the client side, this won't return in time
delay(100)
}
} catch (_: Exception) {
// If aborted, just rethrow or return early
}
ListResourcesResult(resources = emptyList())
}
// Request with 1 msec timeout should fail immediately
@ -900,8 +895,8 @@ class ClientTest {
),
)
// The handler runs on Dispatchers.Default after concurrent dispatch is enabled; await the deferred.
val rootListChangedReceived = CompletableDeferred<Unit>()
// Track notifications
var rootListChangedNotificationReceived = false
val serverSessionResult = CompletableDeferred<ServerSession>()
@ -920,13 +915,16 @@ class ClientTest {
serverSession.setNotificationHandler<RootsListChangedNotification>(
Method.Defined.NotificationsRootsListChanged,
) {
rootListChangedReceived.complete(Unit)
rootListChangedNotificationReceived = true
CompletableDeferred(Unit)
}
client.sendRootsListChanged()
withTimeout(5.seconds) { rootListChangedReceived.await() }
assertTrue(
rootListChangedNotificationReceived,
"Notification should be sent when sendRootsListChanged is called",
)
}
@Test
@ -1124,103 +1122,6 @@ class ClientTest {
assertEquals(elicitationResultContent, result.content)
}
@Test
fun `should reject accepted elicitation content that does not match requested schema`() = runTest {
val requestedSchema = ElicitRequestParams.RequestedSchema(
properties = mapOf("name" to StringSchema()),
required = listOf("name"),
)
val (client, serverSession) = setupElicitationPair {
ElicitResult(action = ElicitResult.Action.Accept, content = buildJsonObject { put("name", 42) })
}
val exception = assertFailsWith<McpException> {
serverSession.createElicitation(message = "mismatch", requestedSchema = requestedSchema)
}
assertTrue(
exception.message.orEmpty()
.startsWith("Elicitation response content does not match requested schema: "),
"Unexpected message: ${exception.message}",
)
assertTrue("'name' must be string" in exception.message.orEmpty(), "Unexpected message: ${exception.message}")
client.close()
}
@Test
fun `should reject accepted elicitation without content when schema has required properties`() = runTest {
val requestedSchema = ElicitRequestParams.RequestedSchema(
properties = mapOf("name" to StringSchema()),
required = listOf("name"),
)
val (client, serverSession) = setupElicitationPair {
ElicitResult(action = ElicitResult.Action.Accept, content = null)
}
val exception = assertFailsWith<McpException> {
serverSession.createElicitation(message = "no content", requestedSchema = requestedSchema)
}
assertTrue(
"must have required property 'name'" in exception.message.orEmpty(),
"Unexpected message: ${exception.message}",
)
client.close()
}
@Test
fun `should accept elicitation without content when schema has no required properties`() = runTest {
val requestedSchema = ElicitRequestParams.RequestedSchema(
properties = mapOf("name" to StringSchema()),
)
val (client, serverSession) = setupElicitationPair {
ElicitResult(action = ElicitResult.Action.Accept, content = null)
}
val result = serverSession.createElicitation(message = "no required", requestedSchema = requestedSchema)
assertEquals(ElicitResult.Action.Accept, result.action)
assertNull(result.content)
client.close()
}
@Test
fun `should pass through declined elicitation without validation`() = runTest {
val requestedSchema = ElicitRequestParams.RequestedSchema(
properties = mapOf("name" to StringSchema()),
required = listOf("name"),
)
val (client, serverSession) = setupElicitationPair {
ElicitResult(action = ElicitResult.Action.Decline)
}
val result = serverSession.createElicitation(message = "decline", requestedSchema = requestedSchema)
assertEquals(ElicitResult.Action.Decline, result.action)
assertNull(result.content)
client.close()
}
@Test
fun `should pass through cancelled elicitation without validation`() = runTest {
val requestedSchema = ElicitRequestParams.RequestedSchema(
properties = mapOf("name" to StringSchema()),
required = listOf("name"),
)
val (client, serverSession) = setupElicitationPair {
ElicitResult(action = ElicitResult.Action.Cancel)
}
val result = serverSession.createElicitation(message = "cancel", requestedSchema = requestedSchema)
assertEquals(ElicitResult.Action.Cancel, result.action)
assertNull(result.content)
client.close()
}
@Test
fun `should apply elicitation defaults for missing fields in empty content`() = runTest {
val schema = defaultsTestSchema()

View file

@ -1,306 +0,0 @@
package io.modelcontextprotocol.kotlin.sdk.integration
import io.kotest.matchers.shouldBe
import io.modelcontextprotocol.kotlin.sdk.ExperimentalMcpApi
import io.modelcontextprotocol.kotlin.sdk.client.Client
import io.modelcontextprotocol.kotlin.sdk.client.ClientOptions
import io.modelcontextprotocol.kotlin.sdk.server.Server
import io.modelcontextprotocol.kotlin.sdk.server.ServerOptions
import io.modelcontextprotocol.kotlin.sdk.server.ServerSession
import io.modelcontextprotocol.kotlin.sdk.shared.InMemoryTransport
import io.modelcontextprotocol.kotlin.sdk.shared.currentRequestHandlerExtra
import io.modelcontextprotocol.kotlin.sdk.testing.ChannelTransport
import io.modelcontextprotocol.kotlin.sdk.types.CallToolRequest
import io.modelcontextprotocol.kotlin.sdk.types.CallToolRequestParams
import io.modelcontextprotocol.kotlin.sdk.types.CallToolResult
import io.modelcontextprotocol.kotlin.sdk.types.ClientCapabilities
import io.modelcontextprotocol.kotlin.sdk.types.CreateMessageRequest
import io.modelcontextprotocol.kotlin.sdk.types.CreateMessageRequestParams
import io.modelcontextprotocol.kotlin.sdk.types.CreateMessageResult
import io.modelcontextprotocol.kotlin.sdk.types.Implementation
import io.modelcontextprotocol.kotlin.sdk.types.InitializeRequest
import io.modelcontextprotocol.kotlin.sdk.types.InitializeRequestParams
import io.modelcontextprotocol.kotlin.sdk.types.InitializedNotification
import io.modelcontextprotocol.kotlin.sdk.types.JSONRPCError
import io.modelcontextprotocol.kotlin.sdk.types.JSONRPCMessage
import io.modelcontextprotocol.kotlin.sdk.types.JSONRPCResponse
import io.modelcontextprotocol.kotlin.sdk.types.LATEST_PROTOCOL_VERSION
import io.modelcontextprotocol.kotlin.sdk.types.ListRootsRequest
import io.modelcontextprotocol.kotlin.sdk.types.ListRootsResult
import io.modelcontextprotocol.kotlin.sdk.types.ListToolsRequest
import io.modelcontextprotocol.kotlin.sdk.types.ListToolsResult
import io.modelcontextprotocol.kotlin.sdk.types.Method
import io.modelcontextprotocol.kotlin.sdk.types.RequestId
import io.modelcontextprotocol.kotlin.sdk.types.Role
import io.modelcontextprotocol.kotlin.sdk.types.SamplingMessage
import io.modelcontextprotocol.kotlin.sdk.types.ServerCapabilities
import io.modelcontextprotocol.kotlin.sdk.types.TextContent
import io.modelcontextprotocol.kotlin.sdk.types.toJSON
import kotlinx.coroutines.CompletableDeferred
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.withTimeout
import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.CopyOnWriteArrayList
import java.util.concurrent.atomic.AtomicReference
import kotlin.coroutines.cancellation.CancellationException
import kotlin.test.Test
import kotlin.test.assertEquals
import kotlin.test.assertFalse
import kotlin.time.Duration.Companion.seconds
/**
* End-to-end tests for concurrent inbound dispatch enabled after the MCP handshake.
*
* Handlers run on a real [kotlinx.coroutines.Dispatchers.Default], so these use [runBlocking] with
* explicit [withTimeout] guards. A virtual test clock would false-fire timeouts against cross-thread
* real dispatchers.
*/
@OptIn(ExperimentalMcpApi::class)
class ProtocolConcurrencyIntegrationTest {
private fun rootsClient(): Client = Client(
clientInfo = Implementation(name = "test client", version = "1.0"),
options = ClientOptions(
capabilities = ClientCapabilities(roots = ClientCapabilities.Roots(listChanged = null)),
),
)
private fun toolsServer(): Server = Server(
serverInfo = Implementation(name = "test server", version = "1.0"),
options = ServerOptions(
capabilities = ServerCapabilities(tools = ServerCapabilities.Tools(listChanged = null)),
),
)
// ChannelTransport delivers inbound messages as soon as its read loop starts, which is before
// start() marks it operational. Creating the session first keeps the client's initialize from
// landing on a transport that would reject the session's response.
private suspend fun connect(
client: Client,
clientTransport: ChannelTransport,
server: Server,
serverTransport: ChannelTransport,
): ServerSession {
val serverSession = server.createSession(serverTransport)
client.connect(clientTransport)
return serverSession
}
// The motivating deadlock over real serial read loops: a server tool handler makes a
// server-to-client request (roots/list) while still handling the client's tools/call.
@Test
fun `server tool handler can call roots list mid-handling without deadlock`() = runBlocking {
val client = rootsClient()
val server = toolsServer()
val (clientTransport, serverTransport) = ChannelTransport.createLinkedPair()
val serverSession = connect(client, clientTransport, server, serverTransport)
// Register the tools/call handler directly on the session so it can call back to the client
// mid-handling. Before concurrent dispatch this would deadlock: the server read loop is
// blocked by the tool handler and can never deliver the roots/list response.
serverSession.setRequestHandler<CallToolRequest>(Method.Defined.ToolsCall) { _, _ ->
val roots = serverSession.request<ListRootsResult>(ListRootsRequest())
CallToolResult(content = listOf(TextContent(text = "roots=${roots.roots.size}")))
}
val result = withTimeout(10.seconds) {
client.callTool(CallToolRequest(CallToolRequestParams("needs-roots")))
}
val text = (result as CallToolResult).content.filterIsInstance<TextContent>().single().text
text shouldBe "roots=0"
client.close()
server.close()
}
// Client-side mirror: a client sampling handler makes a client-to-server request (tools/list)
// while still handling the server's sampling/createMessage.
@Test
fun `client sampling handler can call the server mid-handling without deadlock`() = runBlocking {
val client = Client(
clientInfo = Implementation(name = "test client", version = "1.0"),
options = ClientOptions(
capabilities = ClientCapabilities(sampling = ClientCapabilities.Sampling()),
),
)
val server = toolsServer()
val (clientTransport, serverTransport) = ChannelTransport.createLinkedPair()
client.setRequestHandler<CreateMessageRequest>(Method.Defined.SamplingCreateMessage) { _, _ ->
val tools = client.request<ListToolsResult>(ListToolsRequest())
CreateMessageResult(
role = Role.Assistant,
content = TextContent(text = "tools=${tools.tools.size}"),
model = "test-model",
)
}
val serverSession = connect(client, clientTransport, server, serverTransport)
val result = withTimeout(10.seconds) {
serverSession.request<CreateMessageResult>(
CreateMessageRequest(
CreateMessageRequestParams(
maxTokens = 100,
messages = listOf(SamplingMessage(role = Role.User, content = TextContent(text = "hi"))),
),
),
)
}
val text = result.content.filterIsInstance<TextContent>().single().text
text shouldBe "tools=0"
client.close()
server.close()
}
// Nested-cancel cascade: cancelling the client's tools/call must reach the server tool handler
// over the wire and cascade into its nested roots/list request, cancelling the parked
// client-side roots handler. No tools/call response is ever sent.
@Test
fun `cancelling a tool call cascades into its nested roots list request`() = runBlocking {
val client = rootsClient()
val server = toolsServer()
val (clientTransport, serverTransport) = ChannelTransport.createLinkedPair()
val capturedToolCallId = AtomicReference<RequestId?>(null)
val toolHandlerCancelled = CompletableDeferred<CancellationException>()
val rootsHandlerEntered = CompletableDeferred<Unit>()
val rootsHandlerCancelled = CompletableDeferred<CancellationException>()
val parkForever = CompletableDeferred<ListRootsResult>()
val unexpectedToolCallResponse = CompletableDeferred<Unit>()
// Tap the server->client direction (client transport inbound) to prove no tools/call
// response is ever delivered once the call is cancelled.
clientTransport.onMessage { message ->
if (message is JSONRPCResponse && message.id == capturedToolCallId.get()) {
unexpectedToolCallResponse.complete(Unit)
}
}
val serverSession = connect(client, clientTransport, server, serverTransport)
// Client roots handler parks until cancelled and records the cancellation.
client.setRequestHandler<ListRootsRequest>(Method.Defined.RootsList) { _, _ ->
rootsHandlerEntered.complete(Unit)
try {
parkForever.await()
} catch (e: CancellationException) {
rootsHandlerCancelled.complete(e)
throw e
}
}
// Tool handler makes a nested server->client request and records its own cancellation.
serverSession.setRequestHandler<CallToolRequest>(Method.Defined.ToolsCall) { _, _ ->
val extra = currentRequestHandlerExtra()!!
capturedToolCallId.set(extra.requestId)
try {
extra.sendRequest<ListRootsResult>(ListRootsRequest())
} catch (e: CancellationException) {
toolHandlerCancelled.complete(e)
throw e
}
CallToolResult(content = listOf(TextContent(text = "unreachable")))
}
val call = launch { client.callTool(CallToolRequest(CallToolRequestParams("nested"))) }
withTimeout(10.seconds) { rootsHandlerEntered.await() }
call.cancel(CancellationException("user aborted"))
withTimeout(10.seconds) { toolHandlerCancelled.await() }
withTimeout(10.seconds) { rootsHandlerCancelled.await() }
// The server suppresses the response for a cancelled request, so none is ever sent.
assertFalse(
unexpectedToolCallResponse.isCompleted,
"no tools/call response should be delivered after the call was cancelled",
)
client.close()
server.close()
}
// Init gate plus context element in a high-level handler: initialize and tools/list are
// pipelined before notifications/initialized and answered serially in order. After the
// handshake a high-level addTool handler can observe currentRequestHandlerExtra().
@Test
fun `pipelined initialize and tools list are processed serially then context element is available`(): Unit =
runBlocking {
val server = toolsServer()
server.addTool(name = "ctx-tool", description = "returns its request id") {
val extra = currentRequestHandlerExtra()
checkNotNull(extra) { "currentRequestHandlerExtra() must be non-null inside a tool handler" }
CallToolResult(content = listOf(TextContent(text = extra.requestId.asText())))
}
val (clientTransport, serverTransport) = InMemoryTransport.createLinkedPair()
val arrivalOrder = CopyOnWriteArrayList<RequestId>()
val pending = ConcurrentHashMap<RequestId, CompletableDeferred<JSONRPCMessage>>()
clientTransport.onMessage { message ->
val id = when (message) {
is JSONRPCResponse -> message.id
is JSONRPCError -> message.id
else -> null
}
if (id != null) {
arrivalOrder.add(id)
pending[id]?.complete(message)
}
}
// Drive the client side manually over the raw transport.
server.createSession(serverTransport)
val initRequest = InitializeRequest(
InitializeRequestParams(
protocolVersion = LATEST_PROTOCOL_VERSION,
capabilities = ClientCapabilities(),
clientInfo = Implementation(name = "raw-client", version = "1.0"),
),
).toJSON()
val toolsListRequest = ListToolsRequest().toJSON()
val initAnswered = pending.expect(initRequest.id)
val toolsListAnswered = pending.expect(toolsListRequest.id)
// Pipelined before notifications/initialized: both handled in the serial dispatch phase.
clientTransport.send(initRequest)
clientTransport.send(toolsListRequest)
withTimeout(10.seconds) { initAnswered.await() }
withTimeout(10.seconds) { toolsListAnswered.await() }
// Deterministic serial ordering: initialize answered before tools/list.
assertEquals(
listOf(initRequest.id, toolsListRequest.id),
arrivalOrder.toList(),
"pipelined requests must be answered in arrival order",
)
// Complete the handshake, then call the high-level tool.
clientTransport.send(InitializedNotification().toJSON())
val toolCallRequest = CallToolRequest(CallToolRequestParams("ctx-tool")).toJSON()
val toolCallAnswered = pending.expect(toolCallRequest.id)
clientTransport.send(toolCallRequest)
val toolCallResponse = withTimeout(10.seconds) { toolCallAnswered.await() } as JSONRPCResponse
val text = (toolCallResponse.result as CallToolResult).content.filterIsInstance<TextContent>().single().text
// The handler echoed its request id, proving currentRequestHandlerExtra() was available.
text shouldBe toolCallRequest.id.asText()
}
private fun ConcurrentHashMap<RequestId, CompletableDeferred<JSONRPCMessage>>.expect(
id: RequestId,
): CompletableDeferred<JSONRPCMessage> = CompletableDeferred<JSONRPCMessage>().also { put(id, it) }
private fun RequestId.asText(): String = when (this) {
is RequestId.StringId -> value
is RequestId.NumberId -> value.toString()
}
}

View file

@ -698,8 +698,8 @@ abstract class AbstractPromptIntegrationTest : KotlinTestBase() {
val expectedMessage = "Prompt not found: non-existent-prompt"
withClue("Exception code should be INVALID_PARAMS: ${RPCError.ErrorCode.INVALID_PARAMS}") {
exception.code shouldBe RPCError.ErrorCode.INVALID_PARAMS
withClue("Exception code should be INTERNAL_ERROR: -32603") {
exception.code shouldBe -32603
}
withClue("Unexpected error message for non-existent prompt") {
exception.message shouldBe expectedMessage

View file

@ -77,6 +77,23 @@ abstract class AbstractResourceIntegrationTest : KotlinTestBase() {
)
}
server.addResource(
uri = testResourceUri,
name = testResourceName,
description = testResourceDescription,
mimeType = "text/plain",
) { request ->
ReadResourceResult(
contents = listOf(
TextResourceContents(
text = testResourceContent,
uri = request.params.uri,
mimeType = "text/plain",
),
),
)
}
server.addResource(
uri = binaryResourceUri,
name = binaryResourceName,

View file

@ -3,7 +3,6 @@ package io.modelcontextprotocol.kotlin.sdk.integration.streamablehttp
import io.ktor.server.cio.CIOApplicationEngine
import io.ktor.server.engine.EmbeddedServer
import io.ktor.server.engine.embeddedServer
import io.ktor.server.sse.Heartbeat
import io.modelcontextprotocol.kotlin.sdk.server.Server
import io.modelcontextprotocol.kotlin.sdk.server.ServerOptions
import io.modelcontextprotocol.kotlin.sdk.server.mcpStreamableHttp
@ -18,10 +17,7 @@ import io.ktor.server.cio.CIO as ServerCIO
open class AbstractStreamableHttpIntegrationTest {
suspend fun initTestServer(
name: String? = null,
sseHeartbeatConfig: (Heartbeat.() -> Unit)? = null,
): StreamableHttpTestServer {
suspend fun initTestServer(name: String? = null): StreamableHttpTestServer {
val mcpServer = Server(
Implementation(name = name ?: DEFAULT_SERVER_NAME, version = VERSION),
ServerOptions(
@ -56,7 +52,7 @@ open class AbstractStreamableHttpIntegrationTest {
host = URL,
port = PORT,
) {
mcpStreamableHttp(sseHeartbeatConfig = sseHeartbeatConfig) { mcpServer }
mcpStreamableHttp { mcpServer }
}
return StreamableHttpTestServer(

View file

@ -1,174 +0,0 @@
package io.modelcontextprotocol.kotlin.sdk.integration.streamablehttp
import io.kotest.matchers.shouldBe
import io.ktor.client.HttpClient
import io.ktor.client.plugins.sse.SSE
import io.ktor.client.request.header
import io.ktor.client.request.post
import io.ktor.client.request.prepareGet
import io.ktor.client.request.setBody
import io.ktor.client.statement.bodyAsChannel
import io.ktor.http.ContentType
import io.ktor.http.HttpHeaders
import io.ktor.http.HttpStatusCode
import io.ktor.http.contentType
import io.ktor.sse.ServerSentEvent
import io.ktor.utils.io.ByteReadChannel
import io.ktor.utils.io.readUTF8Line
import io.modelcontextprotocol.kotlin.sdk.types.ClientCapabilities
import io.modelcontextprotocol.kotlin.sdk.types.Implementation
import io.modelcontextprotocol.kotlin.sdk.types.InitializeRequest
import io.modelcontextprotocol.kotlin.sdk.types.InitializeRequestParams
import io.modelcontextprotocol.kotlin.sdk.types.JSONRPCRequest
import io.modelcontextprotocol.kotlin.sdk.types.LATEST_PROTOCOL_VERSION
import io.modelcontextprotocol.kotlin.sdk.types.toJSON
import io.modelcontextprotocol.kotlin.test.utils.actualPort
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.withTimeoutOrNull
import kotlinx.serialization.json.Json
import kotlin.test.Test
import kotlin.test.assertNotNull
import kotlin.time.Duration.Companion.milliseconds
import io.ktor.client.engine.cio.CIO as ClientCIO
private const val SESSION_ID_HEADER = "mcp-session-id"
private const val PROTOCOL_VERSION_HEADER = "mcp-protocol-version"
private const val DEFAULT_HEARTBEAT_LINE = ": heartbeat"
private const val CUSTOM_HEARTBEAT_LINE = ": mcp-heartbeat"
class StreamableHttpHeartbeatIntegrationTest : AbstractStreamableHttpIntegrationTest() {
@Test
fun `GET SSE stream emits configured heartbeat`(): Unit = runBlocking(Dispatchers.IO) {
var server: StreamableHttpTestServer? = null
var httpClient: HttpClient? = null
try {
server = initTestServer("heartbeat-test") {
period = 50.milliseconds
event = ServerSentEvent(comments = "mcp-heartbeat")
}
val mcpUrl = "http://$URL:${server.ktorServer.actualPort()}/mcp"
httpClient = HttpClient(ClientCIO) { install(SSE) }
val sessionId = initializeSession(httpClient, mcpUrl)
httpClient.prepareGet(mcpUrl) {
addSseHeaders(sessionId)
}.execute { response ->
response.status shouldBe HttpStatusCode.OK
response.headers[SESSION_ID_HEADER] shouldBe sessionId
response.bodyAsChannel().readLineMatching(CUSTOM_HEARTBEAT_LINE) shouldBe CUSTOM_HEARTBEAT_LINE
}
} finally {
httpClient?.close()
server?.ktorServer?.stopSuspend(1000, 2000)
}
}
@Test
fun `GET SSE stream does not emit heartbeat by default`(): Unit = runBlocking(Dispatchers.IO) {
var server: StreamableHttpTestServer? = null
var httpClient: HttpClient? = null
try {
server = initTestServer("no-heartbeat-test")
val mcpUrl = "http://$URL:${server.ktorServer.actualPort()}/mcp"
httpClient = HttpClient(ClientCIO) { install(SSE) }
val sessionId = initializeSession(httpClient, mcpUrl)
httpClient.prepareGet(mcpUrl) {
addSseHeaders(sessionId)
}.execute { response ->
response.status shouldBe HttpStatusCode.OK
response.headers[SESSION_ID_HEADER] shouldBe sessionId
response.bodyAsChannel().readLineMatching(timeoutMillis = 150) { line ->
line.isHeartbeatSseLine()
} shouldBe null
}
} finally {
httpClient?.close()
server?.ktorServer?.stopSuspend(1000, 2000)
}
}
@Test
fun `GET SSE stream emits configured heartbeat repeatedly`(): Unit = runBlocking(Dispatchers.IO) {
var server: StreamableHttpTestServer? = null
var httpClient: HttpClient? = null
try {
server = initTestServer("repeating-heartbeat-test") {
period = 50.milliseconds
event = ServerSentEvent(comments = "mcp-heartbeat")
}
val mcpUrl = "http://$URL:${server.ktorServer.actualPort()}/mcp"
httpClient = HttpClient(ClientCIO) { install(SSE) }
val sessionId = initializeSession(httpClient, mcpUrl)
httpClient.prepareGet(mcpUrl) {
addSseHeaders(sessionId)
}.execute { response ->
response.status shouldBe HttpStatusCode.OK
response.headers[SESSION_ID_HEADER] shouldBe sessionId
val channel = response.bodyAsChannel()
channel.readLineMatching(CUSTOM_HEARTBEAT_LINE) shouldBe CUSTOM_HEARTBEAT_LINE
channel.readLineMatching(CUSTOM_HEARTBEAT_LINE) shouldBe CUSTOM_HEARTBEAT_LINE
}
} finally {
httpClient?.close()
server?.ktorServer?.stopSuspend(1000, 2000)
}
}
private suspend fun initializeSession(client: HttpClient, mcpUrl: String): String {
val response = client.post(mcpUrl) {
contentType(ContentType.Application.Json)
header(
HttpHeaders.Accept,
"${ContentType.Application.Json}, ${ContentType.Text.EventStream}",
)
setBody(Json.encodeToString(buildInitPayload()))
}
response.status shouldBe HttpStatusCode.OK
return assertNotNull(response.headers[SESSION_ID_HEADER])
}
private fun buildInitPayload(): JSONRPCRequest = InitializeRequest(
InitializeRequestParams(
protocolVersion = LATEST_PROTOCOL_VERSION,
capabilities = ClientCapabilities(),
clientInfo = Implementation(name = "heartbeat-test-client", version = "1.0.0"),
),
).toJSON()
private fun io.ktor.client.request.HttpRequestBuilder.addSseHeaders(sessionId: String) {
header(HttpHeaders.Accept, ContentType.Text.EventStream.toString())
header(SESSION_ID_HEADER, sessionId)
header(PROTOCOL_VERSION_HEADER, LATEST_PROTOCOL_VERSION)
}
private suspend fun ByteReadChannel.readLineMatching(expectedLine: String, timeoutMillis: Long = 2_000): String? =
readLineMatching(timeoutMillis) { line -> line == expectedLine }
private suspend fun ByteReadChannel.readLineMatching(
timeoutMillis: Long = 2_000,
matches: (String) -> Boolean,
): String? = withTimeoutOrNull(timeoutMillis.milliseconds) {
var line = readUTF8Line()
while (line != null) {
if (matches(line)) return@withTimeoutOrNull line
line = readUTF8Line()
}
null
}
private fun String.isHeartbeatSseLine(): Boolean = this == DEFAULT_HEARTBEAT_LINE ||
this == CUSTOM_HEARTBEAT_LINE ||
(startsWith(":") && contains("heartbeat", ignoreCase = true)) ||
(startsWith("event:") && contains("heartbeat", ignoreCase = true))
}

View file

@ -1,7 +1,5 @@
package io.modelcontextprotocol.kotlin.sdk.server
import io.kotest.assertions.nondeterministic.continually
import io.kotest.assertions.nondeterministic.eventually
import io.kotest.matchers.collections.shouldBeEmpty
import io.kotest.matchers.collections.shouldHaveSize
import io.kotest.matchers.shouldBe
@ -19,9 +17,6 @@ import kotlinx.serialization.json.JsonPrimitive
import org.junit.jupiter.api.Test
import org.junit.jupiter.params.ParameterizedTest
import org.junit.jupiter.params.provider.EnumSource
import java.util.concurrent.CopyOnWriteArrayList
import kotlin.time.Duration.Companion.milliseconds
import kotlin.time.Duration.Companion.seconds
class ClientConnectionLoggingTest : AbstractServerFeaturesTest() {
@ -61,7 +56,7 @@ class ClientConnectionLoggingTest : AbstractServerFeaturesTest() {
@ParameterizedTest
@EnumSource(LoggingLevel::class)
fun `notification should filter logging messages below level`(minLevel: LoggingLevel): Unit = runBlocking {
val receivedMessages = CopyOnWriteArrayList<LoggingMessageNotification>()
val receivedMessages = mutableListOf<LoggingMessageNotification>()
client.setNotificationHandler<LoggingMessageNotification>(Method.Defined.NotificationsMessage) {
receivedMessages.add(it)
CompletableDeferred(Unit)
@ -84,10 +79,8 @@ class ClientConnectionLoggingTest : AbstractServerFeaturesTest() {
client.callTool(CallToolRequest(CallToolRequestParams("test-notification-filtered")))
// Handlers dispatch on Dispatchers.Default after the handshake, so await the expected count
// before asserting order.
val expectedLevels = LoggingLevel.entries.filter { it >= minLevel }
eventually(5.seconds) { receivedMessages shouldHaveSize expectedLevels.size }
receivedMessages shouldHaveSize expectedLevels.size
receivedMessages.map { it.params.level } shouldBe expectedLevels
}
@ -122,7 +115,7 @@ class ClientConnectionLoggingTest : AbstractServerFeaturesTest() {
@ParameterizedTest
@EnumSource(LoggingLevel::class)
fun `sendLoggingMessage should filter messages below level`(minLevel: LoggingLevel): Unit = runBlocking {
val receivedMessages = CopyOnWriteArrayList<LoggingMessageNotification>()
val receivedMessages = mutableListOf<LoggingMessageNotification>()
client.setNotificationHandler<LoggingMessageNotification>(Method.Defined.NotificationsMessage) {
receivedMessages.add(it)
CompletableDeferred(Unit)
@ -145,16 +138,14 @@ class ClientConnectionLoggingTest : AbstractServerFeaturesTest() {
client.callTool(CallToolRequest(CallToolRequestParams("test-logging-level")))
// Handlers dispatch on Dispatchers.Default after the handshake, so await the expected count
// before asserting order.
val expectedLevels = LoggingLevel.entries.filter { it >= minLevel }
eventually(5.seconds) { receivedMessages shouldHaveSize expectedLevels.size }
receivedMessages shouldHaveSize expectedLevels.size
receivedMessages.map { it.params.level } shouldBe expectedLevels
}
@Test
fun `sendLoggingMessage should send no messages when level is set to highest`(): Unit = runBlocking {
val receivedMessages = CopyOnWriteArrayList<LoggingMessageNotification>()
val receivedMessages = mutableListOf<LoggingMessageNotification>()
client.setNotificationHandler<LoggingMessageNotification>(Method.Defined.NotificationsMessage) {
receivedMessages.add(it)
CompletableDeferred(Unit)
@ -176,7 +167,6 @@ class ClientConnectionLoggingTest : AbstractServerFeaturesTest() {
}
client.callTool(CallToolRequest(CallToolRequestParams("test-logging-highest")))
// Handlers dispatch asynchronously, so hold for a window to confirm nothing ever arrives.
continually(500.milliseconds) { receivedMessages.shouldBeEmpty() }
receivedMessages.shouldBeEmpty()
}
}

Some files were not shown because too many files have changed in this diff Show more