feat(mcp): add complete MCP support with remote registry integration (#544)

* feat: add MCP notification types for server lifecycle events

* feat: add MCP remote registry client with TTL caching

Implement McpRegistryClient for fetching MCP servers from
registry.cortex.foundation/mcp with local TTL-based caching.

Features:
- In-memory cache with configurable TTL (default 1 hour)
- Persistent disk cache for offline resilience
- Search by name, description, tags, or category
- Fallback servers when registry is unavailable
- Configurable registry URL and cache directory

* feat: integrate MCP lifecycle notifications with connection manager

* feat: update TUI MCP Manager Modal to support remote registry

- Add RegistryEntry and RegistrySource types in registry.rs
- Add From<RegistryServer> impl to convert remote entries
- Add get_local_registry_entries() and get_remote_server_config()
- Add RegistryLoadState enum in types.rs
- Update SelectFromRegistry mode to include entries and load_state
- Update handlers to work with new RegistryEntry structure
- Update rendering to display category and required env indicators
- Re-export remote registry types for future async loading

Author: echobt

src/cortex-engine/src/mcp/manager.rs

@@ -11,14 +11,29 @@ use std::sync::Arc;
 
 use anyhow::{Result, anyhow};
 use serde_json::Value;
-use tokio::sync::{Mutex, RwLock};
+use tokio::sync::{Mutex, RwLock, mpsc};
 use tracing::warn;
 
 use cortex_mcp_types::{CallToolResult, Resource, Tool};
 
 use super::McpServerConfig;
 use super::client::{ConnectionState, McpClient};
 
+/// MCP lifecycle event for notifications
+#[derive(Debug, Clone)]
+pub enum McpLifecycleEvent {
+    /// Server was added to the manager
+    ServerAdded { name: String },
+    /// Server was connected and tools discovered
+    ServerConnected { name: String, tool_count: usize, tool_names: Vec<String> },
+    /// Server was disconnected
+    ServerDisconnected { name: String },
+    /// Server was removed from the manager
+    ServerRemoved { name: String },
+    /// Connection failed
+    ConnectionFailed { name: String, error: String },
+}
+
 /// Delimiter used to create qualified tool names.
 const TOOL_NAME_DELIMITER: &str = "__";
 
@@ -33,6 +48,8 @@ pub struct McpConnectionManager {
     configs: RwLock<HashMap<String, McpServerConfig>>,
     /// Set of servers currently being started (prevents race conditions).
     starting_servers: Mutex<HashSet<String>>,
+    /// Event sender for lifecycle notifications
+    event_tx: Option<mpsc::UnboundedSender<McpLifecycleEvent>>,
 }
 
 impl Default for McpConnectionManager {
@@ -48,6 +65,29 @@ impl McpConnectionManager {
             clients: RwLock::new(HashMap::new()),
             configs: RwLock::new(HashMap::new()),
             starting_servers: Mutex::new(HashSet::new()),
+            event_tx: None,
+        }
+    }
+
+    /// Create a new connection manager with an event sender
+    pub fn with_event_sender(tx: mpsc::UnboundedSender<McpLifecycleEvent>) -> Self {
+        Self {
+            clients: RwLock::new(HashMap::new()),
+            configs: RwLock::new(HashMap::new()),
+            starting_servers: Mutex::new(HashSet::new()),
+            event_tx: Some(tx),
+        }
+    }
+
+    /// Set the event sender for lifecycle notifications
+    pub fn set_event_sender(&mut self, tx: mpsc::UnboundedSender<McpLifecycleEvent>) {
+        self.event_tx = Some(tx);
+    }
+
+    /// Helper to send an event
+    fn send_event(&self, event: McpLifecycleEvent) {
+        if let Some(ref tx) = self.event_tx {
+            let _ = tx.send(event);
         }
     }
 
@@ -61,11 +101,15 @@ impl McpConnectionManager {
 
         // Create client but don't connect yet
         let client = Arc::new(McpClient::new(config));
-        self.clients.write().await.insert(name, client);
+        self.clients.write().await.insert(name.clone(), client);
+
+        self.send_event(McpLifecycleEvent::ServerAdded { name });
     }
 
     /// Remove a server.
     pub async fn remove_server(&self, name: &str) -> Result<()> {
+        self.send_event(McpLifecycleEvent::ServerRemoved { name: name.to_string() });
+
         // Disconnect if connected
         if let Some(client) = self.clients.write().await.remove(name) {
             client.disconnect().await?;
@@ -150,6 +194,28 @@ impl McpConnectionManager {
             starting.remove(name);
         }
 
+        // Emit lifecycle event based on connection result
+        match &result {
+            Ok(()) => {
+                // After successful connect, get tools and emit event
+                if let Some(client) = self.clients.read().await.get(name).cloned() {
+                    let tools = client.tools().await;
+                    let tool_names: Vec<String> = tools.iter().map(|t| t.name.clone()).collect();
+                    self.send_event(McpLifecycleEvent::ServerConnected {
+                        name: name.to_string(),
+                        tool_count: tools.len(),
+                        tool_names,
+                    });
+                }
+            }
+            Err(e) => {
+                self.send_event(McpLifecycleEvent::ConnectionFailed {
+                    name: name.to_string(),
+                    error: e.to_string(),
+                });
+            }
+        }
+
         result
     }
 
@@ -163,7 +229,13 @@ impl McpConnectionManager {
             .cloned()
             .ok_or_else(|| anyhow!("Server not found: {}", name))?;
 
-        client.disconnect().await
+        let result = client.disconnect().await;
+
+        if result.is_ok() {
+            self.send_event(McpLifecycleEvent::ServerDisconnected { name: name.to_string() });
+        }
+
+        result
     }
 
     /// Connect to all servers.
@@ -402,4 +474,40 @@ mod tests {
         manager.remove_server("test").await.unwrap();
         assert!(manager.server_names().await.is_empty());
     }
+
+    #[tokio::test]
+    async fn test_lifecycle_events() {
+        let (tx, mut rx) = mpsc::unbounded_channel();
+        let manager = McpConnectionManager::with_event_sender(tx);
+
+        let config = McpServerConfig::new("test", "echo");
+        manager.add_server(config).await;
+
+        // Should receive ServerAdded event
+        let event = rx.try_recv();
+        assert!(matches!(event, Ok(McpLifecycleEvent::ServerAdded { name }) if name == "test"));
+
+        manager.remove_server("test").await.unwrap();
+
+        // Should receive ServerRemoved event
+        let event = rx.try_recv();
+        assert!(matches!(event, Ok(McpLifecycleEvent::ServerRemoved { name }) if name == "test"));
+    }
+
+    #[tokio::test]
+    async fn test_manager_with_event_sender() {
+        let (tx, _rx) = mpsc::unbounded_channel();
+        let manager = McpConnectionManager::with_event_sender(tx);
+        assert!(manager.event_tx.is_some());
+    }
+
+    #[tokio::test]
+    async fn test_manager_set_event_sender() {
+        let mut manager = McpConnectionManager::new();
+        assert!(manager.event_tx.is_none());
+
+        let (tx, _rx) = mpsc::unbounded_channel();
+        manager.set_event_sender(tx);
+        assert!(manager.event_tx.is_some());
+    }
 }

src/cortex-engine/src/mcp/mod.rs

@@ -10,6 +10,7 @@ pub mod client;
 pub mod manager;
 pub mod oauth;
 pub mod oauth_callback;
+pub mod registry;
 
 // OAuth exports - these are actively used by cortex-cli
 pub use oauth::{
@@ -25,6 +26,12 @@ pub use oauth_callback::{
 pub use client::{ConnectionState, McpClient};
 pub use manager::{McpConnectionManager, create_qualified_name, parse_qualified_name};
 
+// Registry exports
+pub use registry::{
+    McpRegistryClient, RegistryServer, RegistryInstallConfig, StdioConfig, HttpConfig,
+    REGISTRY_URL, DEFAULT_CACHE_TTL,
+};
+
 use std::collections::HashMap;
 use std::path::PathBuf;
 use std::time::Duration;