#107 Monitor STDERR and refactor StdioClientTransport to use flows (#442)

kotlin-sdk-client/src/commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/client/StdioClientTransport.kt

@@ -1,135 +1,348 @@
 package io.modelcontextprotocol.kotlin.sdk.client
 
 import io.github.oshai.kotlinlogging.KotlinLogging
+import io.modelcontextprotocol.kotlin.sdk.client.StdioClientTransport.StderrSeverity.DEBUG
+import io.modelcontextprotocol.kotlin.sdk.client.StdioClientTransport.StderrSeverity.FATAL
+import io.modelcontextprotocol.kotlin.sdk.client.StdioClientTransport.StderrSeverity.IGNORE
+import io.modelcontextprotocol.kotlin.sdk.client.StdioClientTransport.StderrSeverity.INFO
+import io.modelcontextprotocol.kotlin.sdk.client.StdioClientTransport.StderrSeverity.WARNING
 import io.modelcontextprotocol.kotlin.sdk.internal.IODispatcher
 import io.modelcontextprotocol.kotlin.sdk.shared.AbstractTransport
 import io.modelcontextprotocol.kotlin.sdk.shared.ReadBuffer
 import io.modelcontextprotocol.kotlin.sdk.shared.TransportSendOptions
 import io.modelcontextprotocol.kotlin.sdk.shared.serializeMessage
 import io.modelcontextprotocol.kotlin.sdk.types.JSONRPCMessage
+import io.modelcontextprotocol.kotlin.sdk.types.McpException
+import io.modelcontextprotocol.kotlin.sdk.types.RPCError.ErrorCode.CONNECTION_CLOSED
+import io.modelcontextprotocol.kotlin.sdk.types.RPCError.ErrorCode.INTERNAL_ERROR
 import kotlinx.coroutines.CoroutineName
 import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.Job
 import kotlinx.coroutines.SupervisorJob
+import kotlinx.coroutines.cancel
 import kotlinx.coroutines.cancelAndJoin
 import kotlinx.coroutines.channels.Channel
+import kotlinx.coroutines.channels.ClosedSendChannelException
+import kotlinx.coroutines.channels.ProducerScope
 import kotlinx.coroutines.channels.consumeEach
+import kotlinx.coroutines.flow.channelFlow
 import kotlinx.coroutines.isActive
 import kotlinx.coroutines.launch
+import kotlinx.coroutines.yield
 import kotlinx.io.Buffer
+import kotlinx.io.IOException
 import kotlinx.io.Sink
 import kotlinx.io.Source
 import kotlinx.io.buffered
 import kotlinx.io.readByteArray
 import kotlinx.io.writeString
+import kotlinx.serialization.SerializationException
 import kotlin.concurrent.atomics.AtomicBoolean
 import kotlin.concurrent.atomics.ExperimentalAtomicApi
 import kotlin.coroutines.CoroutineContext
+import kotlin.jvm.JvmOverloads
 
 /**
- * A transport implementation for JSON-RPC communication that leverages standard input and output streams.
+ * A transport implementation for JSON-RPC communication over standard I/O streams.
  *
- * This class reads from an input stream to process incoming JSON-RPC messages and writes JSON-RPC messages
- * to an output stream.
+ * Reads JSON-RPC messages from [input] and writes messages to [output]. Optionally monitors
+ * [error] stream for stderr output with configurable severity handling.
+ *
+ * ## Usage Example
+ * ```kotlin
+ * val process = ProcessBuilder("mcp-server").start()
+ *
+ * val transport = StdioClientTransport(
+ *     input = process.inputStream.asSource().buffered(),
+ *     output = process.outputStream.asSink().buffered(),
+ *     error = process.errorStream.asSource().buffered()
+ * ) { stderrLine ->
+ *     when {
+ *         stderrLine.contains("error", ignoreCase = true) -> StderrSeverity.FATAL
+ *         stderrLine.contains("warning", ignoreCase = true) -> StderrSeverity.WARNING
+ *         else -> StderrSeverity.INFO
+ *     }
+ * }
+ *
+ * transport.start()
+ * ```
  *
  * @param input The input stream where messages are received.
  * @param output The output stream where messages are sent.
+ * @param error Optional error stream for stderr monitoring.
+ * @param sendChannel Channel for outbound messages. Default: buffered channel (capacity 64).
+ * @param classifyStderr Callback to classify stderr lines. Return [StderrSeverity.FATAL] to fail transport,
+ *                       or [StderrSeverity.WARNING] / [StderrSeverity.INFO] / [StderrSeverity.DEBUG]
+ *                       to log, or [StderrSeverity.IGNORE] to discard.
+ *                       Default value: [StderrSeverity.DEBUG].
+ * @see <a href="https://modelcontextprotocol.io/specification/2025-11-25/basic/transports#stdio">MCP Specification</a>
  */
 @OptIn(ExperimentalAtomicApi::class)
