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;