scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-10-12 03:51:01 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 2

fix(server): resolve MCP catalog references

Browse source
This commit is contained in:
Scott Werner 2026-06-30 14:01:25 -04:00
parent b4a2f2665c
commit 0ce756ec64
17 changed files with 662 additions and 69 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
docs/public/api-reference/fabro-api.yaml
View file

@ -4266,6 +4266,9 @@ paths:
responses:
"201":
description: MCP server created
headers:
ETag:
$ref: "#/components/headers/ETag"
content:
application/json:
schema:

33
lib/crates/fabro-config/src/builders.rs
View file

@ -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<SettingsLay
#[derive(Clone, Debug, Default)]
pub struct WorkflowSettingsBuilder {
args: SettingsLayer,
workflow: SettingsLayer,
project: SettingsLayer,
user: SettingsLayer,
server: SettingsLayer,
args: SettingsLayer,
workflow: SettingsLayer,
project: SettingsLayer,
user: SettingsLayer,
server: SettingsLayer,
mcp_server_catalog: HashMap<String, McpServerSettings>,
}
impl WorkflowSettingsBuilder {
@ -561,6 +564,12 @@ impl WorkflowSettingsBuilder {
})
}
#[must_use]
pub fn server_mcp_catalog(mut self, catalog: HashMap<String, McpServerSettings>) -> 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<WorkflowSettings, ResolveErrors> {
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<WorkflowSettings, ResolveErrors> {
Self::from_layer_with_mcp_server_catalog(layer, &HashMap::new())
}
fn from_layer_with_mcp_server_catalog(
layer: &SettingsLayer,
mcp_server_catalog: &HashMap<String, McpServerSettings>,
) -> std::result::Result<WorkflowSettings, ResolveErrors> {
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(

282
lib/crates/fabro-config/src/layers/run.rs
View file

@ -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<bool>,
#[serde(default)]
protocol: McpHttpProtocol,
url: InterpString,
#[serde(default)]
headers: HashMap<String, InterpString>,
#[serde(default)]
startup_timeout: Option<Duration>,
#[serde(default)]
tool_timeout: Option<Duration>,
},
Stdio {
#[serde(default)]
enabled: Option<bool>,
#[serde(default)]
script: Option<InterpString>,
#[serde(default)]
command: Option<Vec<InterpString>>,
#[serde(default)]
env: HashMap<String, InterpString>,
#[serde(default)]
startup_timeout: Option<Duration>,
#[serde(default)]
tool_timeout: Option<Duration>,
},
Sandbox {
#[serde(default)]
enabled: Option<bool>,
#[serde(default)]
protocol: McpHttpProtocol,
#[serde(default)]
script: Option<InterpString>,
#[serde(default)]
command: Option<Vec<InterpString>>,
port: u16,
#[serde(default)]
env: HashMap<String, InterpString>,
#[serde(default)]
startup_timeout: Option<Duration>,
#[serde(default)]
tool_timeout: Option<Duration>,
},
Reference {
id: String,
enabled: Option<bool>,
},
}
impl Serialize for McpEntryLayer {
fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
where
S: Serializer,
{
#[derive(Serialize)]
#[serde(tag = "type", rename_all = "snake_case")]
enum TaggedMcpEntryLayer<'a> {
Http {
enabled: &'a Option<bool>,
protocol: &'a McpHttpProtocol,
url: &'a InterpString,
headers: &'a HashMap<String, InterpString>,
startup_timeout: &'a Option<Duration>,
tool_timeout: &'a Option<Duration>,
},
Stdio {
enabled: &'a Option<bool>,
script: &'a Option<InterpString>,
command: &'a Option<Vec<InterpString>>,
env: &'a HashMap<String, InterpString>,
startup_timeout: &'a Option<Duration>,
tool_timeout: &'a Option<Duration>,
},
Sandbox {
enabled: &'a Option<bool>,
protocol: &'a McpHttpProtocol,
script: &'a Option<InterpString>,
command: &'a Option<Vec<InterpString>>,
port: &'a u16,
env: &'a HashMap<String, InterpString>,
startup_timeout: &'a Option<Duration>,
tool_timeout: &'a Option<Duration>,
},
}
#[derive(Serialize)]
struct McpReferenceLayer<'a> {
id: &'a str,
#[serde(skip_serializing_if = "Option::is_none")]
enabled: &'a Option<bool>,
}
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<D>(deserializer: D) -> Result<Self, D::Error>
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<bool>,
}
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<bool>,
#[serde(default)]
protocol: McpHttpProtocol,
url: InterpString,
#[serde(default)]
headers: HashMap<String, InterpString>,
#[serde(default)]
startup_timeout: Option<Duration>,
#[serde(default)]
tool_timeout: Option<Duration>,
},
Stdio {
#[serde(default)]
enabled: Option<bool>,
#[serde(default)]
script: Option<InterpString>,
#[serde(default)]
command: Option<Vec<InterpString>>,
#[serde(default)]
env: HashMap<String, InterpString>,
#[serde(default)]
startup_timeout: Option<Duration>,
#[serde(default)]
tool_timeout: Option<Duration>,
},
Sandbox {
#[serde(default)]
enabled: Option<bool>,
#[serde(default)]
protocol: McpHttpProtocol,
#[serde(default)]
script: Option<InterpString>,
#[serde(default)]
command: Option<Vec<InterpString>>,
port: u16,
#[serde(default)]
env: HashMap<String, InterpString>,
#[serde(default)]
startup_timeout: Option<Duration>,
#[serde(default)]
tool_timeout: Option<Duration>,
},
}
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.

