feat(mcp): cluster-aware connections + client_name (#906)

* feat(mcp): cluster-aware connections + client_name support

Adds Redis Cluster support and connection metadata to the MCP server:

Cluster mode (--cluster / REDIS_CLUSTER env):
- Uses redis::cluster_async::ClusterConnection for automatic MOVED/ASK
  redirection handling
- Single URL serves as seed node; client discovers cluster topology
- RedisConnection enum wraps both Standalone and Cluster connections,
  implementing ConnectionLike so all existing tools work transparently

Client name (--client-name / REDIS_CLIENT_NAME env):
- Sends CLIENT SETNAME on every new connection (default: "redisctl-mcp")
- MCP connections are identifiable in CLIENT LIST and slowlog

Both options work with profiles and dynamic --database-url connections.

Closes #904, closes #905 (client_name portion).

* style: cargo fmt

Author: joshrotenberg

Cargo.lock

@@ -96,7 +96,7 @@ serial_test = "3.1"
 pretty_assertions = "1.4"
 
 # Redis client
-redis = { version = "0.27", features = ["tokio-comp", "tokio-rustls-comp", "connection-manager"] }
+redis = { version = "0.27", features = ["tokio-comp", "tokio-rustls-comp", "connection-manager", "cluster-async"] }
 
 # External crates (git for dev, version required for crates.io publish)
 redis-cloud = { version = "0.9.5", git = "https://github.com/redis-developer/redis-cloud-rs", branch = "main" }

Cargo.toml

@@ -41,6 +41,8 @@
 //!     CredentialSource::Profiles(vec!["default".to_string()]),
 //!     policy,
 //!     None, // no database URL
+//!     false, // cluster mode
+//!     Some("redisctl-mcp".to_string()), // client name
 //! )?);
 //!
 //! // Use merge to compose sub-routers
@@ -118,8 +120,14 @@ mod tests {
             std::collections::HashMap::new(),
             "test".to_string(),
         ));
-        let state =
-            AppState::new(CredentialSource::Profiles(vec![]), read_only_policy, None).unwrap();
+        let state = AppState::new(
+            CredentialSource::Profiles(vec![]),
+            read_only_policy,
+            None,
+            false,
+            None,
+        )
+        .unwrap();
 
         assert!(!state.is_write_allowed());
     }
@@ -135,7 +143,14 @@ mod tests {
             std::collections::HashMap::new(),
             "test".to_string(),
         ));
-        let state = AppState::new(CredentialSource::Profiles(vec![]), write_policy, None).unwrap();
+        let state = AppState::new(
+            CredentialSource::Profiles(vec![]),
+            write_policy,
+            None,
+            false,
+            None,
+        )
+        .unwrap();
 
         assert!(state.is_write_allowed());
     }
@@ -146,6 +161,8 @@ mod tests {
             CredentialSource::Profiles(vec![]),
             AppState::test_policy(),
             Some("redis://localhost:6379".to_string()),
+            false,
+            None,
         )
         .unwrap();
 
@@ -164,6 +181,8 @@ mod tests {
             ]),
             AppState::test_policy(),
             None,
+            false,
+            None,
         )
         .unwrap();
 
@@ -181,6 +200,8 @@ mod tests {
                 CredentialSource::Profiles(vec![]),
                 AppState::test_policy(),
                 None,
+                false,
+                None,
             )
             .unwrap(),
         );
@@ -231,6 +252,8 @@ mod tests {
                 CredentialSource::Profiles(vec![]),
                 AppState::test_policy(),
                 None,
+                false,
+                None,
             )
             .unwrap(),
         );
@@ -293,6 +316,8 @@ mod tests {
                 CredentialSource::Profiles(vec![]),
                 AppState::test_policy(),
                 Some("redis://localhost:6379".to_string()),
+                false,
+                None,
             )
             .unwrap(),
         );
@@ -316,6 +341,8 @@ mod tests {
                 CredentialSource::Profiles(vec![]),
                 AppState::test_policy(),
                 None,
+                false,
+                None,
             )
             .unwrap(),
         );

crates/redisctl-mcp/src/lib.rs

