Merge pull request #103 from nexxeln/feat/mcp-client-integration

add mcp client support for workers

Author: jamiepine

Cargo.lock

@@ -74,6 +74,7 @@ pin-project = "1"
 
 # Schema validation
 schemars = "0.8"
+rmcp = { version = "0.16", features = ["client", "reqwest", "transport-child-process", "transport-streamable-http-client", "transport-streamable-http-client-reqwest"] }
 
 # Command line (for main.rs)
 clap = { version = "4.5", features = ["derive"] }

Cargo.toml

@@ -185,6 +185,8 @@ impl Worker {
 
         tracing::info!(worker_id = %self.id, task = %self.task, "worker starting");
 
+        let mcp_tools = self.deps.mcp_manager.get_tools().await;
+
         // Create per-worker ToolServer with task tools
         let worker_tool_server = crate::tools::create_worker_tool_server(
             self.deps.agent_id.clone(),
@@ -196,6 +198,7 @@ impl Worker {
             self.brave_search_key.clone(),
             self.deps.runtime_config.workspace_dir.clone(),
             self.deps.runtime_config.instance_dir.clone(),
+            mcp_tools,
         );
 
         let routing = self.deps.runtime_config.routing.load();

src/agent/worker.rs

@@ -134,6 +134,22 @@ pub(super) struct DeleteAgentQuery {
     agent_id: String,
 }
 
+#[derive(Deserialize)]
+pub(super) struct AgentMcpQuery {
+    agent_id: String,
+}
+
+#[derive(Deserialize)]
+pub(super) struct ReconnectMcpRequest {
+    agent_id: String,
+    server_name: String,
+}
+
+#[derive(Serialize)]
+pub(super) struct AgentMcpResponse {
+    servers: Vec<crate::mcp::McpServerStatus>,
+}
+
 /// List all configured agents with their config summaries.
 pub(super) async fn list_agents(State(state): State<Arc<ApiState>>) -> Json<AgentsResponse> {
     let agents = state.agent_configs.load();
@@ -142,6 +158,51 @@ pub(super) async fn list_agents(State(state): State<Arc<ApiState>>) -> Json<Agen
     })
 }
 
