Server incorrectly reports 'Transport error: disconnected' when sending ResourceUpdatedNotification

[byeblack]

Sorry, I misunderstood how the SDK was implemented.

Describe the bug
Server incorrectly reports "Transport error: disconnected" when sending notifications/resources/updated, even though the client successfully receives the data. This creates false error logs and makes it difficult to monitor actual connection issues.

To Reproduce
Steps to reproduce the behavior:

1. Start MCP server
2. Client subscribes to "memo://insights" resource
3. Server sends periodic ResourceUpdatedNotification (every 2 seconds in example)
4. Server logs show "Transport error: disconnected" for each notification
5. Client actually receives all notifications successfully

Expected behavior

1. Server should not log transport errors when notifications are successfully delivered
2. Only actual connection failures should be reported as errors
3. Notification mechanism should distinguish between real disconnections and successful deliveries

Example

    async fn subscribe(
        &self,
        SubscribeRequestParam { uri }: SubscribeRequestParam,
        ctx: RequestContext<RoleServer>,
    ) -> Result<(), McpError> {
        let peer = ctx.peer;

        tokio::spawn(async move {
            let span = tracing::info_span!("subscribe", uri = %uri);
            let _enter = span.enter();

            loop {
                tokio::time::sleep(std::time::Duration::from_secs(2)).await;

                if let Err(e) = peer
                    .notify_resource_updated(ResourceUpdatedNotificationParam { uri: uri.clone() })
                    .await
                {
                    // match e {
                    //     rmcp::ServiceError::Transport(e) => {
                    //         if e.to_string().trim() != "disconnected" {
                    //             tracing::error!(error = %e);
                    //             break;
                    //         }
                    //     }
                    //     _ => {
                    //         tracing::error!(error = %e);
                    //         break;
                    //     }
                    // }

                    tracing::error!(error = %e);
                }
            }
        });

        Ok(())
    }

Logs
Relevant log excerpts showing false positive errors:

{"timestamp":"2025-03-30T23:52:30.259112Z","level":"INFO","fields":{"message":"received request","id":"3","request":"SubscribeRequest(Request { method: SubscribeRequestMethod, params: SubscribeRequestParam { uri: \"memo://insights\" } })"},"threadId":"ThreadId(24)"}
{"timestamp":"2025-03-30T23:52:30.259129Z","level":"INFO","fields":{"message":"response message","id":"3","result":"EmptyResult(EmptyObject)"},"threadId":"ThreadId(24)"}
{"timestamp":"2025-03-30T23:52:30.259137Z","level":"DEBUG","fields":{"message":"new event","evt":"ToSink(Response(EmptyResult(EmptyObject), Number(3)))"},"threadId":"ThreadId(24)"}
{"timestamp":"2025-03-30T23:52:32.271100Z","level":"DEBUG","fields":{"message":"new event","evt":"ProxyMessage(Notification(ResourceUpdatedNotification(Notification { method: ResourceUpdatedNotificationMethod, params: ResourceUpdatedNotificationParam { uri: \"memo://insights\" } }), Sender { inner: Some(Inner { state: State { is_complete: false, is_closed: false, is_rx_task_set: true, is_tx_task_set: false } }) }))"},"threadId":"ThreadId(2)"}
{"timestamp":"2025-03-30T23:52:32.271903Z","level":"ERROR","fields":{"error":"Transport error: disconnected"},"threadId":"ThreadId(2)"}
{"timestamp":"2025-03-30T23:52:34.275746Z","level":"DEBUG","fields":{"message":"new event","evt":"ProxyMessage(Notification(ResourceUpdatedNotification(Notification { method: ResourceUpdatedNotificationMethod, params: ResourceUpdatedNotificationParam { uri: \"memo://insights\" } }), Sender { inner: Some(Inner { state: State { is_complete: false, is_closed: false, is_rx_task_set: true, is_tx_task_set: false } }) }))"},"threadId":"ThreadId(2)"}
{"timestamp":"2025-03-30T23:52:34.275819Z","level":"ERROR","fields":{"error":"Transport error: disconnected"},"threadId":"ThreadId(2)"}
{"timestamp":"2025-03-30T23:52:36.278846Z","level":"DEBUG","fields":{"message":"new event","evt":"ProxyMessage(Notification(ResourceUpdatedNotification(Notification { method: ResourceUpdatedNotificationMethod, params: ResourceUpdatedNotificationParam { uri: \"memo://insights\" } }), Sender { inner: Some(Inner { state: State { is_complete: false, is_closed: false, is_rx_task_set: true, is_tx_task_set: false } }) }))"},"threadId":"ThreadId(2)"}
{"timestamp":"2025-03-30T23:52:36.278922Z","level":"ERROR","fields":{"error":"Transport error: disconnected"},"threadId":"ThreadId(2)"}
{"timestamp":"2025-03-30T23:52:38.290143Z","level":"DEBUG","fields":{"message":"new event","evt":"ProxyMessage(Notification(ResourceUpdatedNotification(Notification { method: ResourceUpdatedNotificationMethod, params: ResourceUpdatedNotificationParam { uri: \"memo://insights\" } }), Sender { inner: Some(Inner { state: State { is_complete: false, is_closed: false, is_rx_task_set: true, is_tx_task_set: false } }) }))"},"threadId":"ThreadId(23)"}
{"timestamp":"2025-03-30T23:52:38.290234Z","level":"ERROR","fields":{"error":"Transport error: disconnected"},"threadId":"ThreadId(23)"}
{"timestamp":"2025-03-30T23:52:40.306348Z","level":"DEBUG","fields":{"message":"new event","evt":"ProxyMessage(Notification(ResourceUpdatedNotification(Notification { method: ResourceUpdatedNotificationMethod, params: ResourceUpdatedNotificationParam { uri: \"memo://insights\" } }), Sender { inner: Some(Inner { state: State { is_complete: false, is_closed: false, is_rx_task_set: true, is_tx_task_set: false } }) }))"},"threadId":"ThreadId(2)"}
{"timestamp":"2025-03-30T23:52:40.306406Z","level":"ERROR","fields":{"error":"Transport error: disconnected"},"threadId":"ThreadId(2)"}

