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

Beginne damit, McpAsyncClient.initialize() und den von Main.kt verwendeten Startpfad von StdioClientTransport nachzuverfolgen, und prüfe, wie der Fehler beim Starten des Prozesses den Reactor-Publisher erreicht. Verwende die bereitgestellte Reproduktion und die Ausgabe des Operators-Loggers, um zu bestätigen, dass der Fehler sofort ausgegeben wird und nicht erst nach initializationTimeout; abgeschlossen ist die Aufgabe, wenn der Fehler doOnError/onErrorResume erreicht, ohne auf das Timeout zu warten.