-public class StdioClientTransport(private val input: Source, private val output: Sink) : AbstractTransport() {
-    private val logger = KotlinLogging.logger {}
-    private val ioCoroutineContext: CoroutineContext = IODispatcher
-    private val scope by lazy {
-        CoroutineScope(ioCoroutineContext + SupervisorJob())
+public class StdioClientTransport @JvmOverloads public constructor(
+    private val input: Source,
+    private val output: Sink,
+    private val error: Source? = null,
+    private val sendChannel: Channel<JSONRPCMessage> = Channel(Channel.BUFFERED),
+    private val classifyStderr: (String) -> StderrSeverity = { DEBUG },
+) : AbstractTransport() {
+
+    private companion object {
+        /**
+         * Buffer size for I/O operations.
+         * 8KB is optimal for most systems (matches default page size).
+         */
+        const val BUFFER_SIZE = 8 * 1024L
+
+        private val logger = KotlinLogging.logger {}
     }
-    private var job: Job? = null
-    private val initialized: AtomicBoolean = AtomicBoolean(false)
-    private val sendChannel = Channel<JSONRPCMessage>(Channel.UNLIMITED)
-    private val readBuffer = ReadBuffer()
+
+    /**
+     * Severity classification for stderr messages.
+     *
+     * - [FATAL]: Calls error handler and terminates transport.
+     * - [WARNING]: Logs at WARN level, transport continues.
+     * - [INFO]: Logs at INFO level, transport continues.
+     * - [DEBUG]: Logs at DEBUG level, transport continues.
+     * - [IGNORE]: Discards message silently, transport continues.
+     */
+    public enum class StderrSeverity { FATAL, WARNING, INFO, DEBUG, IGNORE }
+
+    private val ioCoroutineContext: CoroutineContext = IODispatcher
+    private val scope = CoroutineScope(SupervisorJob() + Dispatchers.Default)
+    private val initialized = AtomicBoolean(false)
+    private val onCloseCalled = AtomicBoolean(false)
 
     override suspend fun start() {
         if (!initialized.compareAndSet(expectedValue = false, newValue = true)) {
             error("StdioClientTransport already started!")
         }
-
         logger.debug { "Starting StdioClientTransport..." }
 
-        val outputStream = output.buffered()
+        // Producers run on IODispatcher for I/O
+        // Collector runs on Default dispatcher for message handling
+        scope.launch(CoroutineName("StdioClientTransport.IO#${hashCode()}")) {
+            var writeJob: Job? = null
+            val mainScope = this
+            try {
+                // Explicitly use ioCoroutineContext for I/O operations
+                writeJob = launch(ioCoroutineContext) {
+                    logger.debug { "Write coroutine started." }
+                    output.buffered().use { sink ->
+                        sendChannel.consumeEach { message ->
+                            sendOutboundMessage(message, sink, mainScope)
+                            yield() // Giving other coroutines a chance to run
+                        }
+                    }
+                }
 
-        job = scope.launch(CoroutineName("StdioClientTransport.IO#${hashCode()}")) {
-            val readJob = launch {
-                logger.debug { "Read coroutine started." }
-                try {
-                    input.use {
-                        while (isActive) {
-                            val buffer = Buffer()
-                            val bytesRead = input.readAtMostTo(buffer, 8192)
-                            if (bytesRead == -1L) break
-                            if (bytesRead > 0L) {
-                                readBuffer.append(buffer.readByteArray())
-                                processReadBuffer()
+                val eventsFlow = channelFlow {
+                    launch(ioCoroutineContext) {
+                        logger.debug { "Read stdin coroutine started." }
+                        val readBuffer = ReadBuffer() // parses bytes into JSONRPCMessage
+                        readSource(stream = ProcessStream.Stdin, source = input, channel = this@channelFlow) { bytes ->
+                            readBuffer.append(bytes)
+                            do {
+                                val msg = readBuffer.readMessage()
+                                msg?.let { send(Event.JsonRpc(msg)) }
+                            } while (msg != null)
+                        }
+                    }.invokeOnCompletion {
+                        logger.debug(it) { "Read stdin coroutine finished." }
+                    }
+
+                    error?.let { source ->
+                        launch(ioCoroutineContext) {
+                            logger.debug { "Read stderr coroutine started." }
+                            readSource(
+                                stream = ProcessStream.Stderr,
+                                source = source,
+                                channel = this@channelFlow,
+                            ) { bytes ->
+                                val str = bytes.decodeToString()
+                                send(Event.StderrEvent(str))
                             }
                         }
                     }
-                } catch (e: Exception) {
-                    _onError.invoke(e)
-                    logger.error(e) { "Error reading from input stream" }
                 }
-            }
 
-            val writeJob = launch {
-                logger.debug { "Write coroutine started." }
-                try {
-                    sendChannel.consumeEach { message ->
-                        val json = serializeMessage(message)
-                        outputStream.writeString(json)
-                        outputStream.flush()
-                    }
-                } catch (e: Throwable) {
-                    if (isActive) {
-                        _onError.invoke(e)
-                        logger.error(e) { "Error writing to output stream" }
-                    }
-                } finally {
-                    output.close()
-                }
-            }
+                // Collect events on handlerCoroutineContext (Dispatchers.Default from parent scope)
+                // No flowOn necessary - collection runs in parent launch context
+                eventsFlow
+                    .collect { event ->
+                        when (event) {
+                            is Event.JsonRpc -> {
+                                handleJSONRPCMessage(event.message)
+                            }
 
-            readJob.join()
-            writeJob.cancelAndJoin()
-            _onClose.invoke()
+                            is Event.StderrEvent -> {
+                                val errorSeverity = classifyStderr(event.message)
+                                when (errorSeverity) {
+                                    FATAL -> {
+                                        runCatching {
+                                            _onError(
+                                                McpException(INTERNAL_ERROR, "Message in StdErr: ${event.message}"),
+                                            )
+                                        }
+                                        stopProcessing("Fatal STDERR message received")
+                                    }
+
+                                    WARNING -> {
+                                        logger.warn { "STDERR message received: ${event.message}" }
+                                    }
+
+                                    INFO -> {
+                                        logger.info { "STDERR message received: ${event.message}" }
+                                    }
+
+                                    DEBUG -> {
+                                        logger.debug { "STDERR message received: ${event.message}" }
+                                    }
+
+                                    IGNORE -> {
+                                        // do nothing
+                                    }
+                                }
+                            }
+
+                            is Event.EOFEvent -> {
+                                if (event.stream == ProcessStream.Stdin) {
+                                    stopProcessing("EOF in ${event.stream}")
+                                }
+                            }
+
+                            is Event.IOErrorEvent -> {
+                                runCatching { _onError(event.cause) }
+                                stopProcessing("IO Error", event.cause)
+                            }
+                        }
+                    }
+            } finally {
+                // Wait for write job to complete before closing, matching old implementation
+                writeJob?.cancelAndJoin()
+                logger.debug { "Transport coroutine completed, calling onClose" }
+                callOnCloseOnce()
+            }
         }
     }
 
     override suspend fun send(message: JSONRPCMessage, options: TransportSendOptions?) {
-        if (!initialized.load()) {
-            error("Transport not started")
+        check(initialized.load()) { "Transport is not started" }
+        check(!onCloseCalled.load()) { "Transport is closed" }
+        @Suppress("TooGenericExceptionCaught", "SwallowedException")
+        try {
+            sendChannel.send(message)
+        } catch (e: ClosedSendChannelException) {
+            logger.debug(e) { "Cannot send message: transport is closed" }
+            throw McpException(CONNECTION_CLOSED, "Transport is closed")
+        } catch (e: McpException) {
+            logger.debug(e) { "Error while sending message: ${e.message}" }
+            throw e
+        } catch (e: Exception) {
+            logger.error(e) { "Error while sending message: ${e.message}" }
+            throw McpException(INTERNAL_ERROR, "Error while sending message: ${e.message}")
         }
-
-        sendChannel.send(message)
     }
 
     override suspend fun close() {
         if (!initialized.compareAndSet(expectedValue = true, newValue = false)) {
-            error("Transport is already closed")
+            return // Already closed
         }
-        job?.cancelAndJoin()
-        input.close()
-        output.close()
-        readBuffer.clear()
-        sendChannel.close()
-        _onClose.invoke()
+        scope.stopProcessing("Closed")
+        scope.coroutineContext[Job]?.join() // Wait for all coroutines to complete
     }
 
-    private suspend fun processReadBuffer() {
-        while (true) {
-            val msg = readBuffer.readMessage() ?: break
-            try {
-                _onMessage.invoke(msg)
-            } catch (e: Throwable) {
-                _onError.invoke(e)
-                logger.error(e) { "Error processing message." }
+    private fun callOnCloseOnce() {
+        if (onCloseCalled.compareAndSet(expectedValue = false, newValue = true)) {
+            runCatching { _onClose() }
+        }
+    }
+
+    private fun sendOutboundMessage(message: JSONRPCMessage, sink: Sink, mainScope: CoroutineScope) {
+        try {
+            val json = serializeMessage(message)
+            sink.writeString(json)
+            sink.flush()
+        } catch (e: SerializationException) {
+            logger.warn(e) { "Can't serialize message" }
+            runCatching { _onError(McpException(INTERNAL_ERROR, "Serialization error")) }
+            mainScope.stopProcessing("Can't serialize message", e)
+        } catch (e: IOException) {
+            logger.warn(e) { "Can't send message" }
+            runCatching { _onError(McpException(CONNECTION_CLOSED, "Can't send message. Connection closed")) }
+            mainScope.stopProcessing("Write I/O failed", e)
+        }
+    }
+
+    private suspend fun handleJSONRPCMessage(msg: JSONRPCMessage) {
+        @Suppress("TooGenericExceptionCaught")
+        try {
+            _onMessage.invoke(msg)
+        } catch (e: Throwable) {
+            logger.error(e) { "Error processing message." }
+            runCatching { _onError.invoke(e) }
+        }
+    }
+
+    private fun CoroutineScope.stopProcessing(reason: String, cause: Throwable? = null) {
+        sendChannel.close() // Stop accepting new messages
+        callOnCloseOnce()
+        cancel(reason, cause) // cancel current coroutine context
+    }
+
+    private suspend fun CoroutineScope.readSource(
+        stream: ProcessStream,
+        source: Source,
+        channel: ProducerScope<Event>,
+        bytesConsumer: suspend (ByteArray) -> Unit,
+    ) {
+        val buffer = Buffer()
+        try {
+            source.use { source ->
+                while (isActive) {
+                    val bytesRead = source.readAtMostTo(buffer, BUFFER_SIZE)
+                    if (bytesRead == -1L) {
+                        logger.debug { "EOF reached in $stream" }
+                        channel.send(Event.EOFEvent(stream))
+                        break
+                    }
+
+                    if (bytesRead > 0L) {
+                        val bytes = buffer.readByteArray()
+                        buffer.clear()
+                        bytesConsumer.invoke(bytes)
+                    }
+
+                    yield() // Giving other coroutines a chance to run
+                }
             }
+        } catch (exception: IOException) {
+            logger.debug(exception) { "IOException while reading stream" }
+            channel.send(Event.IOErrorEvent(stream, exception))
+        } finally {
+            buffer.clear()
+        }
+    }
+
+    private enum class ProcessStream { Stdin, Stderr, Stdout }
+
+    /**
+     * Represents an event in the communication process.
+     *
+     * Events are a sealed hierarchy of different types of communication signals
+     * between processes. These events are used to manage and interpret information
+     * or errors generated during the operation of the associated transport.
+     */
+    private sealed interface Event {
+
+        data class JsonRpc(val message: JSONRPCMessage) : Event
+
+        data class StderrEvent(val message: String) : Event
+
+        data class EOFEvent(val stream: ProcessStream) : Event
+
+        data class IOErrorEvent(val stream: ProcessStream, val cause: Throwable) : Event {
+            override fun toString(): String = "IOErrorEvent(stream=$stream, cause=${cause.message})"
         }
     }
 }

kotlin-sdk-client/src/commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/client/StreamableHttpClientTransport.kt

@@ -105,6 +105,7 @@ public class StreamableHttpClientTransport(
         resumptionToken: String?,
         onResumptionToken: ((String) -> Unit)? = null,
     ) {
+        check(initialized.load()) { "Transport is not started" }
         logger.debug { "Client sending message via POST to $url: ${McpJson.encodeToString(message)}" }
 
         // If we have a resumption token, reconnect the SSE stream with it

kotlin-sdk-client/src/commonTest/kotlin/io/modelcontextprotocol/kotlin/sdk/client/AbstractClientTransportLifecycleTest.kt

@@ -0,0 +1,86 @@
+package io.modelcontextprotocol.kotlin.sdk.client
+
+import io.kotest.assertions.throwables.shouldThrow
+import io.kotest.matchers.shouldBe
+import io.kotest.matchers.string.shouldContain
+import io.modelcontextprotocol.kotlin.sdk.shared.AbstractTransport
+import io.modelcontextprotocol.kotlin.sdk.types.PingRequest
+import io.modelcontextprotocol.kotlin.sdk.types.toJSON
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.test.runTest
+import kotlin.test.BeforeTest
+import kotlin.test.Test
+import kotlin.time.Duration.Companion.milliseconds
+
+abstract class AbstractClientTransportLifecycleTest<T : AbstractTransport> {
+
+    protected lateinit var transport: T
+
+    @BeforeTest
+    fun beforeEach() {
+        transport = createTransport()
+    }
+
+    @Test
+    fun `should throw when started twice`() = runTest {
+        transport.start()
+
+        val exception = shouldThrow<IllegalStateException> {
+            transport.start()
+        }
+        exception.message shouldContain "already started"
+    }
+
+    @Test
+    fun `should be idempotent when closed twice`() = runTest {
+        val transport = createTransport()
+
+        transport.start()
+        transport.close()
+
+        // Second close should not throw
+        transport.close()
+    }
+
+    @Test
+    fun `should throw when sending before start`() = runTest {
+        val transport = createTransport()
+
+        val exception = shouldThrow<IllegalStateException> {
+            transport.send(PingRequest().toJSON())
+        }
+        exception.message shouldContain "not started"
+    }
+
+    @Test
+    fun `should throw when sending after close`() = runTest {
+        val transport = createTransport()
+
+        transport.start()
+        delay(50.milliseconds)
+        transport.close()
+
+        shouldThrow<Exception> {
+            transport.send(PingRequest().toJSON())
+        }
+    }
+
+    @Test
+    fun `should call onClose exactly once`() = runTest {
+        val transport = createTransport()
+
+        var closeCallCount = 0
+        transport.onClose { closeCallCount++ }
+
+        transport.start()
+        delay(50.milliseconds)
+
+        // Multiple close attempts
+        transport.close()
+        transport.close()
+
+        closeCallCount shouldBe 1
+    }
+
+    protected abstract fun createTransport(): T
+}

kotlin-sdk-client/src/commonTest/kotlin/io/modelcontextprotocol/kotlin/sdk/client/stdio/StdioClientTransportLifecycleTest.kt

@@ -0,0 +1,26 @@
+package io.modelcontextprotocol.kotlin.sdk.client.stdio
+
+import io.modelcontextprotocol.kotlin.sdk.client.AbstractClientTransportLifecycleTest
+import io.modelcontextprotocol.kotlin.sdk.client.StdioClientTransport
+import kotlinx.io.Buffer
+import kotlin.test.Ignore
+import kotlin.test.Test
+
+class StdioClientTransportLifecycleTest : AbstractClientTransportLifecycleTest<StdioClientTransport>() {
+
+    /**
+     * Dummy method to make IDE treat this class as a test
+     */
+    @Test
+    @Ignore
+    fun dummyTest() = Unit
+
+    override fun createTransport(): StdioClientTransport {
+        val inputBuffer = Buffer()
+        val outputBuffer = Buffer()
+        return StdioClientTransport(
+            input = inputBuffer,
+            output = outputBuffer,
+        )
+    }
+}

kotlin-sdk-client/src/commonTest/kotlin/io/modelcontextprotocol/kotlin/sdk/client/streamable/http/StreamableHttpClientTransportTest.kt

@@ -0,0 +1,423 @@
+package io.modelcontextprotocol.kotlin.sdk.client.streamable.http
+
+import io.ktor.client.HttpClient
+import io.ktor.client.engine.mock.MockEngine
+import io.ktor.client.engine.mock.MockRequestHandler
+import io.ktor.client.engine.mock.respond
+import io.ktor.client.plugins.sse.SSE
+import io.ktor.http.ContentType
+import io.ktor.http.HttpHeaders
+import io.ktor.http.HttpMethod
+import io.ktor.http.HttpStatusCode
+import io.ktor.http.content.TextContent
+import io.ktor.http.headersOf
+import io.ktor.utils.io.ByteReadChannel
+import io.modelcontextprotocol.kotlin.sdk.client.Client
+import io.modelcontextprotocol.kotlin.sdk.client.StreamableHttpClientTransport
+import io.modelcontextprotocol.kotlin.sdk.types.Implementation
+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.McpJson
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.test.runTest
+import kotlinx.coroutines.withContext
+import kotlinx.coroutines.withTimeout
+import kotlinx.serialization.json.JsonObject
+import kotlinx.serialization.json.JsonPrimitive
+import kotlinx.serialization.json.buildJsonObject
+import kotlin.test.Ignore
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertFailsWith
+import kotlin.test.assertNull
+import kotlin.test.assertTrue
+import kotlin.time.Duration.Companion.seconds
+
+class StreamableHttpClientTransportTest {
+
+    private fun createTransport(handler: MockRequestHandler): StreamableHttpClientTransport {
+        val mockEngine = MockEngine(handler)
+        val httpClient = HttpClient(mockEngine) {
+            install(SSE) {
+                reconnectionTime = 1.seconds
+            }
+        }
+
+        return StreamableHttpClientTransport(httpClient, url = "http://localhost:8080/mcp")
+    }
+
+    @Test
+    fun testSendJsonRpcMessage() = runTest {
+        val message = JSONRPCRequest(
+            id = "test-id",
+            method = "test",
+            params = buildJsonObject { },
+        )
+
+        val transport = createTransport { request ->
+            assertEquals(HttpMethod.Post, request.method)
+            assertEquals("http://localhost:8080/mcp", request.url.toString())
+            assertEquals(ContentType.Application.Json, request.body.contentType)
+
+            val body = (request.body as TextContent).text
+            val decodedMessage = McpJson.decodeFromString<JSONRPCMessage>(body)
+            assertEquals(message, decodedMessage)
+
+            respond(
+                content = "",
+                status = HttpStatusCode.Accepted,
+            )
+        }
+
+        transport.start()
+        transport.send(message)
+        transport.close()
+    }
+
+    @Test
+    fun testStoreSessionId() = runTest {
+        val initMessage = JSONRPCRequest(
+            id = "test-id",
+            method = "initialize",
+            params = buildJsonObject {
+                put(
+                    "clientInfo",
+                    buildJsonObject {
+                        put("name", JsonPrimitive("test-client"))
+                        put("version", JsonPrimitive("1.0"))
+                    },
+                )
+                put("protocolVersion", JsonPrimitive("2025-06-18"))
+            },
+        )
+
+        val transport = createTransport { request ->
+            when (val msg = McpJson.decodeFromString<JSONRPCMessage>((request.body as TextContent).text)) {
+                is JSONRPCRequest if msg.method == "initialize" -> respond(
+                    content = "",
+                    status = HttpStatusCode.OK,
+                    headers = headersOf("mcp-session-id", "test-session-id"),
+                )
+
+                is JSONRPCNotification if msg.method == "test" -> {
+                    assertEquals("test-session-id", request.headers["mcp-session-id"])
+                    respond(
+                        content = "",
+                        status = HttpStatusCode.Accepted,
+                    )
+                }
+
+                else -> error("Unexpected message: $msg")
+            }
+        }
+
+        transport.start()
+        transport.send(initMessage)
+
+        assertEquals("test-session-id", transport.sessionId)
+
+        transport.send(JSONRPCNotification(method = "test"))
+
+        transport.close()
+    }
+
+    @Test
+    fun testTerminateSession() = runTest {
+//        transport.sessionId = "test-session-id"
+
+        val transport = createTransport { request ->
+            assertEquals(HttpMethod.Delete, request.method)
+            assertEquals("test-session-id", request.headers["mcp-session-id"])
+            respond(
+                content = "",
+                status = HttpStatusCode.OK,
+            )
+        }
+
+        transport.start()
+        transport.terminateSession()
+
+        assertNull(transport.sessionId)
+        transport.close()
+    }
+
+    @Test
+    fun testTerminateSessionHandle405() = runTest {
+//        transport.sessionId = "test-session-id"
+
+        val transport = createTransport { request ->
+            assertEquals(HttpMethod.Delete, request.method)
+            respond(
+                content = "",
+                status = HttpStatusCode.MethodNotAllowed,
+            )
+        }
+
+        transport.start()
+        // Should not throw for 405
+        transport.terminateSession()
+
+        // Session ID should still be cleared
+        assertNull(transport.sessionId)
+        transport.close()
+    }
+
+    @Test
+    fun testProtocolVersionHeader() = runTest {
+        val transport = createTransport { request ->
+            assertEquals("2025-06-18", request.headers["mcp-protocol-version"])
+            respond(
+                content = "",
+                status = HttpStatusCode.Accepted,
+            )
+        }
+        transport.protocolVersion = "2025-06-18"
+
+        transport.start()
+        transport.send(JSONRPCNotification(method = "test"))
+        transport.close()
+    }
+
+    // Engine doesn't support SSECapability: https://youtrack.jetbrains.com/issue/KTOR-8177/MockEngine-Add-SSE-support
+    @Ignore
+    @Test
+    fun testNotificationSchemaE2E() = runTest {
+        val receivedMessages = mutableListOf<JSONRPCMessage>()
+        var sseStarted = false
+
+        val transport = createTransport { request ->
+            when (request.method) {
+                HttpMethod.Post if request.body.toString().contains("notifications/initialized") -> {
+                    respond(
+                        content = "",
+                        status = HttpStatusCode.Accepted,
+                        headers = headersOf("mcp-session-id", "notification-test-session"),
+                    )
+                }
+
+                // Handle SSE connection
+                HttpMethod.Get -> {
+                    sseStarted = true
+                    val sseContent = buildString {
+                        // Server sends various notifications
+                        appendLine("event: message")
+                        appendLine("id: 1")
+                        appendLine(
+                            """data: {"jsonrpc":"2.0","method":"notifications/progress","params":{"progressToken":"upload-123","progress":50,"total":100}}""",
+                        )
+                        appendLine()
+
+                        appendLine("event: message")
+                        appendLine("id: 2")
+                        appendLine("""data: {"jsonrpc":"2.0","method":"notifications/resources/list_changed"}""")
+                        appendLine()
+
+                        appendLine("event: message")
+                        appendLine("id: 3")
+                        appendLine("""data: {"jsonrpc":"2.0","method":"notifications/tools/list_changed"}""")
+                        appendLine()
+                    }
+                    respond(
+                        content = ByteReadChannel(sseContent),
+                        status = HttpStatusCode.OK,
+                        headers = headersOf(
+                            HttpHeaders.ContentType,
+                            ContentType.Text.EventStream.toString(),
+                        ),
+                    )
+                }
+
+                // Handle regular notifications
+                HttpMethod.Post -> {
+                    respond(
+                        content = "",
+                        status = HttpStatusCode.Accepted,
+                    )
+                }
+
+                else -> respond("", HttpStatusCode.OK)
+            }
+        }
+
+        transport.onMessage { message ->
+            receivedMessages.add(message)
+        }
+
+        transport.start()
+
+        // Test 1: Send initialized notification to trigger SSE
+        val initializedNotification = JSONRPCNotification(
+            method = "notifications/initialized",
+            params = buildJsonObject {
+                put("protocolVersion", JsonPrimitive("1.0"))
+                put(
+                    "capabilities",
+                    buildJsonObject {
+                        put("tools", JsonPrimitive(true))
+                        put("resources", JsonPrimitive(true))
+                    },
+                )
+            },
+        )
+
+        transport.send(initializedNotification)
+
+        // Verify SSE was triggered
+        assertTrue(sseStarted, "SSE should start after initialized notification")
+
+        // Test 2: Verify received notifications
+        assertEquals(3, receivedMessages.size)
+        assertTrue(receivedMessages.all { it is JSONRPCNotification })
+
+        val notifications = receivedMessages.filterIsInstance<JSONRPCNotification>()
+
+        // Verify progress notification
+        val progressNotif = notifications[0]
+        assertEquals("notifications/progress", progressNotif.method)
+        val progressParams = progressNotif.params as JsonObject
+        assertEquals("upload-123", (progressParams["progressToken"] as JsonPrimitive).content)
+        assertEquals(50, (progressParams["progress"] as JsonPrimitive).content.toInt())
+
+        // Verify list changed notifications
+        assertEquals("notifications/resources/list_changed", notifications[1].method)
+        assertEquals("notifications/tools/list_changed", notifications[2].method)
+
+        // Test 3: Send various client notifications
+        val clientNotifications = listOf(
+            JSONRPCNotification(
+                method = "notifications/progress",
+                params = buildJsonObject {
+                    put("progressToken", JsonPrimitive("download-456"))
+                    put("progress", JsonPrimitive(75))
+                },
+            ),
+            JSONRPCNotification(
+                method = "notifications/cancelled",
+                params = buildJsonObject {
+                    put("requestId", JsonPrimitive("req-789"))
+                    put("reason", JsonPrimitive("user_cancelled"))
+                },
+            ),
+            JSONRPCNotification(
+                method = "notifications/message",
+                params = buildJsonObject {
+                    put("level", JsonPrimitive("info"))
+                    put("message", JsonPrimitive("Operation completed"))
+                    put(
+                        "data",
+                        buildJsonObject {
+                            put("duration", JsonPrimitive(1234))
+                        },
+                    )
+                },
+            ),
+        )
+
+        // Send all client notifications
+        clientNotifications.forEach { notification ->
+            transport.send(notification)
+        }
+
+        // Verify session ID is maintained
+        assertEquals("notification-test-session", transport.sessionId)
+        transport.close()
+    }
+
+    // Engine doesn't support SSECapability: https://youtrack.jetbrains.com/issue/KTOR-8177/MockEngine-Add-SSE-support
+    @Ignore
+    @Test
+    fun testNotificationWithResumptionToken() = runTest {
+        var resumptionTokenReceived: String? = null
+        var lastEventIdSent: String? = null
+
+        val transport = createTransport { request ->
+            // Capture Last-Event-ID header
+            lastEventIdSent = request.headers["Last-Event-ID"]
+
+            when (request.method) {
+                HttpMethod.Get -> {
+                    val sseContent = buildString {
+                        appendLine("event: message")
+                        appendLine("id: resume-100")
+                        appendLine(
+                            """data: {"jsonrpc":"2.0","method":"notifications/resumed","params":{"fromToken":"$lastEventIdSent"}}""",
+                        )
+                        appendLine()
+                    }
+                    respond(
+                        content = ByteReadChannel(sseContent),
+                        status = HttpStatusCode.OK,
+                        headers = headersOf(
+                            HttpHeaders.ContentType,
+                            ContentType.Text.EventStream.toString(),
+                        ),
+                    )
+                }
+
+                else -> respond("", HttpStatusCode.Accepted)
+            }
+        }
+
+        transport.start()
+
+        // Send notification with resumption token
+        transport.send(
+            message = JSONRPCNotification(
+                method = "notifications/test",
+                params = buildJsonObject {
+                    put("data", JsonPrimitive("test-data"))
+                },
+            ),
+            resumptionToken = "previous-token-99",
+            onResumptionToken = { token ->
+                resumptionTokenReceived = token
+            },
+        )
+
+        // Wait for response
+        delay(1.seconds)
+
+        // Verify resumption token was sent in header
+        assertEquals("previous-token-99", lastEventIdSent)
+
+        // Verify new resumption token was received
+        assertEquals("resume-100", resumptionTokenReceived)
+        transport.close()
+    }
+
+    @Test
+    fun testClientConnectWithInvalidJson() = runTest {
+        // Transport under test: respond with invalid JSON for the initialize request
+        val transport = createTransport { _ ->
+            respond(
+                "this is not valid json",
+                status = HttpStatusCode.OK,
+                headers = headersOf(HttpHeaders.ContentType, ContentType.Application.Json.toString()),
+            )
+        }
+
+        val client = Client(
+            clientInfo = Implementation(
+                name = "test-client",
+                version = "1.0",
+            ),
+        )
+
+        try {
+            // Real time-keeping is needed; otherwise Protocol will always throw TimeoutCancellationException in tests
+            assertFailsWith<IllegalStateException>(
+                message = "Expected client.connect to fail on invalid JSON response",
+            ) {
+                withContext(Dispatchers.Default.limitedParallelism(1)) {
+                    withTimeout(5.seconds) {
+                        client.connect(transport)
+                    }
+                }
+            }
+        } finally {
+            transport.close()
+        }
+    }
+}

kotlin-sdk-client/src/commonTest/kotlin/io/modelcontextprotocol/kotlin/sdk/client/streamable/http/StreamingHttpClientTransportLifecycleTest.kt

@@ -0,0 +1,43 @@
+package io.modelcontextprotocol.kotlin.sdk.client.streamable.http
+
+import io.ktor.client.HttpClient
+import io.ktor.client.engine.mock.MockEngine
+import io.ktor.client.engine.mock.respond
+import io.ktor.client.plugins.sse.SSE
+import io.ktor.http.ContentType
+import io.ktor.http.HttpHeaders
+import io.ktor.http.HttpStatusCode
+import io.ktor.http.headersOf
+import io.modelcontextprotocol.kotlin.sdk.client.AbstractClientTransportLifecycleTest
+import io.modelcontextprotocol.kotlin.sdk.client.StreamableHttpClientTransport
+import kotlin.test.Ignore
+import kotlin.test.Test
+import kotlin.time.Duration.Companion.seconds
+
+class StreamingHttpClientTransportLifecycleTest :
+    AbstractClientTransportLifecycleTest<StreamableHttpClientTransport>() {
+
+    /**
+     * Dummy method to make IDE treat this class as a test
+     */
+    @Test
+    @Ignore
+    fun dummyTest() = Unit
+
+    override fun createTransport(): StreamableHttpClientTransport {
+        val mockEngine = MockEngine {
+            respond(
+                "this is not valid json",
+                status = HttpStatusCode.OK,
+                headers = headersOf(HttpHeaders.ContentType, ContentType.Application.Json.toString()),
+            )
+        }
+        val httpClient = HttpClient(mockEngine) {
+            install(SSE) {
+                reconnectionTime = 1.seconds
+            }
+        }
+
+        return StreamableHttpClientTransport(httpClient, url = "http://localhost:8080/mcp")
+    }
+}

kotlin-sdk-client/src/jvmTest/kotlin/io/modelcontextprotocol/kotlin/sdk/client/stdio/StdioClientTransportErrorHandlingTest.kt

@@ -0,0 +1,98 @@
+package io.modelcontextprotocol.kotlin.sdk.client.stdio
+
+import io.kotest.matchers.booleans.shouldBeFalse
+import io.kotest.matchers.shouldBe
+import io.modelcontextprotocol.kotlin.sdk.client.StdioClientTransport
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.test.runTest
+import kotlinx.io.Buffer
+import kotlinx.io.writeString
+import kotlin.concurrent.atomics.AtomicBoolean
+import kotlin.concurrent.atomics.ExperimentalAtomicApi
+import kotlin.test.Test
+import kotlin.time.Duration.Companion.milliseconds
+
+/**
+ * Tests for StdioClientTransport error handling: EOF, IO errors, and edge cases.
+ */
+class StdioClientTransportErrorHandlingTest {
+
+    private lateinit var transport: StdioClientTransport
+
+    @OptIn(ExperimentalAtomicApi::class)
+    @Test
+    fun `should continue on stderr EOF`() = runTest {
+        val stderrBuffer = Buffer()
+        // Empty stderr = immediate EOF
+
+        val inputBuffer = Buffer()
+        inputBuffer.writeString("""data: {"jsonrpc":"2.0","method":"ping","id":1}\n\n""")
+        val outputBuffer = Buffer()
+
+        transport = StdioClientTransport(
+            input = inputBuffer,
+            output = outputBuffer,
+            error = stderrBuffer,
+        )
+
+        val closeCalled = AtomicBoolean(false)
+        transport.onClose { closeCalled.store(true) }
+
+        transport.start()
+        delay(200.milliseconds)
+
+        // Stderr EOF should not close transport
+        closeCalled.load() shouldBe false
+
+        transport.close()
+        closeCalled.load() shouldBe true
+    }
+
+    @Test
+    fun `should call onClose exactly once on error scenarios`() = runTest {
+        val stderrBuffer = Buffer()
+        stderrBuffer.write("FATAL: critical error\n".encodeToByteArray())
+
+        val inputBuffer = Buffer()
+        val outputBuffer = Buffer()
+
+        var closeCallCount = 0
+
+        transport = StdioClientTransport(
+            input = inputBuffer,
+            output = outputBuffer,
+            error = stderrBuffer,
+            classifyStderr = { StdioClientTransport.StderrSeverity.FATAL },
+        )
+
+        transport.onClose { closeCallCount++ }
+
+        transport.start()
+        delay(100.milliseconds)
+
+        // Explicit close after error already closed it
+        transport.close()
+
+        closeCallCount shouldBe 1
+    }
+
+    @Test
+    fun `should handle empty input gracefully`() = runTest {
+        val inputBuffer = Buffer()
+        val outputBuffer = Buffer()
+
+        transport = StdioClientTransport(
+            input = inputBuffer,
+            output = outputBuffer,
+        )
+
+        var errorCalled = false
+        transport.onError { errorCalled = true }
+
+        transport.start()
+        delay(100.milliseconds)
+
+        // Empty input should close cleanly without error
+        errorCalled.shouldBeFalse()
+    }
+}