Additional context

• Issue appears in the notification transport layer implementation
• Current workaround requires ignoring specific transport errors
• Need proper connection state tracking in the notification mechanism
• Affects both SSE and stdio transport implementations

[4t145]

I will have a look on this!

[byeblack]

The following example will reproduce the problem

use anyhow::Result;
use rmcp::ServiceExt;

#[tokio::main]
async fn main() -> Result<()> {
    let transport =
        rmcp::transport::TokioChildProcess::new(tokio::process::Command::new("python").args([
            "server.py",
            "--transport",
            "stdio",
        ]))?;
    let client = ().into_dyn().serve(transport).await?;

    let tools = client.list_all_tools().await?;
    println!("{:?}", tools);
    client.notify_initialized().await?; // Transport error: disconnected
    client.cancel().await?;

    Ok(())
}

// or

#[tokio::main]
async fn main() -> Result<()> {
    let transport = rmcp::transport::SseTransport::start("http://127.0.0.1:8000/sse").await?;
    let client = ().into_dyn().serve(transport).await?;

    let tools = client.list_all_tools().await?;
    println!("{:?}", tools);
    client.notify_initialized().await?; // Transport error: disconnected
    client.cancel().await?;

    Ok(())
}

server.py

from mcp.server.fastmcp import FastMCP
import click


@click.command()
@click.option(
    "--transport",
    type=click.Choice(["stdio", "sse"]),
    default="sse",
)
def main(transport: str):
    mcp = FastMCP("Demo")

    @mcp.tool()
    def add(a: int, b: int) -> int:
        """Add two numbers"""
        return a + b

    @mcp.resource("resource://{city}/weather")
    def get_weather(city: str) -> str:
        return f"Weather for {city}"

    if transport == "sse":
        mcp.settings.host = "127.0.0.1"
        mcp.settings.port = 8000
        mcp.run("sse")
    else:
        mcp.run("stdio")


if __name__ == "__main__":
    main()

[4t145]

This is because the responder did not response correctly when message is successfully send out. I will fix this in #70