fix: preserve negotiated progress responses

Author: DaleSeo

crates/rmcp/src/transport/streamable_http_server/tower.rs

@@ -891,7 +891,13 @@ where
             ))
         })?;
 
-        if self.config.json_response || jsonrpc_http_status(&first) != http::StatusCode::OK {
+        let terminal = matches!(
+            &first,
+            ServerJsonRpcMessage::Response(_) | ServerJsonRpcMessage::Error(_)
+        );
+        if terminal
+            && (self.config.json_response || jsonrpc_http_status(&first) != http::StatusCode::OK)
+        {
             // This message is the whole reply, so `receiver` is dropped here and
             // anything the handler emits afterwards is undeliverable. Cancel it so
             // a still-running handler stops instead of running on unobserved: its

crates/rmcp/tests/test_streamable_http_json_response.rs

@@ -17,6 +17,41 @@ use common::calculator::Calculator;
 
 const INIT_BODY: &str = r#"{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2025-03-26","capabilities":{},"clientInfo":{"name":"test","version":"1.0"}}}"#;
 const CALL_WITH_PROGRESS_BODY: &str = r#"{"jsonrpc":"2.0","id":2,"method":"tools/call","params":{"name":"progress","arguments":{},"_meta":{"progressToken":"progress-test-1"}}}"#;
+const NEGOTIATED_CALL_BODY: &str = r#"{
+    "jsonrpc": "2.0",
+    "id": 2,
+    "method": "tools/call",
+    "params": {
+        "name": "terminal",
+        "arguments": {},
+        "_meta": {
+            "io.modelcontextprotocol/protocolVersion": "2026-07-28",
+            "io.modelcontextprotocol/clientInfo": {
+                "name": "test",
+                "version": "1.0"
+            },
+            "io.modelcontextprotocol/clientCapabilities": {}
+        }
+    }
+}"#;
+const NEGOTIATED_CALL_WITH_PROGRESS_BODY: &str = r#"{
+    "jsonrpc": "2.0",
+    "id": 2,
+    "method": "tools/call",
+    "params": {
+        "name": "progress",
+        "arguments": {},
+        "_meta": {
+            "progressToken": "progress-test-1",
+            "io.modelcontextprotocol/protocolVersion": "2026-07-28",
+            "io.modelcontextprotocol/clientInfo": {
+                "name": "test",
+                "version": "1.0"
+            },
+            "io.modelcontextprotocol/clientCapabilities": {}
+        }
+    }
+}"#;
 
 #[derive(Clone)]
 struct ProgressServer;
@@ -28,22 +63,24 @@ impl ServerHandler for ProgressServer {
 
     async fn call_tool(
         &self,
-        _request: CallToolRequestParams,
+        request: CallToolRequestParams,
         context: RequestContext<rmcp::RoleServer>,
     ) -> Result<CallToolResponse, ErrorData> {
-        let progress_token = context
-            .meta
-            .get_progress_token()
-            .expect("request includes progressToken");
-        context
-            .peer
-            .notify_progress(
-                ProgressNotificationParam::new(progress_token, 50.0)
-                    .with_total(100.0)
-                    .with_message("working"),
-            )
-            .await
-            .expect("progress notification is delivered");
+        if request.name == "progress" {
+            let progress_token = context
+                .meta
+                .get_progress_token()
+                .expect("request includes progressToken");
+            context
+                .peer
+                .notify_progress(
+                    ProgressNotificationParam::new(progress_token, 50.0)
+                        .with_total(100.0)
+                        .with_message("working"),
+                )
+                .await
+                .expect("progress notification is delivered");
+        }
         Ok(CallToolResult::success(vec![ContentBlock::text("done")]).into())
     }
 }
@@ -140,6 +177,49 @@ async fn stateless_json_response_returns_application_json() -> anyhow::Result<()
     Ok(())
 }
 
+#[tokio::test]
+async fn stateless_negotiated_terminal_response_returns_application_json() -> anyhow::Result<()> {
+    let ct = CancellationToken::new();
+    let (client, url, ct) = spawn_progress_server(
+        StreamableHttpServerConfig::default()
+            .with_stateful_mode(false)
+            .with_json_response(true)
+            .with_sse_keep_alive(None)
+            .with_cancellation_token(ct.child_token()),
+    )
+    .await;
+
+    let response = client
+        .post(&url)
+        .header("Content-Type", "application/json")
+        .header("Accept", "application/json, text/event-stream")
+        .header("MCP-Protocol-Version", "2026-07-28")
+        .header("Mcp-Method", "tools/call")
+        .header("Mcp-Name", "terminal")
+        .body(NEGOTIATED_CALL_BODY)
+        .send()
+        .await?;
+
+    assert_eq!(response.status(), 200);
+
+    let content_type = response
+        .headers()
+        .get("content-type")
+        .and_then(|value| value.to_str().ok())
+        .unwrap_or("");
+    assert!(
+        content_type.contains("application/json"),
+        "Expected application/json, got: {content_type}"
+    );
+
+    let body: serde_json::Value = response.json().await?;
+    assert_eq!(body["id"], 2);
+    assert!(body["result"].is_object(), "Expected result object");
+
+    ct.cancel();
+    Ok(())
+}
+
 #[tokio::test]
 async fn stateless_json_response_falls_back_to_sse_for_progress() -> anyhow::Result<()> {
     let ct = CancellationToken::new();
@@ -192,6 +272,61 @@ async fn stateless_json_response_falls_back_to_sse_for_progress() -> anyhow::Res
     Ok(())
 }
 
+#[tokio::test]
+async fn stateless_negotiated_json_response_falls_back_to_sse_for_progress() -> anyhow::Result<()> {
+    let ct = CancellationToken::new();
+    let (client, url, ct) = spawn_progress_server(
+        StreamableHttpServerConfig::default()
+            .with_stateful_mode(false)
+            .with_json_response(true)
+            .with_sse_keep_alive(None)
+            .with_cancellation_token(ct.child_token()),
+    )
+    .await;
+
+    let response = client
+        .post(&url)
+        .header("Content-Type", "application/json")
+        .header("Accept", "application/json, text/event-stream")
+        .header("MCP-Protocol-Version", "2026-07-28")
+        .header("Mcp-Method", "tools/call")
+        .header("Mcp-Name", "progress")
+        .body(NEGOTIATED_CALL_WITH_PROGRESS_BODY)
+        .send()
+        .await?;
+
+    assert_eq!(response.status(), 200);
+
+    let content_type = response
+        .headers()
+        .get("content-type")
+        .and_then(|value| value.to_str().ok())
+        .unwrap_or("");
+    assert!(
+        content_type.contains("text/event-stream"),
+        "Expected SSE fallback, got: {content_type}"
+    );
+
+    let body = response.text().await?;
+    let messages: Vec<serde_json::Value> = body
+        .lines()
+        .filter_map(|line| line.strip_prefix("data:"))
+        .map(str::trim)
+        .filter(|data| !data.is_empty())
+        .map(serde_json::from_str)
+        .collect::<Result<_, _>>()?;
+    assert_eq!(messages.len(), 2, "Expected progress and result: {body}");
+    assert_eq!(messages[0]["method"], "notifications/progress");
+    assert_eq!(messages[1]["id"], 2);
+    assert!(
+        messages[1]["result"].is_object(),
+        "Expected result object: {body}"
+    );
+
+    ct.cancel();
+    Ok(())
+}
+
 #[tokio::test]
 async fn stateless_sse_mode_default_unchanged() -> anyhow::Result<()> {
     let ct = CancellationToken::new();