+/// List MCP connection status for an agent.
+pub(super) async fn list_agent_mcp(
+    State(state): State<Arc<ApiState>>,
+    Query(query): Query<AgentMcpQuery>,
+) -> Result<Json<AgentMcpResponse>, StatusCode> {
+    let managers = state.mcp_managers.load();
+    let manager = managers
+        .get(&query.agent_id)
+        .cloned()
+        .ok_or(StatusCode::NOT_FOUND)?;
+    let servers = manager.statuses().await;
+    Ok(Json(AgentMcpResponse { servers }))
+}
+
+/// Force reconnect for a single MCP server on an agent.
+pub(super) async fn reconnect_agent_mcp(
+    State(state): State<Arc<ApiState>>,
+    Json(request): Json<ReconnectMcpRequest>,
+) -> Result<Json<serde_json::Value>, StatusCode> {
+    let managers = state.mcp_managers.load();
+    let manager = managers
+        .get(&request.agent_id)
+        .cloned()
+        .ok_or(StatusCode::NOT_FOUND)?;
+
+    manager
+        .reconnect(&request.server_name)
+        .await
+        .map_err(|error| {
+            tracing::warn!(
+                %error,
+                agent_id = %request.agent_id,
+                server_name = %request.server_name,
+                "failed to reconnect mcp server"
+            );
+            StatusCode::BAD_REQUEST
+        })?;
+
+    Ok(Json(serde_json::json!({
+        "success": true,
+        "agent_id": request.agent_id,
+        "server_name": request.server_name
+    })))
+}
+
 /// Create a new agent and initialize it live (directories, databases, memory, identity, cron, cortex).
 pub(super) async fn create_agent(
     State(state): State<Arc<ApiState>>,
@@ -233,6 +294,7 @@ pub(super) async fn create_agent(
         ingestion: None,
         cortex: None,
         browser: None,
+        mcp: None,
         brave_search_key: None,
         cron: Vec::new(),
     };
@@ -354,10 +416,14 @@ pub(super) async fn create_agent(
             .clone()
     };
 
+    let mcp_manager = std::sync::Arc::new(crate::mcp::McpManager::new(agent_config.mcp.clone()));
+    mcp_manager.connect_all().await;
+
     let deps = crate::AgentDeps {
         agent_id: arc_agent_id.clone(),
         memory_search: memory_search.clone(),
         llm_manager,
+        mcp_manager: mcp_manager.clone(),
         cron_tool: None,
         runtime_config: runtime_config.clone(),
         event_tx: event_tx.clone(),
@@ -455,6 +521,10 @@ pub(super) async fn create_agent(
         configs.insert(agent_id.clone(), runtime_config);
         state.runtime_configs.store(std::sync::Arc::new(configs));
 
+        let mut mcp_managers = (**state.mcp_managers.load()).clone();
+        mcp_managers.insert(agent_id.clone(), mcp_manager);
+        state.mcp_managers.store(std::sync::Arc::new(mcp_managers));
+
         let mut agent_infos = (**state.agent_configs.load()).clone();
         agent_infos.push(AgentInfo {
             id: agent_config.id.clone(),
@@ -567,6 +637,12 @@ pub(super) async fn delete_agent(
 
     // Remove from all API state maps
     {
+        let mut mcp_managers = (**state.mcp_managers.load()).clone();
+        if let Some(mcp_manager) = mcp_managers.remove(&agent_id) {
+            mcp_manager.disconnect_all().await;
+        }
+        state.mcp_managers.store(std::sync::Arc::new(mcp_managers));
+
         let mut pools = (**state.agent_pools.load()).clone();
         pools.remove(&agent_id);
         state.agent_pools.store(std::sync::Arc::new(pools));

src/api/agents.rs

@@ -329,8 +329,13 @@ pub(super) async fn update_agent_config(
     match crate::config::Config::load_from_path(&config_path) {
         Ok(new_config) => {
             let runtime_configs = state.runtime_configs.load();
-            if let Some(rc) = runtime_configs.get(&request.agent_id) {
-                rc.reload_config(&new_config, &request.agent_id);
+            let mcp_managers = state.mcp_managers.load();
+            if let (Some(rc), Some(mcp_manager)) = (
+                runtime_configs.get(&request.agent_id).cloned(),
+                mcp_managers.get(&request.agent_id).cloned(),
+            ) {
+                rc.reload_config(&new_config, &request.agent_id, &mcp_manager)
+                    .await;
             }
             if request.discord.is_some()
                 && let Some(discord_config) = &new_config.messaging.discord

src/api/config.rs

@@ -51,6 +51,8 @@ pub async fn start_http_server(
                 .post(agents::create_agent)
                 .delete(agents::delete_agent),
         )
+        .route("/agents/mcp", get(agents::list_agent_mcp))
+        .route("/agents/mcp/reconnect", post(agents::reconnect_agent_mcp))
         .route("/agents/overview", get(agents::agent_overview))
         .route("/channels", get(channels::list_channels))
         .route("/channels/messages", get(channels::channel_messages))

src/api/server.rs

@@ -431,8 +431,26 @@ pub(super) async fn update_raw_config(
     match crate::config::Config::load_from_path(&config_path) {
         Ok(new_config) => {
             let runtime_configs = state.runtime_configs.load();
-            for (agent_id, rc) in runtime_configs.iter() {
-                rc.reload_config(&new_config, agent_id);
+            let mcp_managers = state.mcp_managers.load();
+            let reload_targets = runtime_configs
+                .iter()
+                .filter_map(|(agent_id, runtime_config)| {
+                    mcp_managers.get(agent_id).map(|mcp_manager| {
+                        (
+                            agent_id.clone(),
+                            runtime_config.clone(),
+                            mcp_manager.clone(),
+                        )
+                    })
+                })
+                .collect::<Vec<_>>();
+            drop(runtime_configs);
+            drop(mcp_managers);
+
+            for (agent_id, runtime_config, mcp_manager) in reload_targets {
+                runtime_config
+                    .reload_config(&new_config, &agent_id, &mcp_manager)
+                    .await;
             }
         }
         Err(error) => {

src/api/settings.rs

@@ -6,6 +6,7 @@ use crate::agent::status::StatusBlock;
 use crate::config::{Binding, DefaultsConfig, DiscordPermissions, RuntimeConfig, SlackPermissions};
 use crate::cron::{CronStore, Scheduler};
 use crate::llm::LlmManager;
+use crate::mcp::McpManager;
 use crate::memory::{EmbeddingModel, MemorySearch};
 use crate::messaging::MessagingManager;
 use crate::messaging::webchat::WebChatAdapter;
@@ -61,6 +62,8 @@ pub struct ApiState {
     pub cron_schedulers: arc_swap::ArcSwap<HashMap<String, Arc<Scheduler>>>,
     /// Per-agent RuntimeConfig for reading live hot-reloaded configuration.
     pub runtime_configs: ArcSwap<HashMap<String, Arc<RuntimeConfig>>>,
+    /// Per-agent MCP managers for status and reconnect APIs.
+    pub mcp_managers: ArcSwap<HashMap<String, Arc<McpManager>>>,
     /// Shared reference to the Discord permissions ArcSwap (same instance used by the adapter and file watcher).
     pub discord_permissions: RwLock<Option<Arc<ArcSwap<DiscordPermissions>>>>,
     /// Shared reference to the Slack permissions ArcSwap (same instance used by the adapter and file watcher).
@@ -191,6 +194,7 @@ impl ApiState {
             cron_stores: arc_swap::ArcSwap::from_pointee(HashMap::new()),
             cron_schedulers: arc_swap::ArcSwap::from_pointee(HashMap::new()),
             runtime_configs: ArcSwap::from_pointee(HashMap::new()),
+            mcp_managers: ArcSwap::from_pointee(HashMap::new()),
             discord_permissions: RwLock::new(None),
             slack_permissions: RwLock::new(None),
             bindings: RwLock::new(None),
@@ -416,6 +420,11 @@ impl ApiState {
         self.runtime_configs.store(Arc::new(configs));
     }
 
+    /// Set the MCP managers for all agents.
+    pub fn set_mcp_managers(&self, managers: HashMap<String, Arc<McpManager>>) {
+        self.mcp_managers.store(Arc::new(managers));
+    }
+
     /// Share the Discord permissions ArcSwap with the API so reads get hot-reloaded values.
     pub async fn set_discord_permissions(&self, permissions: Arc<ArcSwap<DiscordPermissions>>) {
         *self.discord_permissions.write().await = Some(permissions);