From 55ff3d0601874e7319a150be8e6969baa3f6a4f5 Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Mon, 11 May 2026 16:11:32 -0400 Subject: [PATCH] fix(mcp): reuse CLI server targeting Route MCP API client creation through the same CLI server connection helper used by regular commands. This gives MCP the same Unix socket auto-spawn behavior, install-mode guard, proxy handling, and auth refresh path while removing the duplicate local connection stack. --- Cargo.lock | 2 - lib/crates/fabro-cli/src/commands/mcp/mod.rs | 32 +- lib/crates/fabro-cli/tests/it/cmd/mcp.rs | 138 +++++++++ lib/crates/fabro-mcp-server/Cargo.toml | 2 - lib/crates/fabro-mcp-server/src/lib.rs | 30 +- lib/crates/fabro-mcp-server/src/server.rs | 297 +------------------ 6 files changed, 190 insertions(+), 311 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index aefa4a469..17eceef72 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2045,11 +2045,9 @@ version = "0.230.0-nightly.0" dependencies = [ "anyhow", "chrono", - "dirs", "fabro-api", "fabro-client", "fabro-config", - "fabro-http", "fabro-manifest", "fabro-server", "fabro-types", diff --git a/lib/crates/fabro-cli/src/commands/mcp/mod.rs b/lib/crates/fabro-cli/src/commands/mcp/mod.rs index 6f047a73f..f6a64a0ca 100644 --- a/lib/crates/fabro-cli/src/commands/mcp/mod.rs +++ b/lib/crates/fabro-cli/src/commands/mcp/mod.rs @@ -4,7 +4,7 @@ use anyhow::{Context as _, Result}; use crate::args::{McpAgent, McpCommand, McpNamespace, ServerConnectionArgs}; use crate::command_context::CommandContext; -use crate::user_config; +use crate::server_client; pub(crate) async fn dispatch(ns: McpNamespace, base_ctx: &CommandContext) -> Result<()> { match ns.command { @@ -28,14 +28,30 @@ fn server_settings( connection: &ServerConnectionArgs, ) -> Result { let connection_ctx = base_ctx.with_connection(connection)?; - let server_target = user_config::resolve_nondefault_server_target( - &connection.target, - connection_ctx.user_settings(), - )?; + let target = connection.target.clone(); + let user_settings = connection_ctx.user_settings().clone(); + let storage_dir = connection_ctx.storage_dir().to_path_buf(); + let base_config_path = connection_ctx.base_config_path().to_path_buf(); + let config_path = base_config_path.clone(); + let client_factory: fabro_mcp_server::FabroClientFactory = std::sync::Arc::new(move || { + let target = target.clone(); + let user_settings = user_settings.clone(); + let storage_dir = storage_dir.clone(); + let base_config_path = base_config_path.clone(); + let future: fabro_mcp_server::FabroClientFuture = Box::pin(async move { + server_client::connect_server_with_settings( + &target, + &user_settings, + &storage_dir, + &base_config_path, + ) + .await + }); + future + }); Ok(fabro_mcp_server::FabroMcpServerSettings { - server_target, - storage_dir: connection_ctx.storage_dir().to_path_buf(), - config_path: connection_ctx.base_config_path().to_path_buf(), + client_factory, + config_path, cwd: base_ctx.cwd().to_path_buf(), }) } diff --git a/lib/crates/fabro-cli/tests/it/cmd/mcp.rs b/lib/crates/fabro-cli/tests/it/cmd/mcp.rs index 8001035f3..8d9e7e834 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/mcp.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/mcp.rs @@ -621,6 +621,112 @@ async fn mcp_run_tools_use_default_local_server_without_server_flag() { .expect("MCP client should shut down"); } +#[tokio::test(flavor = "multi_thread")] +async fn mcp_configured_unix_target_auto_spawns_like_cli() { + let mut context = test_context!(); + let isolated = fabro_test::isolated_storage_dir(); + let storage_dir = isolated.path().join("storage"); + let socket_path = isolated.path().join("configured.sock"); + write_mcp_server_settings(&mut context, &storage_dir, Some(&socket_path)); + let workflow = context.install_fixture("simple.fabro"); + + assert!(!socket_path.exists()); + let client = spawn_mcp_client(&context, &[]).await; + let run_id = create_mcp_run(&client, workflow, false).await; + let search = call_tool_json( + &client, + "fabro_run_search", + serde_json::json!({ "run_ids": [run_id], "first": 1 }), + ) + .await; + + assert_eq!(search["runs"][0]["run_id"], run_id); + assert!( + socket_path.exists(), + "configured Unix socket should be auto-spawned" + ); + client + .shutdown() + .await + .expect("MCP client should shut down"); +} + +#[tokio::test(flavor = "multi_thread")] +async fn mcp_explicit_unix_target_auto_spawns_like_cli() { + let mut context = test_context!(); + let isolated = fabro_test::isolated_storage_dir(); + let storage_dir = isolated.path().join("storage"); + let socket_path = isolated.path().join("explicit.sock"); + write_mcp_server_settings(&mut context, &storage_dir, None); + let workflow = context.install_fixture("simple.fabro"); + let storage_arg = storage_dir.display().to_string(); + let socket_arg = socket_path.display().to_string(); + + assert!(!socket_path.exists()); + let client = spawn_mcp_client(&context, &[ + "--storage-dir", + &storage_arg, + "--server", + &socket_arg, + ]) + .await; + let run_id = create_mcp_run(&client, workflow, false).await; + let search = call_tool_json( + &client, + "fabro_run_search", + serde_json::json!({ "run_ids": [run_id], "first": 1 }), + ) + .await; + + assert_eq!(search["runs"][0]["run_id"], run_id); + assert!( + socket_path.exists(), + "explicit Unix socket should be auto-spawned" + ); + client + .shutdown() + .await + .expect("MCP client should shut down"); +} + +#[tokio::test(flavor = "multi_thread")] +async fn mcp_missing_default_settings_reports_configure_first_error_and_stays_alive() { + let home = tempfile::tempdir().unwrap(); + let workspace = tempfile::tempdir().unwrap(); + let mut env = fabro_test::isolated_env(home.path()); + env.insert( + "FABRO_HOME".to_string(), + home.path().join(".fabro").display().to_string(), + ); + let client = spawn_mcp_client_from_fixture(McpStdioFixture { + command: vec![ + env!("CARGO_BIN_EXE_fabro").to_string(), + "mcp".to_string(), + "start".to_string(), + ], + env, + current_dir: workspace.path().to_path_buf(), + }) + .await; + + let error = call_tool_error_text( + &client, + "fabro_run_search", + serde_json::json!({ "run_ids": ["missing"], "first": 1 }), + ) + .await; + + assert!( + error.contains("Cannot reach Fabro server: no settings.toml configured."), + "{error}" + ); + assert_eq!(client.list_tools().await.unwrap().len(), 5); + client + .shutdown() + .await + .expect("MCP client should shut down"); +} + #[tokio::test(flavor = "multi_thread")] async fn mcp_search_filters_status_dates_and_paginates() { let context = test_context!(); @@ -1975,6 +2081,38 @@ fn mcp_stdio_fixture(context: &fabro_test::TestContext, extra_args: &[&str]) -> } } +fn write_mcp_server_settings( + context: &mut fabro_test::TestContext, + storage_dir: &Path, + socket_path: Option<&Path>, +) { + context.manage_storage_dir(storage_dir); + let cli_target = socket_path.map_or_else(String::new, |path| { + format!( + r#" +[cli.target] +type = "unix" +path = "{}" +"#, + path.display() + ) + }); + context.write_home( + ".fabro/settings.toml", + format!( + r#"_version = 1 + +[server.storage] +root = "{}" + +[server.auth] +methods = ["dev-token"] +{cli_target}"#, + storage_dir.display() + ), + ); +} + async fn spawn_mcp_client(context: &fabro_test::TestContext, extra_args: &[&str]) -> McpClient { let fixture = mcp_stdio_fixture(context, extra_args); spawn_mcp_client_from_fixture(fixture).await diff --git a/lib/crates/fabro-mcp-server/Cargo.toml b/lib/crates/fabro-mcp-server/Cargo.toml index c10f1e0e7..15f481431 100644 --- a/lib/crates/fabro-mcp-server/Cargo.toml +++ b/lib/crates/fabro-mcp-server/Cargo.toml @@ -15,10 +15,8 @@ workspace = true [dependencies] anyhow.workspace = true chrono = { workspace = true, features = ["serde"] } -dirs.workspace = true fabro-api = { path = "../fabro-api" } fabro-client = { path = "../fabro-client" } -fabro-http.workspace = true fabro-manifest = { path = "../fabro-manifest" } fabro-config = { path = "../fabro-config" } fabro-server = { path = "../fabro-server" } diff --git a/lib/crates/fabro-mcp-server/src/lib.rs b/lib/crates/fabro-mcp-server/src/lib.rs index 4c2ed94d3..b8351bab5 100644 --- a/lib/crates/fabro-mcp-server/src/lib.rs +++ b/lib/crates/fabro-mcp-server/src/lib.rs @@ -2,18 +2,36 @@ mod config; mod run_tools; mod server; +use std::future::Future; use std::path::PathBuf; +use std::pin::Pin; +use std::sync::Arc; +use anyhow::Result; pub use config::{config_json, init_agent}; -use fabro_client::ServerTarget; +use fabro_client::Client; pub use server::start; -#[derive(Debug, Clone)] +pub type FabroClientFuture = Pin> + Send>>; + +pub type FabroClientFactory = Arc FabroClientFuture + Send + Sync>; + +#[derive(Clone)] pub struct FabroMcpServerSettings { - pub server_target: Option, - pub storage_dir: PathBuf, - pub config_path: PathBuf, - pub cwd: PathBuf, + pub client_factory: FabroClientFactory, + pub config_path: PathBuf, + pub cwd: PathBuf, +} + +impl std::fmt::Debug for FabroMcpServerSettings { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter + .debug_struct("FabroMcpServerSettings") + .field("client_factory", &"") + .field("config_path", &self.config_path) + .field("cwd", &self.cwd) + .finish() + } } #[derive(Debug, Clone, Default)] diff --git a/lib/crates/fabro-mcp-server/src/server.rs b/lib/crates/fabro-mcp-server/src/server.rs index 74fe1e7e5..e82b2e404 100644 --- a/lib/crates/fabro-mcp-server/src/server.rs +++ b/lib/crates/fabro-mcp-server/src/server.rs @@ -1,33 +1,17 @@ -use std::net::IpAddr; -use std::path::{Path, PathBuf}; +use std::path::PathBuf; use std::sync::Arc; -use std::time::Duration; -use anyhow::{Context as _, Result, anyhow}; -use fabro_client::{ - AuthEntry, AuthStore, Client, Credential, OAuthSession, ServerTarget, TransportConnector, - apply_bearer_token_auth, -}; -use fabro_config::bind::Bind; -use fabro_config::daemon::ServerDaemon; -use fabro_config::{RuntimeDirectory, Storage}; -use fabro_util::dev_token; -use fabro_util::version::FABRO_VERSION; +use anyhow::Result; +use fabro_client::Client; use rmcp::handler::server::router::tool::ToolRouter; use rmcp::handler::server::wrapper::Parameters; use rmcp::model::{CallToolResult, ServerCapabilities, ServerInfo}; use rmcp::transport::stdio; use rmcp::{ErrorData, ServerHandler, serve_server, tool, tool_handler, tool_router}; -use tokio::process::Command as TokioCommand; use tokio::sync::OnceCell; -use tokio::task::yield_now; -use tokio::time::sleep; use crate::{FabroMcpServerSettings, run_tools}; -const CLIENT_REQUEST_TIMEOUT: Duration = Duration::from_secs(30); -const SERVER_START_TIMEOUT: Duration = Duration::from_secs(8); - #[derive(Clone)] pub(crate) struct FabroMcpServer { settings: Arc, @@ -176,7 +160,7 @@ impl FabroMcpServer { async fn client(&self) -> Result, run_tools::ToolError> { self.client .get_or_try_init(|| async { - client_from_settings(&self.settings) + (self.settings.client_factory)() .await .map(Arc::new) .map_err(|err| run_tools::ToolError::from_anyhow(&err)) @@ -185,276 +169,3 @@ impl FabroMcpServer { .map(Arc::clone) } } - -async fn client_from_settings(settings: &FabroMcpServerSettings) -> Result { - yield_now().await; - if let Some(server) = settings.server_target.as_ref() { - return connect_target(server, settings).await; - } - connect_local_server(settings).await -} - -async fn connect_target( - target: &ServerTarget, - settings: &FabroMcpServerSettings, -) -> Result { - let auth_store = AuthStore::default(); - let mut credential = resolve_target_credential_with_store(target, &auth_store)?; - if credential.is_none() && target.is_unix_socket() { - let runtime_token_path = Storage::new(&settings.storage_dir) - .runtime_directory() - .dev_token_path(); - credential = dev_token::read_dev_token_file(&runtime_token_path).map(Credential::DevToken); - } - let oauth_session = refreshable_oauth(target, &auth_store, credential.as_ref()); - let mut builder = Client::builder() - .target(target.clone()) - .transport_connector(target_transport_connector(target.clone())) - .request_timeout(CLIENT_REQUEST_TIMEOUT); - if let Some(credential) = credential { - builder = builder.credential(credential); - } - if let Some(oauth_session) = oauth_session { - builder = builder.oauth_session(oauth_session); - } - builder - .connect() - .await - .context("failed to connect Fabro API") -} - -async fn connect_local_server(settings: &FabroMcpServerSettings) -> Result { - let bind = ensure_local_server_running(&settings.storage_dir, &settings.config_path).await?; - match bind { - Bind::Unix(path) => { - let token = wait_for_runtime_dev_token( - &Storage::new(&settings.storage_dir) - .runtime_directory() - .dev_token_path(), - ) - .await?; - let http_client = connect_bind_http_client(&Bind::Unix(path), Some(&token)).await?; - Client::builder() - .transport("http://fabro", http_client) - .request_timeout(CLIENT_REQUEST_TIMEOUT) - .connect() - .await - } - Bind::Tcp(addr) => { - let target = ServerTarget::http_url(format!("http://{addr}"))?; - let auth_store = AuthStore::default(); - let credential = resolve_target_credential_with_store(&target, &auth_store)?; - let oauth_session = refreshable_oauth(&target, &auth_store, credential.as_ref()); - let mut builder = Client::builder() - .target(target.clone()) - .transport_connector(target_transport_connector(target)) - .request_timeout(CLIENT_REQUEST_TIMEOUT); - if let Some(credential) = credential { - builder = builder.credential(credential); - } - if let Some(oauth_session) = oauth_session { - builder = builder.oauth_session(oauth_session); - } - builder.connect().await - } - } -} - -async fn ensure_local_server_running(storage_dir: &Path, config_path: &Path) -> Result { - let runtime_directory = RuntimeDirectory::new(storage_dir); - if let Some(existing) = ServerDaemon::load_running(&runtime_directory)? { - return Ok(existing.bind); - } - - let exe = std::env::current_exe().context("resolving current fabro executable path")?; - let status = TokioCommand::new(exe) - .args(["server", "start", "--no-web", "--storage-dir"]) - .arg(storage_dir) - .arg("--config") - .arg(config_path) - .stdout(std::process::Stdio::null()) - .stderr(std::process::Stdio::null()) - .stdin(std::process::Stdio::null()) - .status() - .await - .context("starting local Fabro server")?; - if !status.success() { - return Err(anyhow!("fabro server start exited with status {status}")); - } - - let deadline = std::time::Instant::now() + SERVER_START_TIMEOUT; - while std::time::Instant::now() < deadline { - if let Some(running) = ServerDaemon::load_running(&runtime_directory)? { - return Ok(running.bind); - } - sleep(Duration::from_millis(50)).await; - } - Err(anyhow!( - "Fabro server started but no active record was found for {}", - storage_dir.display() - )) -} - -async fn wait_for_runtime_dev_token(path: &Path) -> Result { - let deadline = std::time::Instant::now() + SERVER_START_TIMEOUT; - while std::time::Instant::now() < deadline { - if let Some(token) = dev_token::read_dev_token_file(path) { - return Ok(token); - } - sleep(Duration::from_millis(50)).await; - } - Err(anyhow!( - "runtime dev token did not become available at {}", - path.display() - )) -} - -fn resolve_target_credential_with_store( - target: &ServerTarget, - store: &AuthStore, -) -> Result> { - let Some(entry) = store.get(target)? else { - return Ok(None); - }; - let now = chrono::Utc::now(); - match entry { - AuthEntry::DevToken(entry) => Ok(Some(Credential::DevToken(entry.token))), - AuthEntry::OAuth(entry) - if entry.access_token_expires_at > now || entry.refresh_token_expires_at > now => - { - Ok(Some(Credential::OAuth(entry))) - } - AuthEntry::OAuth(_) => Ok(None), - } -} - -fn refreshable_oauth( - target: &ServerTarget, - auth_store: &AuthStore, - credential: Option<&Credential>, -) -> Option { - matches!(credential, Some(Credential::OAuth(_))) - .then(|| OAuthSession::new(target.clone(), auth_store.clone())) -} - -fn target_transport_connector(target: ServerTarget) -> TransportConnector { - TransportConnector::new(move |bearer_token| { - let target = target.clone(); - async move { connect_target_transport(&target, bearer_token.as_deref()) } - }) -} - -fn connect_target_transport( - target: &ServerTarget, - bearer_token: Option<&str>, -) -> Result<(fabro_http::HttpClient, String)> { - if let Some(api_url) = target.as_http_url() { - let mut builder = cli_compatible_http_client_builder(); - if should_bypass_proxy_for_http_target(api_url) { - builder = builder.no_proxy(); - } - if let Some(token) = bearer_token { - builder = apply_bearer_token_auth(builder, token)?; - } - return Ok((builder.build()?, api_url.to_string())); - } - - let Some(path) = target.as_unix_socket_path() else { - return Err(anyhow!( - "server target must be an http(s) URL or absolute Unix socket path" - )); - }; - let mut builder = fabro_http::HttpClientBuilder::new() - .unix_socket(path) - .no_proxy(); - if let Some(token) = bearer_token { - builder = apply_bearer_token_auth(builder, token)?; - } - Ok((builder.build()?, "http://fabro".to_string())) -} - -fn cli_compatible_http_client_builder() -> fabro_http::HttpClientBuilder { - fabro_http::HttpClientBuilder::new().user_agent(format!("fabro-cli/{FABRO_VERSION}")) -} - -#[expect( - clippy::disallowed_types, - reason = "Proxy bypass classification parses a configured raw API target and does not log credential-bearing URLs." -)] -fn should_bypass_proxy_for_http_target(api_url: &str) -> bool { - let Ok(url) = fabro_http::Url::parse(api_url) else { - return false; - }; - let Some(host) = url.host_str() else { - return false; - }; - if host.eq_ignore_ascii_case("localhost") { - return true; - } - host.trim_matches(['[', ']']) - .parse::() - .is_ok_and(|ip| ip.is_loopback()) -} - -async fn connect_bind_http_client( - bind: &Bind, - bearer_token: Option<&str>, -) -> Result { - let (client, health_url) = match bind { - Bind::Unix(path) => { - let mut builder = fabro_http::HttpClientBuilder::new() - .unix_socket(path) - .no_proxy(); - if let Some(token) = bearer_token { - builder = apply_bearer_token_auth(builder, token)?; - } - (builder.build()?, "http://fabro/health".to_string()) - } - Bind::Tcp(addr) => { - let mut builder = fabro_http::HttpClientBuilder::new().no_proxy(); - if let Some(token) = bearer_token { - builder = apply_bearer_token_auth(builder, token)?; - } - (builder.build()?, format!("http://{addr}/health")) - } - }; - let deadline = std::time::Instant::now() + SERVER_START_TIMEOUT; - let mut last_error = None; - while std::time::Instant::now() < deadline { - match client.get(&health_url).send().await { - Ok(response) if response.status().is_success() => return Ok(client), - Ok(response) => last_error = Some(anyhow!("health returned {}", response.status())), - Err(err) => last_error = Some(anyhow!(err)), - } - sleep(Duration::from_millis(50)).await; - } - Err(last_error.unwrap_or_else(|| anyhow!("Fabro server did not become ready in time"))) -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn explicit_http_proxy_bypass_matches_cli_for_local_targets() { - assert!(should_bypass_proxy_for_http_target( - "http://localhost:3000/api/v1" - )); - assert!(should_bypass_proxy_for_http_target( - "http://127.0.0.1:3000/api/v1" - )); - assert!(should_bypass_proxy_for_http_target( - "http://[::1]:3000/api/v1" - )); - } - - #[test] - fn explicit_http_proxy_bypass_matches_cli_for_remote_targets() { - assert!(!should_bypass_proxy_for_http_target( - "https://fabro.example.test/api/v1" - )); - assert!(!should_bypass_proxy_for_http_target( - "http://192.0.2.44:3000/api/v1" - )); - } -}