feat(cli): add mcp run create and search tools

This commit is contained in:
Bryan Helmkamp 2026-05-11 08:15:52 -04:00
parent f425e51acb
commit e6b891147c
No known key found for this signature in database
5 changed files with 533 additions and 17 deletions

5
Cargo.lock generated
View file

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

View file

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

View file

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

View file

@ -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<FabroRunCreateParams> for ValidatedCreateRuns {
fn try_from(params: FabroRunCreateParams) -> Result<Self, Self::Error> {
validate_len("runs", params.runs.len(), 1, 50)?;
for spec in &params.runs {
for (key, value) in &spec.inputs {
json_to_toml_value(key, value)?;
}
}
Ok(Self { runs: params.runs })
}
}
@ -113,6 +122,12 @@ impl TryFrom<FabroRunSearchParams> 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<Client>,
_base_cwd: &Path,
_params: ValidatedCreateRuns,
client: Arc<Client>,
base_cwd: &Path,
params: ValidatedCreateRuns,
) -> ToolResult<CreateRunsResult> {
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<Client>,
_params: ValidatedSearchRuns,
client: Arc<Client>,
params: ValidatedSearchRuns,
) -> ToolResult<SearchRunsResult> {
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::<Vec<_>>();
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<types::RunManifest> {
if let Some(run_id) = spec.run_id.as_deref() {
run_id.parse::<RunId>().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<types::ManifestArgs> {
let label = spec
.labels
.iter()
.map(|(key, value)| format!("{key}={value}"))
.collect::<Vec<_>>();
let input = spec
.inputs
.iter()
.map(|(key, value)| format!("{key}={value}"))
.collect::<Vec<_>>();
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<toml::Value> {
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::<ToolResult<Vec<_>>>()
.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<DateTime<Utc>> {
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"));
}
}

View file

@ -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<Client> {
async fn client_from_settings(settings: &McpServerSettings) -> Result<Client> {
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()))
}