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

Inizia tracciando McpAsyncClient.initialize() e il percorso di avvio di StdioClientTransport utilizzato da Main.kt, verificando come l’errore di avvio del processo raggiunge il publisher Reactor. Usa la riproduzione fornita e l’output del logger di Operators per confermare che l’errore viene emesso immediatamente, anziché solo dopo initializationTimeout; il lavoro è completato quando l’errore raggiunge doOnError/onErrorResume senza attendere il timeout.