fix(transport): make message channel capacity configurable for all transports (#171)

* merge

* Add channel_capacity option to transports

* Allow clippy too_many_arguments and simplify doc

Add #[allow(clippy::too_many_arguments)] to MCPStream::create and
MCPStream::create_with_ack to silence the lint for many parameters.

Update TransportOptions doc comment to state "Defaults to 36"
instead of referencing the constant name.

Author: hashemix

crates/rust-mcp-transport/src/client_sse.rs

@@ -39,6 +39,7 @@ const SHUTDOWN_TIMEOUT_SECONDS: u64 = 5;
 pub struct ClientSseTransportOptions {
     pub request_timeout: Duration,
     pub max_line_length: usize,
+    pub channel_capacity: usize,
     pub retry_delay: Option<Duration>,
     pub max_retries: Option<usize>,
     pub custom_headers: Option<HashMap<String, String>>,
@@ -50,6 +51,7 @@ impl Default for ClientSseTransportOptions {
         Self {
             request_timeout: TransportOptions::default().timeout,
             max_line_length: TransportOptions::default().max_line_length,
+            channel_capacity: TransportOptions::default().channel_capacity,
             retry_delay: None,
             max_retries: None,
             custom_headers: None,
@@ -72,6 +74,8 @@ where
     request_timeout: Duration,
     /// Maximum line length for incoming messages
     max_line_length: usize,
+    /// Capacity of the incoming-message channel buffer
+    channel_capacity: usize,
     /// HTTP client for making requests
     client: Client,
     /// URL for the SSE endpoint
@@ -134,6 +138,7 @@ where
             is_shut_down: Mutex::new(false),
             request_timeout: options.request_timeout,
             max_line_length: options.max_line_length,
+            channel_capacity: options.channel_capacity,
             custom_headers: headers,
             sse_task: tokio::sync::RwLock::new(None),
             post_task: tokio::sync::RwLock::new(None),
@@ -330,6 +335,7 @@ where
             self.request_timeout,
             self.max_line_length,
             cancellation_token,
+            self.channel_capacity,
         );
 
         self.set_message_sender(sender).await;

crates/rust-mcp-transport/src/client_streamable_http.rs

@@ -54,6 +54,7 @@ impl StreamableTransportOptions {
 pub struct RequestOptions {
     pub request_timeout: Duration,
     pub max_line_length: usize,
+    pub channel_capacity: usize,
     pub retry_delay: Option<Duration>,
     pub max_retries: Option<usize>,
     pub custom_headers: Option<HashMap<String, String>>,
@@ -64,6 +65,7 @@ impl Default for RequestOptions {
         Self {
             request_timeout: TransportOptions::default().timeout,
             max_line_length: TransportOptions::default().max_line_length,
+            channel_capacity: TransportOptions::default().channel_capacity,
             retry_delay: None,
             max_retries: None,
             custom_headers: None,
@@ -83,6 +85,8 @@ where
     request_timeout: Duration,
     /// Maximum line length for incoming messages
     max_line_length: usize,
+    /// Capacity of the incoming-message channel buffer
+    channel_capacity: usize,
     /// HTTP client for making requests
     client: Client,
     /// URL for the SSE endpoint
@@ -124,6 +128,7 @@ where
             is_shut_down: Mutex::new(false),
             request_timeout: options.request_options.request_timeout,
             max_line_length: options.request_options.max_line_length,
+            channel_capacity: options.request_options.channel_capacity,
             client,
             mcp_server_url,
             retry_delay: options
@@ -294,6 +299,7 @@ where
                 self.request_timeout,
                 self.max_line_length,
                 cancellation_token,
+                self.channel_capacity,
             );
 
             self.set_message_sender(sender).await;
@@ -380,6 +386,7 @@ where
                 self.request_timeout,
                 self.max_line_length,
                 cancellation_token,
+                self.channel_capacity,
             );
 
             self.set_message_sender(sender).await;

crates/rust-mcp-transport/src/mcp_stream.rs

@@ -12,7 +12,9 @@ use tokio::{
     sync::Mutex,
 };
 
-const CHANNEL_CAPACITY: usize = 36;
+/// Default capacity of the incoming-message channel. Used when callers do not
+/// override it (see [`crate::TransportOptions::channel_capacity`]).
+pub(crate) const DEFAULT_MESSAGE_CHANNEL_CAPACITY: usize = 36;
 
 pub struct MCPStream {}
 
@@ -25,6 +27,7 @@ impl MCPStream {
     /// - A `Pin<Box<dyn Stream<Item = R> + Send>>`: A stream that yields items of type `R`.
     /// - A `MessageDispatcher<R>`: A sender that can be used to send messages of type `R`.
     /// - An `IoStream`: An error handling stream for managing error I/O (stderr).
+    #[allow(clippy::too_many_arguments)]
     pub fn create<X, R>(
         readable: Pin<Box<dyn tokio::io::AsyncRead + Send + Sync>>,
         writable: Mutex<Pin<Box<dyn tokio::io::AsyncWrite + Send + Sync>>>,
@@ -33,6 +36,7 @@ impl MCPStream {
         request_timeout: Duration,
         max_line_length: usize,
         cancellation_token: CancellationToken,
+        channel_capacity: usize,
     ) -> (
         tokio_stream::wrappers::ReceiverStream<X>,
         MessageDispatcher<R>,
@@ -42,7 +46,7 @@ impl MCPStream {
         R: Clone + Send + Sync + serde::de::DeserializeOwned + 'static,
         X: Clone + Send + Sync + serde::de::DeserializeOwned + 'static,
     {
-        let (tx, rx) = tokio::sync::mpsc::channel::<X>(CHANNEL_CAPACITY);
+        let (tx, rx) = tokio::sync::mpsc::channel::<X>(channel_capacity);
         let stream = tokio_stream::wrappers::ReceiverStream::new(rx);
 
         let reader_token = cancellation_token.clone();
@@ -55,6 +59,7 @@ impl MCPStream {
         (stream, sender, error_io)
     }
 
+    #[allow(clippy::too_many_arguments)]
     pub fn create_with_ack<X, R>(
         readable: Pin<Box<dyn tokio::io::AsyncRead + Send + Sync>>,
         writable: tokio::sync::mpsc::Sender<(
@@ -66,6 +71,7 @@ impl MCPStream {
         request_timeout: Duration,
         max_line_length: usize,
         cancellation_token: CancellationToken,
+        channel_capacity: usize,
     ) -> (
         tokio_stream::wrappers::ReceiverStream<X>,
         MessageDispatcher<R>,
@@ -75,7 +81,7 @@ impl MCPStream {
         R: Clone + Send + Sync + serde::de::DeserializeOwned + 'static,
         X: Clone + Send + Sync + serde::de::DeserializeOwned + 'static,
     {
-        let (tx, rx) = tokio::sync::mpsc::channel::<X>(CHANNEL_CAPACITY);
+        let (tx, rx) = tokio::sync::mpsc::channel::<X>(channel_capacity);
         let stream = tokio_stream::wrappers::ReceiverStream::new(rx);
 
         let reader_token = cancellation_token.clone();

crates/rust-mcp-transport/src/sse.rs

@@ -190,6 +190,7 @@ impl Transport<ClientMessages, MessageFromServer, ClientMessage, ServerMessages,
             self.options.timeout,
             self.options.max_line_length,
             cancellation_token,
+            self.options.channel_capacity,
         );
 
         if let (Some(session_id), Some(stream_id), Some(event_store)) = (

crates/rust-mcp-transport/src/stdio.rs

@@ -224,6 +224,7 @@ where
                 self.options.timeout,
                 self.options.max_line_length,
                 cancellation_token,
+                self.options.channel_capacity,
             );
 
             self.set_message_sender(sender).await;
@@ -239,6 +240,7 @@ where
                 self.options.timeout,
                 self.options.max_line_length,
                 cancellation_token,
+                self.options.channel_capacity,
             );
 
             self.set_message_sender(sender).await;

crates/rust-mcp-transport/src/transport.rs

@@ -44,12 +44,19 @@ pub struct TransportOptions {
     /// large tool results or responses, increase this value.
     /// Default: 16 MiB.
     pub max_line_length: usize,
+
+    /// Capacity of the incoming-message channel buffer.
+    ///
+    /// A larger value smooths out head-of-line jitter under bursty traffic at
+    /// the cost of more buffered memory. Defaults to 36.
+    pub channel_capacity: usize,
 }
 impl Default for TransportOptions {
     fn default() -> Self {
         Self {
             timeout: Duration::from_millis(DEFAULT_TIMEOUT_MSEC),
             max_line_length: DEFAULT_MAX_LINE_LENGTH,
+            channel_capacity: crate::mcp_stream::DEFAULT_MESSAGE_CHANNEL_CAPACITY,
         }
     }
 }
@@ -197,3 +204,25 @@ where
 //         Ok(self)
 //     }
 // }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn default_channel_capacity_matches_constant() {
+        assert_eq!(
+            TransportOptions::default().channel_capacity,
+            crate::mcp_stream::DEFAULT_MESSAGE_CHANNEL_CAPACITY
+        );
+    }
+
+    #[test]
+    fn channel_capacity_is_overridable() {
+        let options = TransportOptions {
+            channel_capacity: 256,
+            ..Default::default()
+        };
+        assert_eq!(options.channel_capacity, 256);
+    }
+}