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)
}

Comienza siguiendo McpAsyncClient.initialize() y la ruta de inicio de StdioClientTransport utilizada por Main.kt, comprobando cómo el fallo al iniciar el proceso llega al publisher de Reactor. Usa la reproducción proporcionada y la salida del logger de Operators para confirmar que el fallo se emite inmediatamente, en lugar de solo después de initializationTimeout; se considera terminado cuando el error llega a doOnError/onErrorResume sin esperar al tiempo de espera.