diff --git a/docs/public/api-reference/fabro-api.yaml b/docs/public/api-reference/fabro-api.yaml index 7e0484dc0..80561ae6c 100644 --- a/docs/public/api-reference/fabro-api.yaml +++ b/docs/public/api-reference/fabro-api.yaml @@ -4266,6 +4266,9 @@ paths: responses: "201": description: MCP server created + headers: + ETag: + $ref: "#/components/headers/ETag" content: application/json: schema: diff --git a/lib/crates/fabro-config/src/builders.rs b/lib/crates/fabro-config/src/builders.rs index 7e3b206e8..0b411f113 100644 --- a/lib/crates/fabro-config/src/builders.rs +++ b/lib/crates/fabro-config/src/builders.rs @@ -1,8 +1,9 @@ -use std::collections::BTreeMap; +use std::collections::{BTreeMap, HashMap}; use std::fmt; use std::path::Path; use fabro_model::catalog as model_catalog; +use fabro_types::settings::run::McpServerSettings; use fabro_types::settings::{RunNamespace, WorkflowNamespace}; use fabro_types::{ServerSettings, UserSettings, WorkflowSettings}; use fabro_util::error::SharedError; @@ -210,6 +211,7 @@ impl RunSettingsBuilder { let run = resolve_run( &layer.run.clone().unwrap_or_default(), &layer.environments, + &HashMap::new(), &mut errors, ); finish_result(run, "failed to resolve run settings", errors) @@ -444,11 +446,12 @@ fn parse_settings_toml(source: &str, kind: SettingsSource) -> Result, } impl WorkflowSettingsBuilder { @@ -561,6 +564,12 @@ impl WorkflowSettingsBuilder { }) } + #[must_use] + pub fn server_mcp_catalog(mut self, catalog: HashMap) -> Self { + self.mcp_server_catalog = catalog; + self + } + #[must_use] pub fn run_overrides(self, run: RunLayer) -> Self { self.args_layer(SettingsLayer { @@ -598,11 +607,20 @@ impl WorkflowSettingsBuilder { } pub fn build(self) -> std::result::Result { - Self::from_layer(&self.build_layer()) + let mcp_server_catalog = self.mcp_server_catalog.clone(); + let layer = self.build_layer(); + Self::from_layer_with_mcp_server_catalog(&layer, &mcp_server_catalog) } pub(crate) fn from_layer( layer: &SettingsLayer, + ) -> std::result::Result { + Self::from_layer_with_mcp_server_catalog(layer, &HashMap::new()) + } + + fn from_layer_with_mcp_server_catalog( + layer: &SettingsLayer, + mcp_server_catalog: &HashMap, ) -> std::result::Result { let layer = layer.clone().combine(DEFAULTS_LAYER.clone()); let mut errors = Vec::new(); @@ -611,6 +629,7 @@ impl WorkflowSettingsBuilder { let run = resolve_run( &layer.run.clone().unwrap_or_default(), &layer.environments, + mcp_server_catalog, &mut errors, ); finish_dense_result( diff --git a/lib/crates/fabro-config/src/layers/run.rs b/lib/crates/fabro-config/src/layers/run.rs index cc849fb23..d02ad1289 100644 --- a/lib/crates/fabro-config/src/layers/run.rs +++ b/lib/crates/fabro-config/src/layers/run.rs @@ -6,7 +6,8 @@ use fabro_types::settings::run::{ AgentPermissions, ApprovalMode, HookEvent, McpHttpProtocol, MergeStrategy, RunMode, }; use fabro_types::settings::{Duration, InterpString, ModelRef}; -use serde::{Deserialize, Serialize}; +use serde::de::{self, Deserializer}; +use serde::{Deserialize, Serialize, Serializer}; use super::combine::Combine; use super::environment::RunEnvironmentLayer; @@ -418,53 +419,141 @@ pub struct RunAgentLayer { /// A single MCP entry. `type` selects the transport; `script`/`command` are /// mutually exclusive for process-launching transports. Non-launching HTTP /// transports use neither field. -#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] -#[serde(deny_unknown_fields, tag = "type", rename_all = "snake_case")] +#[derive(Debug, Clone, PartialEq)] pub enum McpEntryLayer { Http { - #[serde(default)] enabled: Option, - #[serde(default)] protocol: McpHttpProtocol, url: InterpString, - #[serde(default)] headers: HashMap, - #[serde(default)] startup_timeout: Option, - #[serde(default)] tool_timeout: Option, }, Stdio { - #[serde(default)] enabled: Option, - #[serde(default)] script: Option, - #[serde(default)] command: Option>, - #[serde(default)] env: HashMap, - #[serde(default)] startup_timeout: Option, - #[serde(default)] tool_timeout: Option, }, Sandbox { - #[serde(default)] enabled: Option, - #[serde(default)] protocol: McpHttpProtocol, - #[serde(default)] script: Option, - #[serde(default)] command: Option>, port: u16, - #[serde(default)] env: HashMap, - #[serde(default)] startup_timeout: Option, - #[serde(default)] tool_timeout: Option, }, + Reference { + id: String, + enabled: Option, + }, +} + +impl Serialize for McpEntryLayer { + fn serialize(&self, serializer: S) -> Result + where + S: Serializer, + { + #[derive(Serialize)] + #[serde(tag = "type", rename_all = "snake_case")] + enum TaggedMcpEntryLayer<'a> { + Http { + enabled: &'a Option, + protocol: &'a McpHttpProtocol, + url: &'a InterpString, + headers: &'a HashMap, + startup_timeout: &'a Option, + tool_timeout: &'a Option, + }, + Stdio { + enabled: &'a Option, + script: &'a Option, + command: &'a Option>, + env: &'a HashMap, + startup_timeout: &'a Option, + tool_timeout: &'a Option, + }, + Sandbox { + enabled: &'a Option, + protocol: &'a McpHttpProtocol, + script: &'a Option, + command: &'a Option>, + port: &'a u16, + env: &'a HashMap, + startup_timeout: &'a Option, + tool_timeout: &'a Option, + }, + } + + #[derive(Serialize)] + struct McpReferenceLayer<'a> { + id: &'a str, + #[serde(skip_serializing_if = "Option::is_none")] + enabled: &'a Option, + } + + match self { + Self::Http { + enabled, + protocol, + url, + headers, + startup_timeout, + tool_timeout, + } => TaggedMcpEntryLayer::Http { + enabled, + protocol, + url, + headers, + startup_timeout, + tool_timeout, + } + .serialize(serializer), + Self::Stdio { + enabled, + script, + command, + env, + startup_timeout, + tool_timeout, + } => TaggedMcpEntryLayer::Stdio { + enabled, + script, + command, + env, + startup_timeout, + tool_timeout, + } + .serialize(serializer), + Self::Sandbox { + enabled, + protocol, + script, + command, + port, + env, + startup_timeout, + tool_timeout, + } => TaggedMcpEntryLayer::Sandbox { + enabled, + protocol, + script, + command, + port, + env, + startup_timeout, + tool_timeout, + } + .serialize(serializer), + Self::Reference { id, enabled } => { + McpReferenceLayer { id, enabled }.serialize(serializer) + } + } + } } impl McpEntryLayer { @@ -475,12 +564,161 @@ impl McpEntryLayer { let enabled = match self { Self::Http { enabled, .. } | Self::Stdio { enabled, .. } - | Self::Sandbox { enabled, .. } => enabled, + | Self::Sandbox { enabled, .. } + | Self::Reference { enabled, .. } => enabled, }; enabled.unwrap_or(true) } } +impl<'de> Deserialize<'de> for McpEntryLayer { + fn deserialize(deserializer: D) -> Result + where + D: Deserializer<'de>, + { + let value = serde_json::Value::deserialize(deserializer)?; + let serde_json::Value::Object(map) = &value else { + return Err(de::Error::custom("MCP entry must be a table")); + }; + + let has_reference_fields = map.contains_key("id"); + let has_inline_fields = map.contains_key("type") + || map.contains_key("script") + || map.contains_key("command") + || map.contains_key("url") + || map.contains_key("headers") + || map.contains_key("port") + || map.contains_key("env") + || map.contains_key("startup_timeout") + || map.contains_key("tool_timeout"); + + if has_reference_fields && has_inline_fields { + return Err(de::Error::custom( + "MCP entry cannot mix catalog reference fields (`id`, `enabled`) with inline \ + server fields", + )); + } + + if has_reference_fields { + #[derive(Deserialize)] + #[serde(deny_unknown_fields)] + struct McpReferenceLayer { + id: String, + #[serde(default)] + enabled: Option, + } + + let reference: McpReferenceLayer = + serde_json::from_value(value).map_err(de::Error::custom)?; + return Ok(Self::Reference { + id: reference.id, + enabled: reference.enabled, + }); + } + + #[derive(Deserialize)] + #[serde(deny_unknown_fields, tag = "type", rename_all = "snake_case")] + enum TaggedMcpEntryLayer { + Http { + #[serde(default)] + enabled: Option, + #[serde(default)] + protocol: McpHttpProtocol, + url: InterpString, + #[serde(default)] + headers: HashMap, + #[serde(default)] + startup_timeout: Option, + #[serde(default)] + tool_timeout: Option, + }, + Stdio { + #[serde(default)] + enabled: Option, + #[serde(default)] + script: Option, + #[serde(default)] + command: Option>, + #[serde(default)] + env: HashMap, + #[serde(default)] + startup_timeout: Option, + #[serde(default)] + tool_timeout: Option, + }, + Sandbox { + #[serde(default)] + enabled: Option, + #[serde(default)] + protocol: McpHttpProtocol, + #[serde(default)] + script: Option, + #[serde(default)] + command: Option>, + port: u16, + #[serde(default)] + env: HashMap, + #[serde(default)] + startup_timeout: Option, + #[serde(default)] + tool_timeout: Option, + }, + } + + match serde_json::from_value(value).map_err(de::Error::custom)? { + TaggedMcpEntryLayer::Http { + enabled, + protocol, + url, + headers, + startup_timeout, + tool_timeout, + } => Ok(Self::Http { + enabled, + protocol, + url, + headers, + startup_timeout, + tool_timeout, + }), + TaggedMcpEntryLayer::Stdio { + enabled, + script, + command, + env, + startup_timeout, + tool_timeout, + } => Ok(Self::Stdio { + enabled, + script, + command, + env, + startup_timeout, + tool_timeout, + }), + TaggedMcpEntryLayer::Sandbox { + enabled, + protocol, + script, + command, + port, + env, + startup_timeout, + tool_timeout, + } => Ok(Self::Sandbox { + enabled, + protocol, + script, + command, + port, + env, + startup_timeout, + tool_timeout, + }), + } + } +} + /// A run hook entry. Exactly one of `script`, `command`, `url`, `prompt`, or /// `agent` fields determines the hook behavior. The `id` field, when set, is /// used for cross-layer replace-by-id merging. diff --git a/lib/crates/fabro-config/src/resolve/cli.rs b/lib/crates/fabro-config/src/resolve/cli.rs index 0c555b2ba..2b77d7b0d 100644 --- a/lib/crates/fabro-config/src/resolve/cli.rs +++ b/lib/crates/fabro-config/src/resolve/cli.rs @@ -12,7 +12,7 @@ pub fn resolve_cli(layer: &CliLayer, errors: &mut Vec) -> CliNames auth: CliAuthSettings { strategy: layer.auth.as_ref().and_then(|auth| auth.strategy), }, - exec: resolve_exec(layer.exec.as_ref()), + exec: resolve_exec(layer.exec.as_ref(), errors), output: CliOutputSettings { format: layer .output @@ -62,7 +62,7 @@ fn resolve_target( } } -fn resolve_exec(exec: Option<&CliExecLayer>) -> CliExecSettings { +fn resolve_exec(exec: Option<&CliExecLayer>, errors: &mut Vec) -> CliExecSettings { let exec = exec.expect("defaults.toml should provide cli.exec defaults"); let model = exec.model.as_ref(); @@ -86,7 +86,9 @@ fn resolve_exec(exec: Option<&CliExecLayer>) -> CliExecSettings { agent: CliExecAgentSettings { permissions: exec.agent.as_ref().and_then(|agent| agent.permissions), mcps: exec.agent.as_ref().and_then(|agent| { - (!agent.mcps.is_empty()).then(|| super::run::resolve_enabled_mcps(&agent.mcps)) + (!agent.mcps.is_empty()).then(|| { + super::run::resolve_enabled_mcps(&agent.mcps, "cli.exec.agent.mcps", errors) + }) }), }, } diff --git a/lib/crates/fabro-config/src/resolve/run.rs b/lib/crates/fabro-config/src/resolve/run.rs index 7452a3ff5..45f496b53 100644 --- a/lib/crates/fabro-config/src/resolve/run.rs +++ b/lib/crates/fabro-config/src/resolve/run.rs @@ -24,6 +24,7 @@ use crate::{ pub fn resolve_run( layer: &RunLayer, environments: &MergeMap, + mcp_server_catalog: &HashMap, errors: &mut Vec, ) -> RunNamespace { let clone = resolve_clone(layer.clone.as_ref()); @@ -73,7 +74,7 @@ pub fn resolve_run( .map(|(name, route)| (name.clone(), resolve_notification_route(route))) .collect(), interviews: resolve_interviews(layer.interviews.as_ref()), - agent: resolve_agent(layer.agent.as_ref()), + agent: resolve_agent(layer.agent.as_ref(), mcp_server_catalog, errors), hooks: layer .hooks .iter() @@ -280,7 +281,11 @@ fn resolve_interview_provider(provider: &InterviewProviderLayer) -> InterviewPro } } -fn resolve_agent(agent: Option<&RunAgentLayer>) -> RunAgentSettings { +fn resolve_agent( + agent: Option<&RunAgentLayer>, + mcp_server_catalog: &HashMap, + errors: &mut Vec, +) -> RunAgentSettings { let Some(agent) = agent else { return RunAgentSettings::default(); }; @@ -288,28 +293,67 @@ fn resolve_agent(agent: Option<&RunAgentLayer>) -> RunAgentSettings { RunAgentSettings { fabro_tools: agent.fabro_tools.unwrap_or(false), permissions: agent.permissions, - mcps: enabled_mcp_settings(&agent.mcps) - .map(|(name, settings)| (name, ResolvedMcpEntry::Resolved(settings))) - .collect(), + mcps: resolve_mcp_entries(&agent.mcps, mcp_server_catalog, errors), } } +fn resolve_mcp_entries( + mcps: &StickyMap, + mcp_server_catalog: &HashMap, + errors: &mut Vec, +) -> HashMap { + let mut resolved = HashMap::new(); + for (name, entry) in mcps.iter() { + if !entry.is_enabled() { + continue; + } + match entry { + McpEntryLayer::Reference { id, .. } => match mcp_server_catalog.get(id) { + Some(server) => { + resolved.insert(name.clone(), ResolvedMcpEntry::Resolved(server.clone())); + } + None => { + errors.push(ResolveError::Invalid { + path: format!("run.agent.mcps.{name}.id"), + reason: format!("unknown MCP server: {id}"), + }); + } + }, + _ => { + resolved.insert( + name.clone(), + ResolvedMcpEntry::Resolved(resolve_mcp_entry(name, entry)), + ); + } + } + } + resolved +} + /// Resolve an agent layer's inline MCP entries into runtime settings, dropping /// any entry with an explicit `enabled = false`. Shared by the `run.agent` and /// `cli.exec.agent` resolution paths so the enable check lives in one place and /// any future inline-MCP site inherits it for free. pub(crate) fn resolve_enabled_mcps( mcps: &StickyMap, + path: &str, + errors: &mut Vec, ) -> HashMap { - enabled_mcp_settings(mcps).collect() -} - -fn enabled_mcp_settings( - mcps: &StickyMap, -) -> impl Iterator + '_ { - mcps.iter() - .filter(|(_, entry)| entry.is_enabled()) - .map(|(name, entry)| (name.clone(), resolve_mcp_entry(name, entry))) + let mut resolved = HashMap::new(); + for (name, entry) in mcps.iter() { + if !entry.is_enabled() { + continue; + } + if matches!(entry, McpEntryLayer::Reference { .. }) { + errors.push(ResolveError::Invalid { + path: format!("{path}.{name}.id"), + reason: "MCP catalog references are only supported in run.agent.mcps".to_string(), + }); + continue; + } + resolved.insert(name.clone(), resolve_mcp_entry(name, entry)); + } + resolved } #[expect( @@ -361,6 +405,9 @@ pub(crate) fn resolve_mcp_entry(name: &str, entry: &McpEntryLayer) -> McpServerS .map(|(key, value)| (key.clone(), value.as_source())) .collect(), }, + McpEntryLayer::Reference { .. } => { + unreachable!("MCP catalog references are resolved before inline entry conversion") + } }; let (startup_timeout_secs, tool_timeout_secs) = match entry { @@ -382,6 +429,9 @@ pub(crate) fn resolve_mcp_entry(name: &str, entry: &McpEntryLayer) -> McpServerS startup_timeout.map_or(10, |timeout| timeout.as_std().as_secs()), tool_timeout.map_or(60, |timeout| timeout.as_std().as_secs()), ), + McpEntryLayer::Reference { .. } => { + unreachable!("MCP catalog references are resolved before inline entry conversion") + } }; McpServerSettings { diff --git a/lib/crates/fabro-config/src/tests/resolve_run.rs b/lib/crates/fabro-config/src/tests/resolve_run.rs index e79e5f9c2..cdeea2751 100644 --- a/lib/crates/fabro-config/src/tests/resolve_run.rs +++ b/lib/crates/fabro-config/src/tests/resolve_run.rs @@ -1017,10 +1017,13 @@ mod run_agent_mcps { //! Layer + resolver tests for `[run.agent.mcps]`: same-key replacement //! across layers (`StickyMap`) and honoring `enabled = false`. - use fabro_types::settings::run::{McpTransport, ResolvedMcpEntry}; + use std::collections::HashMap; + + use fabro_types::settings::run::{McpServerSettings, McpTransport, ResolvedMcpEntry}; - use crate::SettingsLayer; use crate::layers::Combine; + use crate::tests::seeded_environment_catalog; + use crate::{RunLayer, SettingsLayer, WorkflowSettingsBuilder}; fn parse_settings(source: &str) -> SettingsLayer { source @@ -1036,6 +1039,21 @@ mod run_agent_mcps { } } + fn mcp_catalog() -> HashMap { + HashMap::from([("sentry".to_string(), McpServerSettings { + name: "sentry".to_string(), + transport: McpTransport::Http { + protocol: fabro_types::settings::run::McpHttpProtocol::default(), + url: "https://sentry.example.com/mcp".to_string(), + headers: HashMap::from([( + "Authorization".to_string(), + "{{ secrets.SENTRY_TOKEN }}".to_string(), + )]), + }, + ..McpServerSettings::default() + })]) + } + #[test] fn higher_layer_replaces_same_key_mcp_entry() { // Both layers define `[run.agent.mcps.fs]`; the higher (workflow) @@ -1129,6 +1147,97 @@ command = ["fs-server"] ); } + #[test] + fn catalog_reference_resolves_from_server_catalog() { + let settings = WorkflowSettingsBuilder::new() + .server_manifest_defaults(RunLayer::default(), seeded_environment_catalog()) + .server_mcp_catalog(mcp_catalog()) + .workflow_toml( + r#" +_version = 1 + +[run.agent.mcps.error_tracker] +id = "sentry" +"#, + ) + .expect("settings should parse") + .build() + .expect("settings should resolve"); + + let server = settings.run.agent.mcps["error_tracker"] + .as_resolved() + .expect("catalog reference should resolve to a concrete server"); + assert_eq!(server.name, "sentry"); + match &server.transport { + McpTransport::Http { url, headers, .. } => { + assert_eq!(url, "https://sentry.example.com/mcp"); + assert_eq!( + headers.get("Authorization").map(String::as_str), + Some("{{ secrets.SENTRY_TOKEN }}") + ); + } + other => panic!("expected HTTP catalog transport, got {other:?}"), + } + } + + #[test] + fn disabled_catalog_reference_is_skipped_without_lookup() { + let settings = WorkflowSettingsBuilder::new() + .server_manifest_defaults(RunLayer::default(), seeded_environment_catalog()) + .workflow_toml( + r#" +_version = 1 + +[run.agent.mcps.optional] +id = "missing" +enabled = false +"#, + ) + .expect("settings should parse") + .build() + .expect("disabled reference should not require a catalog entry"); + + assert!(!settings.run.agent.mcps.contains_key("optional")); + } + + #[test] + fn missing_catalog_reference_is_a_resolve_error() { + let err = WorkflowSettingsBuilder::new() + .server_manifest_defaults(RunLayer::default(), seeded_environment_catalog()) + .workflow_toml( + r#" +_version = 1 + +[run.agent.mcps.missing] +id = "missing" +"#, + ) + .expect("settings should parse") + .build() + .expect_err("missing catalog entry should fail resolution"); + + assert!( + err.to_string().contains("unknown MCP server: missing"), + "unexpected error: {err}" + ); + } + + #[test] + fn catalog_reference_serializes_without_transport_tag() { + let settings = parse_settings( + r#" +_version = 1 + +[run.agent.mcps.error_tracker] +id = "sentry" +"#, + ); + let entry = &settings.run.unwrap().agent.unwrap().mcps["error_tracker"]; + let value = serde_json::to_value(entry).expect("MCP reference should serialize"); + + assert_eq!(value, serde_json::json!({ "id": "sentry" })); + } + #[test] fn higher_layer_disable_shadows_lower_layer_entry() { // Lower layer enables `fs`; higher layer redefines the same key with diff --git a/lib/crates/fabro-mcp-store/src/store.rs b/lib/crates/fabro-mcp-store/src/store.rs index c8554a386..d9122ca86 100644 --- a/lib/crates/fabro-mcp-store/src/store.rs +++ b/lib/crates/fabro-mcp-store/src/store.rs @@ -1,14 +1,16 @@ use std::collections::HashMap; use std::io::ErrorKind; use std::path::{Path, PathBuf}; +use std::sync::RwLock; use std::time::{SystemTime, UNIX_EPOCH}; use fabro_types::{ McpServerDefinition, McpServerDraft, McpServerId, McpServerReplace, McpServerRevision, + McpServerView, }; use tokio::fs; use tokio::io::AsyncWriteExt as _; -use tokio::sync::{Mutex, RwLock}; +use tokio::sync::Mutex; use crate::error::McpServerStoreError; use crate::model; @@ -41,25 +43,53 @@ impl McpServerStore { }) } + fn read_defs( + &self, + ) -> std::sync::RwLockReadGuard<'_, HashMap> { + self.defs.read().expect("mcp server store lock poisoned") + } + + fn write_defs( + &self, + ) -> std::sync::RwLockWriteGuard<'_, HashMap> { + self.defs.write().expect("mcp server store lock poisoned") + } + pub async fn list(&self) -> Vec { - let defs = self.defs.read().await; + let defs = self.read_defs(); let mut values = defs.values().cloned().collect::>(); values.sort_by(|left, right| left.id.cmp(&right.id)); values } + pub async fn list_views(&self) -> Vec { + let defs = self.read_defs(); + let mut values = defs.values().map(McpServerView::from).collect::>(); + values.sort_by(|left, right| left.id.cmp(&right.id)); + values + } + + pub fn catalog_settings( + &self, + ) -> HashMap { + let defs = self.read_defs(); + defs.iter() + .map(|(id, definition)| (id.to_string(), server_settings_from_definition(definition))) + .collect() + } + /// Sorted ids only, without cloning the (potentially sensitive) env/header /// maps carried by full definitions. Used by missing-reference errors to /// list available ids cheaply. pub async fn ids(&self) -> Vec { - let defs = self.defs.read().await; + let defs = self.read_defs(); let mut ids = defs.keys().cloned().collect::>(); ids.sort(); ids } pub async fn get(&self, id: &McpServerId) -> Option { - self.defs.read().await.get(id).cloned() + self.read_defs().get(id).cloned() } pub async fn create( @@ -68,7 +98,7 @@ impl McpServerStore { ) -> Result { let (id, replace) = draft.into(); let _mutation = self.mutations.lock().await; - if self.defs.read().await.contains_key(&id) { + if self.read_defs().contains_key(&id) { return Err(McpServerStoreError::AlreadyExists { id }); } let (definition, bytes) = model::definition_from_replace(id.clone(), replace)?; @@ -78,7 +108,7 @@ impl McpServerStore { .await .map_err(|err| create_error_for(id.clone(), err))?; - let mut defs = self.defs.write().await; + let mut defs = self.write_defs(); defs.insert(id, definition.clone()); Ok(definition) } @@ -91,13 +121,13 @@ impl McpServerStore { ) -> Result { let _mutation = self.mutations.lock().await; { - let defs = self.defs.read().await; + let defs = self.read_defs(); check_revision(&defs, id, expected)?; } let (definition, bytes) = model::definition_from_replace(id.clone(), replace)?; write_atomic(&self.dir, &definition_path(&self.dir, id), &bytes).await?; - let mut defs = self.defs.write().await; + let mut defs = self.write_defs(); defs.insert(id.clone(), definition.clone()); Ok(definition) } @@ -109,7 +139,7 @@ impl McpServerStore { ) -> Result<(), McpServerStoreError> { let _mutation = self.mutations.lock().await; { - let defs = self.defs.read().await; + let defs = self.read_defs(); check_revision(&defs, id, expected)?; } @@ -117,7 +147,7 @@ impl McpServerStore { fs::remove_file(&path) .await .map_err(|err| McpServerStoreError::io(path, err))?; - let mut defs = self.defs.write().await; + let mut defs = self.write_defs(); defs.remove(id); Ok(()) } @@ -141,6 +171,19 @@ fn check_revision( Ok(()) } +fn server_settings_from_definition( + definition: &McpServerDefinition, +) -> fabro_types::settings::run::McpServerSettings { + fabro_types::settings::run::McpServerSettings { + name: definition.id.to_string(), + transport: definition.transport.clone(), + current_dir: None, + clear_env: false, + startup_timeout_secs: definition.startup_timeout_secs, + tool_timeout_secs: definition.tool_timeout_secs, + } +} + #[expect( clippy::disallowed_methods, reason = "MCP server directory scan runs once at startup, before the runtime needs to make progress; std::fs avoids needing a Tokio runtime for the caller." diff --git a/lib/crates/fabro-server/src/manifest_validation.rs b/lib/crates/fabro-server/src/manifest_validation.rs index 40c432d92..baca3f014 100644 --- a/lib/crates/fabro-server/src/manifest_validation.rs +++ b/lib/crates/fabro-server/src/manifest_validation.rs @@ -1,3 +1,4 @@ +use std::collections::HashMap; use std::sync::Arc; use anyhow::Result; @@ -30,6 +31,7 @@ pub fn validate_manifest_with_environment_defaults( let prepared = run_manifest::prepare_manifest_with_environment_defaults( manifest_run_defaults, manifest_environment_defaults, + &HashMap::new(), manifest, )?; let validated = diff --git a/lib/crates/fabro-server/src/run_manifest.rs b/lib/crates/fabro-server/src/run_manifest.rs index 72deeb2e4..acbdbcbdc 100644 --- a/lib/crates/fabro-server/src/run_manifest.rs +++ b/lib/crates/fabro-server/src/run_manifest.rs @@ -27,7 +27,7 @@ use fabro_sandbox::{DockerSandboxOptions, Sandbox, SandboxSpec}; use fabro_static::EnvVars; use fabro_types::settings::cli::OutputVerbosity; use fabro_types::settings::interp::InterpString; -use fabro_types::settings::run::{EnvironmentProvider, RunGoal, RunNamespace}; +use fabro_types::settings::run::{EnvironmentProvider, McpServerSettings, RunGoal, RunNamespace}; use fabro_types::{ ManifestPath, RunId, RunProvenance, SandboxProviderKind, ServerSettings, WorkflowSettings, }; @@ -77,6 +77,7 @@ pub(crate) fn manifest_run_defaults(run: Option<&RunLayer>) -> RunLayer { pub(crate) fn prepare_manifest_with_environment_defaults( manifest_run_defaults: &RunLayer, manifest_environment_defaults: &MergeMap, + manifest_mcp_server_catalog: &HashMap, manifest: &types::RunManifest, ) -> Result { if manifest.version != 1 { @@ -95,10 +96,12 @@ pub(crate) fn prepare_manifest_with_environment_defaults( let args_overrides = manifest_args_overrides(manifest.args.as_ref()).context("failed to parse manifest args")?; - let mut workflow_settings_builder = WorkflowSettingsBuilder::new().server_manifest_defaults( - manifest_run_defaults.clone(), - manifest_environment_defaults.clone(), - ); + let mut workflow_settings_builder = WorkflowSettingsBuilder::new() + .server_manifest_defaults( + manifest_run_defaults.clone(), + manifest_environment_defaults.clone(), + ) + .server_mcp_catalog(manifest_mcp_server_catalog.clone()); if let Some(run) = args_overrides.run { workflow_settings_builder = workflow_settings_builder.run_overrides(run); } @@ -1442,6 +1445,7 @@ mod tests { super::prepare_manifest_with_environment_defaults( manifest_run_defaults, &environment_defaults_fixture(), + &HashMap::new(), manifest, ) } @@ -1511,6 +1515,7 @@ enabled = {clone_enabled} let prepared = super::prepare_manifest_with_environment_defaults( &manifest_run_defaults(Some(&default_settings_fixture())), &environment_defaults, + &HashMap::new(), &manifest, ) .unwrap(); diff --git a/lib/crates/fabro-server/src/server.rs b/lib/crates/fabro-server/src/server.rs index c6aacb6ef..37d67bb9d 100644 --- a/lib/crates/fabro-server/src/server.rs +++ b/lib/crates/fabro-server/src/server.rs @@ -1343,11 +1343,12 @@ impl AppState { .clone() } - pub(crate) fn refresh_manifest_run_settings_from_environment_catalog(&self) { + pub(crate) fn refresh_manifest_run_settings_from_catalogs(&self) { let manifest_run_defaults = self.manifest_run_defaults(); let manifest_run_settings = resolve_manifest_run_settings_with_catalog( manifest_run_defaults.as_ref(), &self.stores.environments, + &self.stores.mcp_servers, ); *self .manifest_run_settings @@ -1355,6 +1356,10 @@ impl AppState { .expect("manifest run settings lock poisoned") = manifest_run_settings; } + pub(crate) fn refresh_manifest_run_settings_from_environment_catalog(&self) { + self.refresh_manifest_run_settings_from_catalogs(); + } + fn http_client(&self) -> Result { match &self.http_client { Some(client) => Ok(client.clone()), @@ -1589,6 +1594,7 @@ impl AppState { let manifest_run_settings = resolve_manifest_run_settings_with_catalog( manifest_run_defaults.as_ref(), &self.stores.environments, + &self.stores.mcp_servers, ); let catalog = Arc::new( Catalog::from_builtin_with_overrides(&llm_catalog_settings) @@ -2182,12 +2188,14 @@ fn build_prune_plan( fn resolve_manifest_run_settings_with_catalog( manifest_run_defaults: &RunLayer, environment_store: &EnvironmentStore, + mcp_server_store: &McpServerStore, ) -> std::result::Result { WorkflowSettingsBuilder::new() .server_manifest_defaults( manifest_run_defaults.clone(), (*environment_store.catalog_layer()).clone(), ) + .server_mcp_catalog(mcp_server_store.catalog_settings()) .build() .map(|settings| settings.run) .map_err(|err| SharedError::new(anyhow::Error::msg(err.to_string()))) @@ -2367,6 +2375,7 @@ pub(crate) fn build_app_state(config: AppStateConfig) -> anyhow::Result Response { let manifest_run_defaults = state.manifest_run_defaults(); let manifest_environment_defaults = state.environment_store().catalog_layer(); + let manifest_mcp_server_catalog = state.mcp_server_store().catalog_settings(); let prepared = match run_manifest::prepare_manifest_with_environment_defaults( manifest_run_defaults.as_ref(), manifest_environment_defaults.as_ref(), + &manifest_mcp_server_catalog, &req.manifest, ) { Ok(prepared) => prepared, diff --git a/lib/crates/fabro-server/src/server/handler/mcp_servers.rs b/lib/crates/fabro-server/src/server/handler/mcp_servers.rs index f14d444e0..dc0d1ce07 100644 --- a/lib/crates/fabro-server/src/server/handler/mcp_servers.rs +++ b/lib/crates/fabro-server/src/server/handler/mcp_servers.rs @@ -39,9 +39,8 @@ pub(super) fn routes() -> Router> { } async fn list_mcp_servers(_auth: RequiredUser, State(state): State>) -> Response { - let definitions = state.mcp_server_store().list().await; - let total = definitions.len(); - let data: Vec = definitions.into_iter().map(McpServerView::from).collect(); + let data = state.mcp_server_store().list_views().await; + let total = data.len(); ( StatusCode::OK, Json(McpServerListResponse { @@ -63,6 +62,7 @@ async fn create_mcp_server( // point the user at `{{ secrets.NAME }}` is undecided. Until then literals // are accepted verbatim and secret references are stored as written. let definition = state.mcp_server_store().create(draft).await?; + state.refresh_manifest_run_settings_from_catalogs(); Ok(mcp_server_with_etag_response( StatusCode::CREATED, definition, @@ -96,6 +96,7 @@ async fn replace_mcp_server( .mcp_server_store() .replace(&id, &expected, replacement) .await?; + state.refresh_manifest_run_settings_from_catalogs(); Ok(mcp_server_with_etag_response(StatusCode::OK, definition)) } @@ -108,6 +109,7 @@ async fn delete_mcp_server( let id = parse_path_id(id)?; let expected = parse_required_if_match(&headers, "mcp server", &id)?; state.mcp_server_store().delete(&id, &expected).await?; + state.refresh_manifest_run_settings_from_catalogs(); Ok(StatusCode::NO_CONTENT.into_response()) } diff --git a/lib/crates/fabro-server/src/server/handler/runs.rs b/lib/crates/fabro-server/src/server/handler/runs.rs index 4fddbb4fc..89bec59a8 100644 --- a/lib/crates/fabro-server/src/server/handler/runs.rs +++ b/lib/crates/fabro-server/src/server/handler/runs.rs @@ -655,9 +655,11 @@ pub(crate) async fn create_run_from_manifest( } = request; let manifest_run_defaults = state.manifest_run_defaults(); let manifest_environment_defaults = state.environment_store().catalog_layer(); + let manifest_mcp_server_catalog = state.mcp_server_store().catalog_settings(); let mut prepared = match run_manifest::prepare_manifest_with_environment_defaults( manifest_run_defaults.as_ref(), manifest_environment_defaults.as_ref(), + &manifest_mcp_server_catalog, &manifest, ) { Ok(prepared) => prepared, @@ -921,9 +923,11 @@ async fn run_preflight( ) -> Response { let manifest_run_defaults = state.manifest_run_defaults(); let manifest_environment_defaults = state.environment_store().catalog_layer(); + let manifest_mcp_server_catalog = state.mcp_server_store().catalog_settings(); let mut prepared = match run_manifest::prepare_manifest_with_environment_defaults( manifest_run_defaults.as_ref(), manifest_environment_defaults.as_ref(), + &manifest_mcp_server_catalog, &req, ) { Ok(prepared) => prepared, @@ -969,9 +973,11 @@ async fn validate_run_manifest( ) -> Response { let manifest_run_defaults = state.manifest_run_defaults(); let manifest_environment_defaults = state.environment_store().catalog_layer(); + let manifest_mcp_server_catalog = state.mcp_server_store().catalog_settings(); let mut prepared = match run_manifest::prepare_manifest_with_environment_defaults( manifest_run_defaults.as_ref(), manifest_environment_defaults.as_ref(), + &manifest_mcp_server_catalog, &req, ) { Ok(prepared) => prepared, diff --git a/lib/crates/fabro-server/src/server/tests.rs b/lib/crates/fabro-server/src/server/tests.rs index 07b58ac2e..7220bc0d6 100644 --- a/lib/crates/fabro-server/src/server/tests.rs +++ b/lib/crates/fabro-server/src/server/tests.rs @@ -1257,6 +1257,8 @@ fn system_sandbox_provider_uses_manifest_defaults() { .expect("seed built-in environments"); let environment_store = EnvironmentStore::load(&environment_dir, true).expect("environment store should load"); + let mcp_server_store = + McpServerStore::load(temp.path().join("mcps")).expect("mcp server store should load"); let source = r#" _version = 1 @@ -1266,6 +1268,7 @@ id = "default" let manifest_run_settings = resolve_manifest_run_settings_with_catalog( &run_manifest::manifest_run_defaults(Some(&manifest_run_defaults_from_toml(source))), &environment_store, + &mcp_server_store, ); assert_eq!(system_sandbox_provider(&manifest_run_settings), "daytona"); @@ -1276,6 +1279,8 @@ fn system_sandbox_provider_defaults_when_manifest_run_settings_do_not_resolve() let temp = tempfile::tempdir().unwrap(); let environment_store = EnvironmentStore::load(temp.path().join("environments"), true) .expect("environment store should load"); + let mcp_server_store = + McpServerStore::load(temp.path().join("mcps")).expect("mcp server store should load"); let source = r#" _version = 1 @@ -1285,6 +1290,7 @@ id = "missing" let manifest_run_settings = resolve_manifest_run_settings_with_catalog( &run_manifest::manifest_run_defaults(Some(&manifest_run_defaults_from_toml(source))), &environment_store, + &mcp_server_store, ); assert_eq!( diff --git a/lib/crates/fabro-server/tests/it/api/mcp_servers.rs b/lib/crates/fabro-server/tests/it/api/mcp_servers.rs index 1227b3ae0..50e78b2a3 100644 --- a/lib/crates/fabro-server/tests/it/api/mcp_servers.rs +++ b/lib/crates/fabro-server/tests/it/api/mcp_servers.rs @@ -7,7 +7,9 @@ use fabro_server::test_support::{TestAppStateBuilder, build_test_router, test_au use serde_json::{Value, json}; use tower::ServiceExt; -use crate::helpers::{checked_response, response_json, response_status}; +use crate::helpers::{ + MINIMAL_DOT, checked_response, minimal_manifest_json, response_json, response_status, +}; fn mcp_server_body(id: &str, display_name: &str) -> Value { json!({ @@ -529,6 +531,25 @@ async fn unknown_transport_type_is_unprocessable() { .await; } +#[tokio::test] +async fn unknown_transport_field_is_unprocessable() { + let (app, _temp_dir, _mcp_dir) = mcp_server_app(); + let mut body = mcp_server_body("sentry", "Sentry"); + body["transport"]["unexpected"] = json!("typo"); + + let response = app + .oneshot(json_request(Method::POST, "/mcp-servers", &body)) + .await + .expect("unknown transport field create should respond"); + + response_status( + response, + StatusCode::UNPROCESSABLE_ENTITY, + "POST /api/v1/mcp-servers unknown transport field", + ) + .await; +} + #[tokio::test] async fn invalid_mcp_server_id_is_bad_request() { let (app, _temp_dir, _mcp_dir) = mcp_server_app(); @@ -546,6 +567,31 @@ async fn invalid_mcp_server_id_is_bad_request() { .await; } +#[tokio::test] +async fn created_mcp_server_can_be_referenced_by_manifest_validation() { + let (app, _temp_dir, _mcp_dir) = mcp_server_app(); + create_mcp_server(&app, "sentry", "Sentry").await; + + let mut manifest = minimal_manifest_json(MINIMAL_DOT); + manifest["workflows"]["workflow.fabro"]["config"] = json!({ + "path": "workflow.toml", + "source": r#" +_version = 1 + +[run.agent.mcps.sentry] +id = "sentry" +"# + }); + + let response = app + .oneshot(json_request(Method::POST, "/validate", &manifest)) + .await + .expect("manifest validation should respond"); + let body = response_json(response, StatusCode::OK, "POST /api/v1/validate").await; + + assert_eq!(body["ok"], true); +} + #[tokio::test] async fn mcp_server_store_malformed_persisted_toml_fails_startup() { let temp_dir = tempfile::tempdir().expect("mcp server test tempdir should be created"); diff --git a/lib/crates/fabro-types/src/mcp_store.rs b/lib/crates/fabro-types/src/mcp_store.rs index 69b1a6fdd..18cedbe42 100644 --- a/lib/crates/fabro-types/src/mcp_store.rs +++ b/lib/crates/fabro-types/src/mcp_store.rs @@ -116,6 +116,20 @@ impl From for McpServerView { } } +impl From<&McpServerDefinition> for McpServerView { + fn from(value: &McpServerDefinition) -> Self { + Self { + id: value.id.clone(), + revision: value.revision.clone(), + display_name: value.display_name.clone(), + description: value.description.clone(), + transport: (&value.transport).into(), + startup_timeout_secs: value.startup_timeout_secs, + tool_timeout_secs: value.tool_timeout_secs, + } + } +} + /// The read-side projection of [`McpTransport`]. It mirrors the transport shape /// but replaces the `env`/`headers` value maps with sorted `env_keys`/ /// `header_keys` name lists, so secret-bearing values are never returned. @@ -172,6 +186,37 @@ impl From for McpTransportView { } } +impl From<&McpTransport> for McpTransportView { + fn from(value: &McpTransport) -> Self { + match value { + McpTransport::Stdio { command, env } => Self::Stdio { + command: command.clone(), + env_keys: sorted_keys_ref(env), + }, + McpTransport::Http { + protocol, + url, + headers, + } => Self::Http { + protocol: *protocol, + url: url.clone(), + header_keys: sorted_keys_ref(headers), + }, + McpTransport::Sandbox { + protocol, + command, + port, + env, + } => Self::Sandbox { + protocol: *protocol, + command: command.clone(), + port: *port, + env_keys: sorted_keys_ref(env), + }, + } + } +} + /// Collect a value map's keys into a sorted list, so the projection is /// deterministic regardless of the map's iteration order. fn sorted_keys(map: HashMap) -> Vec { @@ -180,6 +225,12 @@ fn sorted_keys(map: HashMap) -> Vec { keys } +fn sorted_keys_ref(map: &HashMap) -> Vec { + let mut keys: Vec = map.keys().cloned().collect(); + keys.sort(); + keys +} + /// Validation errors for MCP server domain fields. #[derive(Debug, Clone, PartialEq, Eq)] pub enum McpServerValidationError { diff --git a/lib/crates/fabro-types/src/settings/run.rs b/lib/crates/fabro-types/src/settings/run.rs index c34fab921..b4f3b4896 100644 --- a/lib/crates/fabro-types/src/settings/run.rs +++ b/lib/crates/fabro-types/src/settings/run.rs @@ -1557,7 +1557,7 @@ mod resolve_transport_env_tests { } #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] -#[serde(tag = "type", rename_all = "snake_case")] +#[serde(tag = "type", rename_all = "snake_case", deny_unknown_fields)] pub enum McpTransport { Stdio { command: Vec,