admin/openfang
1
0
Fork 0
mirror of https://github.com/RightNow-AI/openfang.git synced 2026-07-30 06:32:17 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #778 from BaseDatum/feat/rmcp-protocol

Browse Source 
feat: replace hand-rolled MCP transport with rmcp SDK
This commit is contained in:
Jaber Jaber
2026-03-27 22:00:18 +03:00
committed by GitHub
parent b6cb4cc2d9 9e2853a5f8
commit 3a64e322ad
31 changed files with 781 additions and 650 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+190 -23
View File
@@ -384,6 +384,28 @@ version = "1.5.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c08606f8c3cbf4ce6ec8e28fb0014a2c086708fe954eaa885384a6165172e7e8"
[[package]]
name = "aws-lc-rs"
version = "1.16.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a054912289d18629dc78375ba2c3726a3afe3ff71b4edba9dedfca0e3446d1fc"
dependencies = [
"aws-lc-sys",
"zeroize",
]
[[package]]
name = "aws-lc-sys"
version = "0.39.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1fa7e52a4c5c547c741610a2c6f123f3881e409b714cd27e6798ef020c514f0a"
dependencies = [
"cc",
"cmake",
"dunce",
"fs_extra",
]
[[package]]
name = "axum"
version = "0.8.8"
@@ -860,6 +882,15 @@ version = "1.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c8d4a3bb8b1e0c1050499d1815f5ab16d04f0959b233085fb31653fbfc9d98f9"
[[package]]
name = "cmake"
version = "0.1.57"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "75443c44cd6b379beb8c5b45d85d0773baf31cce901fe7bb252f4eff3008ef7d"
dependencies = [
"cc",
]
[[package]]
name = "cobs"
version = "0.3.0"
@@ -881,7 +912,7 @@ version = "3.1.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "faf9468729b8cbcea668e36183cb69d317348c2e08e994829fb56ebfdfbaac34"
dependencies = [
"windows-sys 0.52.0",
"windows-sys 0.59.0",
]
[[package]]
@@ -1788,7 +1819,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb"
dependencies = [
"libc",
"windows-sys 0.52.0",
"windows-sys 0.59.0",
]
[[package]]
@@ -1958,6 +1989,12 @@ dependencies = [
"percent-encoding",
]
[[package]]
name = "fs_extra"
version = "1.3.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "42703706b716c37f96a77aea830392ad231f44c9e9a67872fa5548707e11b11c"
[[package]]
name = "futf"
version = "0.1.5"
@@ -3608,6 +3645,18 @@ version = "1.0.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "650eef8c711430f1a879fdd01d4745a7deea475becfb90269c06775983bbf086"
[[package]]
name = "nix"
version = "0.31.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "5d6d0705320c1e6ba1d912b5e37cf18071b6c2e9b7fa8215a1e8a7651966f5d3"
dependencies = [
"bitflags 2.11.0",
"cfg-if",
"cfg_aliases",
"libc",
]
[[package]]
name = "no-std-compat"
version = "0.4.1"
@@ -4106,6 +4155,7 @@ dependencies = [
"async-trait",
"chrono",
"openfang-types",
"reqwest 0.12.28",
"rmp-serde",
"rusqlite",
"serde",
@@ -4149,11 +4199,13 @@ dependencies = [
"dashmap",
"futures",
"hex",
"http",
"openfang-memory",
"openfang-skills",
"openfang-types",
"regex-lite",
"reqwest 0.12.28",
"rmcp",
"rusqlite",
"serde",
"serde_json",
@@ -4267,6 +4319,15 @@ version = "0.2.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7c87def4c32ab89d880effc9e097653c8da5d6ef28e6b539d313baaacfbafcbe"
[[package]]
name = "openssl-src"
version = "300.5.5+3.5.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "3f1787d533e03597a7934fd0a765f0d28e94ecc5fb7789f8053b1e699a56f709"
dependencies = [
"cc",
]
[[package]]
name = "openssl-sys"
version = "0.9.112"
@@ -4275,6 +4336,7 @@ checksum = "57d55af3b3e226502be1526dfdba67ab0e9c96fc293004e79576b2b9edb0dbdb"
dependencies = [
"cc",
"libc",
"openssl-src",
"pkg-config",
"vcpkg",
]
@@ -4896,6 +4958,20 @@ dependencies = [
"unicode-ident",
]
[[package]]
name = "process-wrap"
version = "9.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "2e842efad9119158434d193c6682e2ebee4b44d6ad801d7b349623b3f57cdf55"
dependencies = [
"futures",
"indexmap 2.13.0",
"nix",
"tokio",
"tracing",
"windows 0.62.2",
]
[[package]]
name = "prost"
version = "0.13.5"
@@ -5017,6 +5093,7 @@ version = "0.11.14"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "434b42fec591c96ef50e21e886936e66d3cc3f737104fdb9b737c40ffb94c098"
dependencies = [
"aws-lc-rs",
"bytes",
"getrandom 0.3.4",
"lru-slab",
@@ -5420,6 +5497,7 @@ dependencies = [
"log",
"percent-encoding",
"pin-project-lite",
"quinn",
"rustls",
"rustls-pki-types",
"rustls-platform-verifier",
@@ -5477,6 +5555,29 @@ dependencies = [
"windows-sys 0.52.0",
]
[[package]]
name = "rmcp"
version = "1.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ba6b9d2f0efe2258b23767f1f9e0054cfbcac9c2d6f81a031214143096d7864f"
dependencies = [
"async-trait",
"chrono",
"futures",
"http",
"pin-project-lite",
"process-wrap",
"reqwest 0.13.2",
"serde",
"serde_json",
"sse-stream",
"thiserror 2.0.18",
"tokio",
"tokio-stream",
"tokio-util",
"tracing",
]
[[package]]
name = "rmp"
version = "0.8.15"
@@ -5564,7 +5665,7 @@ dependencies = [
"errno",
"libc",
"linux-raw-sys 0.12.1",
"windows-sys 0.52.0",
"windows-sys 0.59.0",
]
[[package]]
@@ -5573,6 +5674,7 @@ version = "0.23.37"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "758025cb5fccfd3bc2fd74708fd4682be41d99e5dff73c377c0646c6012c73a4"
dependencies = [
"aws-lc-rs",
"log",
"once_cell",
"ring",
@@ -5622,7 +5724,7 @@ dependencies = [
"security-framework",
"security-framework-sys",
"webpki-root-certs",
"windows-sys 0.52.0",
"windows-sys 0.59.0",
]
[[package]]
@@ -5633,10 +5735,11 @@ checksum = "f87165f0995f63a9fbeea62b64d10b4d9d8e78ec6d7d51fb2125fda7bb36788f"
[[package]]
name = "rustls-webpki"
version = "0.103.9"
version = "0.103.10"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d7df23109aa6c1567d1c575b9952556388da57401e4ace1d15f79eedad0d8f53"
checksum = "df33b2b81ac578cabaf06b89b0631153a3f416b0a886e8a7a1707fb51abbd1ef"
dependencies = [
"aws-lc-rs",
"ring",
"rustls-pki-types",
"untrusted",
@@ -6243,6 +6346,19 @@ dependencies = [
"der",
]
[[package]]
name = "sse-stream"
version = "0.2.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "eb4dc4d33c68ec1f27d386b5610a351922656e1fdf5c05bbaad930cd1519479a"
dependencies = [
"bytes",
"futures-util",
"http-body",
"http-body-util",
"pin-project-lite",
]
[[package]]
name = "stable_deref_trait"
version = "1.2.1"
@@ -6449,7 +6565,7 @@ dependencies = [
"tao-macros",
"unicode-segmentation",
"url",
"windows",
"windows 0.61.3",
"windows-core 0.61.2",
"windows-version",
"x11-dl",
@@ -6468,9 +6584,9 @@ dependencies = [
[[package]]
name = "tar"
version = "0.4.44"
version = "0.4.45"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1d863878d212c87a19c1a610eb53bb01fe12951c0501cf5a0d65f724914a667a"
checksum = "22692a6476a21fa75fdfc11d452fda482af402c008cdbaf3476414e122040973"
dependencies = [
"filetime",
"libc",
@@ -6538,7 +6654,7 @@ dependencies = [
"webkit2gtk",
"webview2-com",
"window-vibrancy",
"windows",
"windows 0.61.3",
]
[[package]]
@@ -6800,7 +6916,7 @@ dependencies = [
"url",
"webkit2gtk",
"webview2-com",
"windows",
"windows 0.61.3",
]
[[package]]
@@ -6825,7 +6941,7 @@ dependencies = [
"url",
"webkit2gtk",
"webview2-com",
"windows",
"windows 0.61.3",
"wry",
]
@@ -6886,7 +7002,7 @@ checksum = "0b1e66e07de489fe43a46678dd0b8df65e0c973909df1b60ba33874e297ba9b9"
dependencies = [
"quick-xml 0.37.5",
"thiserror 2.0.18",
"windows",
"windows 0.61.3",
"windows-version",
]
@@ -6900,7 +7016,7 @@ dependencies = [
"getrandom 0.4.2",
"once_cell",
"rustix 1.1.4",
"windows-sys 0.52.0",
"windows-sys 0.59.0",
]
[[package]]
@@ -7452,7 +7568,7 @@ checksum = "f2f6fb2847f6742cd76af783a2a2c49e9375d0a111c7bef6f71cd9e738c72d6e"
dependencies = [
"memoffset",
"tempfile",
"windows-sys 0.59.0",
"windows-sys 0.60.2",
]
[[package]]
@@ -8299,7 +8415,7 @@ checksum = "7130243a7a5b33c54a444e54842e6a9e133de08b5ad7b5861cd8ed9a6a5bc96a"
dependencies = [
"webview2-com-macros",
"webview2-com-sys",
"windows",
"windows 0.61.3",
"windows-core 0.61.2",
"windows-implement",
"windows-interface",
@@ -8323,7 +8439,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "381336cfffd772377d291702245447a5251a2ffa5bad679c99e61bc48bacbf9c"
dependencies = [
"thiserror 2.0.18",
"windows",
"windows 0.61.3",
"windows-core 0.61.2",
]
@@ -8349,7 +8465,7 @@ version = "0.1.11"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22"
dependencies = [
"windows-sys 0.52.0",
"windows-sys 0.59.0",
]
[[package]]
@@ -8399,11 +8515,23 @@ version = "0.61.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9babd3a767a4c1aef6900409f85f5d53ce2544ccdfaa86dad48c91782c6d6893"
dependencies = [
"windows-collections",
"windows-collections 0.2.0",
"windows-core 0.61.2",
"windows-future",
"windows-future 0.2.1",
"windows-link 0.1.3",
"windows-numerics",
"windows-numerics 0.2.0",
]
[[package]]
name = "windows"
version = "0.62.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "527fadee13e0c05939a6a05d5bd6eec6cd2e3dbd648b9f8e447c6518133d8580"
dependencies = [
"windows-collections 0.3.2",
"windows-core 0.62.2",
"windows-future 0.3.2",
"windows-numerics 0.3.1",
]
[[package]]
@@ -8415,6 +8543,15 @@ dependencies = [
"windows-core 0.61.2",
]
[[package]]
name = "windows-collections"
version = "0.3.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "23b2d95af1a8a14a3c7367e1ed4fc9c20e0a26e79551b1454d72583c97cc6610"
dependencies = [
"windows-core 0.62.2",
]
[[package]]
name = "windows-core"
version = "0.61.2"
@@ -8449,7 +8586,18 @@ checksum = "fc6a41e98427b19fe4b73c550f060b59fa592d7d686537eebf9385621bfbad8e"
dependencies = [
"windows-core 0.61.2",
"windows-link 0.1.3",
"windows-threading",
"windows-threading 0.1.0",
]
[[package]]
name = "windows-future"
version = "0.3.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e1d6f90251fe18a279739e78025bd6ddc52a7e22f921070ccdc67dde84c605cb"
dependencies = [
"windows-core 0.62.2",
"windows-link 0.2.1",
"windows-threading 0.2.1",
]
[[package]]
@@ -8496,6 +8644,16 @@ dependencies = [
"windows-link 0.1.3",
]
[[package]]
name = "windows-numerics"
version = "0.3.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6e2e40844ac143cdb44aead537bbf727de9b044e107a0f1220392177d15b0f26"
dependencies = [
"windows-core 0.62.2",
"windows-link 0.2.1",
]
[[package]]
name = "windows-result"
version = "0.3.4"
@@ -8634,6 +8792,15 @@ dependencies = [
"windows-link 0.1.3",
]
[[package]]
name = "windows-threading"
version = "0.2.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "3949bd5b99cafdf1c7ca86b43ca564028dfe27d66958f2470940f73d86d75b37"
dependencies = [
"windows-link 0.2.1",
]
[[package]]
name = "windows-version"
version = "0.1.7"
@@ -8977,7 +9144,7 @@ dependencies = [
"webkit2gtk",
"webkit2gtk-sys",
"webview2-com",
"windows",
"windows 0.61.3",
"windows-core 0.61.2",
"windows-version",
"x11-dl",
Cargo.toml
+3
View File
@@ -130,6 +130,9 @@ html-escape = "0.2"
# Lightweight regex
regex-lite = "0.1"
# MCP SDK (official Rust implementation)
rmcp = { version = "1.2", default-features = false, features = ["client", "transport-child-process", "transport-streamable-http-client-reqwest", "reqwest"] }
# Socket options (SO_REUSEADDR)
socket2 = "0.5"
crates/openfang-api/src/routes.rs
+12 -2
View File
@@ -367,7 +367,11 @@ pub async fn send_message(
// (not as a separate session message which the LLM may not process).
let content_blocks = if !req.attachments.is_empty() {
let image_blocks = resolve_attachments(&req.attachments);
if image_blocks.is_empty() { None } else { Some(image_blocks) }
if image_blocks.is_empty() {
None
} else {
Some(image_blocks)
}
} else {
None
};
@@ -3129,7 +3133,7 @@ pub async fn list_templates() -> impl IntoResponse {
let manifest_content = std::fs::read_to_string(&manifest_path).ok();
let description = manifest_content
.as_ref()
.and_then(|content| toml::from_str::<AgentManifest>(&content).ok())
.and_then(|content| toml::from_str::<AgentManifest>(content).ok())
.map(|m| m.description)
.unwrap_or_default();
@@ -4871,6 +4875,12 @@ pub async fn list_mcp_servers(State(state): State<Arc<AppState>>) -> impl IntoRe
"url": url,
})
}
openfang_types::config::McpTransportEntry::Http { url } => {
serde_json::json!({
"type": "http",
"url": url,
})
}
};
serde_json::json!({
"name": s.name,
crates/openfang-api/src/ws.rs
+6 -2
View File
@@ -1003,7 +1003,9 @@ fn map_stream_event(event: &StreamEvent, verbose: VerboseLevel) -> Option<serde_
"id": id,
"tool": name,
})),
StreamEvent::ToolUseEnd { id, name, input, .. } if name == "canvas_present" => {
StreamEvent::ToolUseEnd {
id, name, input, ..
} if name == "canvas_present" => {
let html = input.get("html").and_then(|v| v.as_str()).unwrap_or("");
let title = input
.get("title")
@@ -1017,7 +1019,9 @@ fn map_stream_event(event: &StreamEvent, verbose: VerboseLevel) -> Option<serde_
"title": title,
}))
}
StreamEvent::ToolUseEnd { id, name, input, .. } => match verbose {
StreamEvent::ToolUseEnd {
id, name, input, ..
} => match verbose {
VerboseLevel::Off => None,
VerboseLevel::On => {
let input_preview: String = serde_json::to_string(input)
crates/openfang-api/tests/api_integration_test.rs
+2
View File
@@ -78,6 +78,7 @@ async fn start_test_server_with_provider(
shutdown_notify: Arc::new(tokio::sync::Notify::new()),
clawhub_cache: dashmap::DashMap::new(),
provider_probe_cache: openfang_runtime::provider_health::ProbeCache::new(),
budget_config: Arc::new(tokio::sync::RwLock::new(Default::default())),
});
let app = Router::new()
@@ -707,6 +708,7 @@ async fn start_test_server_with_auth(api_key: &str) -> TestServer {
shutdown_notify: Arc::new(tokio::sync::Notify::new()),
clawhub_cache: dashmap::DashMap::new(),
provider_probe_cache: openfang_runtime::provider_health::ProbeCache::new(),
budget_config: Arc::new(tokio::sync::RwLock::new(Default::default())),
});
let api_key = state.kernel.config.api_key.trim().to_string();
crates/openfang-api/tests/daemon_lifecycle_test.rs
+2
View File
@@ -115,6 +115,7 @@ async fn test_full_daemon_lifecycle() {
shutdown_notify: Arc::new(tokio::sync::Notify::new()),
clawhub_cache: dashmap::DashMap::new(),
provider_probe_cache: openfang_runtime::provider_health::ProbeCache::new(),
budget_config: Arc::new(tokio::sync::RwLock::new(Default::default())),
});
let app = Router::new()
@@ -240,6 +241,7 @@ async fn test_server_immediate_responsiveness() {
shutdown_notify: Arc::new(tokio::sync::Notify::new()),
clawhub_cache: dashmap::DashMap::new(),
provider_probe_cache: openfang_runtime::provider_health::ProbeCache::new(),
budget_config: Arc::new(tokio::sync::RwLock::new(Default::default())),
});
let app = Router::new()
crates/openfang-api/tests/load_test.rs
+1
View File
@@ -59,6 +59,7 @@ async fn start_test_server() -> TestServer {
shutdown_notify: Arc::new(tokio::sync::Notify::new()),
clawhub_cache: dashmap::DashMap::new(),
provider_probe_cache: openfang_runtime::provider_health::ProbeCache::new(),
budget_config: Arc::new(tokio::sync::RwLock::new(Default::default())),
});
let app = Router::new()
crates/openfang-channels/src/feishu.rs
+43 -18
View File
@@ -1792,10 +1792,8 @@ mod tests {
#[test]
fn test_feishu_websocket_adapter_creation() {
let adapter = FeishuAdapter::new_websocket(
"cli_abc123".to_string(),
"app-secret-456".to_string(),
);
let adapter =
FeishuAdapter::new_websocket("cli_abc123".to_string(), "app-secret-456".to_string());
assert_eq!(adapter.name(), "feishu");
assert_eq!(adapter.connection_mode, FeishuConnectionMode::WebSocket);
assert_eq!(adapter.webhook_port, 0); // not used in WS mode
@@ -1803,8 +1801,7 @@ mod tests {
#[test]
fn test_connection_mode_default_is_webhook() {
let adapter =
FeishuAdapter::new("cli_abc123".to_string(), "secret".to_string(), 9000);
let adapter = FeishuAdapter::new("cli_abc123".to_string(), "secret".to_string(), 9000);
assert_eq!(adapter.connection_mode, FeishuConnectionMode::Webhook);
}
@@ -1836,14 +1833,32 @@ mod tests {
#[test]
fn test_combine_payload_multi_part() {
let headers_seq0 = vec![
FeishuWsHeader { key: "sum".to_string(), value: "2".to_string() },
FeishuWsHeader { key: "seq".to_string(), value: "0".to_string() },
FeishuWsHeader { key: "message_id".to_string(), value: "msg1".to_string() },
FeishuWsHeader {
key: "sum".to_string(),
value: "2".to_string(),
},
FeishuWsHeader {
key: "seq".to_string(),
value: "0".to_string(),
},
FeishuWsHeader {
key: "message_id".to_string(),
value: "msg1".to_string(),
},
];
let headers_seq1 = vec![
FeishuWsHeader { key: "sum".to_string(), value: "2".to_string() },
FeishuWsHeader { key: "seq".to_string(), value: "1".to_string() },
FeishuWsHeader { key: "message_id".to_string(), value: "msg1".to_string() },
FeishuWsHeader {
key: "sum".to_string(),
value: "2".to_string(),
},
FeishuWsHeader {
key: "seq".to_string(),
value: "1".to_string(),
},
FeishuWsHeader {
key: "message_id".to_string(),
value: "msg1".to_string(),
},
];
let mut parts = HashMap::new();
// First part — not yet complete
@@ -1856,10 +1871,14 @@ mod tests {
#[test]
fn test_ws_header_lookup() {
let headers = vec![
FeishuWsHeader { key: "service_id".to_string(), value: "svc_123".to_string() },
];
assert_eq!(ws_header(&headers, "service_id"), Some("svc_123".to_string()));
let headers = vec![FeishuWsHeader {
key: "service_id".to_string(),
value: "svc_123".to_string(),
}];
assert_eq!(
ws_header(&headers, "service_id"),
Some("svc_123".to_string())
);
assert_eq!(ws_header(&headers, "missing"), None);
}
@@ -1881,8 +1900,14 @@ mod tests {
service: 1,
method: 2,
headers: vec![
FeishuWsHeader { key: "log_id".to_string(), value: "log_abc".to_string() },
FeishuWsHeader { key: "type".to_string(), value: "event".to_string() },
FeishuWsHeader {
key: "log_id".to_string(),
value: "log_abc".to_string(),
},
FeishuWsHeader {
key: "type".to_string(),
value: "event".to_string(),
},
],
payload: Some(vec![]),
payload_encoding: None,
crates/openfang-channels/src/whatsapp.rs
+6 -3
View File
@@ -258,7 +258,8 @@ impl ChannelAdapter for WhatsAppAdapter {
"https://graph.facebook.com/v21.0/{}/messages",
self.phone_number_id
);
let resp = self.client
let resp = self
.client
.post(&api_url)
.bearer_auth(&*self.access_token)
.json(&body)
@@ -284,7 +285,8 @@ impl ChannelAdapter for WhatsAppAdapter {
"https://graph.facebook.com/v21.0/{}/messages",
self.phone_number_id
);
let resp = self.client
let resp = self
.client
.post(&api_url)
.bearer_auth(&*self.access_token)
.json(&body)
@@ -310,7 +312,8 @@ impl ChannelAdapter for WhatsAppAdapter {
"https://graph.facebook.com/v21.0/{}/messages",
self.phone_number_id
);
let resp = self.client
let resp = self
.client
.post(&api_url)
.bearer_auth(&*self.access_token)
.json(&body)
crates/openfang-channels/tests/bridge_integration_test.rs
+14 -3
View File
@@ -495,7 +495,10 @@ async fn test_bridge_manager_lifecycle() {
assert_eq!(sent.len(), 5, "Expected 5 responses, got {}", sent.len());
for (i, (_, text)) in sent.iter().enumerate() {
assert!(text.contains(&format!("message {i}")), "Expected 'message {i}' in: {text}");
assert!(
text.contains(&format!("message {i}")),
"Expected 'message {i}' in: {text}"
);
}
// Stop — should complete without hanging
@@ -544,11 +547,19 @@ async fn test_bridge_multiple_adapters() {
let tg_sent = tg_ref.get_sent();
assert_eq!(tg_sent.len(), 1);
assert!(tg_sent[0].1.contains("from telegram"), "Expected 'from telegram' in: {}", tg_sent[0].1);
assert!(
tg_sent[0].1.contains("from telegram"),
"Expected 'from telegram' in: {}",
tg_sent[0].1
);
let dc_sent = dc_ref.get_sent();
assert_eq!(dc_sent.len(), 1);
assert!(dc_sent[0].1.contains("from discord"), "Expected 'from discord' in: {}", dc_sent[0].1);
assert!(
dc_sent[0].1.contains("from discord"),
"Expected 'from discord' in: {}",
dc_sent[0].1
);
manager.stop().await;
}
crates/openfang-cli/src/main.rs
+3 -3
View File
@@ -886,7 +886,6 @@ fn write_stdout_safe(msg: &str) {
}
fn main() {
// Load ~/.openfang/.env into process environment (system env takes priority).
dotenv::load_dotenv();
@@ -2596,7 +2595,8 @@ decay_rate = 0.05
checks.push(serde_json::json!({"check": "mcp_server_config", "status": "warn", "name": server.name}));
}
}
openfang_types::config::McpTransportEntry::Sse { url } => {
openfang_types::config::McpTransportEntry::Sse { url }
| openfang_types::config::McpTransportEntry::Http { url } => {
if url.is_empty() {
if !json {
ui::check_warn(&format!(
@@ -6948,4 +6948,4 @@ args = ["-y", "@modelcontextprotocol/server-github"]
assert!(!is_openfang_path_line("# openfang config", dir));
assert!(!is_openfang_path_line("alias of=openfang", dir));
}
}
}
crates/openfang-extensions/src/bundled.rs
+3
View File
@@ -146,6 +146,9 @@ mod tests {
crate::McpTransportTemplate::Sse { .. } => {
panic!("{} unexpectedly uses SSE transport", id);
}
crate::McpTransportTemplate::Http { .. } => {
panic!("{} unexpectedly uses HTTP transport", id);
}
}
}
}
crates/openfang-extensions/src/lib.rs
+3
View File
@@ -88,6 +88,9 @@ pub enum McpTransportTemplate {
Sse {
url: String,
},
Http {
url: String,
},
}
/// An environment variable required by an integration.
crates/openfang-extensions/src/registry.rs
+4
View File
@@ -191,6 +191,9 @@ impl IntegrationRegistry {
crate::McpTransportTemplate::Sse { url } => {
McpTransportEntry::Sse { url: url.clone() }
}
crate::McpTransportTemplate::Http { url } => {
McpTransportEntry::Http { url: url.clone() }
}
};
let env: Vec<String> = template
.required_env
@@ -202,6 +205,7 @@ impl IntegrationRegistry {
transport,
timeout_secs: 30,
env,
headers: Vec::new(),
})
})
.collect()
crates/openfang-hands/src/bundled.rs
+99 -25
View File
@@ -166,7 +166,7 @@ mod tests {
assert!(def.tools.contains(&"event_publish".to_string()));
assert!(!def.settings.is_empty());
assert!(!def.dashboard.metrics.is_empty());
assert_eq!(def.agent.max_iterations, Some(80));
assert_eq!(def.agent.max_iterations, Some(25));
}
#[test]
@@ -315,38 +315,112 @@ mod tests {
assert_eq!(def.category, crate::HandCategory::Security);
assert!(def.skill_content.is_some());
// Required env vars
assert!(!def.requires.is_empty(), "infisical-sync must declare env var requirements");
assert!(
!def.requires.is_empty(),
"infisical-sync must declare env var requirements"
);
let req_keys: Vec<&str> = def.requires.iter().map(|r| r.key.as_str()).collect();
assert!(req_keys.contains(&"INFISICAL_URL"), "must require INFISICAL_URL");
assert!(req_keys.contains(&"INFISICAL_CLIENT_ID"), "must require INFISICAL_CLIENT_ID");
assert!(req_keys.contains(&"INFISICAL_CLIENT_SECRET"), "must require INFISICAL_CLIENT_SECRET");
assert!(
req_keys.contains(&"INFISICAL_URL"),
"must require INFISICAL_URL"
);
assert!(
req_keys.contains(&"INFISICAL_CLIENT_ID"),
"must require INFISICAL_CLIENT_ID"
);
assert!(
req_keys.contains(&"INFISICAL_CLIENT_SECRET"),
"must require INFISICAL_CLIENT_SECRET"
);
// Einstein scheduling tools
assert!(def.tools.contains(&"schedule_create".to_string()), "must have schedule_create");
assert!(def.tools.contains(&"schedule_list".to_string()), "must have schedule_list");
assert!(def.tools.contains(&"schedule_delete".to_string()), "must have schedule_delete");
assert!(
def.tools.contains(&"schedule_create".to_string()),
"must have schedule_create"
);
assert!(
def.tools.contains(&"schedule_list".to_string()),
"must have schedule_list"
);
assert!(
def.tools.contains(&"schedule_delete".to_string()),
"must have schedule_delete"
);
// Memory tools
assert!(def.tools.contains(&"memory_store".to_string()), "must have memory_store");
assert!(def.tools.contains(&"memory_recall".to_string()), "must have memory_recall");
assert!(
def.tools.contains(&"memory_store".to_string()),
"must have memory_store"
);
assert!(
def.tools.contains(&"memory_recall".to_string()),
"must have memory_recall"
);
// Knowledge graph tools
assert!(def.tools.contains(&"knowledge_add_entity".to_string()), "must have knowledge_add_entity");
assert!(def.tools.contains(&"knowledge_add_relation".to_string()), "must have knowledge_add_relation");
assert!(def.tools.contains(&"knowledge_query".to_string()), "must have knowledge_query");
assert!(
def.tools.contains(&"knowledge_add_entity".to_string()),
"must have knowledge_add_entity"
);
assert!(
def.tools.contains(&"knowledge_add_relation".to_string()),
"must have knowledge_add_relation"
);
assert!(
def.tools.contains(&"knowledge_query".to_string()),
"must have knowledge_query"
);
// Event bus
assert!(def.tools.contains(&"event_publish".to_string()), "must have event_publish");
assert!(
def.tools.contains(&"event_publish".to_string()),
"must have event_publish"
);
// Infisical-specific tools
assert!(def.tools.contains(&"shell_exec".to_string()), "must have shell_exec");
assert!(def.tools.contains(&"vault_set".to_string()), "must have vault_set");
assert!(def.tools.contains(&"vault_get".to_string()), "must have vault_get");
assert!(def.tools.contains(&"vault_list".to_string()), "must have vault_list");
assert!(def.tools.contains(&"vault_delete".to_string()), "must have vault_delete");
assert!(
def.tools.contains(&"shell_exec".to_string()),
"must have shell_exec"
);
assert!(
def.tools.contains(&"vault_set".to_string()),
"must have vault_set"
);
assert!(
def.tools.contains(&"vault_get".to_string()),
"must have vault_get"
);
assert!(
def.tools.contains(&"vault_list".to_string()),
"must have vault_list"
);
assert!(
def.tools.contains(&"vault_delete".to_string()),
"must have vault_delete"
);
// Dashboard
assert!(!def.dashboard.metrics.is_empty(), "must have dashboard metrics");
let metric_keys: Vec<&str> = def.dashboard.metrics.iter().map(|m| m.memory_key.as_str()).collect();
assert!(metric_keys.contains(&"infisical_sync_secrets_count"), "must have secrets_count metric");
assert!(metric_keys.contains(&"infisical_sync_last_sync"), "must have last_sync metric");
assert!(
!def.dashboard.metrics.is_empty(),
"must have dashboard metrics"
);
let metric_keys: Vec<&str> = def
.dashboard
.metrics
.iter()
.map(|m| m.memory_key.as_str())
.collect();
assert!(
metric_keys.contains(&"infisical_sync_secrets_count"),
"must have secrets_count metric"
);
assert!(
metric_keys.contains(&"infisical_sync_last_sync"),
"must have last_sync metric"
);
// Agent config
assert!(!def.agent.system_prompt.is_empty(), "must have system_prompt");
assert!(def.agent.temperature < 0.2, "security hand should use low temperature");
assert!(
!def.agent.system_prompt.is_empty(),
"must have system_prompt"
);
assert!(
def.agent.temperature < 0.2,
"security hand should use low temperature"
);
}
#[test]
crates/openfang-hands/src/registry.rs
+1 -3
View File
@@ -382,9 +382,7 @@ impl HandRegistry {
// Only non-optional requirements gate readiness.
// Optional requirements (e.g. chromium for browser hand) are nice-to-have;
// missing them results in "degraded" status but not "requirements not met".
let requirements_met = reqs
.iter()
.all(|(req, ok)| *ok || req.optional);
let requirements_met = reqs.iter().all(|(req, ok)| *ok || req.optional);
// A hand is active if at least one instance is in Active status.
let active = self
crates/openfang-kernel/src/heartbeat.rs
+42 -9
View File
@@ -169,7 +169,8 @@ pub fn check_agents(registry: &AgentRegistry, config: &HeartbeatConfig) -> Vec<H
// Periodic / Hand agents with long schedule intervals (e.g. 3600s) are
// also covered: they sit idle between ticks and their `last_active` stays
// near `created_at` until the first tick fires.
let never_active = (entry_ref.last_active - entry_ref.created_at).num_seconds() <= IDLE_GRACE_SECS;
let never_active =
(entry_ref.last_active - entry_ref.created_at).num_seconds() <= IDLE_GRACE_SECS;
if never_active && entry_ref.state == AgentState::Running {
debug!(
@@ -299,7 +300,12 @@ mod tests {
use std::collections::HashMap;
/// Helper: build a minimal AgentEntry for heartbeat tests.
fn make_entry(name: &str, state: AgentState, created_at: chrono::DateTime<Utc>, last_active: chrono::DateTime<Utc>) -> AgentEntry {
fn make_entry(
name: &str,
state: AgentState,
created_at: chrono::DateTime<Utc>,
last_active: chrono::DateTime<Utc>,
) -> AgentEntry {
AgentEntry {
id: AgentId::new(),
name: name.to_string(),
@@ -351,14 +357,22 @@ mod tests {
// statuses because it was never genuinely active.
let registry = crate::registry::AgentRegistry::new();
let five_min_ago = Utc::now() - Duration::seconds(300);
let idle_agent = make_entry("idle-agent", AgentState::Running, five_min_ago, five_min_ago);
let idle_agent = make_entry(
"idle-agent",
AgentState::Running,
five_min_ago,
five_min_ago,
);
registry.register(idle_agent).unwrap();
let config = HeartbeatConfig::default(); // timeout = 180s
let statuses = check_agents(&registry, &config);
// The idle agent should be skipped entirely
assert!(statuses.is_empty(), "idle agent should be skipped by heartbeat");
assert!(
statuses.is_empty(),
"idle agent should be skipped by heartbeat"
);
}
#[test]
@@ -368,14 +382,22 @@ mod tests {
let registry = crate::registry::AgentRegistry::new();
let ten_min_ago = Utc::now() - Duration::seconds(600);
let five_min_ago = Utc::now() - Duration::seconds(300);
let active_agent = make_entry("active-agent", AgentState::Running, ten_min_ago, five_min_ago);
let active_agent = make_entry(
"active-agent",
AgentState::Running,
ten_min_ago,
five_min_ago,
);
registry.register(active_agent).unwrap();
let config = HeartbeatConfig::default(); // timeout = 180s, inactive = ~300s
let statuses = check_agents(&registry, &config);
assert_eq!(statuses.len(), 1);
assert!(statuses[0].unresponsive, "active agent past timeout should be unresponsive");
assert!(
statuses[0].unresponsive,
"active agent past timeout should be unresponsive"
);
}
#[test]
@@ -391,7 +413,10 @@ mod tests {
let statuses = check_agents(&registry, &config);
assert_eq!(statuses.len(), 1);
assert!(!statuses[0].unresponsive, "recently active agent should not be unresponsive");
assert!(
!statuses[0].unresponsive,
"recently active agent should not be unresponsive"
);
}
#[test]
@@ -400,14 +425,22 @@ mod tests {
// even if it was never genuinely active.
let registry = crate::registry::AgentRegistry::new();
let five_min_ago = Utc::now() - Duration::seconds(300);
let crashed_agent = make_entry("crashed-idle", AgentState::Crashed, five_min_ago, five_min_ago);
let crashed_agent = make_entry(
"crashed-idle",
AgentState::Crashed,
five_min_ago,
five_min_ago,
);
registry.register(crashed_agent).unwrap();
let config = HeartbeatConfig::default();
let statuses = check_agents(&registry, &config);
assert_eq!(statuses.len(), 1);
assert!(statuses[0].unresponsive, "crashed agent should be marked unresponsive");
assert!(
statuses[0].unresponsive,
"crashed agent should be marked unresponsive"
);
}
#[test]
crates/openfang-kernel/src/kernel.rs
+34 -5
View File
@@ -1835,7 +1835,10 @@ impl OpenFangKernel {
granted_tools: tools.iter().map(|t| t.name.clone()).collect(),
recalled_memories: vec![],
skill_summary: Self::build_skill_summary_from(&skill_snapshot, &manifest.skills),
skill_prompt_context: Self::collect_prompt_context_from(&skill_snapshot, &manifest.skills),
skill_prompt_context: Self::collect_prompt_context_from(
&skill_snapshot,
&manifest.skills,
),
mcp_summary: if mcp_tool_count > 0 {
self.build_mcp_summary(&manifest.mcp_servers)
} else {
@@ -2393,7 +2396,10 @@ impl OpenFangKernel {
granted_tools: tools.iter().map(|t| t.name.clone()).collect(),
recalled_memories: vec![], // Recalled in agent_loop, not here
skill_summary: Self::build_skill_summary_from(&skill_snapshot, &manifest.skills),
skill_prompt_context: Self::collect_prompt_context_from(&skill_snapshot, &manifest.skills),
skill_prompt_context: Self::collect_prompt_context_from(
&skill_snapshot,
&manifest.skills,
),
mcp_summary: if mcp_tool_count > 0 {
self.build_mcp_summary(&manifest.mcp_servers)
} else {
@@ -4703,6 +4709,7 @@ impl OpenFangKernel {
args: args.clone(),
},
McpTransportEntry::Sse { url } => McpTransport::Sse { url: url.clone() },
McpTransportEntry::Http { url } => McpTransport::Http { url: url.clone() },
};
// Resolve env vars from vault/dotenv before passing to MCP subprocess.
@@ -4721,6 +4728,7 @@ impl OpenFangKernel {
transport,
timeout_secs: server_config.timeout_secs,
env: server_config.env.clone(),
headers: server_config.headers.clone(),
};
match McpConnection::connect(mcp_config).await {
@@ -4822,6 +4830,7 @@ impl OpenFangKernel {
args: args.clone(),
},
McpTransportEntry::Sse { url } => McpTransport::Sse { url: url.clone() },
McpTransportEntry::Http { url } => McpTransport::Http { url: url.clone() },
};
let mcp_config = McpServerConfig {
@@ -4829,6 +4838,7 @@ impl OpenFangKernel {
transport,
timeout_secs: server_config.timeout_secs,
env: server_config.env.clone(),
headers: server_config.headers.clone(),
};
self.extension_health.register(&server_config.name);
@@ -4940,6 +4950,7 @@ impl OpenFangKernel {
args: args.clone(),
},
McpTransportEntry::Sse { url } => McpTransport::Sse { url: url.clone() },
McpTransportEntry::Http { url } => McpTransport::Http { url: url.clone() },
};
let mcp_config = McpServerConfig {
@@ -4947,6 +4958,7 @@ impl OpenFangKernel {
transport,
timeout_secs: server_config.timeout_secs,
env: server_config.env.clone(),
headers: server_config.headers.clone(),
};
match McpConnection::connect(mcp_config).await {
@@ -5032,7 +5044,16 @@ impl OpenFangKernel {
agent_id: AgentId,
skill_snapshot: Option<&openfang_skills::registry::SkillRegistry>,
) -> Vec<ToolDefinition> {
let all_builtins = builtin_tool_definitions();
let all_builtins = if self.config.browser.enabled {
builtin_tool_definitions()
} else {
// When built-in browser is disabled (replaced by an external
// browser MCP server such as CamoFox), filter out browser_* tools.
builtin_tool_definitions()
.into_iter()
.filter(|t| !t.name.starts_with("browser_"))
.collect()
};
// Look up agent entry for profile, skill/MCP allowlists, and declared tools
let entry = self.registry.get(agent_id);
@@ -5165,10 +5186,18 @@ impl OpenFangKernel {
.unwrap_or_default();
if !tool_allowlist.is_empty() {
all_tools.retain(|t| tool_allowlist.iter().any(|a| a.to_lowercase() == t.name.to_lowercase()));
all_tools.retain(|t| {
tool_allowlist
.iter()
.any(|a| a.to_lowercase() == t.name.to_lowercase())
});
}
if !tool_blocklist.is_empty() {
all_tools.retain(|t| !tool_blocklist.iter().any(|b| b.to_lowercase() == t.name.to_lowercase()));
all_tools.retain(|t| {
!tool_blocklist
.iter()
.any(|b| b.to_lowercase() == t.name.to_lowercase())
});
}
// Step 5: Remove shell_exec if exec_policy denies it.
crates/openfang-memory/src/knowledge.rs
+2 -10
View File
@@ -100,11 +100,7 @@ impl KnowledgeStore {
let mut idx = 1;
if let Some(ref source) = pattern.source {
sql.push_str(&format!(
" AND (s.id = ?{} OR s.name = ?{})",
idx,
idx + 1
));
sql.push_str(&format!(" AND (s.id = ?{} OR s.name = ?{})", idx, idx + 1));
params.push(Box::new(source.clone()));
params.push(Box::new(source.clone()));
idx += 2;
@@ -117,11 +113,7 @@ impl KnowledgeStore {
idx += 1;
}
if let Some(ref target) = pattern.target {
sql.push_str(&format!(
" AND (t.id = ?{} OR t.name = ?{})",
idx,
idx + 1
));
sql.push_str(&format!(" AND (t.id = ?{} OR t.name = ?{})", idx, idx + 1));
params.push(Box::new(target.clone()));
params.push(Box::new(target.clone()));
idx += 2;
crates/openfang-memory/src/semantic.rs
+8 -16
View File
@@ -163,14 +163,7 @@ impl SemanticStore {
}
Err(e) => {
warn!(error = %e, "HTTP memory store failed, falling back to SQLite");
self.remember_sqlite(
agent_id,
content,
source,
scope,
metadata.clone(),
None,
)
self.remember_sqlite(agent_id, content, source, scope, metadata.clone(), None)
}
}
}
@@ -446,17 +439,13 @@ impl SemanticStore {
let created_at = r
.created_at
.map(|ms| {
chrono::DateTime::from_timestamp_millis(ms as i64)
.unwrap_or_else(|| Utc::now())
chrono::DateTime::from_timestamp_millis(ms as i64).unwrap_or_else(Utc::now)
})
.unwrap_or_else(|| Utc::now());
.unwrap_or_else(Utc::now);
MemoryFragment {
id: MemoryId::new(),
agent_id: filter
.as_ref()
.and_then(|f| f.agent_id)
.unwrap_or_else(AgentId::new),
agent_id: filter.as_ref().and_then(|f| f.agent_id).unwrap_or_default(),
content: r.content,
embedding: None,
metadata: HashMap::new(),
@@ -470,7 +459,10 @@ impl SemanticStore {
})
.collect();
debug!(count = fragments.len(), "Recalled memories via HTTP backend");
debug!(
count = fragments.len(),
"Recalled memories via HTTP backend"
);
Ok(fragments)
}
}
crates/openfang-memory/src/substrate.rs
+11 -3
View File
@@ -43,7 +43,11 @@ impl MemorySubstrate {
/// When `memory_config.backend == "http"` and `http_url`/`http_token_env` are set,
/// the semantic store routes `remember`/`recall` to the memory-api gateway.
/// All other stores (KV, knowledge graph, sessions) remain local SQLite.
pub fn open(db_path: &Path, decay_rate: f32, memory_config: &MemoryConfig) -> OpenFangResult<Self> {
pub fn open(
db_path: &Path,
decay_rate: f32,
memory_config: &MemoryConfig,
) -> OpenFangResult<Self> {
let conn = Connection::open(db_path).map_err(|e| OpenFangError::Memory(e.to_string()))?;
conn.execute_batch("PRAGMA journal_mode=WAL; PRAGMA busy_timeout=5000;")
.map_err(|e| OpenFangError::Memory(e.to_string()))?;
@@ -70,13 +74,17 @@ impl MemorySubstrate {
) -> SemanticStore {
#[cfg(feature = "http-memory")]
if memory_config.backend == "http" {
if let (Some(url), Some(token_env)) = (&memory_config.http_url, &memory_config.http_token_env) {
if let (Some(url), Some(token_env)) =
(&memory_config.http_url, &memory_config.http_token_env)
{
match crate::http_client::MemoryApiClient::new(url, token_env) {
Ok(client) => {
// Best-effort health check on startup
match client.health_check() {
Ok(()) => info!(url = %url, "HTTP memory backend connected"),
Err(e) => warn!(url = %url, error = %e, "HTTP memory backend health check failed, will retry on use"),
Err(e) => {
warn!(url = %url, error = %e, "HTTP memory backend health check failed, will retry on use")
}
}
return SemanticStore::new_with_http(conn, client);
}
crates/openfang-runtime/Cargo.toml
+2
View File
@@ -30,6 +30,8 @@ zeroize = { workspace = true }
dashmap = { workspace = true }
regex-lite = { workspace = true }
rusqlite = { workspace = true }
rmcp = { workspace = true }
http = "1"
tokio-tungstenite = "0.24"
shlex = "1"
crates/openfang-runtime/src/agent_loop.rs
+35 -8
View File
@@ -57,8 +57,15 @@ fn phantom_action_detected(text: &str) -> bool {
let lower = text.to_lowercase();
let action_verbs = ["sent ", "posted ", "emailed ", "delivered ", "forwarded "];
let channel_refs = [
"telegram", "whatsapp", "slack", "discord", "email", "channel",
"message sent", "successfully sent", "has been sent",
"telegram",
"whatsapp",
"slack",
"discord",
"email",
"channel",
"message sent",
"successfully sent",
"has been sent",
];
let has_action = action_verbs.iter().any(|v| lower.contains(v));
let has_channel = channel_refs.iter().any(|c| lower.contains(c));
@@ -272,7 +279,9 @@ pub async fn run_agent_loop(
// The LLM already received them via llm_messages above.
for msg in session.messages.iter_mut() {
if let MessageContent::Blocks(blocks) = &mut msg.content {
let had_images = blocks.iter().any(|b| matches!(b, ContentBlock::Image { .. }));
let had_images = blocks
.iter()
.any(|b| matches!(b, ContentBlock::Image { .. }));
if had_images {
blocks.retain(|b| !matches!(b, ContentBlock::Image { .. }));
if blocks.is_empty() {
@@ -389,7 +398,14 @@ pub async fn run_agent_loop(
// Call LLM with retry, error classification, and circuit breaker
let provider_name = manifest.model.provider.as_str();
let mut response = call_with_retry(&*driver, request, Some(provider_name), None, &manifest.fallback_models).await?;
let mut response = call_with_retry(
&*driver,
request,
Some(provider_name),
None,
&manifest.fallback_models,
)
.await?;
total_usage.input_tokens += response.usage.input_tokens;
total_usage.output_tokens += response.usage.output_tokens;
@@ -460,7 +476,10 @@ pub async fn run_agent_loop(
// One-shot retry: if the LLM returns empty text with no tool use,
// try once more before accepting the empty result.
// Triggers on first call OR when input_tokens=0 (silently failed request).
if text.trim().is_empty() && response.tool_calls.is_empty() && !response.has_any_content() {
if text.trim().is_empty()
&& response.tool_calls.is_empty()
&& !response.has_any_content()
{
let is_silent_failure =
response.usage.input_tokens == 0 && response.usage.output_tokens == 0;
if iteration == 0 || is_silent_failure {
@@ -505,7 +524,10 @@ pub async fn run_agent_loop(
// channel action (send, post, email, etc.) but never actually
// called the corresponding tool, re-prompt once to force real
// tool usage instead of hallucinated completion.
let text = if !any_tools_executed && iteration == 0 && phantom_action_detected(&text) {
let text = if !any_tools_executed
&& iteration == 0
&& phantom_action_detected(&text)
{
warn!(agent = %manifest.name, "Phantom action detected — re-prompting for real tool use");
messages.push(Message::assistant(text));
messages.push(Message::user(
@@ -1419,7 +1441,9 @@ pub async fn run_agent_loop_streaming(
// The LLM already received them via llm_messages above.
for msg in session.messages.iter_mut() {
if let MessageContent::Blocks(blocks) = &mut msg.content {
let had_images = blocks.iter().any(|b| matches!(b, ContentBlock::Image { .. }));
let had_images = blocks
.iter()
.any(|b| matches!(b, ContentBlock::Image { .. }));
if had_images {
blocks.retain(|b| !matches!(b, ContentBlock::Image { .. }));
if blocks.is_empty() {
@@ -1620,7 +1644,10 @@ pub async fn run_agent_loop_streaming(
// One-shot retry: if the LLM returns empty text with no tool use,
// try once more before accepting the empty result.
// Triggers on first call OR when input_tokens=0 (silently failed request).
if text.trim().is_empty() && response.tool_calls.is_empty() && !response.has_any_content() {
if text.trim().is_empty()
&& response.tool_calls.is_empty()
&& !response.has_any_content()
{
let is_silent_failure =
response.usage.input_tokens == 0 && response.usage.output_tokens == 0;
if iteration == 0 || is_silent_failure {
crates/openfang-runtime/src/drivers/anthropic.rs
+4 -6
View File
@@ -471,9 +471,8 @@ impl LlmDriver for AnthropicDriver {
input_json,
}) = blocks.get(block_idx)
{
let input: serde_json::Value =
serde_json::from_str(input_json)
.unwrap_or_else(|_| serde_json::json!({}));
let input: serde_json::Value = serde_json::from_str(input_json)
.unwrap_or_else(|_| serde_json::json!({}));
let _ = tx
.send(StreamEvent::ToolUseEnd {
id: id.clone(),
@@ -521,9 +520,8 @@ impl LlmDriver for AnthropicDriver {
name,
input_json,
} => {
let input: serde_json::Value =
serde_json::from_str(&input_json)
.unwrap_or_else(|_| serde_json::json!({}));
let input: serde_json::Value = serde_json::from_str(&input_json)
.unwrap_or_else(|_| serde_json::json!({}));
content.push(ContentBlock::ToolUse {
id: id.clone(),
name: name.clone(),
crates/openfang-runtime/src/drivers/claude_code.rs
+3 -7
View File
@@ -545,15 +545,11 @@ impl LlmDriver for ClaudeCodeDriver {
.join("")
})
.unwrap_or_default();
let text_chunk =
if !chunk.is_empty() { chunk } else { nested };
let text_chunk = if !chunk.is_empty() { chunk } else { nested };
if !text_chunk.is_empty() {
full_text.push_str(&text_chunk);
let _ = tx
.send(StreamEvent::TextDelta {
text: text_chunk,
})
.await;
let _ =
tx.send(StreamEvent::TextDelta { text: text_chunk }).await;
}
}
"result" | "done" | "complete" => {
crates/openfang-runtime/src/drivers/openai.rs
+8 -6
View File
@@ -681,9 +681,8 @@ impl LlmDriver for OpenAIDriver {
if let Some(calls) = choice.message.tool_calls {
for call in calls {
let input: serde_json::Value =
serde_json::from_str(&call.function.arguments)
.unwrap_or_else(|_| serde_json::json!({}));
let input: serde_json::Value = serde_json::from_str(&call.function.arguments)
.unwrap_or_else(|_| serde_json::json!({}));
content.push(ContentBlock::ToolUse {
id: call.id.clone(),
name: call.function.name.clone(),
@@ -1150,7 +1149,10 @@ impl LlmDriver for OpenAIDriver {
}
// Reasoning/thinking content delta (DeepSeek-R1, Qwen3 via LM Studio/Ollama)
if let Some(reasoning) = delta["reasoning_content"].as_str().or_else(|| delta["reasoning"].as_str()) {
if let Some(reasoning) = delta["reasoning_content"]
.as_str()
.or_else(|| delta["reasoning"].as_str())
{
if !reasoning.is_empty() {
reasoning_content.push_str(reasoning);
let _ = tx
@@ -1315,8 +1317,8 @@ impl LlmDriver for OpenAIDriver {
}
for (id, name, arguments) in &tool_accum {
let input: serde_json::Value = serde_json::from_str(arguments)
.unwrap_or_else(|_| serde_json::json!({}));
let input: serde_json::Value =
serde_json::from_str(arguments).unwrap_or_else(|_| serde_json::json!({}));
content.push(ContentBlock::ToolUse {
id: id.clone(),
name: name.clone(),
crates/openfang-runtime/src/drivers/vertex.rs
+4 -1
View File
@@ -449,7 +449,10 @@ fn convert_response(resp: VertexResponse) -> Result<CompletionResponse, LlmError
for part in vertex_content.parts {
match part {
VertexPart::Text { text } => {
content.push(ContentBlock::Text { text, provider_metadata: None });
content.push(ContentBlock::Text {
text,
provider_metadata: None,
});
}
VertexPart::FunctionCall { function_call } => {
tool_calls.push(ToolCall {
crates/openfang-runtime/src/mcp.rs
+202 -491
View File
@@ -1,16 +1,20 @@
//! MCP (Model Context Protocol) client — connect to external MCP servers.
//!
//! MCP uses JSON-RPC 2.0 over stdio or HTTP+SSE. This module lets OpenFang
//! agents use tools from any MCP server (100+ available: GitHub, filesystem,
//! databases, APIs, etc.).
//! Uses the official `rmcp` SDK for protocol handling. Supports:
//! - **stdio**: subprocess with JSON-RPC over stdin/stdout
//! - **sse**: deprecated HTTP+SSE transport (protocol version 2024-11-05)
//! - **http**: Streamable HTTP transport (protocol version 2025-03-26+)
//!
//! All MCP tools are namespaced with `mcp_{server}_{tool}` to prevent collisions.
use http::{HeaderName, HeaderValue};
use openfang_types::tool::ToolDefinition;
use rmcp::model::{CallToolRequestParams, ClientCapabilities, ClientInfo, Implementation};
use rmcp::service::RunningService;
use rmcp::{RoleClient, ServiceExt};
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::process::Stdio;
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use std::sync::Arc;
use tracing::{debug, info};
// ---------------------------------------------------------------------------
@@ -30,6 +34,12 @@ pub struct McpServerConfig {
/// Environment variables to pass through to the subprocess (sandboxed).
#[serde(default)]
pub env: Vec<String>,
/// Extra HTTP headers to send with every SSE / Streamable-HTTP request.
/// Each entry is `"Header-Name: value"`. Useful for authentication
/// (`Authorization: Bearer <token>`), API keys (`X-Api-Key: ...`),
/// or any custom headers required by a remote MCP server.
#[serde(default)]
pub headers: Vec<String>,
}
fn default_timeout() -> u64 {
@@ -46,8 +56,14 @@ pub enum McpTransport {
#[serde(default)]
args: Vec<String>,
},
/// HTTP Server-Sent Events.
/// Deprecated HTTP+SSE transport (protocol version 2024-11-05).
/// Uses POST for sending and SSE for receiving.
Sse { url: String },
/// Streamable HTTP transport (MCP 2025-03-26+).
/// Single endpoint, client MUST send Accept: application/json, text/event-stream.
/// Server responds with either JSON or SSE stream.
/// Supports Mcp-Session-Id for session management.
Http { url: String },
}
// ---------------------------------------------------------------------------
@@ -64,59 +80,9 @@ pub struct McpConnection {
/// Needed because `normalize_name` replaces hyphens with underscores,
/// but the server expects the original name (e.g. "list-connections").
original_names: HashMap<String, String>,
/// Transport handle for sending requests.
transport: McpTransportHandle,
/// Next JSON-RPC request ID.
next_id: u64,
}
/// Transport handle — abstraction over stdio subprocess or HTTP.
enum McpTransportHandle {
Stdio {
child: Box<tokio::process::Child>,
stdin: tokio::process::ChildStdin,
stdout: BufReader<tokio::process::ChildStdout>,
},
Sse {
client: reqwest::Client,
url: String,
},
}
/// JSON-RPC 2.0 request.
#[derive(Serialize)]
struct JsonRpcRequest {
jsonrpc: &'static str,
id: u64,
method: String,
#[serde(skip_serializing_if = "Option::is_none")]
params: Option<serde_json::Value>,
}
/// JSON-RPC 2.0 response.
#[derive(Deserialize)]
struct JsonRpcResponse {
#[allow(dead_code)]
jsonrpc: String,
#[allow(dead_code)]
id: Option<u64>,
result: Option<serde_json::Value>,
error: Option<JsonRpcError>,
}
/// JSON-RPC 2.0 error object.
#[derive(Debug, Deserialize)]
pub struct JsonRpcError {
pub code: i64,
pub message: String,
#[allow(dead_code)]
pub data: Option<serde_json::Value>,
}
impl std::fmt::Display for JsonRpcError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "JSON-RPC error {}: {}", self.code, self.message)
}
/// The rmcp client handle — type-erased because the concrete type
/// depends on which transport was used (stdio vs HTTP).
client: RunningService<RoleClient, ClientInfo>,
}
// ---------------------------------------------------------------------------
@@ -126,13 +92,17 @@ impl std::fmt::Display for JsonRpcError {
impl McpConnection {
/// Connect to an MCP server, perform handshake, and discover tools.
pub async fn connect(config: McpServerConfig) -> Result<Self, String> {
let transport = match &config.transport {
let client_info = ClientInfo::new(
ClientCapabilities::default(),
Implementation::new("openfang", env!("CARGO_PKG_VERSION")),
);
let client = match &config.transport {
McpTransport::Stdio { command, args } => {
Self::connect_stdio(command, args, &config.env).await?
Self::connect_stdio(command, args, &config.env, client_info).await?
}
McpTransport::Sse { url } => {
// SSRF check: reject private/localhost URLs unless explicitly configured
Self::connect_sse(url).await?
McpTransport::Sse { url } | McpTransport::Http { url } => {
Self::connect_http(url, &config.headers, client_info).await?
}
};
@@ -140,13 +110,9 @@ impl McpConnection {
config,
tools: Vec::new(),
original_names: HashMap::new(),
transport,
next_id: 1,
client,
};
// Initialize handshake
conn.initialize().await?;
// Discover tools
conn.discover_tools().await?;
@@ -159,76 +125,34 @@ impl McpConnection {
Ok(conn)
}
/// Send the MCP `initialize` handshake.
async fn initialize(&mut self) -> Result<(), String> {
let params = serde_json::json!({
"protocolVersion": "2024-11-05",
"capabilities": {},
"clientInfo": {
"name": "openfang",
"version": env!("CARGO_PKG_VERSION")
}
});
let response = self.send_request("initialize", Some(params)).await?;
if let Some(result) = response {
debug!(
server = %self.config.name,
server_info = %result,
"MCP initialize response"
);
}
// Send initialized notification (no response expected)
self.send_notification("notifications/initialized", None)
.await?;
Ok(())
}
/// Discover available tools via `tools/list`.
async fn discover_tools(&mut self) -> Result<(), String> {
let response = self.send_request("tools/list", None).await?;
let tools = self
.client
.list_all_tools()
.await
.map_err(|e| format!("Failed to list MCP tools: {e}"))?;
if let Some(result) = response {
if let Some(tools_array) = result.get("tools").and_then(|t| t.as_array()) {
let server_name = &self.config.name;
for tool in tools_array {
let raw_name = tool["name"].as_str().unwrap_or("unnamed");
let description = tool["description"].as_str().unwrap_or("");
let input_schema = tool
.get("inputSchema")
.cloned()
.and_then(|v| {
// Ensure input_schema is a JSON object. MCP servers may
// return it as a string, null, or omit it entirely.
match &v {
serde_json::Value::Object(_) => Some(v),
serde_json::Value::String(s) => {
serde_json::from_str::<serde_json::Value>(s)
.ok()
.filter(|p| p.is_object())
}
_ => None,
}
})
.unwrap_or(serde_json::json!({"type": "object"}));
let server_name = &self.config.name;
for tool in &tools {
let raw_name = &tool.name;
let description = tool.description.as_deref().unwrap_or("");
// Namespace: mcp_{server}_{tool}
let namespaced = format_mcp_tool_name(server_name, raw_name);
let input_schema = serde_json::to_value(&tool.input_schema)
.unwrap_or(serde_json::json!({"type": "object"}));
// Store original name so we can send it back to the server
self.original_names
.insert(namespaced.clone(), raw_name.to_string());
// Namespace: mcp_{server}_{tool}
let namespaced = format_mcp_tool_name(server_name, raw_name);
self.tools.push(ToolDefinition {
name: namespaced,
description: format!("[MCP:{server_name}] {description}"),
input_schema,
});
}
}
// Store original name so we can send it back to the server
self.original_names
.insert(namespaced.clone(), raw_name.to_string());
self.tools.push(ToolDefinition {
name: namespaced,
description: format!("[MCP:{server_name}] {description}"),
input_schema,
});
}
Ok(())
@@ -243,40 +167,38 @@ impl McpConnection {
arguments: &serde_json::Value,
) -> Result<String, String> {
// Look up the original tool name from the server (preserves hyphens etc.)
let raw_name = self
let raw_name: String = self
.original_names
.get(name)
.map(|s| s.as_str())
.or_else(|| strip_mcp_prefix(&self.config.name, name))
.unwrap_or(name);
.cloned()
.or_else(|| strip_mcp_prefix(&self.config.name, name).map(|s| s.to_string()))
.unwrap_or_else(|| name.to_string());
let params = serde_json::json!({
"name": raw_name,
"arguments": arguments,
});
let args = arguments.as_object().cloned().unwrap_or_default();
let response = self.send_request("tools/call", Some(params)).await?;
debug!(tool = %raw_name, server = %self.config.name, "MCP tool call");
match response {
Some(result) => {
// Extract text content from the response
if let Some(content) = result.get("content").and_then(|c| c.as_array()) {
let texts: Vec<&str> = content
.iter()
.filter_map(|item| {
if item["type"].as_str() == Some("text") {
item["text"].as_str()
} else {
None
}
})
.collect();
Ok(texts.join("\n"))
} else {
Ok(result.to_string())
}
}
None => Err("No result from MCP tools/call".to_string()),
let params = CallToolRequestParams::new(raw_name).with_arguments(args);
let result = self
.client
.call_tool(params)
.await
.map_err(|e| format!("MCP tool call failed: {e}"))?;
// Extract text content from the response.
// `Content` is `Annotated<RawContent>` which Derefs to `RawContent`.
let texts: Vec<&str> = result
.content
.iter()
.filter_map(|item| item.as_text().map(|tc| tc.text.as_str()))
.collect();
if texts.is_empty() {
// Fallback: serialize the entire result
Ok(serde_json::to_string(&result).unwrap_or_default())
} else {
Ok(texts.join("\n"))
}
}
@@ -290,294 +212,124 @@ impl McpConnection {
&self.config.name
}
// --- Transport helpers ---
async fn send_request(
&mut self,
method: &str,
params: Option<serde_json::Value>,
) -> Result<Option<serde_json::Value>, String> {
let id = self.next_id;
self.next_id += 1;
let request = JsonRpcRequest {
jsonrpc: "2.0",
id,
method: method.to_string(),
params,
};
let request_json = serde_json::to_string(&request)
.map_err(|e| format!("Failed to serialize request: {e}"))?;
debug!(method, id, "MCP request");
match &mut self.transport {
McpTransportHandle::Stdio { stdin, stdout, .. } => {
// Write request + newline
stdin
.write_all(request_json.as_bytes())
.await
.map_err(|e| format!("Failed to write to MCP stdin: {e}"))?;
stdin
.write_all(b"\n")
.await
.map_err(|e| format!("Failed to write newline: {e}"))?;
stdin
.flush()
.await
.map_err(|e| format!("Failed to flush stdin: {e}"))?;
// Read response lines until we find one matching our request ID.
// MCP servers may send notifications or log lines before the
// actual response.
let timeout_dur = tokio::time::Duration::from_secs(self.config.timeout_secs);
let deadline = tokio::time::Instant::now() + timeout_dur;
loop {
let mut line = String::new();
let remaining = deadline.saturating_duration_since(tokio::time::Instant::now());
if remaining.is_zero() {
return Err("MCP request timed out".to_string());
}
match tokio::time::timeout(remaining, stdout.read_line(&mut line)).await {
Ok(Ok(0)) => return Err("MCP server closed connection".to_string()),
Ok(Ok(_)) => {}
Ok(Err(e)) => return Err(format!("Failed to read MCP response: {e}")),
Err(_) => return Err("MCP request timed out".to_string()),
}
let trimmed = line.trim();
if trimmed.is_empty() {
continue;
}
// Try to parse as JSON-RPC response
let parsed: Result<JsonRpcResponse, _> = serde_json::from_str(trimmed);
match parsed {
Ok(response) if response.id == Some(id) => {
if let Some(err) = response.error {
return Err(format!("{err}"));
}
return Ok(response.result);
}
Ok(_other) => {
// Response for a different ID or notification — skip
debug!("MCP: skipping non-matching response line");
continue;
}
Err(_) => {
// Not valid JSON-RPC — skip (could be a log line)
debug!("MCP: skipping non-JSON line from server");
continue;
}
}
}
}
McpTransportHandle::Sse { client, url } => {
let response = client
.post(url.as_str())
.json(&request)
.timeout(std::time::Duration::from_secs(self.config.timeout_secs))
.send()
.await
.map_err(|e| format!("MCP SSE request failed: {e}"))?;
if !response.status().is_success() {
return Err(format!("MCP SSE returned {}", response.status()));
}
let body = response
.text()
.await
.map_err(|e| format!("Failed to read SSE response: {e}"))?;
// Handle Streamable HTTP MCP responses that use SSE framing
// (e.g. "event: message\ndata: {...}\n\n"). Extract the JSON
// from the last `data:` line if the body looks like SSE.
let json_body = if body.trim_start().starts_with("event:") || body.trim_start().starts_with("data:") {
body.lines()
.filter_map(|line| line.strip_prefix("data: ").or_else(|| line.strip_prefix("data:")))
.filter(|s| !s.is_empty())
.last()
.unwrap_or(&body)
.to_string()
} else {
body
};
let rpc_response: JsonRpcResponse = serde_json::from_str(&json_body)
.map_err(|e| format!("Invalid MCP SSE JSON-RPC response: {e}"))?;
if let Some(err) = rpc_response.error {
return Err(format!("{err}"));
}
Ok(rpc_response.result)
}
}
}
async fn send_notification(
&mut self,
method: &str,
params: Option<serde_json::Value>,
) -> Result<(), String> {
let notification = serde_json::json!({
"jsonrpc": "2.0",
"method": method,
"params": params.unwrap_or(serde_json::json!({})),
});
let json = serde_json::to_string(&notification)
.map_err(|e| format!("Failed to serialize notification: {e}"))?;
match &mut self.transport {
McpTransportHandle::Stdio { stdin, .. } => {
stdin
.write_all(json.as_bytes())
.await
.map_err(|e| format!("Write notification: {e}"))?;
stdin
.write_all(b"\n")
.await
.map_err(|e| format!("Write newline: {e}"))?;
stdin.flush().await.map_err(|e| format!("Flush: {e}"))?;
}
McpTransportHandle::Sse { client, url } => {
let _ = client.post(url.as_str()).json(&notification).send().await;
}
}
Ok(())
}
// -- Transport constructors -----------------------------------------------
/// Connect using stdio transport (subprocess).
async fn connect_stdio(
command: &str,
args: &[String],
env_whitelist: &[String],
) -> Result<McpTransportHandle, String> {
client_info: ClientInfo,
) -> Result<RunningService<RoleClient, ClientInfo>, String> {
use rmcp::transport::{ConfigureCommandExt, TokioChildProcess};
use tokio::process::Command;
// Validate command path (no path traversal)
if command.contains("..") {
return Err("MCP command path contains '..': rejected".to_string());
}
// On Windows, npm/npx install as .cmd batch wrappers. Detect and adapt.
let resolved_command: String = if cfg!(windows) {
// If the user already specified .cmd/.bat, use as-is
if command.ends_with(".cmd") || command.ends_with(".bat") {
command.to_string()
} else {
// Check if the .cmd variant exists on PATH
let cmd_variant = format!("{command}.cmd");
let has_cmd = std::env::var("PATH")
.unwrap_or_default()
.split(';')
.any(|dir| std::path::Path::new(dir).join(&cmd_variant).exists());
if has_cmd {
cmd_variant
} else {
command.to_string()
let cmd_str = command.to_string();
let args_vec: Vec<String> = args.to_vec();
let env_list: Vec<String> = env_whitelist.to_vec();
let transport = TokioChildProcess::new(Command::new(&cmd_str).configure(move |cmd| {
for arg in &args_vec {
cmd.arg(arg);
}
// Sandbox: clear environment, only pass whitelisted vars
cmd.env_clear();
for var_name in &env_list {
if let Ok(val) = std::env::var(var_name) {
cmd.env(var_name, val);
}
}
} else {
command.to_string()
// Always pass PATH for binary resolution
if let Ok(path) = std::env::var("PATH") {
cmd.env("PATH", path);
}
// On Windows, npm/node need extra vars
if cfg!(windows) {
for var in &[
"APPDATA",
"LOCALAPPDATA",
"USERPROFILE",
"SystemRoot",
"TEMP",
"TMP",
"HOME",
"HOMEDRIVE",
"HOMEPATH",
] {
if let Ok(val) = std::env::var(var) {
cmd.env(var, val);
}
}
}
}))
.map_err(|e| format!("Failed to spawn MCP server '{cmd_str}': {e}"))?;
let client = client_info
.serve(transport)
.await
.map_err(|e| format!("MCP stdio handshake failed: {e}"))?;
Ok(client)
}
/// Connect using Streamable HTTP transport (or SSE fallback via the same endpoint).
///
/// The `rmcp` SDK's `StreamableHttpClientTransport` handles the full
/// Streamable HTTP protocol: Accept headers, Mcp-Session-Id tracking,
/// SSE stream parsing, and content-type negotiation.
async fn connect_http(
url: &str,
headers: &[String],
client_info: ClientInfo,
) -> Result<RunningService<RoleClient, ClientInfo>, String> {
use rmcp::transport::streamable_http_client::StreamableHttpClientTransportConfig;
use rmcp::transport::StreamableHttpClientTransport;
Self::check_ssrf(url)?;
// Parse custom headers (e.g., "Authorization: Bearer <token>").
let mut custom_headers: HashMap<HeaderName, HeaderValue> = HashMap::new();
for header_str in headers {
if let Some((name, value)) = header_str.split_once(':') {
let name = name.trim();
let value = value.trim();
if let (Ok(hn), Ok(hv)) = (
HeaderName::from_bytes(name.as_bytes()),
HeaderValue::from_str(value),
) {
custom_headers.insert(hn, hv);
}
}
}
let config = StreamableHttpClientTransportConfig {
uri: Arc::from(url),
custom_headers,
..Default::default()
};
let mut cmd = tokio::process::Command::new(&resolved_command);
cmd.args(args);
cmd.stdin(Stdio::piped());
cmd.stdout(Stdio::piped());
cmd.stderr(Stdio::piped());
let transport = StreamableHttpClientTransport::from_config(config);
// Sandbox: clear environment, only pass whitelisted vars
cmd.env_clear();
for var_name in env_whitelist {
if let Ok(val) = std::env::var(var_name) {
cmd.env(var_name, val);
}
}
// Always pass PATH for binary resolution
if let Ok(path) = std::env::var("PATH") {
cmd.env("PATH", path);
}
// On Windows, npm/node need APPDATA, USERPROFILE, LOCALAPPDATA, and SystemRoot
if cfg!(windows) {
for var in &[
"APPDATA",
"LOCALAPPDATA",
"USERPROFILE",
"SystemRoot",
"TEMP",
"TMP",
"HOME",
"HOMEDRIVE",
"HOMEPATH",
] {
if let Ok(val) = std::env::var(var) {
cmd.env(var, val);
}
}
}
let client = client_info
.serve(transport)
.await
.map_err(|e| format!("MCP HTTP connection failed: {e}"))?;
let mut child = cmd
.spawn()
.map_err(|e| format!("Failed to spawn MCP server '{resolved_command}': {e}"))?;
// Log stderr in background for debugging MCP server issues
if let Some(stderr) = child.stderr.take() {
let cmd_name = resolved_command.clone();
tokio::spawn(async move {
use tokio::io::AsyncBufReadExt;
let reader = tokio::io::BufReader::new(stderr);
let mut lines = reader.lines();
while let Ok(Some(line)) = lines.next_line().await {
tracing::debug!(mcp_server = %cmd_name, "stderr: {line}");
}
});
}
let stdin = child
.stdin
.take()
.ok_or("Failed to capture MCP server stdin")?;
let stdout = child
.stdout
.take()
.ok_or("Failed to capture MCP server stdout")?;
Ok(McpTransportHandle::Stdio {
child: Box::new(child),
stdin,
stdout: BufReader::new(stdout),
})
Ok(client)
}
async fn connect_sse(url: &str) -> Result<McpTransportHandle, String> {
// Basic SSRF check: reject obviously private URLs
/// Basic SSRF check: reject obviously private/metadata URLs.
fn check_ssrf(url: &str) -> Result<(), String> {
let lower = url.to_lowercase();
if lower.contains("169.254.169.254") || lower.contains("metadata.google") {
return Err("SSRF: MCP SSE URL targets metadata endpoint".to_string());
}
let client = reqwest::Client::builder()
.timeout(std::time::Duration::from_secs(30))
.build()
.map_err(|e| format!("Failed to create HTTP client: {e}"))?;
Ok(McpTransportHandle::Sse {
client,
url: url.to_string(),
})
}
}
impl Drop for McpConnection {
fn drop(&mut self) {
if let McpTransportHandle::Stdio { ref mut child, .. } = self.transport {
// Best-effort kill of the subprocess
let _ = child.start_kill();
return Err("SSRF: MCP URL targets metadata endpoint".to_string());
}
Ok(())
}
}
@@ -668,16 +420,12 @@ mod tests {
#[test]
fn test_hyphenated_tool_name_preserved() {
// Tool names with hyphens get normalized to underscores for namespacing,
// but original_names map preserves the original for call_tool dispatch.
let namespaced = format_mcp_tool_name("sqlcl", "list-connections");
assert_eq!(namespaced, "mcp_sqlcl_list_connections");
// Simulate what discover_tools does
let mut original_names = HashMap::new();
original_names.insert(namespaced.clone(), "list-connections".to_string());
// call_tool should resolve to original hyphenated name
let raw = original_names
.get(&namespaced)
.map(|s| s.as_str())
@@ -696,25 +444,21 @@ mod tests {
#[test]
fn test_extract_mcp_server_from_known_with_hyphens() {
// Server "bocha-search" normalized to "bocha_search" in tool prefix
let servers = vec!["bocha-search", "github"];
let tool = "mcp_bocha_search_bocha_web_search";
assert_eq!(
extract_mcp_server_from_known(tool, &servers),
Some("bocha-search")
);
// Simple server name still works
assert_eq!(
extract_mcp_server_from_known("mcp_github_create_issue", &servers),
Some("github")
);
// Non-MCP tool returns None
assert_eq!(extract_mcp_server_from_known("file_read", &servers), None);
}
#[test]
fn test_extract_mcp_server_from_known_longest_match() {
// "my-api" and "my-api-v2" — should match the longer one
let servers = vec!["my-api", "my-api-v2"];
assert_eq!(
extract_mcp_server_from_known("mcp_my_api_v2_get_users", &servers),
@@ -726,60 +470,6 @@ mod tests {
);
}
#[test]
fn test_mcp_jsonrpc_initialize() {
// Verify the initialize request structure
let request = JsonRpcRequest {
jsonrpc: "2.0",
id: 1,
method: "initialize".to_string(),
params: Some(serde_json::json!({
"protocolVersion": "2024-11-05",
"capabilities": {},
"clientInfo": {
"name": "openfang",
"version": "0.1.0"
}
})),
};
let json = serde_json::to_string(&request).unwrap();
assert!(json.contains("initialize"));
assert!(json.contains("protocolVersion"));
assert!(json.contains("openfang"));
}
#[test]
fn test_mcp_jsonrpc_tools_list() {
// Simulate a tools/list response
let response_json = r#"{
"jsonrpc": "2.0",
"id": 2,
"result": {
"tools": [
{
"name": "create_issue",
"description": "Create a GitHub issue",
"inputSchema": {
"type": "object",
"properties": {
"title": {"type": "string"},
"body": {"type": "string"}
},
"required": ["title"]
}
}
]
}
}"#;
let response: JsonRpcResponse = serde_json::from_str(response_json).unwrap();
assert!(response.error.is_none());
let result = response.result.unwrap();
let tools = result["tools"].as_array().unwrap();
assert_eq!(tools.len(), 1);
assert_eq!(tools[0]["name"].as_str().unwrap(), "create_issue");
}
#[test]
fn test_mcp_transport_config_serde() {
let config = McpServerConfig {
@@ -793,6 +483,7 @@ mod tests {
},
timeout_secs: 30,
env: vec!["GITHUB_PERSONAL_ACCESS_TOKEN".to_string()],
headers: vec![],
};
let json = serde_json::to_string(&config).unwrap();
@@ -817,6 +508,7 @@ mod tests {
},
timeout_secs: 60,
env: vec![],
headers: vec![],
};
let json = serde_json::to_string(&sse_config).unwrap();
let back: McpServerConfig = serde_json::from_str(&json).unwrap();
@@ -824,5 +516,24 @@ mod tests {
McpTransport::Sse { url } => assert_eq!(url, "https://example.com/mcp"),
_ => panic!("Expected SSE transport"),
}
// HTTP (Streamable HTTP) variant
let http_config = McpServerConfig {
name: "atlassian".to_string(),
transport: McpTransport::Http {
url: "https://mcp.atlassian.com/v1/mcp".to_string(),
},
timeout_secs: 120,
env: vec![],
headers: vec!["Authorization: Bearer test-token-456".to_string()],
};
let json = serde_json::to_string(&http_config).unwrap();
let back: McpServerConfig = serde_json::from_str(&json).unwrap();
match back.transport {
McpTransport::Http { url } => {
assert_eq!(url, "https://mcp.atlassian.com/v1/mcp")
}
_ => panic!("Expected Http transport"),
}
}
}
crates/openfang-runtime/src/tool_runner.rs
+6 -3
View File
@@ -3651,10 +3651,13 @@ mod tests {
None, // process_manager
)
.await;
// Should NOT be "Permission denied" — it should normalize to file_write
// and pass the capability check. It will fail for other reasons (path validation).
// Should NOT be the capability-check denial — it should normalize to file_write
// and pass the capability check. It may fail for other reasons (path validation,
// OS-level errors), but not the agent capability gate.
assert!(
!result.content.contains("Permission denied"),
!result
.content
.contains("does not have capability to use tool"),
"fs-write should normalize to file_write and pass capability check, got: {}",
result.content
);
crates/openfang-skills/src/registry.rs
+17 -3
View File
@@ -636,7 +636,12 @@ input_schema = { type = "object" }
registry.load_all().unwrap();
assert_eq!(registry.count(), 1);
assert_eq!(
registry.get("shared-skill").unwrap().manifest.skill.description,
registry
.get("shared-skill")
.unwrap()
.manifest
.skill
.description,
"Global version"
);
@@ -645,9 +650,18 @@ input_schema = { type = "object" }
snapshot.load_workspace_skills(ws_dir.path()).unwrap();
// The workspace version must override the global version
assert_eq!(snapshot.count(), 1, "Duplicate should be overwritten, not added");
assert_eq!(
snapshot.get("shared-skill").unwrap().manifest.skill.description,
snapshot.count(),
1,
"Duplicate should be overwritten, not added"
);
assert_eq!(
snapshot
.get("shared-skill")
.unwrap()
.manifest
.skill
.description,
"Workspace override version",
"Workspace skill must override global skill (#808)"
);
crates/openfang-types/src/config.rs
+11
View File
@@ -310,6 +310,10 @@ impl Default for WebFetchConfig {
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(default)]
pub struct BrowserConfig {
/// Enable the built-in CDP browser tools (browser_navigate, browser_click,
/// etc.). Set to `false` when using an external browser MCP server such as
/// CamoFox, which replaces these tools with its own set.
pub enabled: bool,
/// Run browser in headless mode (no visible window).
pub headless: bool,
/// Viewport width in pixels.
@@ -329,6 +333,7 @@ pub struct BrowserConfig {
impl Default for BrowserConfig {
fn default() -> Self {
Self {
enabled: true,
headless: true,
viewport_width: 1280,
viewport_height: 720,
@@ -1224,6 +1229,10 @@ pub struct McpServerConfigEntry {
/// Environment variables to pass through (e.g., ["GITHUB_PERSONAL_ACCESS_TOKEN"]).
#[serde(default)]
pub env: Vec<String>,
/// Extra HTTP headers for SSE / Streamable-HTTP transports.
/// Each entry is `"Header-Name: value"` (e.g., `"Authorization: Bearer <token>"`).
#[serde(default)]
pub headers: Vec<String>,
}
fn default_mcp_timeout() -> u64 {
@@ -1242,6 +1251,8 @@ pub enum McpTransportEntry {
},
/// HTTP Server-Sent Events.
Sse { url: String },
/// Streamable HTTP (MCP 2025-03-26+).
Http { url: String },
}
/// A2A (Agent-to-Agent) protocol configuration.