fix(stdio): #573, #574 customize dispatchers & scope for stdio server transport configuration by kpavlov · Pull Request #584 · modelcontextprotocol/kotlin-sdk

kpavlov · 2026-03-03T22:04:02Z

Add a customizable builder to StdioServerTransport

Scope lifecycle is fixed (StdioServerTransport: The coroutine scope that's used in this transport is completely detached from the coroutine tree #574 resolved), processing dispatcher defaults to Dispatchers.Default (StdioServerTransport: Request handlers should not run on Dispatchers.IO #573 resolved), scope context is clean (no spurious dispatcher stacking), and @Volatile on the three job vars addresses the visibility race.
- Introduce a Configuration class for StdioServerTransport to improve API flexibility and readability. Now one can provide a dispatcher for IO running on Virtual Threads
- Updated transport initialization to use a builder block for configuring parameters such as I/O streams, buffer sizes, dispatchers, and parent coroutine scope.
- Added integration test validating the builder functionality.
Added InjectDispatcher rule to detekt exclusions for test folders in detekt.yml. Removed InjectDispatcher issues from detekt baseline files.

NB! This PR doesn't fully resolve race conditions on close(). It will be resolved in the course of #518

Motivation and Context

#573, #574

How Has This Been Tested?

Integration test / regression tests

Breaking Changes

No, only API is only extended

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

I have read the MCP Documentation
My code follows the repository's style guidelines
New and existing tests pass locally
I have added appropriate error handling
I have added or updated documentation as needed

Additional context

…port Scope lifecycle is fixed (#574 resolved), processing dispatcher defaults to `Dispatchers.Default` (#573 resolved), scope context is clean (no spurious dispatcher stacking), and @volatile on the three job vars addresses the visibility race. - Introduce a `Configuration` class for `StdioServerTransport` to improve API flexibility and readability. - Updated transport initialization to use a builder block for configuring parameters such as I/O streams, buffer sizes, dispatchers, and parent coroutine scope. Test: Added integration test validating the builder functionality.

… baselines - Added `InjectDispatcher` rule to detekt exclusions for test folders in `detekt.yml`. - Removed resolved `InjectDispatcher` issues from baseline files.

codecov-commenter · 2026-03-03T23:38:02Z

Codecov Report

❌ Patch coverage is 88.33333% with 7 lines in your changes missing coverage. Please review.
✅ All tests successful. No failed tests found.

Files with missing lines	Patch %	Lines
...protocol/kotlin/sdk/server/StdioServerTransport.kt	88.33%	1 Missing and 6 partials ⚠️

📢 Thoughts on this report? Let us know!

devcrocod

Please take a look at my comments

I’m not sure we need to extend the Configuration for stdio if it’s enough to just pass dispatcher for io

devcrocod · 2026-03-04T12:53:52Z

...rver/src/commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/server/StdioServerTransport.kt

+ * val transport = StdioServerTransport {
+ *   source = System.`in`.asInput(),
+ *   sink =  System.out.asSink(),
+ * }


It seems to me this is an incorrect example
as I remember, need to call buffered

Creating buffered instance happens internally here

I just noticed that commas are also used here

devcrocod · 2026-03-04T12:55:23Z

...rver/src/commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/server/StdioServerTransport.kt

 */
 @OptIn(ExperimentalAtomicApi::class)
-public class StdioServerTransport(private val inputStream: Source, outputStream: Sink) : AbstractTransport() {
+public class StdioServerTransport(block: Configuration.() -> Unit) : AbstractTransport() {


Isn’t Configuration class used in streamable constructor?
Why is a builder used here then?

What do you mean? StreamableHttpServerTransport uses different configuration class. This is StdioServerTransport.Configuration

You are using two different API approaches to create transport

https://github.com/modelcontextprotocol/kotlin-sdk/blob/3e7ea39bf70556154699bdfe483ca049a16b99aa/kotlin-sdk-server/src/commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/server/StreamableHttpServerTransport.kt#L75C1-L75C85

Accordingly, we need to settle on a single approach: either update/add a builder for streamable, or make the changes here for stdio

devcrocod · 2026-03-04T12:55:53Z

...rver/src/commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/server/StdioServerTransport.kt

+    public class Configuration internal constructor(
+        public var source: Source? = null,
+        public var sink: Sink? = null,
+        public var readBufferSize: Long = READ_BUFFER_SIZE,


This doesn’t need to be made public

devcrocod · 2026-03-04T12:56:30Z

...rver/src/commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/server/StdioServerTransport.kt

+        public var readChannelBufferSize: Int = Channel.UNLIMITED,
+        public var writeChannelBufferSize: Int = Channel.UNLIMITED,


This also shouldn’t be public

I disagree with this. Buffer sizes should be configurable for production systems. The default value of UNLIMITED is a clear performance issue (OOM risk). I am not changing it at this time to avoid unexpected changes for users but this is a defect for sure.

The default value of UNLIMITED is a clear performance issue

That makes sense
At the same time, I'd still suggest keeping it private and changing this to BUFFERED. Leaving it as UNLIMITED seems unreasonable to me, sooner or later we'll likely have to change it anyway

Later on, we can add the ability to configure this parameter, which will be easier to do within an already agreed-upon API

I'd also suggest renaming it, since it's essentially a capacity

If we want to configure backpressure, let's pass the entire Channel as a parameter.

devcrocod · 2026-03-04T13:00:48Z

...rver/src/commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/server/StdioServerTransport.kt

+        public var readingJobDispatcher: CoroutineDispatcher = IODispatcher,
+        public var writingJobDispatcher: CoroutineDispatcher = IODispatcher,
+        public var processingJobDispatcher: CoroutineDispatcher = Dispatchers.Default,


Why are there three dispatchers if one would be enough?

One dispatcher is not enough. Here is why:

processingJobDispatcher for issue StdioServerTransport: Request handlers should not run on Dispatchers.IO #573.

readingJobDispatcher and writingJobDispatcher should be configurable to run IO jobs on virtual threads. Dispatchers.IO uses platform threads by default.

I have considered having only one dispatcher for IO, but rejected this option for the sake of flexibility to implement the bulkhead pattern. This is crucial for high-load applications where read and write thread pools should be isolated.

@kpavlov better do val stdioDispatcher = ioDispatcher.limitedParallelism(2) to have dedicated threads for your jobs. The don't consume the global IO dispatcher pool in this case. So you don't need to pass a dispatcher for each job (parameters should specify a kind of dispatcher for IO, but not every single dispatcher for every job)

or you pass an IO dispatcher as a parameter and write a doc that at least 2 threads must be allocated in the dispatcher. So either a user passes a simple Dispatchers.IO, or also manually makes Dispatchers.IO.limitedParallelism(2)

The approach you implemented does not follow from either the attached issue or the original one
Splitting into io and default, along with the ability to pass a value for io, fully addresses the request
agree with @nerzhulart

devcrocod · 2026-03-04T13:01:41Z

...rver/src/commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/server/StdioServerTransport.kt

+        public var processingJobDispatcher: CoroutineDispatcher = Dispatchers.Default,
+        public var readChannelBufferSize: Int = Channel.UNLIMITED,
+        public var writeChannelBufferSize: Int = Channel.UNLIMITED,
+        public var coroutineScope: CoroutineScope? = null,


Why does coroutineScope need to be public?

I'd pass a scope as a constructor parameter

The coroutine scope that's used in this transport is completely detached from the coroutine tree, which may cause leaks. Please add an optional CoroutineScope parameter

agree

This is public, because it's builder's field. It is copied to the private val field in constructor and the builder got distroyed

devcrocod · 2026-03-04T13:08:12Z

...rver/src/commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/server/StdioServerTransport.kt

+
+    @Volatile
    private var readingJob: Job? = null
+
+    @Volatile
    private var sendingJob: Job? = null
-    private var processingJob: Job? = null

-    private val coroutineContext: CoroutineContext = IODispatcher + SupervisorJob()
-    private val scope = CoroutineScope(coroutineContext)
-    private val readChannel = Channel<ByteArray>(Channel.UNLIMITED)
-    private val writeChannel = Channel<JSONRPCMessage>(Channel.UNLIMITED)
-    private val outputSink = outputStream.buffered()
+    @Volatile
+    private var processingJob: Job? = null


Can we get rid of storing the Job fields and use a scope instead?
It seems to me that this would eliminate potential race conditions

Let us address this issue within the scope of #518.

The close() method already contains a race condition which this PR does not exacerbate. The test is passing.

We should retain the current algorithm and pattern for the time being.

You are already making a change here, so instead of adding volatile, it could simply be removed jobs

I don't quite follow the reasoning here, if you're already doing a refactor, why not address this as well? Is there some constraint I’m missing?

devcrocod · 2026-03-04T13:09:35Z

...ver/src/jvmTest/kotlin/io/modelcontextprotocol/kotlin/sdk/server/StdioServerTransportTest.kt

        printOutput = output.asSink().buffered()
    }

+    @OptIn(InternalAPI::class)


I think it’s better not to use InternalAPI from other libraries unless there’s a clear necessity

I agree with you in general. This is for clientDispatcher and is not causing any issues here.

kpavlov

@devcrocod, thanks for your review! I’ve replied to your comments.
@e5l, I’d be grateful if you could also take a look.

kpavlov · 2026-03-04T14:18:15Z

...rver/src/commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/server/StdioServerTransport.kt

+ * val transport = StdioServerTransport {
+ *   source = System.`in`.asInput(),
+ *   sink =  System.out.asSink(),
+ * }


Creating buffered instance happens internally here

kpavlov · 2026-03-04T14:19:40Z

...rver/src/commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/server/StdioServerTransport.kt

 */
 @OptIn(ExperimentalAtomicApi::class)
-public class StdioServerTransport(private val inputStream: Source, outputStream: Sink) : AbstractTransport() {
+public class StdioServerTransport(block: Configuration.() -> Unit) : AbstractTransport() {


What do you mean? StreamableHttpServerTransport uses different configuration class. This is StdioServerTransport.Configuration

kpavlov · 2026-03-04T14:22:35Z

...rver/src/commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/server/StdioServerTransport.kt

+    public class Configuration internal constructor(
+        public var source: Source? = null,
+        public var sink: Sink? = null,
+        public var readBufferSize: Long = READ_BUFFER_SIZE,


This is your opinion. There is a similar request for StreamableHttpTransport (#521). So, making buffer size customizable makes sense to me. Might be useful for performance tuning.

Of course, this is just my opinion, I'm not claiming it's truth
And I’m not saying that this is a bad change
At the same time, it locks api. As I mentioned in another comment, we can always add something later without trouble, but removing or changing it afterwards would be more problematic
If we want to pass such things through the main config, then that's ok

kpavlov · 2026-03-04T14:26:51Z

...rver/src/commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/server/StdioServerTransport.kt

+        public var readingJobDispatcher: CoroutineDispatcher = IODispatcher,
+        public var writingJobDispatcher: CoroutineDispatcher = IODispatcher,
+        public var processingJobDispatcher: CoroutineDispatcher = Dispatchers.Default,


One dispatcher is not enough. Here is why:

processingJobDispatcher for issue StdioServerTransport: Request handlers should not run on Dispatchers.IO #573.

readingJobDispatcher and writingJobDispatcher should be configurable to run IO jobs on virtual threads. Dispatchers.IO uses platform threads by default.

I have considered having only one dispatcher for IO, but rejected this option for the sake of flexibility to implement the bulkhead pattern. This is crucial for high-load applications where read and write thread pools should be isolated.

kpavlov · 2026-03-04T14:30:02Z

...rver/src/commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/server/StdioServerTransport.kt

+        public var readChannelBufferSize: Int = Channel.UNLIMITED,
+        public var writeChannelBufferSize: Int = Channel.UNLIMITED,


I disagree with this. Buffer sizes should be configurable for production systems. The default value of UNLIMITED is a clear performance issue (OOM risk). I am not changing it at this time to avoid unexpected changes for users but this is a defect for sure.

kpavlov · 2026-03-04T14:30:38Z

...rver/src/commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/server/StdioServerTransport.kt

+        public var processingJobDispatcher: CoroutineDispatcher = Dispatchers.Default,
+        public var readChannelBufferSize: Int = Channel.UNLIMITED,
+        public var writeChannelBufferSize: Int = Channel.UNLIMITED,
+        public var coroutineScope: CoroutineScope? = null,


kpavlov · 2026-03-04T14:38:58Z

...rver/src/commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/server/StdioServerTransport.kt

+
+    @Volatile
    private var readingJob: Job? = null
+
+    @Volatile
    private var sendingJob: Job? = null
-    private var processingJob: Job? = null

-    private val coroutineContext: CoroutineContext = IODispatcher + SupervisorJob()
-    private val scope = CoroutineScope(coroutineContext)
-    private val readChannel = Channel<ByteArray>(Channel.UNLIMITED)
-    private val writeChannel = Channel<JSONRPCMessage>(Channel.UNLIMITED)
-    private val outputSink = outputStream.buffered()
+    @Volatile
+    private var processingJob: Job? = null


Let us address this issue within the scope of #518.

The close() method already contains a race condition which this PR does not exacerbate. The test is passing.

We should retain the current algorithm and pattern for the time being.

kpavlov · 2026-03-04T14:40:43Z

...ver/src/jvmTest/kotlin/io/modelcontextprotocol/kotlin/sdk/server/StdioServerTransportTest.kt

        printOutput = output.asSink().buffered()
    }

+    @OptIn(InternalAPI::class)


I agree with you in general. This is for clientDispatcher and is not causing any issues here.

kpavlov · 2026-03-04T15:10:11Z

cc: @i1bro

devcrocod · 2026-03-04T15:41:45Z

This PR introduces quite a few changes beyond addressing the specific issue
In my opinion, these changes are not ideal at the moment:

The resulting api is not very consistent and doesn’t look well designed
As noted in agents.md, the code should follow solid principles. In this case, it exposes too many internal implementation details
It makes it harder for us to change the internal implementation or change api

I think it would be better not to add many things to the public api without a request for them. We can always add this later without breaking compatibility, but removing or changing it afterwards would be much hard

kpavlov · 2026-03-04T17:07:31Z

@devcrocod, where exactly this pr worsening existing design?
The PR is solving specific issues that were discovered by users. The rewriting of the transport is not the goal.
Could you please elaborate and provide constructive feedback?

nerzhulart · 2026-03-04T19:49:08Z

...rver/src/commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/server/StdioServerTransport.kt

+     * @property coroutineScope The [CoroutineScope] used for managing coroutines.
+     */
+    @Suppress("LongParameterList")
+    public class Configuration internal constructor(


I'd not use such kind of builders. IMHO they look fancy but from use-site it's usually overengineering. I'd prefer put necessary parameters as parameters, like source and sink, as well as parent scope and io dispatcher (that's not presented in KMP). The default dispatcher can be overidden by passing a specific scope with the necessary dispatcher, but I'm sure the default one fits always. All the rest optional/default parmeters like buffer size I'd hide under StdioOptions data class with default instance if possible

nerzhulart · 2026-03-04T19:57:55Z

...rver/src/commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/server/StdioServerTransport.kt

+        public var readingJobDispatcher: CoroutineDispatcher = IODispatcher,
+        public var writingJobDispatcher: CoroutineDispatcher = IODispatcher,
+        public var processingJobDispatcher: CoroutineDispatcher = Dispatchers.Default,
+        public var readChannelBufferSize: Int = Channel.UNLIMITED,


when doing channel sizes configurable always specify where backpressure occurs (if occurs) and add tests for the limited cases

nerzhulart · 2026-03-04T20:00:43Z

...rver/src/commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/server/StdioServerTransport.kt

    private val readBuffer = ReadBuffer()
    private val initialized: AtomicBoolean = AtomicBoolean(false)
+
+    @Volatile


I don't think you even need these fields for jobs. Just run an orchestration job in init and inside run all necessary jobs - read, write, processing together, then you may join them. In this case you get structured concurrency that joins in a single job. Less state (especially something like volatile) less problems

nerzhulart · 2026-03-04T20:02:57Z

...rver/src/commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/server/StdioServerTransport.kt

+    }.apply {
+        invokeOnCompletion { cause ->
            logJobCompletion("Message sending", cause)
            if (cause is CancellationException) {


I'd try to control cancellation via channels

nerzhulart · 2026-03-04T20:06:19Z

...rver/src/commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/server/StdioServerTransport.kt

+                source.close()
            }.onFailure { logger.warn(it) { "Failed to close stdin" } }

            readingJob?.cancel()


when you hardly cancel the jobs they don't have a chance to terminate gracefully, I think better to close the channels and streams and handle semantic exceptions from them

devcrocod · 2026-03-04T20:40:01Z

@kpavlov I replied to each comment again in case additional details were needed

kpavlov · 2026-03-04T20:57:41Z

@e5l @devcrocod @nerzhulart I would appreciate it if you could share a link to the official Kotlin guide on writing idiomatic APIs. Until then, I am confused, as the answers appear to be based on personal preferences.

…ove `Configuration` class - Replaced the builder-based setup with a streamlined primary constructor for `StdioServerTransport`. - Removed the `Configuration` class for reduced complexity and enhanced readability. - Updated `StdioServerTransportTest` to reflect the refactored initialization. - Refactor cleanup.

kpavlov added 2 commits March 4, 2026 00:12

chore(detekt): exclude InjectDispatcher for test folders and update…

570184d

… baselines - Added `InjectDispatcher` rule to detekt exclusions for test folders in `detekt.yml`. - Removed resolved `InjectDispatcher` issues from baseline files.

kpavlov force-pushed the kpavlov/573-574-stdio-server-transport branch from c4cf6ac to 570184d Compare March 3, 2026 22:12

kpavlov requested review from devcrocod and e5l March 3, 2026 22:28

kpavlov added bugfix Something was fixed 🎉 labels Mar 3, 2026

kpavlov marked this pull request as ready for review March 3, 2026 22:28

devcrocod requested changes Mar 4, 2026

View reviewed changes

kpavlov commented Mar 4, 2026

View reviewed changes

nerzhulart reviewed Mar 4, 2026

View reviewed changes

nerzhulart suggested changes Mar 4, 2026

View reviewed changes

kpavlov marked this pull request as draft March 4, 2026 21:25

kpavlov force-pushed the kpavlov/573-574-stdio-server-transport branch 2 times, most recently from 132ebd9 to ac552f7 Compare March 4, 2026 21:37

kpavlov force-pushed the kpavlov/573-574-stdio-server-transport branch from ac552f7 to 4693529 Compare March 5, 2026 09:10

		public var readChannelBufferSize: Int = Channel.UNLIMITED,
		public var writeChannelBufferSize: Int = Channel.UNLIMITED,

Conversation

kpavlov commented Mar 3, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Motivation and Context

How Has This Been Tested?

Breaking Changes

Types of changes

Checklist

Additional context

Uh oh!

codecov-commenter commented Mar 3, 2026

Codecov Report

Uh oh!

devcrocod left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

devcrocod Mar 4, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

kpavlov Mar 4, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

devcrocod Mar 4, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

kpavlov Mar 4, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

kpavlov commented Mar 3, 2026 •

edited

Loading

devcrocod left a comment •

edited

Loading

devcrocod Mar 4, 2026 •

edited

Loading

kpavlov Mar 4, 2026 •

edited

Loading

devcrocod Mar 4, 2026 •

edited

Loading

kpavlov Mar 4, 2026 •

edited

Loading

kpavlov Mar 4, 2026 •

edited

Loading

kpavlov Mar 4, 2026 •

edited

Loading