Initialization errors in McpAsyncClient not emitted immediately

Bug Description
When subscribing to exceptions from McpAsyncClient.initialize() using onErrorResumein order to catch initialization failures immediately, nothing is emitted. I suspect this is caused by McpAsyncClient not propagating errors downstream.

I’m new to Reactor and don’t know how to fix this. Any simple tips or pointers, like which class or method handles error propagation, would be really helpful so I can send a PR.

Also, I believe this issue may be related to:

• https://github.com/modelcontextprotocol/java-sdk/issues/275
• https://github.com/modelcontextprotocol/java-sdk/issues/377
• https://github.com/modelcontextprotocol/java-sdk/issues/151

Environment

JDK: zulu-21
gradle: 8.10.2
IntelliJ IDEA 2025.1.1.1 (Community Edition) (K2 mode)

io.modelcontextprotocol java-sdk commit: https://github.com/modelcontextprotocol/java-sdk/commit/8a2f97f1b1995880c492159dd1b4c50266d0b579

Steps to reproduce

1. Clone and install the repo, checkout https://github.com/modelcontextprotocol/java-sdk/commit/8a2f97f1b1995880c492159dd1b4c50266d0b579
2. Run the script provided below
3. Observe that no immediate exception is thrown

Expected behavior

• When I initialize McpClient with an incorrect transport
• And internally, the reactor.core.publisher.Operators logger reports:
  java.lang.RuntimeException: Failed to start process with command: [a, a]
• Then I expect the initialization to fail immediately - not after my configured timeout

Minimal Complete Reproducible example
Main.kt:

import io.modelcontextprotocol.client.transport.StdioClientTransport
import io.modelcontextprotocol.client.transport.ServerParameters
import kotlinx.coroutines.reactive.awaitSingle
import mu.KotlinLogging
import reactor.core.publisher.Mono
import java.time.Duration


private val logger = KotlinLogging.logger {}

suspend fun main() {
    val params = ServerParameters.builder("a").args("b").build()

    val transport = StdioClientTransport(params)

    val client = io.modelcontextprotocol.client.McpClient
        .async(transport)
        .requestTimeout(Duration.ofSeconds(6))
        .initializationTimeout(Duration.ofSeconds(5))
        .build()

    client.initialize()
        .doOnError { t -> logger.error { t }; return@doOnError }
        .onErrorResume { return@onErrorResume Mono.error(it) }
        .awaitSingle()
}

build.gradle.kts:

plugins {
    kotlin("jvm") version "2.1.20"
}

group = "com.anything.mcp"
version = "1.0-SNAPSHOT"

repositories {
    mavenLocal()
    mavenCentral()
}

dependencies {
    implementation("io.modelcontextprotocol.sdk:mcp:0.11.0-SNAPSHOT")
    implementation("io.modelcontextprotocol.sdk:mcp-spring-webflux:0.11.0-SNAPSHOT")
    implementation("io.github.microutils:kotlin-logging:3.0.5")
    implementation("org.jetbrains.kotlinx:kotlinx-coroutines-reactor:1.8.0")
}

kotlin {
    jvmToolchain(21)
}

Bắt đầu bằng cách lần theo McpAsyncClient.initialize() và đường dẫn khởi động của StdioClientTransport được Main.kt sử dụng, kiểm tra cách lỗi khởi động tiến trình đến được Reactor publisher. Sử dụng bản tái hiện được cung cấp và đầu ra của Operators logger để xác nhận rằng lỗi được phát ra ngay lập tức thay vì chỉ sau initializationTimeout; hoàn tất khi lỗi đến được doOnError/onErrorResume mà không phải chờ timeout.