| 1 | //! HTTP MCP transport. |
| 2 | //! |
| 3 | //! Speaks Streamable HTTP first and falls back to the legacy SSE endpoint |
| 4 | //! when the server rejects the newer protocol, plus the header/token/OAuth |
| 5 | //! resolution shared by both HTTP-flavoured transports. |
| 6 | |
| 7 | use std::collections::HashMap; |
| 8 | use std::time::Duration; |
| 9 | |
| 10 | use anyhow::Result; |
| 11 | |
| 12 | use super::headers::{apply_safe_custom_headers, with_default_mcp_http_headers}; |
| 13 | use super::http_client::McpHttpClient; |
| 14 | use super::sse::SseTransport; |
| 15 | use super::streamable_http::{StreamableHttpTransport, StreamableSendError}; |
| 16 | use super::{McpServerConfig, McpTransport, ReviewedPluginMcpSource, oauth}; |
| 17 | pub(super) struct HttpTransport { |
| 18 | mode: HttpTransportMode, |
| 19 | client: McpHttpClient, |
| 20 | base_url: String, |
| 21 | auth: McpHttpAuth, |
| 22 | cancel_token: tokio_util::sync::CancellationToken, |
| 23 | endpoint_timeout: Duration, |
| 24 | } |
| 25 | |
| 26 | enum HttpTransportMode { |
| 27 | Streamable(StreamableHttpTransport), |
| 28 | Sse(SseTransport), |
| 29 | } |
| 30 | |
| 31 | #[derive(Clone, Default)] |
| 32 | pub(super) struct McpHttpAuth { |
| 33 | pub(super) server_name: String, |
| 34 | pub(super) headers: HashMap<String, String>, |
| 35 | pub(super) env_headers: HashMap<String, String>, |
| 36 | pub(super) bearer_token_env_var: Option<String>, |
| 37 | pub(super) oauth: Option<oauth::McpOAuthRuntime>, |
| 38 | /// Whether the server's *configuration* routes authentication through |
| 39 | /// OAuth, independent of whether a credential is cached yet: a URL-based |
| 40 | /// server that is neither plugin-contributed nor supplied a manual |
| 41 | /// bearer/Authorization credential (#6030). |
| 42 | /// |
| 43 | /// This is [`oauth::server_supports_oauth_login`] — the same predicate the |
| 44 | /// login flow itself is gated on — so the recovery copy it selects |
| 45 | /// (`/mcp login <name>`) names a command that will actually run. A live |
| 46 | /// [`Self::oauth`] runtime always implies it: the runtime is only built |
| 47 | /// for a server that passes this predicate. A first-run OAuth server has |
| 48 | /// no runtime yet, which is exactly the case that used to fall through to |
| 49 | /// the bearer-token copy. |
| 50 | pub(super) oauth_configured: bool, |
| 51 | pub(super) suppress_server_error_details: bool, |
| 52 | pub(super) reviewed_plugin: Option<ReviewedPluginMcpSource>, |
| 53 | } |
| 54 | |
| 55 | impl McpHttpAuth { |
| 56 | pub(super) fn from_config( |
| 57 | server_name: &str, |
| 58 | config: &McpServerConfig, |
| 59 | oauth: Option<oauth::McpOAuthRuntime>, |
| 60 | ) -> Self { |
| 61 | Self { |
| 62 | server_name: server_name.to_string(), |
| 63 | headers: config.headers.clone(), |
| 64 | env_headers: config.env_headers.clone(), |
| 65 | bearer_token_env_var: config.bearer_token_env_var.clone(), |
| 66 | oauth, |
| 67 | oauth_configured: oauth::server_supports_oauth_login(config), |
| 68 | suppress_server_error_details: config.reviewed_plugin.is_some(), |
| 69 | reviewed_plugin: config.reviewed_plugin.clone(), |
| 70 | } |
| 71 | } |
| 72 | |
| 73 | pub(super) fn server_error_preview(&self, preview: &str) -> String { |
| 74 | if self.suppress_server_error_details { |
| 75 | "<server details suppressed for reviewed plugin>".to_string() |
| 76 | } else { |
| 77 | preview.to_string() |
| 78 | } |
| 79 | } |
| 80 | |
| 81 | pub(super) async fn resolved_headers(&self) -> Result<HashMap<String, String>> { |
| 82 | if let Some(source) = self.reviewed_plugin.as_ref() { |
| 83 | source.validate_before_use(&self.server_name, "authenticate request to")?; |
| 84 | } |
| 85 | let mut headers = self.headers.clone(); |
| 86 | for (name, env_var) in &self.env_headers { |
| 87 | let value = self.reviewed_plugin.as_ref().map_or_else( |
| 88 | || std::env::var(env_var), |
| 89 | |source| source.host_environment.var(env_var), |
| 90 | ); |
| 91 | if let Ok(value) = value |
| 92 | && !value.trim().is_empty() |
| 93 | { |
| 94 | headers.insert(name.clone(), value); |
| 95 | } |
| 96 | } |
| 97 | if !mcp_headers_have_authorization(&headers) |
| 98 | && let Some(env_var) = self.bearer_token_env_var.as_deref() |
| 99 | && let Ok(token) = self.reviewed_plugin.as_ref().map_or_else( |
| 100 | || std::env::var(env_var), |
| 101 | |source| source.host_environment.var(env_var), |
| 102 | ) |
| 103 | { |
| 104 | let token = token.trim(); |
| 105 | if !token.is_empty() { |
| 106 | headers.insert("Authorization".to_string(), format!("Bearer {token}")); |
| 107 | } |
| 108 | } |
| 109 | if !mcp_headers_have_authorization(&headers) |
| 110 | && let Some(oauth) = &self.oauth |
| 111 | { |
| 112 | let authorization = match oauth.authorization_header().await { |
| 113 | Ok(authorization) => authorization, |
| 114 | Err(_) if self.suppress_server_error_details => { |
| 115 | anyhow::bail!( |
| 116 | "Reviewed plugin MCP authentication failed (provider details suppressed)" |
| 117 | ) |
| 118 | } |
| 119 | Err(error) => return Err(error), |
| 120 | }; |
| 121 | if let Some(value) = authorization { |
| 122 | headers.insert("Authorization".to_string(), value); |
| 123 | } |
| 124 | } |
| 125 | Ok(headers) |
| 126 | } |
| 127 | } |
| 128 | |
| 129 | pub(super) fn mcp_headers_have_authorization(headers: &HashMap<String, String>) -> bool { |
| 130 | headers |
| 131 | .keys() |
| 132 | .any(|key| key.trim().eq_ignore_ascii_case("authorization")) |
| 133 | } |
| 134 | |
| 135 | impl HttpTransport { |
| 136 | pub(super) fn new( |
| 137 | client: McpHttpClient, |
| 138 | url: String, |
| 139 | auth: McpHttpAuth, |
| 140 | cancel_token: tokio_util::sync::CancellationToken, |
| 141 | endpoint_timeout: Duration, |
| 142 | ) -> Self { |
| 143 | Self { |
| 144 | mode: HttpTransportMode::Streamable(StreamableHttpTransport::new( |
| 145 | client.clone(), |
| 146 | url.clone(), |
| 147 | auth.clone(), |
| 148 | )), |
| 149 | client, |
| 150 | base_url: url, |
| 151 | auth, |
| 152 | cancel_token, |
| 153 | endpoint_timeout, |
| 154 | } |
| 155 | } |
| 156 | |
| 157 | async fn switch_to_sse_and_send(&mut self, msg: Vec<u8>) -> Result<()> { |
| 158 | let mut sse = SseTransport::connect( |
| 159 | self.client.clone(), |
| 160 | self.base_url.clone(), |
| 161 | self.auth.clone(), |
| 162 | self.cancel_token.clone(), |
| 163 | self.endpoint_timeout, |
| 164 | ) |
| 165 | .await?; |
| 166 | sse.send(msg).await?; |
| 167 | self.mode = HttpTransportMode::Sse(sse); |
| 168 | Ok(()) |
| 169 | } |
| 170 | |
| 171 | /// Best-effort session-establishment GET preflight. |
| 172 | /// |
| 173 | /// Per the Streamable HTTP spec, the server may return an |
| 174 | /// `Mcp-Session-Id` header on the `initialize` response (the normal |
| 175 | /// path handled inside [`StreamableHttpTransport::send`] above). |
| 176 | /// However some servers (e.g. Hindsight, #1629) **require** a session |
| 177 | /// ID on every POST including `initialize`, creating a chicken-and-egg |
| 178 | /// problem. For those servers we send a short-lived GET before the |
| 179 | /// first POST: if the server returns a session ID in the GET response |
| 180 | /// it will be captured by the header-reading code in |
| 181 | /// [`StreamableHttpTransport::send`] just as if it came from a POST |
| 182 | /// response. |
| 183 | /// |
| 184 | /// This is intentionally best-effort: |
| 185 | /// * The GET uses a tight per-request inner timeout so it never |
| 186 | /// blocks connection startup for long. |
| 187 | /// * If the server doesn't support GET (405, 404, …) we log a debug |
| 188 | /// line and move on — the `initialize` POST will proceed without a |
| 189 | /// session ID. |
| 190 | /// * If the server opens an SSE stream in response (the GET from old |
| 191 | /// SSE transport), we read only the headers, then discard the body |
| 192 | /// so the SSE stream is torn down. The actual SSE path uses a |
| 193 | /// dedicated `SseTransport` and is triggered by the incompatible- |
| 194 | /// status fallback in [`HttpTransport::send`]. |
| 195 | pub(super) async fn try_establish_session(&mut self) -> Result<()> { |
| 196 | let cancel = self.cancel_token.clone(); |
| 197 | let transport = match &mut self.mode { |
| 198 | HttpTransportMode::Streamable(t) => t, |
| 199 | // Already on SSE — session is implicit via the long-lived GET. |
| 200 | HttpTransportMode::Sse(_) => return Ok(()), |
| 201 | }; |
| 202 | |
| 203 | let headers = tokio::select! { |
| 204 | biased; |
| 205 | _ = cancel.cancelled() => { |
| 206 | anyhow::bail!("MCP session preflight cancelled after plugin authority changed") |
| 207 | } |
| 208 | headers = transport.auth.resolved_headers() => headers?, |
| 209 | }; |
| 210 | let request = apply_safe_custom_headers( |
| 211 | with_default_mcp_http_headers(transport.client.get(&transport.url), false), |
| 212 | &headers, |
| 213 | ); |
| 214 | let response = tokio::select! { |
| 215 | biased; |
| 216 | _ = cancel.cancelled() => { |
| 217 | anyhow::bail!("MCP session preflight cancelled after plugin authority changed") |
| 218 | } |
| 219 | response = tokio::time::timeout(Duration::from_secs(5), transport.client.send(request)) => { |
| 220 | response |
| 221 | .map_err(|_| anyhow::anyhow!("GET timeout"))? |
| 222 | .map_err(|e| anyhow::anyhow!("GET error: {e}"))? |
| 223 | } |
| 224 | }; |
| 225 | |
| 226 | // Capture session ID from the GET response so subsequent POSTs |
| 227 | // (including `initialize`) can include it. This is the same |
| 228 | // header-reading logic that would be hit inside |
| 229 | // `StreamableHttpTransport::send` for POST responses, but since |
| 230 | // the GET is sent before any POST we do it here directly. |
| 231 | if let Some(sid) = response |
| 232 | .headers() |
| 233 | .get("Mcp-Session-Id") |
| 234 | .and_then(|v| v.to_str().ok()) |
| 235 | && transport.session_id.as_deref() != Some(sid) |
| 236 | { |
| 237 | let session_ref = crate::utils::redacted_identifier_for_log(sid); |
| 238 | tracing::debug!(target: "mcp", session = %session_ref, "captured MCP session ID via GET preflight"); |
| 239 | transport.session_id = Some(sid.to_string()); |
| 240 | } |
| 241 | |
| 242 | // We only care about the response headers — discard the body. |
| 243 | // If the server opened an SSE stream in response (some servers |
| 244 | // do this on GET), it will be torn down when response is dropped. |
| 245 | drop(response); |
| 246 | |
| 247 | Ok(()) |
| 248 | } |
| 249 | } |
| 250 | |
| 251 | #[async_trait::async_trait] |
| 252 | impl McpTransport for HttpTransport { |
| 253 | fn set_protocol_version(&mut self, version: &str) { |
| 254 | // Only Streamable HTTP carries the MCP-Protocol-Version header; the |
| 255 | // legacy SSE transport predates it and ignores the negotiation result. |
| 256 | if let HttpTransportMode::Streamable(transport) = &mut self.mode { |
| 257 | transport.set_protocol_version(version); |
| 258 | } |
| 259 | } |
| 260 | |
| 261 | async fn send(&mut self, msg: Vec<u8>) -> Result<()> { |
| 262 | match &mut self.mode { |
| 263 | HttpTransportMode::Streamable(transport) => match transport.send(msg.clone()).await { |
| 264 | Ok(()) => Ok(()), |
| 265 | Err(StreamableSendError::Incompatible(detail)) => { |
| 266 | tracing::debug!( |
| 267 | "MCP Streamable HTTP unavailable; falling back to SSE endpoint discovery: {}", |
| 268 | detail |
| 269 | ); |
| 270 | self.switch_to_sse_and_send(msg).await |
| 271 | } |
| 272 | Err(StreamableSendError::StaleSession(detail)) => { |
| 273 | if let HttpTransportMode::Streamable(transport) = &mut self.mode { |
| 274 | tracing::debug!( |
| 275 | target: "mcp", |
| 276 | error = %detail, |
| 277 | "MCP Streamable HTTP session expired; clearing cached session ID" |
| 278 | ); |
| 279 | transport.session_id = None; |
| 280 | } |
| 281 | Err(anyhow::anyhow!( |
| 282 | "MCP Streamable HTTP session expired; retry with a new session required ({detail})" |
| 283 | )) |
| 284 | } |
| 285 | Err(StreamableSendError::Other(err)) => Err(err), |
| 286 | }, |
| 287 | HttpTransportMode::Sse(transport) => transport.send(msg).await, |
| 288 | } |
| 289 | } |
| 290 | |
| 291 | async fn recv(&mut self) -> Result<Vec<u8>> { |
| 292 | match &mut self.mode { |
| 293 | HttpTransportMode::Streamable(transport) => transport.recv().await, |
| 294 | HttpTransportMode::Sse(transport) => transport.recv().await, |
| 295 | } |
| 296 | } |
| 297 | |
| 298 | async fn shutdown(&mut self) { |
| 299 | if let HttpTransportMode::Sse(transport) = &mut self.mode { |
| 300 | transport.shutdown().await; |
| 301 | } |
| 302 | } |
| 303 | } |
| 304 |