fix(rmcp-client): accept empty notification responses

codex-rs/rmcp-client/src/http_client_adapter.rs

@@ -215,6 +215,9 @@ impl StreamableHttpClient for StreamableHttpClientAdapter {
             }
             Some(content_type) if content_type.starts_with(JSON_MIME_TYPE) => {
                 let body = collect_body(&mut body_stream).await?;
+                if body.is_empty() && matches!(message, JsonRpcMessage::Notification(_)) {
+                    return Ok(StreamableHttpPostResponse::Accepted);
+                }
                 let message: ServerJsonRpcMessage =
                     serde_json::from_slice(&body).map_err(StreamableHttpError::Deserialize)?;
                 Ok(StreamableHttpPostResponse::Json(message, session_id))

codex-rs/rmcp-client/tests/streamable_http_recovery.rs

@@ -18,7 +18,9 @@ use serde_json::Value;
 
 use streamable_http_test_support::arm_initialize_post_failure;
 use streamable_http_test_support::arm_initialize_post_json_rpc_failure;
+use streamable_http_test_support::arm_initialize_post_response;
 use streamable_http_test_support::arm_initialized_notification_post_json_rpc_failure;
+use streamable_http_test_support::arm_initialized_notification_post_response;
 use streamable_http_test_support::arm_session_post_failure;
 use streamable_http_test_support::arm_session_post_json_rpc_failure;
 use streamable_http_test_support::call_echo_tool;
@@ -148,6 +150,39 @@ async fn streamable_http_initialize_retries_json_rpc_transient_status() -> anyho
     Ok(())
 }
 
+#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+async fn streamable_http_rejects_empty_ok_for_initialize_request() -> anyhow::Result<()> {
+    let (_server, base_url) = spawn_streamable_http_server().await?;
+
+    arm_initialize_post_response(&base_url, /*status*/ 200, /*remaining*/ 10, "").await?;
+
+    let Err(_) = create_client(&base_url).await else {
+        anyhow::bail!("empty response unexpectedly accepted for initialize request");
+    };
+
+    Ok(())
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+async fn streamable_http_accepts_empty_ok_for_initialized_notification() -> anyhow::Result<()> {
+    let (_server, base_url) = spawn_streamable_http_server().await?;
+
+    arm_initialized_notification_post_response(
+        &base_url, /*status*/ 200, /*remaining*/ 10, "",
+    )
+    .await?;
+
+    let client = create_client(&base_url).await?;
+    let result = call_echo_tool(&client, "after-empty-notification-response").await?;
+
+    assert_eq!(
+        result,
+        expected_echo_result("after-empty-notification-response")
+    );
+
+    Ok(())
+}
+
 #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
 async fn streamable_http_retries_initialized_notification_status() -> anyhow::Result<()> {
     let (_server, base_url) = spawn_streamable_http_server().await?;

codex-rs/rmcp-client/tests/streamable_http_test_support.rs

@@ -229,6 +229,24 @@ pub(crate) async fn arm_initialized_notification_post_json_rpc_failure(
     base_url: &str,
     status: u16,
     remaining: usize,
+) -> anyhow::Result<()> {
+    let body = json!({
+        "jsonrpc": "2.0",
+        "id": 1,
+        "error": {
+            "code": -32000,
+            "message": "transient session failure",
+        },
+    })
+    .to_string();
+    arm_initialized_notification_post_response(base_url, status, remaining, &body).await
+}
+
+pub(crate) async fn arm_initialized_notification_post_response(
+    base_url: &str,
+    status: u16,
+    remaining: usize,
+    body: &str,
 ) -> anyhow::Result<()> {
     let response = reqwest::Client::new()
         .post(format!(
@@ -238,14 +256,7 @@ pub(crate) async fn arm_initialized_notification_post_json_rpc_failure(
             "status": status,
             "remaining": remaining,
             "content_type": "application/json",
-            "body": json!({
-                "jsonrpc": "2.0",
-                "id": 1,
-                "error": {
-                    "code": -32000,
-                    "message": "transient session failure",
-                },
-            }).to_string(),
+            "body": body,
         }))
         .send()
         .await?;
@@ -276,6 +287,24 @@ pub(crate) async fn arm_initialize_post_json_rpc_failure(
     base_url: &str,
     status: u16,
     remaining: usize,
+) -> anyhow::Result<()> {
+    let body = json!({
+        "jsonrpc": "2.0",
+        "id": 1,
+        "error": {
+            "code": -32000,
+            "message": "transient initialize failure",
+        },
+    })
+    .to_string();
+    arm_initialize_post_response(base_url, status, remaining, &body).await
+}
+
+pub(crate) async fn arm_initialize_post_response(
+    base_url: &str,
+    status: u16,
+    remaining: usize,
+    body: &str,
 ) -> anyhow::Result<()> {
     let response = reqwest::Client::new()
         .post(format!("{base_url}{INITIALIZE_POST_FAILURE_CONTROL_PATH}"))
@@ -283,14 +312,7 @@ pub(crate) async fn arm_initialize_post_json_rpc_failure(
             "status": status,
             "remaining": remaining,
             "content_type": "application/json",
-            "body": json!({
-                "jsonrpc": "2.0",
-                "id": 1,
-                "error": {
-                    "code": -32000,
-                    "message": "transient initialize failure",
-                },
-            }).to_string(),
+            "body": body,
         }))
         .send()
         .await?;