Share MCP subscription sessions with daemon-managed transports

src/adapters/mcp/client.rs

@@ -137,24 +137,6 @@ impl McpStdioClient {
         self.transport.kill_and_wait(timeout).await
     }
 
-    pub async fn take_tool_list_changed(&mut self) -> bool {
-        let notifications = self.transport.drain_notifications().await;
-        let mut tool_list_changed = false;
-
-        for notification in notifications {
-            if notification.method == "notifications/tools/list_changed" {
-                tool_list_changed = true;
-            } else {
-                tracing::debug!(
-                    method = %notification.method,
-                    "Ignoring unsupported MCP stdio notification"
-                );
-            }
-        }
-
-        tool_list_changed
-    }
-
     pub async fn drain_notifications(&mut self) -> Vec<JsonRpcNotification> {
         self.transport.drain_notifications().await
     }

src/adapters/mcp/http_transport.rs

@@ -68,6 +68,9 @@ pub enum ProbeAuthFailureCode {
 pub struct McpHttpTransport {
     /// HTTP client
     client: Client,
+    /// Dedicated client for the long-lived SSE event stream so it does not
+    /// contend with regular JSON-RPC POST requests on the same session.
+    event_stream_client: Client,
     /// Server URL
     server_url: String,
     /// Request ID counter
@@ -159,9 +162,14 @@ impl McpHttpTransport {
 
         let client =
             build_resilient_http_client(std::time::Duration::from_secs(30), "MCP HTTP transport")?;
+        let event_stream_client = build_resilient_http_client(
+            std::time::Duration::from_secs(LEGACY_SSE_STREAM_TIMEOUT_SECS),
+            "MCP HTTP event stream",
+        )?;
 
         Ok(Self {
             client,
+            event_stream_client,
             server_url: url,
             next_id: Arc::new(Mutex::new(1i64)),
             session_id: Arc::new(Mutex::new(None)),
@@ -1012,8 +1020,6 @@ impl McpHttpTransport {
     }
 
     pub async fn subscribe_resource(&self, uri: &str) -> Result<()> {
-        self.ensure_event_stream().await?;
-
         let params = serde_json::json!({
             "uri": uri
         });
@@ -1151,7 +1157,7 @@ impl McpHttpTransport {
         let resolved = Self::resolve_request_auth("GET", &self.server_url, profile.as_ref())?;
 
         let mut req = self
-            .client
+            .event_stream_client
             .get(&resolved.url)
             .header("Accept", "text/event-stream");
         if let Some(session_id) = session_id {
@@ -1789,6 +1795,13 @@ impl McpRemoteTransport {
         }
     }
 
+    pub async fn ensure_notification_stream(&self) -> Result<()> {
+        match self {
+            Self::Streamable(transport) => transport.ensure_event_stream().await,
+            Self::Legacy(_) => Ok(()),
+        }
+    }
+
     pub async fn unsubscribe_resource(&self, uri: &str) -> Result<()> {
         match self {
             Self::Streamable(transport) => transport.unsubscribe_resource(uri).await,