feat(mcp): enables remote mcp (#2836)

Author: dingfeli

Cargo.lock

@@ -129,7 +129,7 @@ winnow = "=0.6.2"
 winreg = "0.55.0"
 schemars = "1.0.4"
 jsonschema = "0.30.0"
-rmcp = { version = "0.6.0", features = ["client", "transport-child-process"] }
+rmcp = { version = "0.6.3", features = ["client", "transport-sse-client-reqwest", "reqwest", "transport-streamable-http-client-reqwest", "transport-child-process", "tower", "auth"] }
 
 [workspace.lints.rust]
 future_incompatible = "warn"

Cargo.toml

@@ -44,6 +44,10 @@ pub enum UpdateEventMessage {
         result: Result<ListResourceTemplatesResult>,
         peer: Option<Peer<RoleClient>>,
     },
+    OauthLink {
+        server_name: String,
+        link: String,
+    },
     InitStart {
         server_name: String,
     },
@@ -146,6 +150,17 @@ impl Messenger for ServerMessenger {
             .map_err(|e| MessengerError::Custom(e.to_string()))?)
     }
 
+    async fn send_oauth_link(&self, link: String) -> MessengerResult {
+        Ok(self
+            .update_event_sender
+            .send(UpdateEventMessage::OauthLink {
+                server_name: self.server_name.clone(),
+                link,
+            })
+            .await
+            .map_err(|e| MessengerError::Custom(e.to_string()))?)
+    }
+
     async fn send_init_msg(&self) -> MessengerResult {
         Ok(self
             .update_event_sender

crates/chat-cli/src/cli/chat/server_messenger.rs

@@ -90,6 +90,7 @@ use crate::database::settings::Setting;
 use crate::mcp_client::messenger::Messenger;
 use crate::mcp_client::{
     InitializedMcpClient,
+    InnerService,
     McpClientService,
 };
 use crate::os::Os;
@@ -137,6 +138,11 @@ enum LoadingMsg {
     /// This is sent when all tool initialization is complete or when the application is shutting
     /// down.
     Terminate { still_loading: Vec<String> },
+    /// Indicates that a server requires user authentication and provides a sign-in link.
+    /// This message is used to notify the user about authentication requirements for MCP servers
+    /// that need OAuth or other authentication methods. Contains the server name and the
+    /// authentication message (typically a URL or instructions).
+    SignInNotice { name: String },
 }
 
 /// Used to denote the loading outcome associated with a server.
@@ -630,19 +636,29 @@ impl ToolManager {
                         let server_name_clone = server_name.clone();
                         tokio::spawn(async move {
                             match handle.await {
-                                Ok(Ok(client)) => match client.cancel().await {
-                                    Ok(_) => info!("Server {server_name_clone} evicted due to agent swap"),
-                                    Err(e) => error!("Server {server_name_clone} has failed to cancel: {e}"),
+                                Ok(Ok(client)) => {
+                                    let InnerService::Original(client) = client.inner_service else {
+                                        unreachable!();
+                                    };
+                                    match client.cancel().await {
+                                        Ok(_) => info!("Server {server_name_clone} evicted due to agent swap"),
+                                        Err(e) => error!("Server {server_name_clone} has failed to cancel: {e}"),
+                                    }
                                 },
                                 Ok(Err(_)) | Err(_) => {
                                     error!("Server {server_name_clone} has failed to cancel");
                                 },
                             }
                         });
                     },
-                    InitializedMcpClient::Ready(running_service) => match running_service.cancel().await {
-                        Ok(_) => info!("Server {server_name} evicted due to agent swap"),
-                        Err(e) => error!("Server {server_name} has failed to cancel: {e}"),
+                    InitializedMcpClient::Ready(running_service) => {
+                        let InnerService::Original(client) = running_service.inner_service else {
+                            unreachable!();
+                        };
+                        match client.cancel().await {
+                            Ok(_) => info!("Server {server_name} evicted due to agent swap"),
+                            Err(e) => error!("Server {server_name} has failed to cancel: {e}"),
+                        }
                     },
                 }
             }
@@ -869,17 +885,16 @@ impl ToolManager {
                     });
                 };
 
-                let running_service = (*client.get_running_service().await.map_err(|e| ToolResult {
+                let running_service = client.get_running_service().await.map_err(|e| ToolResult {
                     tool_use_id: value.id.clone(),
                     content: vec![ToolResultContentBlock::Text(format!("Mcp tool client not ready: {e}"))],
                     status: ToolResultStatus::Error,
-                })?)
-                .clone();
+                })?;
 
                 Tool::Custom(CustomTool {
                     name: tool_name.to_owned(),
                     server_name: server_name.to_owned(),
-                    client: running_service,
+                    client: running_service.clone(),
                     params: value.args.as_object().cloned(),
                 })
             },
@@ -1170,6 +1185,15 @@ fn spawn_display_task(
                                 execute!(output, style::Print("\n"),)?;
                                 break;
                             },
