From 1d5777824c617450ac2ce685d15b52c99ef69db3 Mon Sep 17 00:00:00 2001 From: jr42 Date: Thu, 26 Mar 2026 22:47:31 +0100 Subject: [PATCH] fix(mcp): handle 202 Accepted and wire session manager for Streamable HTTP (#1437) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(mcp): handle 202 Accepted for Streamable HTTP notifications The MCP Streamable HTTP spec requires servers to respond with 202 Accepted (empty body) for JSON-RPC notifications like `notifications/initialized`. The HTTP transport tried to parse this empty body as JSON, which failed and broke the session handshake — subsequent requests like `tools/list` were rejected because the server considered the session uninitialized. Add an early return for 202 responses that produces an empty McpResponse without attempting body parsing. Fixes #1436 Co-Authored-By: Claude Opus 4.6 (1M context) * fix(mcp): wire session manager into transport for non-OAuth HTTP clients The factory used McpClient::new_with_config().with_session_manager() which only set the session manager on the client, not on the HttpMcpTransport. The transport never captured Mcp-Session-Id from responses, so subsequent requests lacked the header and the server rejected them as uninitialized. Fix by constructing the HttpMcpTransport with the session manager before wrapping it in Arc, matching the pattern already used by new_authenticated(). Co-Authored-By: Claude Opus 4.6 (1M context) * refactor(mcp): deduplicate factory HTTP path, gate dead-code methods as test-only - Collapse the two identical non-OAuth HTTP branches in `create_client_from_config()` into one (early-return for the authenticated path, fall through for the common case). - Gate `McpClient::new_with_config()` and `McpClient::with_session_manager()` as `#[cfg(test)]` — the factory was their only production caller and no longer uses them. Both methods silently skip wiring the session manager into the transport, which was the root cause of #1436. - Add doc warnings on both methods explaining the footgun. Co-Authored-By: Claude Opus 4.6 (1M context) --------- Co-authored-by: Claude Opus 4.6 (1M context) Co-authored-by: ilblackdragon@gmail.com --- src/tools/mcp/client.rs | 20 +++++- src/tools/mcp/factory.rs | 117 +++++++++++++++++++++++++++----- src/tools/mcp/http_transport.rs | 28 ++++++++ 3 files changed, 148 insertions(+), 17 deletions(-) diff --git a/src/tools/mcp/client.rs b/src/tools/mcp/client.rs index 148f5a86..32c5767d 100644 --- a/src/tools/mcp/client.rs +++ b/src/tools/mcp/client.rs @@ -117,6 +117,11 @@ impl McpClient { /// The config must use HTTP transport (the default); for stdio/UDS use `new_with_transport`. /// /// Returns an error if the config uses a non-HTTP transport. + /// + /// **Note:** The session manager is NOT wired into the transport. For + /// production use, prefer `create_client_from_config()` which constructs + /// the transport with session tracking. + #[cfg(test)] pub fn new_with_config(config: McpServerConfig) -> Result { if !matches!( config.effective_transport(), @@ -214,7 +219,14 @@ impl McpClient { } } - /// Attach a session manager for Streamable HTTP session tracking. + /// Attach a session manager to the **client** only. + /// + /// **Warning:** This does NOT wire the session manager into the underlying + /// `HttpMcpTransport`, so the transport will not capture `Mcp-Session-Id` + /// from responses. For production use, construct the transport with + /// `HttpMcpTransport::with_session_manager()` and pass it to + /// `new_with_transport()` instead. See `create_client_from_config()`. + #[cfg(test)] pub fn with_session_manager(mut self, session_manager: Arc) -> Self { self.session_manager = Some(session_manager); self @@ -235,6 +247,12 @@ impl McpClient { self.session_manager.is_some() } + /// Get the underlying transport (test-only). + #[cfg(test)] + pub(crate) fn transport(&self) -> &Arc { + &self.transport + } + /// Get the next request ID. fn next_request_id(&self) -> u64 { self.next_id.fetch_add(1, Ordering::SeqCst) diff --git a/src/tools/mcp/factory.rs b/src/tools/mcp/factory.rs index c31c5051..bbb09256 100644 --- a/src/tools/mcp/factory.rs +++ b/src/tools/mcp/factory.rs @@ -7,6 +7,7 @@ use std::sync::Arc; use crate::secrets::SecretsStore; use crate::tools::mcp::config::{EffectiveTransport, McpServerConfig}; +use crate::tools::mcp::http_transport::HttpMcpTransport; use crate::tools::mcp::{McpClient, McpProcessManager, McpSessionManager, McpTransport}; /// Error returned when MCP client creation fails. @@ -78,33 +79,37 @@ pub async fn create_client_from_config( Err(McpFactoryError::UnixNotSupported { name: server_name }) } EffectiveTransport::Http => { + // Authenticated (OAuth) path: tokens exist or server requires auth. if let Some(ref secrets) = secrets { let has_tokens = crate::tools::mcp::is_authenticated(&server, secrets, user_id).await; if has_tokens || server.requires_auth() { - Ok(McpClient::new_authenticated( + return Ok(McpClient::new_authenticated( server, Arc::clone(session_manager), Arc::clone(secrets), user_id, - )) - } else { - Ok(McpClient::new_with_config(server) - .map_err(|e| McpFactoryError::InvalidConfig { - name: server_name.clone(), - reason: e.to_string(), - })? - .with_session_manager(Arc::clone(session_manager))) + )); } - } else { - Ok(McpClient::new_with_config(server) - .map_err(|e| McpFactoryError::InvalidConfig { - name: server_name, - reason: e.to_string(), - })? - .with_session_manager(Arc::clone(session_manager))) } + + // Non-OAuth HTTP: wire the session manager into the *transport* so + // it captures `Mcp-Session-Id` from responses. Passing it only to + // the client (via `with_session_manager`) is not enough — the + // transport must know about it to read/write the header. + let transport = Arc::new( + HttpMcpTransport::new(server.url.clone(), server.name.clone()) + .with_session_manager(Arc::clone(session_manager)), + ); + Ok(McpClient::new_with_transport( + server.name.clone(), + transport, + Some(Arc::clone(session_manager)), + secrets, + user_id, + Some(server), + )) } } } @@ -134,4 +139,84 @@ mod tests { "non-OAuth HTTP clients must carry a session manager" ); } + + /// Regression test: the factory must wire the session manager into the + /// *transport*, not just the client. Otherwise the transport never + /// captures `Mcp-Session-Id` from responses and subsequent requests + /// lack the header, causing the server to reject them. + #[tokio::test] + async fn test_factory_non_oauth_http_transport_captures_session_id() { + use axum::http::header::HeaderName; + use axum::{Router, http::StatusCode, response::IntoResponse, routing::post}; + use tokio::net::TcpListener; + + const SESSION_ID: &str = "test-session-abc123"; + + async fn session_echo() -> impl IntoResponse { + let body = serde_json::json!({ + "jsonrpc": "2.0", + "id": 1, + "result": {} + }) + .to_string(); + ( + StatusCode::OK, + [( + HeaderName::from_static("mcp-session-id"), + SESSION_ID.to_string(), + )], + body, + ) + } + + let app = Router::new().route("/", post(session_echo)); + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let url = format!("http://127.0.0.1:{}", addr.port()); + + tokio::spawn(async move { + axum::serve(listener, app).await.unwrap(); + }); + + let server = McpServerConfig::new("session-test", &url); + let session_manager = Arc::new(McpSessionManager::new()); + let process_manager = Arc::new(McpProcessManager::new()); + + let client = create_client_from_config( + server, + &session_manager, + &process_manager, + None, + "test-user", + ) + .await + .expect("factory should succeed for HTTP config"); + + // Pre-create a session entry so that update_session_id has something to update. + // In production, the MCP initialize handshake calls get_or_create before responses arrive. + session_manager.get_or_create("session-test", &url).await; + + // Send a request through the client's transport to trigger session capture. + use crate::tools::mcp::protocol::McpRequest; + let request = McpRequest { + jsonrpc: "2.0".to_string(), + id: Some(1), + method: "test".to_string(), + params: Some(serde_json::json!({})), + }; + let headers = std::collections::HashMap::new(); + client + .transport() + .send(&request, &headers) + .await + .expect("request should succeed"); + + // Verify the session manager captured the session ID from the response. + let captured = session_manager.get_session_id("session-test").await; + assert_eq!( + captured.as_deref(), + Some(SESSION_ID), + "transport must capture Mcp-Session-Id into session manager" + ); + } } diff --git a/src/tools/mcp/http_transport.rs b/src/tools/mcp/http_transport.rs index 59873ce4..ea3e1c03 100644 --- a/src/tools/mcp/http_transport.rs +++ b/src/tools/mcp/http_transport.rs @@ -494,6 +494,34 @@ mod tests { assert_eq!(echoed["authorization"], "Bearer oauth-token"); } + /// Regression test for #1436: 202 Accepted responses for notifications + /// were parsed as JSON, causing "Failed to parse MCP response" errors + /// that broke the MCP session handshake. + #[tokio::test] + async fn test_wire_202_accepted_for_notification() { + use axum::{Router, http::StatusCode, routing::post}; + use tokio::net::TcpListener; + + async fn accept_notification() -> StatusCode { + StatusCode::ACCEPTED + } + + let app = Router::new().route("/", post(accept_notification)); + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let url = format!("http://127.0.0.1:{}", addr.port()); + + tokio::spawn(async move { + axum::serve(listener, app).await.unwrap(); + }); + + let transport = HttpMcpTransport::new(&url, "test-202"); + let request = McpRequest::initialized_notification(); + let response = transport.send(&request, &HashMap::new()).await.unwrap(); + assert!(response.result.is_none()); + assert!(response.error.is_none()); + } + #[tokio::test] async fn test_wire_custom_auth_preserved_when_no_per_request_auth() { let (url, _handle) = spawn_echo_server().await;