diff --git a/Cargo.lock b/Cargo.lock index 6f7ded875..c946b7520 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2025,13 +2025,18 @@ name = "fabro-mcp-server" version = "0.230.0-nightly.0" dependencies = [ "anyhow", + "chrono", "dirs", + "fabro-api", "fabro-client", + "fabro-http", + "fabro-types", "rmcp", "schemars 1.2.1", "serde", "serde_json", "tokio", + "toml 0.8.23", ] [[package]] diff --git a/lib/crates/fabro-cli/tests/it/cmd/mcp.rs b/lib/crates/fabro-cli/tests/it/cmd/mcp.rs index cb0a5f61b..4635a515a 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/mcp.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/mcp.rs @@ -16,6 +16,8 @@ use fabro_mcp::client::McpClient; use fabro_mcp::config::{McpServerSettings, McpTransport}; use fabro_test::{fabro_json_snapshot, fabro_snapshot, test_context}; +use crate::support::{RealAuthHarness, TEST_DEV_TOKEN, seed_dev_token_auth}; + #[test] fn help() { let context = test_context!(); @@ -417,6 +419,71 @@ async fn stdio_startup_and_list_tools_is_fast() { assert!(start.elapsed() < std::time::Duration::from_secs(2)); } +#[tokio::test(flavor = "multi_thread")] +async fn mcp_create_and_search_manage_real_runs_with_cli_auth() { + let context = test_context!(); + let harness = + RealAuthHarness::start_with_dev_token(fabro_test::GitHubAppState::default()).await; + let target_url = harness.api_target(); + let target: fabro_client::ServerTarget = target_url.parse().unwrap(); + seed_dev_token_auth(&context.home_dir, &target, TEST_DEV_TOKEN); + let workflow = context.install_fixture("simple.fabro"); + + let client = spawn_mcp_client(&context, &["--server", &target_url]).await; + + let create = call_tool_json( + &client, + "fabro_run_create", + serde_json::json!({ + "runs": [{ + "workflow": workflow, + "dry_run": true, + "auto_approve": true, + "labels": { "source": "mcp-test" } + }] + }), + ) + .await; + let run_id = create["runs"][0]["run_id"].as_str().unwrap().to_string(); + assert_eq!(create["runs"][0]["started"], true); + + let search = call_tool_json( + &client, + "fabro_run_search", + serde_json::json!({ + "run_ids": [run_id], + "labels": { "source": "mcp-test" }, + "first": 10 + }), + ) + .await; + fabro_json_snapshot!(context, normalize_run_search(search), @r#" + { + "runs": [ + { + "run_id": "[RUN_ID]", + "workflow_name": "Simple", + "workflow_slug": "simple", + "status": "queued", + "archived": false, + "created_at": "[TIMESTAMP]", + "started_at": null, + "completed_at": null, + "labels": { + "source": "mcp-test" + }, + "source_directory": "[SOURCE_DIRECTORY]", + "repo_origin_url": null, + "goal": "Run the Fabro workflow." + } + ], + "next_cursor": null + } + "#); + + harness.shutdown().await; +} + fn expected_claude_config_path(home_dir: &Path) -> PathBuf { #[cfg(target_os = "macos")] { @@ -488,3 +555,48 @@ async fn spawn_mcp_client(context: &fabro_test::TestContext, extra_args: &[&str] .expect("MCP server should initialize"); client } + +async fn call_tool_json( + client: &McpClient, + name: &str, + arguments: serde_json::Value, +) -> serde_json::Value { + let result = client + .call_tool(name, arguments, std::time::Duration::from_secs(30)) + .await + .expect("tool call should complete"); + assert_ne!( + result.is_error, + Some(true), + "tool returned error: {result:?}" + ); + let text = result + .content + .first() + .and_then(|content| serde_json::to_value(content).ok()) + .and_then(|content| content["text"].as_str().map(ToOwned::to_owned)) + .expect("tool result should include text fallback"); + assert!(!text.starts_with('{') && !text.starts_with('[')); + result + .structured_content + .expect("tool result should include structured content") +} + +fn normalize_run_search(mut value: serde_json::Value) -> serde_json::Value { + if let Some(runs) = value["runs"].as_array_mut() { + for run in runs { + run["run_id"] = serde_json::json!("[RUN_ID]"); + run["created_at"] = serde_json::json!("[TIMESTAMP]"); + if run["started_at"].is_string() { + run["started_at"] = serde_json::json!("[TIMESTAMP]"); + } + if run["completed_at"].is_string() { + run["completed_at"] = serde_json::json!("[TIMESTAMP]"); + } + if run["source_directory"].is_string() { + run["source_directory"] = serde_json::json!("[SOURCE_DIRECTORY]"); + } + } + } + value +} diff --git a/lib/crates/fabro-mcp-server/Cargo.toml b/lib/crates/fabro-mcp-server/Cargo.toml index 963d8aabc..b391116d4 100644 --- a/lib/crates/fabro-mcp-server/Cargo.toml +++ b/lib/crates/fabro-mcp-server/Cargo.toml @@ -14,10 +14,15 @@ 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-types = { path = "../fabro-types" } rmcp = { workspace = true, features = ["server", "macros", "schemars", "transport-io"] } schemars = "1.2.1" serde.workspace = true serde_json.workspace = true tokio.workspace = true +toml.workspace = true diff --git a/lib/crates/fabro-mcp-server/src/run_tools.rs b/lib/crates/fabro-mcp-server/src/run_tools.rs index 7164f5885..97684c20e 100644 --- a/lib/crates/fabro-mcp-server/src/run_tools.rs +++ b/lib/crates/fabro-mcp-server/src/run_tools.rs @@ -7,11 +7,15 @@ use std::collections::HashMap; use std::path::{Path, PathBuf}; use std::sync::Arc; +use chrono::{DateTime, NaiveDate, Utc}; +use fabro_api::types; use fabro_client::Client; +use fabro_types::{Run, RunId, RunStatus}; use rmcp::model::{CallToolResult, Content}; use schemars::JsonSchema; use serde::{Deserialize, Serialize}; use serde_json::Value; +use tokio::fs; use tokio::task::yield_now; #[derive(Debug)] @@ -71,6 +75,11 @@ impl TryFrom for ValidatedCreateRuns { fn try_from(params: FabroRunCreateParams) -> Result { validate_len("runs", params.runs.len(), 1, 50)?; + for spec in ¶ms.runs { + for (key, value) in &spec.inputs { + json_to_toml_value(key, value)?; + } + } Ok(Self { runs: params.runs }) } } @@ -113,6 +122,12 @@ impl TryFrom for ValidatedSearchRuns { if params.first.is_some_and(|first| first > 100) { return Err(ToolError::message("first must be <= 100")); } + if let Some(created_after) = params.created_after.as_deref() { + parse_datetime_filter("created_after", created_after)?; + } + if let Some(created_before) = params.created_before.as_deref() { + parse_datetime_filter("created_before", created_before)?; + } Ok(Self { raw: params }) } } @@ -300,24 +315,105 @@ pub(crate) struct RunEventResult { } pub(crate) async fn create_runs( - _client: Arc, - _base_cwd: &Path, - _params: ValidatedCreateRuns, + client: Arc, + base_cwd: &Path, + params: ValidatedCreateRuns, ) -> ToolResult { - yield_now().await; - Err(ToolError::message( - "fabro_run_create is not implemented yet", - )) + let mut created = Vec::with_capacity(params.runs.len()); + for spec in params.runs { + let cwd = spec.cwd.clone().unwrap_or_else(|| base_cwd.to_path_buf()); + let manifest = build_run_manifest(&spec, &cwd).await?; + let run_id = client + .create_run_from_manifest(manifest) + .await + .map_err(|err| ToolError::from_anyhow(&err))?; + let started = spec.start.unwrap_or(true); + if started { + client + .start_run(&run_id, false) + .await + .map_err(|err| ToolError::from_anyhow(&err))?; + } + let summary = client + .retrieve_run(&run_id) + .await + .map_err(|err| ToolError::from_anyhow(&err))?; + created.push(CreatedRunResult { + run_id: summary.id.to_string(), + workflow: spec.workflow, + started, + status: run_status_kind(summary.lifecycle.status).to_string(), + }); + } + Ok(CreateRunsResult { runs: created }) } pub(crate) async fn search_runs( - _client: Arc, - _params: ValidatedSearchRuns, + client: Arc, + params: ValidatedSearchRuns, ) -> ToolResult { - yield_now().await; - Err(ToolError::message( - "fabro_run_search is not implemented yet", - )) + let raw = params.raw; + let mut runs = client + .list_store_runs() + .await + .map_err(|err| ToolError::from_anyhow(&err))?; + runs.sort_by(|a, b| { + b.timestamps + .created_at + .cmp(&a.timestamps.created_at) + .then_with(|| b.id.to_string().cmp(&a.id.to_string())) + }); + + if let Some(after) = raw.after.as_deref() { + if let Some(position) = runs.iter().position(|run| run.id.to_string() == after) { + runs = runs.into_iter().skip(position + 1).collect(); + } + } + + if let Some(run_ids) = raw.run_ids.as_ref() { + runs.retain(|run| run_ids.iter().any(|id| id == &run.id.to_string())); + } + if let Some(workflow) = raw.workflow.as_deref() { + runs.retain(|run| { + run.workflow.name == workflow || run.workflow.slug.as_deref() == Some(workflow) + }); + } + if let Some(labels) = raw.labels.as_ref() { + runs.retain(|run| { + labels + .iter() + .all(|(key, value)| run.labels.get(key) == Some(value)) + }); + } + if let Some(status) = raw.status.as_ref() { + runs.retain(|run| { + status + .iter() + .any(|status| status == run_status_kind(run.lifecycle.status)) + }); + } + if let Some(archived) = raw.archived { + runs.retain(|run| run.lifecycle.archived == archived); + } + if let Some(created_after) = raw.created_after.as_deref() { + let cutoff = parse_datetime_filter("created_after", created_after)?; + runs.retain(|run| run.timestamps.created_at >= cutoff); + } + if let Some(created_before) = raw.created_before.as_deref() { + let cutoff = parse_datetime_filter("created_before", created_before)?; + runs.retain(|run| run.timestamps.created_at <= cutoff); + } + + let first = raw.first.unwrap_or(20).min(100); + let has_more = runs.len() > first; + let page = runs.into_iter().take(first).collect::>(); + let next_cursor = has_more + .then(|| page.last().map(|run| run.id.to_string())) + .flatten(); + Ok(SearchRunsResult { + runs: page.iter().map(run_summary_result).collect(), + next_cursor, + }) } pub(crate) async fn interact_run( @@ -417,3 +513,239 @@ fn validate_len(name: &str, len: usize, min: usize, max: usize) -> ToolResult<() fn format_tool_error(err: &anyhow::Error) -> String { format!("{err:#}") } + +async fn build_run_manifest(spec: &CreateRunSpec, cwd: &Path) -> ToolResult { + if let Some(run_id) = spec.run_id.as_deref() { + run_id.parse::().map_err(|err| { + ToolError::message(format!("run_id must be a valid Fabro run id: {err}")) + })?; + } + let workflow_path = resolve_workflow_path(&spec.workflow, cwd); + let manifest_cwd = manifest_cwd_for_workflow(cwd, &workflow_path); + let workflow_key = workflow_path + .strip_prefix(&manifest_cwd) + .unwrap_or(&workflow_path) + .display() + .to_string(); + let source = fs::read_to_string(&workflow_path).await.map_err(|err| { + ToolError::message(format!( + "failed to read workflow {}: {err}", + workflow_path.display() + )) + })?; + let workflows = HashMap::from([(workflow_key.clone(), types::ManifestWorkflow { + config: None, + files: HashMap::new(), + source, + })]); + Ok(types::RunManifest { + args: mcp_manifest_args(spec), + configs: Vec::new(), + cwd: manifest_cwd.display().to_string(), + git: None, + goal: Some(types::ManifestGoal { + path: None, + text: spec + .goal + .clone() + .unwrap_or_else(|| "Run the Fabro workflow.".to_string()), + type_: types::ManifestGoalType::Value, + }), + run_id: spec.run_id.clone(), + target: types::ManifestTarget { + identifier: spec.workflow.clone(), + path: workflow_key, + }, + title: None, + version: 1, + workflows, + }) +} + +fn resolve_workflow_path(workflow: &str, cwd: &Path) -> PathBuf { + let path = PathBuf::from(workflow); + if path.is_absolute() { + path + } else { + cwd.join(path) + } +} + +fn manifest_cwd_for_workflow(cwd: &Path, workflow_path: &Path) -> PathBuf { + if workflow_path.strip_prefix(cwd).is_ok() { + cwd.to_path_buf() + } else { + workflow_path + .parent() + .map_or_else(|| cwd.to_path_buf(), Path::to_path_buf) + } +} + +fn mcp_manifest_args(spec: &CreateRunSpec) -> Option { + let label = spec + .labels + .iter() + .map(|(key, value)| format!("{key}={value}")) + .collect::>(); + let input = spec + .inputs + .iter() + .map(|(key, value)| format!("{key}={value}")) + .collect::>(); + let payload = types::ManifestArgs { + auto_approve: spec.auto_approve.filter(|value| *value), + docker_image: None, + dry_run: spec.dry_run.filter(|value| *value), + input, + label, + model: spec.model.clone(), + preserve_sandbox: spec.preserve_sandbox.filter(|value| *value), + provider: spec.provider.clone(), + sandbox: spec.sandbox.clone(), + verbose: None, + }; + (!mcp_manifest_args_is_empty(&payload)).then_some(payload) +} + +fn mcp_manifest_args_is_empty(args: &types::ManifestArgs) -> bool { + args.auto_approve.is_none() + && args.docker_image.is_none() + && args.dry_run.is_none() + && args.input.is_empty() + && args.label.is_empty() + && args.model.is_none() + && args.preserve_sandbox.is_none() + && args.provider.is_none() + && args.sandbox.is_none() + && args.verbose.is_none() +} + +fn json_to_toml_value(key: &str, value: &Value) -> ToolResult { + match value { + Value::Null => Err(ToolError::message(format!( + "input `{key}` cannot be null; use a string, boolean, number, array, or object" + ))), + Value::Bool(value) => Ok(toml::Value::Boolean(*value)), + Value::Number(value) => { + if let Some(integer) = value.as_i64() { + Ok(toml::Value::Integer(integer)) + } else if let Some(float) = value.as_f64() { + Ok(toml::Value::Float(float)) + } else { + Err(ToolError::message(format!( + "input `{key}` contains a number outside TOML's supported range" + ))) + } + } + Value::String(value) => Ok(toml::Value::String(value.clone())), + Value::Array(values) => values + .iter() + .map(|value| json_to_toml_value(key, value)) + .collect::>>() + .map(toml::Value::Array), + Value::Object(values) => { + let mut table = toml::Table::new(); + for (child_key, child_value) in values { + table.insert(child_key.clone(), json_to_toml_value(key, child_value)?); + } + Ok(toml::Value::Table(table)) + } + } +} + +fn run_summary_result(run: &Run) -> RunSummaryResult { + RunSummaryResult { + run_id: run.id.to_string(), + workflow_name: run.workflow.name.clone(), + workflow_slug: run.workflow.slug.clone(), + status: run_status_kind(run.lifecycle.status).to_string(), + archived: run.lifecycle.archived, + created_at: run.timestamps.created_at.to_rfc3339(), + started_at: run + .timestamps + .started_at + .map(|timestamp| timestamp.to_rfc3339()), + completed_at: run + .timestamps + .completed_at + .map(|timestamp| timestamp.to_rfc3339()), + labels: run.labels.clone(), + source_directory: run.source_directory.clone(), + repo_origin_url: run + .repository + .as_ref() + .and_then(|repository| repository.origin_url.clone()), + goal: run.goal.clone(), + } +} + +fn parse_datetime_filter(name: &str, raw: &str) -> ToolResult> { + if let Ok(timestamp) = DateTime::parse_from_rfc3339(raw) { + return Ok(timestamp.with_timezone(&Utc)); + } + let date = NaiveDate::parse_from_str(raw, "%Y-%m-%d").map_err(|err| { + ToolError::message(format!("{name} must be RFC3339 or YYYY-MM-DD: {err}")) + })?; + let datetime = date + .and_hms_opt(0, 0, 0) + .ok_or_else(|| ToolError::message(format!("{name} contains an invalid date")))?; + Ok(DateTime::from_naive_utc_and_offset(datetime, Utc)) +} + +fn run_status_kind(status: RunStatus) -> &'static str { + match status { + RunStatus::Submitted => "submitted", + RunStatus::Queued => "queued", + RunStatus::Starting => "starting", + RunStatus::Running => "running", + RunStatus::Blocked { .. } => "blocked", + RunStatus::Paused { .. } => "paused", + RunStatus::Removing => "removing", + RunStatus::Succeeded { .. } => "succeeded", + RunStatus::Failed { .. } => "failed", + RunStatus::Dead => "dead", + } +} + +#[cfg(test)] +mod tests { + use serde_json::json; + + use super::*; + + #[test] + fn json_inputs_convert_to_toml_values() { + let cases = [ + (json!("hello"), toml::Value::String("hello".to_string())), + (json!(true), toml::Value::Boolean(true)), + (json!(42), toml::Value::Integer(42)), + (json!(0.5), toml::Value::Float(0.5)), + ( + json!(["a", 1]), + toml::Value::Array(vec![ + toml::Value::String("a".to_string()), + toml::Value::Integer(1), + ]), + ), + ( + json!({ "enabled": true, "count": 2 }), + toml::Value::Table(toml::Table::from_iter([ + ("enabled".to_string(), toml::Value::Boolean(true)), + ("count".to_string(), toml::Value::Integer(2)), + ])), + ), + ]; + + for (json, expected) in cases { + assert_eq!(json_to_toml_value("input", &json).unwrap(), expected); + } + } + + #[test] + fn json_input_null_is_rejected_with_key_name() { + let err = json_to_toml_value("goal", &Value::Null).unwrap_err(); + + assert!(err.as_str().contains("goal")); + assert!(err.as_str().contains("null")); + } +} diff --git a/lib/crates/fabro-mcp-server/src/server.rs b/lib/crates/fabro-mcp-server/src/server.rs index 59b18c6c1..f4f6e66b9 100644 --- a/lib/crates/fabro-mcp-server/src/server.rs +++ b/lib/crates/fabro-mcp-server/src/server.rs @@ -1,8 +1,11 @@ use std::path::PathBuf; use std::sync::Arc; -use anyhow::{Result, anyhow}; -use fabro_client::Client; +use anyhow::{Context as _, Result, anyhow}; +use fabro_client::{ + AuthEntry, AuthStore, Client, Credential, ServerTarget, TransportConnector, + apply_bearer_token_auth, +}; use rmcp::handler::server::router::tool::ToolRouter; use rmcp::handler::server::wrapper::Parameters; use rmcp::model::{CallToolResult, ServerCapabilities, ServerInfo}; @@ -171,7 +174,66 @@ impl FabroMcpServer { } } -async fn client_from_settings(_settings: &McpServerSettings) -> Result { +async fn client_from_settings(settings: &McpServerSettings) -> Result { yield_now().await; - Err(anyhow!("fabro MCP API client is not implemented yet")) + let Some(server) = settings.config.server.as_ref() else { + return Err(anyhow!( + "fabro mcp start requires --server for run tools in this release" + )); + }; + let target: ServerTarget = server.parse()?; + let credential = AuthStore::new(settings.home_dir.join(".fabro").join("auth.json")) + .get(&target)? + .map(credential_from_auth_entry); + let mut builder = Client::builder() + .target(target.clone()) + .transport_connector(target_transport_connector(target)) + .request_timeout(std::time::Duration::from_secs(30)); + if let Some(credential) = credential { + builder = builder.credential(credential); + } + builder + .connect() + .await + .context("failed to connect Fabro API") +} + +fn credential_from_auth_entry(entry: AuthEntry) -> Credential { + match entry { + AuthEntry::OAuth(entry) => Credential::OAuth(entry), + AuthEntry::DevToken(entry) => Credential::DevToken(entry.token), + } +} + +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 = fabro_http::HttpClientBuilder::new().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())) }