# feat: Add kotlin-sdk-testing module with channel-based transport implementation - Introduced `kotlin-sdk-testing` module for testing utilities in the MCP Kotlin SDK. - Implemented `ChannelTransport`, providing JSON-RPC transport via Kotlin coroutines channels. <!-- Provide a brief summary of your changes --> ## Motivation and Context This ChannelTransport can be used for testing Server implementation without using real transports. ## How Has This Been Tested? Unit test ## Breaking Changes No ## Types of changes - [ ] Bug fix (non-breaking change which fixes an issue) - [x] New feature (non-breaking change which adds functionality) - [ ] Breaking change (fix or feature that would cause existing functionality to change) - [x] Documentation update ## 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 - [x] I have added appropriate error handling - [x] I have added or updated documentation as needed
2.5 KiB
2.5 KiB
Module kotlin-sdk-testing
kotlin-sdk-testing provides testing utilities for MCP client-server interactions. It eliminates the need for external
processes or network connections by offering coroutine-safe in-memory transport implementations that connect clients and servers directly.
Typical usage
- Create linked transport pair with
ChannelTransport.createLinkedPair(). - Connect client and server concurrently using the paired transports.
- Configure server request handlers.
- Execute client operations and verify results.
Complete test example
@OptIn(ExperimentalMcpApi::class)
@Test
fun testClientServerCommunication(): Unit = runBlocking {
val server = Server(
Implementation(name = "test server", version = "1.0"),
ServerOptions(capabilities = ServerCapabilities(resources = ServerCapabilities.Resources())),
)
val (clientTransport, serverTransport) = ChannelTransport.createLinkedPair()
val client = Client(clientInfo = Implementation(name = "test client", version = "1.0"))
val serverSessionResult = CompletableDeferred<ServerSession>()
listOf(
launch { client.connect(clientTransport) },
launch { serverSessionResult.complete(server.createSession(serverTransport)) },
).joinAll()
val serverSession = serverSessionResult.await()
serverSession.setRequestHandler<InitializeRequest>(Method.Defined.Initialize) { _, _ ->
InitializeResult(
protocolVersion = LATEST_PROTOCOL_VERSION,
capabilities = ServerCapabilities(
resources = ServerCapabilities.Resources(null, null),
tools = ServerCapabilities.Tools(null),
),
serverInfo = Implementation(name = "test", version = "1.0"),
)
}
serverSession.setRequestHandler<ListResourcesRequest>(Method.Defined.ResourcesList) { _, _ ->
ListResourcesResult(
resources = listOf(Resource(uri = "/foo/bar", name = "foo-bar-resource")),
)
}
// Test client operations
client.listResources() shouldNotBeNull {
this.resources shouldHaveSize 1
}
// Clean up resources
client.close()
server.close()
}
Note: In production tests, wrap client/server operations in try-finally blocks to ensure proper resource cleanup even if assertions fail.
Use this module for unit and integration testing of MCP implementations without managing external server processes, network ports, or I/O streams. The channel-based transport ensures tests run instantly with full isolation between test cases.