@@ -180,6 +180,14 @@ struct Args {
     #[arg(long, env = "REDIS_URL")]
     database_url: Option<String>,
 
+    /// Enable Redis Cluster mode (handles MOVED/ASK redirections)
+    #[arg(long, env = "REDIS_CLUSTER")]
+    cluster: bool,
+
+    /// Client name for CLIENT SETNAME (identifies MCP connections in CLIENT LIST)
+    #[arg(long, env = "REDIS_CLIENT_NAME", default_value = "redisctl-mcp")]
+    client_name: Option<String>,
+
     /// Toolsets to enable (default: all compiled-in).
     /// Use bare names for all sub-modules: cloud,enterprise,database,app.
     /// Use colon syntax for specific sub-modules: cloud:subscriptions,cloud:networking.
@@ -551,6 +559,8 @@ async fn main() -> Result<()> {
         credential_source,
         policy.clone(),
         args.database_url.clone(),
+        args.cluster,
+        args.client_name.clone(),
     )?);
 
     // Resolve skills directory
@@ -1023,6 +1033,8 @@ mod tests {
                 state::CredentialSource::Profiles(vec![]),
                 AppState::test_policy(),
                 None,
+                false,
+                None,
             )
             .unwrap(),
         )

crates/redisctl-mcp/src/main.rs

@@ -36,7 +36,7 @@ pub struct CachedClients {
     #[cfg(feature = "enterprise")]
     pub enterprise: HashMap<String, EnterpriseClient>,
     #[cfg(feature = "database")]
-    pub database: HashMap<String, redis::aio::MultiplexedConnection>,
+    pub database: HashMap<String, crate::tools::redis::RedisConnection>,
 }
 
 /// Shared application state
