scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-09-07 08:27:12 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

fix(mcp): reuse CLI server targeting

Browse source 
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.
This commit is contained in:
Bryan Helmkamp 2026-05-11 16:11:32 -04:00
parent 6d016d69a1
commit 55ff3d0601
No known key found for this signature in database
6 changed files with 190 additions and 311 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
Cargo.lock generated
View file

@ -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",

32
lib/crates/fabro-cli/src/commands/mcp/mod.rs
View file

@ -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<fabro_mcp_server::FabroMcpServerSettings> {
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(),
})
}

138
lib/crates/fabro-cli/tests/it/cmd/mcp.rs
View file

@ -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

2
lib/crates/fabro-mcp-server/Cargo.toml
View file

@ -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" }

30
lib/crates/fabro-mcp-server/src/lib.rs
View file

@ -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<Box<dyn Future<Output = Result<Client>> + Send>>;
pub type FabroClientFactory = Arc<dyn Fn() -> FabroClientFuture + Send + Sync>;
#[derive(Clone)]
pub struct FabroMcpServerSettings {
pub server_target: Option<ServerTarget>,
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", &"<factory>")
.field("config_path", &self.config_path)
.field("cwd", &self.cwd)
.finish()
}
}
#[derive(Debug, Clone, Default)]

297
lib/crates/fabro-mcp-server/src/server.rs
View file

@ -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<FabroMcpServerSettings>,
@ -176,7 +160,7 @@ impl FabroMcpServer {
async fn client(&self) -> Result<Arc<Client>, 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<Client> {
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<Client> {
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<Client> {
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<Bind> {
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<String> {
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<Option<Credential>> {
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<OAuthSession> {
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::<IpAddr>()
.is_ok_and(|ip| ip.is_loopback())
}
async fn connect_bind_http_client(
bind: &Bind,
bearer_token: Option<&str>,
) -> Result<fabro_http::HttpClient> {
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"
));
}
}