Skip to content

fix: concurrent message processing for all transports #610

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
rnett wants to merge 1 commit into from
Closed

fix: concurrent message processing for all transports #610

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 11 additions & 3 deletions AGENTS.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -77,6 +77,9 @@ Follow these rules to keep changes safe, comprehensible, and easy to maintain.
- **Prioritize test readability**
- Avoid creating too many test methods; use parametrized tests when testing multiple similar scenarios
- When running tests on Kotlin Multiplatform projects, run JVM tests only unless asked for other platforms
- **Concurrency in Tests**: Always use thread-safe collections (e.g., `Mutex`-protected lists or `Channel`) when
collecting messages from transports that process messages concurrently in the background (like those inheriting from
`AbstractTransport`). Using non-thread-safe `MutableList` will lead to flaky tests or missing messages.

### Test Framework Stack

Expand All @@ -93,9 +96,12 @@ Follow these rules to keep changes safe, comprehensible, and easy to maintain.
- **Ktor MockEngine**: For HTTP client mocking (`io.ktor:ktor-client-mock`)
- **Java tests**: Use JUnit5, Mockito, AssertJ core
- **Serialization test utilities** (`io.modelcontextprotocol.kotlin.test.utils`):
- `verifySerialization(value, json, expectedJson)` — serializes, asserts match, round-trips back; use for most serialization tests
- `verifyDeserialization(json, payload)` — deserializes from JSON, re-serializes, asserts match; returns the object for further assertions
- Always test both empty/null/omitted and non-null cases for nullable fields; `McpJson` has `explicitNulls = false` so null properties must be absent from JSON, not `null`
- `verifySerialization(value, json, expectedJson)` — serializes, asserts match, round-trips back; use for most
serialization tests
- `verifyDeserialization(json, payload)` — deserializes from JSON, re-serializes, asserts match; returns the object
for further assertions
- Always test both empty/null/omitted and non-null cases for nullable fields; `McpJson` has `explicitNulls = false`
so null properties must be absent from JSON, not `null`

### Kotest Patterns