8
lib/crates/fabro-config/src/resolve/cli.rs
View file

@ -12,7 +12,7 @@ pub fn resolve_cli(layer: &CliLayer, errors: &mut Vec<ResolveError>) -> 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<ResolveError>) -> 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)
})
}),
},
}

78
lib/crates/fabro-config/src/resolve/run.rs
View file

@ -24,6 +24,7 @@ use crate::{
pub fn resolve_run(
layer: &RunLayer,
environments: &MergeMap<EnvironmentLayer>,
mcp_server_catalog: &HashMap<String, McpServerSettings>,
errors: &mut Vec<ResolveError>,
) -> 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<String, McpServerSettings>,
errors: &mut Vec<ResolveError>,
) -> 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<McpEntryLayer>,
mcp_server_catalog: &HashMap<String, McpServerSettings>,
errors: &mut Vec<ResolveError>,
) -> HashMap<String, ResolvedMcpEntry> {
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<McpEntryLayer>,
path: &str,
errors: &mut Vec<ResolveError>,
) -> HashMap<String, McpServerSettings> {
enabled_mcp_settings(mcps).collect()
}
fn enabled_mcp_settings(
mcps: &StickyMap<McpEntryLayer>,
) -> impl Iterator<Item = (String, McpServerSettings)> + '_ {
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 {

113
lib/crates/fabro-config/src/tests/resolve_run.rs
View file

@ -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<String, McpServerSettings> {
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

63
lib/crates/fabro-mcp-store/src/store.rs
View file

@ -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<McpServerId, McpServerDefinition>> {
self.defs.read().expect("mcp server store lock poisoned")
}
fn write_defs(
&self,
) -> std::sync::RwLockWriteGuard<'_, HashMap<McpServerId, McpServerDefinition>> {
self.defs.write().expect("mcp server store lock poisoned")
}
pub async fn list(&self) -> Vec<McpServerDefinition> {
let defs = self.defs.read().await;
let defs = self.read_defs();
let mut values = defs.values().cloned().collect::<Vec<_>>();
values.sort_by(|left, right| left.id.cmp(&right.id));
values
}
pub async fn list_views(&self) -> Vec<McpServerView> {
let defs = self.read_defs();
let mut values = defs.values().map(McpServerView::from).collect::<Vec<_>>();
values.sort_by(|left, right| left.id.cmp(&right.id));
values
}
pub fn catalog_settings(
&self,
) -> HashMap<String, fabro_types::settings::run::McpServerSettings> {
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<McpServerId> {
let defs = self.defs.read().await;
let defs = self.read_defs();
let mut ids = defs.keys().cloned().collect::<Vec<_>>();
ids.sort();
ids
}
pub async fn get(&self, id: &McpServerId) -> Option<McpServerDefinition> {
self.defs.read().await.get(id).cloned()
self.read_defs().get(id).cloned()
}
pub async fn create(
@ -68,7 +98,7 @@ impl McpServerStore {
) -> Result<McpServerDefinition, McpServerStoreError> {
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<McpServerDefinition, McpServerStoreError> {
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."

2
lib/crates/fabro-server/src/manifest_validation.rs
View file

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

15
lib/crates/fabro-server/src/run_manifest.rs
View file

@ -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<EnvironmentLayer>,
manifest_mcp_server_catalog: &HashMap<String, McpServerSettings>,
manifest: &types::RunManifest,
) -> Result<PreparedManifest> {
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();

11
lib/crates/fabro-server/src/server.rs
View file

@ -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<fabro_http::HttpClient, fabro_http::HttpClientBuildError> {
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<RunNamespace, SharedError> {
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<Arc<AppS
let current_manifest_run_settings = resolve_manifest_run_settings_with_catalog(
current_manifest_run_defaults.as_ref(),
&environment_store,
&mcp_server_store,
);
let current_catalog = Arc::new(
Catalog::from_builtin_with_overrides(&resolved_settings.llm_catalog_settings)

2
lib/crates/fabro-server/src/server/handler/graph.rs
View file

@ -42,9 +42,11 @@ async fn render_graph_from_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 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,

8
lib/crates/fabro-server/src/server/handler/mcp_servers.rs
View file

@ -39,9 +39,8 @@ pub(super) fn routes() -> Router<Arc<AppState>> {
}
async fn list_mcp_servers(_auth: RequiredUser, State(state): State<Arc<AppState>>) -> Response {
let definitions = state.mcp_server_store().list().await;
let total = definitions.len();
let data: Vec<McpServerView> = 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())
}

6
lib/crates/fabro-server/src/server/handler/runs.rs
View file

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

6
lib/crates/fabro-server/src/server/tests.rs
View file

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

48
lib/crates/fabro-server/tests/it/api/mcp_servers.rs
View file

@ -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");

51
lib/crates/fabro-types/src/mcp_store.rs
View file

@ -116,6 +116,20 @@ impl From<McpServerDefinition> 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<McpTransport> 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<String, String>) -> Vec<String> {
@ -180,6 +225,12 @@ fn sorted_keys(map: HashMap<String, String>) -> Vec<String> {
keys
}
fn sorted_keys_ref(map: &HashMap<String, String>) -> Vec<String> {
let mut keys: Vec<String> = map.keys().cloned().collect();
keys.sort();
keys
}
/// Validation errors for MCP server domain fields.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum McpServerValidationError {

2
lib/crates/fabro-types/src/settings/run.rs
View file

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