+                            LoadingMsg::SignInNotice { name } => {
+                                execute!(
+                                    output,
+                                    cursor::MoveToColumn(0),
+                                    cursor::MoveUp(1),
+                                    terminal::Clear(terminal::ClearType::CurrentLine),
+                                )?;
+                                queue_oauth_message(&name, &mut output)?;
+                            },
                         },
                         Err(_e) => {
                             spinner_logo_idx = (spinner_logo_idx + 1) % SPINNER_CHARS.len();
@@ -1595,6 +1619,35 @@ fn spawn_orchestrator_task(
                 },
                 UpdateEventMessage::ListResourcesResult { .. } => {},
                 UpdateEventMessage::ResourceTemplatesListResult { .. } => {},
+                UpdateEventMessage::OauthLink { server_name, link } => {
+                    let mut buf_writer = BufWriter::new(&mut *record_temp_buf);
+                    let msg = eyre::eyre!(link);
+                    let _ = queue_oauth_message_with_link(server_name.as_str(), &msg, &mut buf_writer);
+                    let _ = buf_writer.flush();
+                    drop(buf_writer);
+                    let record_str = String::from_utf8_lossy(record_temp_buf).to_string();
+                    let record = LoadingRecord::Warn(record_str.clone());
+                    load_record
+                        .lock()
+                        .await
+                        .entry(server_name.clone())
+                        .and_modify(|load_record| {
+                            load_record.push(record.clone());
+                        })
+                        .or_insert(vec![record]);
+                    if let Some(sender) = &loading_status_sender {
+                        let msg = LoadingMsg::SignInNotice {
+                            name: server_name.clone(),
+                        };
+                        if let Err(e) = sender.send(msg).await {
+                            warn!(
+                                "Error sending update message to display task: {:?}\nAssume display task has completed",
+                                e
+                            );
+                            loading_status_sender.take();
+                        }
+                    }
+                },
                 UpdateEventMessage::InitStart { server_name, .. } => {
                     pending.write().await.insert(server_name.clone());
                     loading_servers.insert(server_name, std::time::Instant::now());
@@ -1876,6 +1929,34 @@ fn queue_failure_message(
     )?)
 }
 
+fn queue_oauth_message(name: &str, output: &mut impl Write) -> eyre::Result<()> {
+    Ok(queue!(
+        output,
+        style::SetForegroundColor(style::Color::Yellow),
+        style::Print("⚠ "),
+        style::SetForegroundColor(style::Color::Blue),
+        style::Print(name),
+        style::ResetColor,
+        style::Print(" requires OAuth authentication. Use /mcp to see the auth link\n"),
+    )?)
+}
+
+fn queue_oauth_message_with_link(name: &str, msg: &eyre::Report, output: &mut impl Write) -> eyre::Result<()> {
+    Ok(queue!(
+        output,
+        style::SetForegroundColor(style::Color::Yellow),
+        style::Print("⚠ "),
+        style::SetForegroundColor(style::Color::Blue),
+        style::Print(name),
+        style::ResetColor,
+        style::Print(" requires OAuth authentication. Follow this link to proceed: \n"),
+        style::SetForegroundColor(style::Color::Yellow),
+        style::Print(msg),
+        style::ResetColor,
+        style::Print("\n")
+    )?)
+}
+
 fn queue_warn_message(name: &str, msg: &eyre::Report, time: &str, output: &mut impl Write) -> eyre::Result<()> {
     Ok(queue!(
         output,

crates/chat-cli/src/cli/chat/tool_manager.rs

@@ -7,7 +7,6 @@ use crossterm::{
     style,
 };
 use eyre::Result;
-use rmcp::RoleClient;
 use rmcp::model::CallToolRequestParam;
 use schemars::JsonSchema;
 use serde::{
@@ -23,14 +22,39 @@ use crate::cli::agent::{
 };
 use crate::cli::chat::CONTINUATION_LINE;
 use crate::cli::chat::token_counter::TokenCounter;
+use crate::mcp_client::RunningService;
 use crate::os::Os;
 use crate::util::MCP_SERVER_TOOL_DELIMITER;
 use crate::util::pattern_matching::matches_any_pattern;
 
-// TODO: support http transport type
+#[derive(Clone, Serialize, Deserialize, Debug, Eq, PartialEq, JsonSchema)]
+#[serde(rename_all = "camelCase")]
+pub enum TransportType {
+    /// Standard input/output transport (default)
+    Stdio,
+    /// HTTP transport for web-based communication
+    Http,
+}
+
+impl Default for TransportType {
+    fn default() -> Self {
+        Self::Stdio
+    }
+}
+
 #[derive(Clone, Serialize, Deserialize, Debug, Eq, PartialEq, JsonSchema)]
 pub struct CustomToolConfig {
+    /// The type of transport the mcp server is expecting
+    #[serde(default)]
+    pub r#type: TransportType,
+    /// The URL endpoint for HTTP-based MCP servers
+    #[serde(default)]
+    pub url: String,
+    /// HTTP headers to include when communicating with HTTP-based MCP servers
+    #[serde(default)]
+    pub headers: HashMap<String, String>,
     /// The command string used to initialize the mcp server
+    #[serde(default)]
     pub command: String,
     /// A list of arguments to be used to run the command with
     #[serde(default)]
@@ -64,20 +88,20 @@ pub struct CustomTool {
     /// prefixed to the tool name when presented to the model for disambiguation.
     pub server_name: String,
     /// Reference to the client that manages communication with the tool's server process.
-    pub client: rmcp::Peer<RoleClient>,
+    pub client: RunningService,
     /// Optional parameters to pass to the tool when invoking the method.
     /// Structured as a JSON value to accommodate various parameter types and structures.
     pub params: Option<serde_json::Map<String, serde_json::Value>>,
 }
 
 impl CustomTool {
-    pub async fn invoke(&self, _os: &Os, _updates: impl Write) -> Result<InvokeOutput> {
+    pub async fn invoke(&self, _os: &Os, _updates: &mut impl Write) -> Result<InvokeOutput> {
         let params = CallToolRequestParam {
             name: Cow::from(self.name.clone()),
             arguments: self.params.clone(),
         };
 
-        let resp = self.client.call_tool(params).await?;
+        let resp = self.client.call_tool(params.clone()).await?;
 
         if resp.is_error.is_none_or(|v| !v) {
             Ok(InvokeOutput {