Expand Down Expand Up @@ -169,6 +175,8 @@ prop.shouldNotBeNull {
- Use Kotlinx Serialization with explicit `@Serializable` annotations
- JSON config is defined in `jsonUtils.kt` as `McpJson` — use it consistently
- Register custom serializers in companion objects
- **SSE Data Concatenation**: When parsing Server-Sent Events (SSE) data, always ensure that multiple `data:` lines are
concatenated with a newline (`\n`) separator, as per the SSE specification.

### Error Handling

Expand Down
232 changes: 119 additions & 113 deletions ...ration-test/src/commonTest/kotlin/io/modelcontextprotocol/kotlin/sdk/client/ClientTest.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -61,32 +61,33 @@ class ClientTest {
@Test
fun `should initialize with matching protocol version`() = runTest {
var initialised = false
val clientTransport = object : AbstractTransport() {
override suspend fun start() {}

override suspend fun send(message: JSONRPCMessage, options: TransportSendOptions?) {
if (message !is JSONRPCRequest) return
initialised = true
val result = InitializeResult(
protocolVersion = LATEST_PROTOCOL_VERSION,
capabilities = ServerCapabilities(),
serverInfo = Implementation(
name = "test",
version = "1.0",
),
)

val response = JSONRPCResponse(
id = message.id,
result = result,
)

_onMessage.invoke(response)
}
val clientTransport =
object : AbstractTransport(backgroundScope.coroutineContext, backgroundScope.coroutineContext) {
override suspend fun start() {}

override suspend fun send(message: JSONRPCMessage, options: TransportSendOptions?) {
if (message !is JSONRPCRequest) return
initialised = true
val result = InitializeResult(
protocolVersion = LATEST_PROTOCOL_VERSION,
capabilities = ServerCapabilities(),
serverInfo = Implementation(
name = "test",
version = "1.0",
),
)

val response = JSONRPCResponse(
id = message.id,
result = result,
)

handleMessage(response)
}

override suspend fun close() {
override suspend fun close() {
}
}
}

val client = Client(
clientInfo = Implementation(
Expand All @@ -107,32 +108,33 @@ class ClientTest {
@Test
fun `should initialize with supported older protocol version`() = runTest {
val oldVersion = SUPPORTED_PROTOCOL_VERSIONS[1]
val clientTransport = object : AbstractTransport() {
override suspend fun start() {}

override suspend fun send(message: JSONRPCMessage, options: TransportSendOptions?) {
if (message !is JSONRPCRequest) return
check(message.method == Method.Defined.Initialize.value)

val result = InitializeResult(
protocolVersion = oldVersion,
capabilities = ServerCapabilities(),
serverInfo = Implementation(
name = "test",
version = "1.0",
),
)

val response = JSONRPCResponse(
id = message.id,
result = result,
)
_onMessage.invoke(response)
}
val clientTransport =
object : AbstractTransport(backgroundScope.coroutineContext, backgroundScope.coroutineContext) {
override suspend fun start() {}

override suspend fun send(message: JSONRPCMessage, options: TransportSendOptions?) {
if (message !is JSONRPCRequest) return
check(message.method == Method.Defined.Initialize.value)

val result = InitializeResult(
protocolVersion = oldVersion,
capabilities = ServerCapabilities(),
serverInfo = Implementation(
name = "test",
version = "1.0",
),
)

val response = JSONRPCResponse(
id = message.id,
result = result,
)
handleMessage(response)
}

override suspend fun close() {
override suspend fun close() {
}
}
}

val client = Client(
clientInfo = Implementation(
Expand All @@ -156,34 +158,35 @@ class ClientTest {
@Test
fun `should reject unsupported protocol version`() = runTest {
var closed = false
val clientTransport = object : AbstractTransport() {
override suspend fun start() {}

override suspend fun send(message: JSONRPCMessage, options: TransportSendOptions?) {
if (message !is JSONRPCRequest) return
check(message.method == Method.Defined.Initialize.value)

val result = InitializeResult(
protocolVersion = "invalid-version",
capabilities = ServerCapabilities(),
serverInfo = Implementation(
name = "test",
version = "1.0",
),
)

val response = JSONRPCResponse(
id = message.id,
result = result,
)

_onMessage.invoke(response)
}
val clientTransport =
object : AbstractTransport(backgroundScope.coroutineContext, backgroundScope.coroutineContext) {
override suspend fun start() {}

override suspend fun send(message: JSONRPCMessage, options: TransportSendOptions?) {
if (message !is JSONRPCRequest) return
check(message.method == Method.Defined.Initialize.value)

val result = InitializeResult(
protocolVersion = "invalid-version",
capabilities = ServerCapabilities(),
serverInfo = Implementation(
name = "test",
version = "1.0",
),
)

val response = JSONRPCResponse(
id = message.id,
result = result,
)

handleMessage(response)
}

override suspend fun close() {
closed = true
override suspend fun close() {
closed = true
}
}
}

val client = Client(
clientInfo = Implementation(
Expand All @@ -203,19 +206,20 @@ class ClientTest {
@Test
fun `should reject due to non cancellation exception`() = runTest {
var closed = false
val failingTransport = object : AbstractTransport() {
override suspend fun start() {}

override suspend fun send(message: JSONRPCMessage, options: TransportSendOptions?) {
if (message !is JSONRPCRequest) return
check(message.method == Method.Defined.Initialize.value)
throw IllegalStateException("Test error")
}
val failingTransport =
object : AbstractTransport(backgroundScope.coroutineContext, backgroundScope.coroutineContext) {
override suspend fun start() {}

override suspend fun send(message: JSONRPCMessage, options: TransportSendOptions?) {
if (message !is JSONRPCRequest) return
check(message.method == Method.Defined.Initialize.value)
throw IllegalStateException("Test error")
}

override suspend fun close() {
closed = true
override suspend fun close() {
closed = true
}
}
}

val client = Client(
clientInfo = Implementation(
Expand All @@ -237,22 +241,23 @@ class ClientTest {
@Test
fun `should rethrow McpException as is`() = runTest {
var closed = false
val failingTransport = object : AbstractTransport() {
override suspend fun start() {}

override suspend fun send(message: JSONRPCMessage, options: TransportSendOptions?) {
if (message !is JSONRPCRequest) return
check(message.method == Method.Defined.Initialize.value)
throw McpException(
code = -32600,
message = "Invalid Request",
)
}
val failingTransport =
object : AbstractTransport(backgroundScope.coroutineContext, backgroundScope.coroutineContext) {
override suspend fun start() {}

override suspend fun send(message: JSONRPCMessage, options: TransportSendOptions?) {
if (message !is JSONRPCRequest) return
check(message.method == Method.Defined.Initialize.value)
throw McpException(
code = -32600,
message = "Invalid Request",
)
}

override suspend fun close() {
closed = true
override suspend fun close() {
closed = true
}
}
}

val client = Client(
clientInfo = Implementation(
Expand All @@ -275,22 +280,23 @@ class ClientTest {
@Test
fun `should rethrow StreamableHttpError as is`() = runTest {
var closed = false
val failingTransport = object : AbstractTransport() {
override suspend fun start() {}

override suspend fun send(message: JSONRPCMessage, options: TransportSendOptions?) {
if (message !is JSONRPCRequest) return
check(message.method == Method.Defined.Initialize.value)
throw StreamableHttpError(
code = 500,
message = "Internal Server Error",
)
}
val failingTransport =
object : AbstractTransport(backgroundScope.coroutineContext, backgroundScope.coroutineContext) {
override suspend fun start() {}

override suspend fun send(message: JSONRPCMessage, options: TransportSendOptions?) {
if (message !is JSONRPCRequest) return
check(message.method == Method.Defined.Initialize.value)
throw StreamableHttpError(
code = 500,
message = "Internal Server Error",
)
}

override suspend fun close() {
closed = true
override suspend fun close() {
closed = true
}
}
}

val client = Client(
clientInfo = Implementation(
Expand Down
12 changes: 9 additions & 3 deletions ...commonTest/kotlin/io/modelcontextprotocol/kotlin/sdk/integration/InMemoryTransportTest.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,8 @@ import io.modelcontextprotocol.kotlin.sdk.types.ResourceUpdatedNotification
import io.modelcontextprotocol.kotlin.sdk.types.ResourceUpdatedNotificationParams
import io.modelcontextprotocol.kotlin.sdk.types.ToolListChangedNotification
import io.modelcontextprotocol.kotlin.sdk.types.toJSON
import kotlinx.coroutines.sync.Mutex
import kotlinx.coroutines.sync.withLock
import kotlinx.coroutines.test.runTest
import kotlin.test.BeforeTest
import kotlin.test.Test
Expand Down Expand Up @@ -44,15 +46,16 @@ class InMemoryTransportTest {

@Test
fun `should send message from client to server`() = runTest {
val (client, server) = InMemoryTransport.createLinkedPair(backgroundScope.coroutineContext)
val message = InitializedNotification()

var receivedMessage: JSONRPCMessage? = null
serverTransport.onMessage { msg ->
server.onMessage { msg ->
receivedMessage = msg
}

val rpcNotification = message.toJSON()
clientTransport.send(rpcNotification)
client.send(rpcNotification)
assertEquals(rpcNotification, receivedMessage)
}

Expand Down Expand Up @@ -190,8 +193,11 @@ class InMemoryTransportTest {
)

val receivedMessages = mutableListOf<JSONRPCMessage>()
val mutex = Mutex()
clientTransport.onMessage { msg ->
receivedMessages.add(msg)
mutex.withLock {
receivedMessages.add(msg)
}
}

notifications.forEach { notification ->
Expand Down
Loading
Loading