@@ -47,6 +47,10 @@ pub struct AppState {
     pub policy: Arc<Policy>,
     /// Optional Redis database URL for direct connections
     pub database_url: Option<String>,
+    /// Enable Redis Cluster mode (handles MOVED/ASK redirections)
+    pub cluster: bool,
+    /// Client name for CLIENT SETNAME (identifies connections in CLIENT LIST)
+    pub client_name: Option<String>,
     /// redisctl config (for profile-based auth)
     config: Option<Config>,
     /// Configured profiles (for multi-cluster support)
@@ -65,6 +69,8 @@ impl AppState {
         credential_source: CredentialSource,
         policy: Arc<Policy>,
         database_url: Option<String>,
+        cluster: bool,
+        client_name: Option<String>,
     ) -> Result<Self> {
         // Extract profiles list
         let profiles = match &credential_source {
@@ -82,6 +88,8 @@ impl AppState {
             credential_source,
             policy,
             database_url,
+            cluster,
+            client_name,
             config,
             profiles,
             clients: RwLock::new(CachedClients {
@@ -352,11 +360,16 @@ impl AppState {
     ///
     /// Connections are cached by URL. If a cached connection fails a PING
     /// health check, it is evicted and a fresh connection is created.
+    ///
+    /// Returns a `RedisConnection` (standalone or cluster) based on the
+    /// `cluster` flag in AppState.
     #[cfg(feature = "database")]
     pub async fn redis_connection_for_url(
         &self,
         url: &str,
-    ) -> Result<redis::aio::MultiplexedConnection> {
+    ) -> Result<crate::tools::redis::RedisConnection> {
+        use crate::tools::redis::RedisConnection;
+
         // Check cache first
         {
             let clients = self.clients.read().await;
@@ -375,11 +388,42 @@ impl AppState {
         }
 
         // Create new connection (or reconnect after eviction)
-        let client = redis::Client::open(url).context("Failed to create Redis client")?;
-        let conn = client
-            .get_multiplexed_async_connection()
-            .await
-            .context("Failed to connect to Redis")?;
+        let conn = if self.cluster {
+            let client = redis::cluster::ClusterClient::new(vec![url])
+                .context("Failed to create Redis cluster client")?;
+            let mut cluster_conn = client
+                .get_async_connection()
+                .await
+                .context("Failed to connect to Redis cluster")?;
+
+            // Set client name if configured
+            if let Some(ref name) = self.client_name {
+                let _ = redis::cmd("CLIENT")
+                    .arg("SETNAME")
+                    .arg(name)
+                    .query_async::<String>(&mut cluster_conn)
+                    .await;
+            }
+
+            RedisConnection::Cluster(cluster_conn)
+        } else {
+            let client = redis::Client::open(url).context("Failed to create Redis client")?;
+            let mut standalone_conn = client
+                .get_multiplexed_async_connection()
+                .await
+                .context("Failed to connect to Redis")?;
+
+            // Set client name if configured
+            if let Some(ref name) = self.client_name {
+                let _ = redis::cmd("CLIENT")
+                    .arg("SETNAME")
+                    .arg(name)
+                    .query_async::<String>(&mut standalone_conn)
+                    .await;
+            }
+
+            RedisConnection::Standalone(standalone_conn)
+        };
 
         // Cache it
         {
@@ -449,6 +493,8 @@ impl Clone for AppState {
             credential_source: self.credential_source.clone(),
             policy: self.policy.clone(),
             database_url: self.database_url.clone(),
+            cluster: self.cluster,
+            client_name: self.client_name.clone(),
             config: self.config.clone(),
             profiles: self.profiles.clone(),
             clients: RwLock::new(CachedClients {
@@ -486,6 +532,8 @@ impl AppState {
             credential_source: CredentialSource::Profiles(vec![]),
             policy: Self::test_policy(),
             database_url: None,
+            cluster: false,
+            client_name: None,
             config: None,
             profiles: vec![],
             clients: RwLock::new(CachedClients {
@@ -509,6 +557,8 @@ impl AppState {
             credential_source: CredentialSource::Profiles(vec![]),
             policy: Self::test_policy(),
             database_url: None,
+            cluster: false,
+            client_name: None,
             config: None,
             profiles: vec![],
             clients: RwLock::new(CachedClients {
@@ -534,6 +584,8 @@ impl AppState {
             credential_source: CredentialSource::Profiles(vec![]),
             policy: Self::test_policy(),
             database_url: None,
+            cluster: false,
+            client_name: None,
             config: None,
             profiles: vec![],
             clients: RwLock::new(CachedClients {

crates/redisctl-mcp/src/state.rs

@@ -0,0 +1,43 @@
+//! Unified Redis connection type supporting both standalone and cluster modes.
+
+use redis::aio::{ConnectionLike, MultiplexedConnection};
+use redis::cluster_async::ClusterConnection;
+use redis::{Cmd, RedisFuture, Value};
+
+/// A Redis connection that transparently handles both standalone and cluster topologies.
+///
+/// Implements `ConnectionLike` so it can be used anywhere a Redis connection is expected
+/// (`cmd.query_async()`, `pipe.query_async()`, etc.).
+#[derive(Clone)]
+pub enum RedisConnection {
+    Standalone(MultiplexedConnection),
+    Cluster(ClusterConnection),
+}
+
+impl ConnectionLike for RedisConnection {
+    fn req_packed_command<'a>(&'a mut self, cmd: &'a Cmd) -> RedisFuture<'a, Value> {
+        match self {
+            RedisConnection::Standalone(conn) => conn.req_packed_command(cmd),
+            RedisConnection::Cluster(conn) => conn.req_packed_command(cmd),
+        }
+    }
+
+    fn req_packed_commands<'a>(
+        &'a mut self,
+        cmd: &'a redis::Pipeline,
+        offset: usize,
+        count: usize,
+    ) -> RedisFuture<'a, Vec<Value>> {
+        match self {
+            RedisConnection::Standalone(conn) => conn.req_packed_commands(cmd, offset, count),
+            RedisConnection::Cluster(conn) => conn.req_packed_commands(cmd, offset, count),
+        }
+    }
+
+    fn get_db(&self) -> i64 {
+        match self {
+            RedisConnection::Standalone(conn) => conn.get_db(),
+            RedisConnection::Cluster(conn) => conn.get_db(),
+        }
+    }
+}

crates/redisctl-mcp/src/tools/redis/connection.rs

@@ -2,6 +2,7 @@
 
 mod aliases;
 mod bulk;
+mod connection;
 mod diagnostics;
 mod json;
 mod keys;
@@ -10,6 +11,8 @@ mod search;
 mod server;
 mod structures;
 
+pub(crate) use connection::RedisConnection;
+
 #[allow(unused_imports)]
 pub use aliases::*;
 #[allow(unused_imports)]
@@ -145,15 +148,15 @@ pub(crate) fn resolve_redis_url(
     })
 }
 
-/// Resolve a Redis URL and return a cached multiplexed connection.
+/// Resolve a Redis URL and return a cached connection (standalone or cluster).
 ///
 /// This is the main entry point for tool handlers. It resolves the URL from
 /// the input parameters, then returns a pooled connection (creating one if needed).
 pub(crate) async fn get_connection(
     url: Option<String>,
     profile: Option<&str>,
     state: &AppState,
-) -> Result<redis::aio::MultiplexedConnection, ToolError> {
+) -> Result<RedisConnection, ToolError> {
     let url = resolve_redis_url(url, profile, state)?;
     state
         .redis_connection_for_url(&url)