Initialization errors in McpAsyncClient not emitted immediately #393

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:

• Non initialized McpServerSession hangs and timesout instead of providing helpful error message #275
• Error : No handler registered for notification method: notifications/stderr #377
• webclient onErrorResume does not work #151

Environment

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

io.modelcontextprotocol java-sdk commit: 8a2f97f

Steps to reproduce

1. Clone and install the repo, checkout 8a2f97f
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)
}