diff --git a/src/openhuman/config/mod.rs b/src/openhuman/config/mod.rs index da019c7c7..82bffc4bb 100644 --- a/src/openhuman/config/mod.rs +++ b/src/openhuman/config/mod.rs @@ -29,7 +29,8 @@ pub use schema::{ CurlConfig, DelegateAgentConfig, DictationActivationMode, DictationConfig, DiscordConfig, DockerRuntimeConfig, EmbeddingRouteConfig, GitbooksConfig, HeartbeatConfig, HttpRequestConfig, IMessageConfig, IntegrationToggle, IntegrationsConfig, LarkConfig, LearningConfig, LlmBackend, - LocalAiConfig, MatrixConfig, MeetConfig, MemoryConfig, MemoryTreeConfig, ModelRouteConfig, + LocalAiConfig, MatrixConfig, McpAuthConfig, McpClientConfig, McpClientIdentityConfig, + McpServerConfig, MeetConfig, MemoryConfig, MemoryTreeConfig, ModelRouteConfig, MultimodalConfig, ObservabilityConfig, ProxyConfig, ProxyScope, ReflectionSource, ReliabilityConfig, ResourceLimitsConfig, RuntimeConfig, SandboxBackend, SandboxConfig, SchedulerConfig, SchedulerGateConfig, SchedulerGateMode, ScreenIntelligenceConfig, diff --git a/src/openhuman/config/schema/mod.rs b/src/openhuman/config/schema/mod.rs index 9bad5167f..49b57a1fb 100644 --- a/src/openhuman/config/schema/mod.rs +++ b/src/openhuman/config/schema/mod.rs @@ -68,8 +68,9 @@ pub use storage_memory::{ }; pub use tools::{ BrowserComputerUseConfig, BrowserConfig, ComposioConfig, ComputerControlConfig, CurlConfig, - GitbooksConfig, HttpRequestConfig, IntegrationToggle, IntegrationsConfig, MultimodalConfig, - SecretsConfig, SeltzConfig, WebSearchConfig, COMPOSIO_MODE_BACKEND, COMPOSIO_MODE_DIRECT, + GitbooksConfig, HttpRequestConfig, IntegrationToggle, IntegrationsConfig, McpAuthConfig, + McpClientConfig, McpClientIdentityConfig, McpServerConfig, MultimodalConfig, SecretsConfig, + SeltzConfig, WebSearchConfig, COMPOSIO_MODE_BACKEND, COMPOSIO_MODE_DIRECT, }; pub use update::{UpdateConfig, UpdateRestartStrategy}; mod voice_server; diff --git a/src/openhuman/config/schema/tools.rs b/src/openhuman/config/schema/tools.rs index ebd30086d..333d0997c 100644 --- a/src/openhuman/config/schema/tools.rs +++ b/src/openhuman/config/schema/tools.rs @@ -3,6 +3,7 @@ use super::defaults; use schemars::JsonSchema; use serde::{Deserialize, Serialize}; +use std::collections::HashMap; #[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] #[serde(default)] @@ -223,6 +224,147 @@ impl Default for GitbooksConfig { } } +#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] +#[serde(default)] +pub struct McpServerConfig { + /// Stable server slug used by the agent-facing bridge tools. + #[serde(default)] + pub name: String, + /// MCP endpoint URL. Current implementation supports stateless + /// Streamable HTTP / JSON responses. + #[serde(default)] + pub endpoint: String, + /// Optional stdio command for local MCP servers. When set, the + /// client launches this command as a subprocess and speaks newline- + /// delimited JSON-RPC over stdin/stdout per the MCP stdio transport. + #[serde(default)] + pub command: String, + /// Command-line arguments for stdio MCP servers. + #[serde(default)] + pub args: Vec, + /// Extra environment variables for stdio MCP servers. MCP stdio auth + /// is typically passed this way. + #[serde(default)] + pub env: HashMap, + /// Optional working directory for stdio MCP servers. + #[serde(default)] + pub cwd: Option, + /// Optional human-readable description shown in bridge tool output. + #[serde(default)] + pub description: Option, + /// Whether this server should be exposed to the MCP bridge tools. + #[serde(default = "defaults::default_true")] + pub enabled: bool, + /// Per-request timeout in seconds. + #[serde(default = "default_mcp_timeout_secs")] + pub timeout_secs: u64, + /// Optional auth strategy applied to outbound requests for this + /// server. Useful for API-key and pre-provisioned bearer-token + /// flows; interactive OAuth discovery is handled by the client + /// transport separately when a server returns an auth challenge. + #[serde(default)] + pub auth: McpAuthConfig, +} + +fn default_mcp_timeout_secs() -> u64 { + 30 +} + +impl Default for McpServerConfig { + fn default() -> Self { + Self { + name: String::new(), + endpoint: String::new(), + command: String::new(), + args: Vec::new(), + env: HashMap::new(), + cwd: None, + description: None, + enabled: defaults::default_true(), + timeout_secs: default_mcp_timeout_secs(), + auth: McpAuthConfig::None, + } + } +} + +#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] +#[serde(tag = "kind", rename_all = "snake_case")] +pub enum McpAuthConfig { + None, + BearerToken { token: String }, + Basic { username: String, password: String }, + Header { name: String, value: String }, + QueryParam { name: String, value: String }, +} + +impl Default for McpAuthConfig { + fn default() -> Self { + Self::None + } +} + +#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] +#[serde(default)] +pub struct McpClientIdentityConfig { + /// Client name sent during `initialize.clientInfo.name`. + #[serde(default = "default_mcp_client_name")] + pub name: String, + /// Client title sent during `initialize.clientInfo.title`. + #[serde(default = "default_mcp_client_title")] + pub title: String, + /// Client version sent during `initialize.clientInfo.version`. + #[serde(default = "default_mcp_client_version")] + pub version: String, +} + +fn default_mcp_client_name() -> String { + "openhuman-core".into() +} + +fn default_mcp_client_title() -> String { + "OpenHuman Core MCP Client".into() +} + +fn default_mcp_client_version() -> String { + env!("CARGO_PKG_VERSION").into() +} + +impl Default for McpClientIdentityConfig { + fn default() -> Self { + Self { + name: default_mcp_client_name(), + title: default_mcp_client_title(), + version: default_mcp_client_version(), + } + } +} + +#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] +#[serde(default)] +pub struct McpClientConfig { + /// When `true`, register the generic MCP bridge tools and expose + /// configured remote MCP servers to the agent runtime. + #[serde(default = "defaults::default_true")] + pub enabled: bool, + /// Named remote MCP servers accessible via `mcp_list_*` / + /// `mcp_call_tool`. + #[serde(default)] + pub servers: Vec, + /// Identity block sent during initialize. + #[serde(default)] + pub client_identity: McpClientIdentityConfig, +} + +impl Default for McpClientConfig { + fn default() -> Self { + Self { + enabled: defaults::default_true(), + servers: Vec::new(), + client_identity: McpClientIdentityConfig::default(), + } + } +} + #[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] #[serde(default)] pub struct SeltzConfig { diff --git a/src/openhuman/config/schema/types.rs b/src/openhuman/config/schema/types.rs index 6e63bd4c6..d409bd24d 100644 --- a/src/openhuman/config/schema/types.rs +++ b/src/openhuman/config/schema/types.rs @@ -140,6 +140,9 @@ pub struct Config { #[serde(default)] pub gitbooks: GitbooksConfig, + #[serde(default)] + pub mcp_client: McpClientConfig, + #[serde(default)] pub multimodal: MultimodalConfig, @@ -419,6 +422,7 @@ impl Default for Config { http_request: HttpRequestConfig::default(), curl: CurlConfig::default(), gitbooks: GitbooksConfig::default(), + mcp_client: McpClientConfig::default(), multimodal: MultimodalConfig::default(), seltz: SeltzConfig::default(), web_search: WebSearchConfig::default(), diff --git a/src/openhuman/mcp_client/client.rs b/src/openhuman/mcp_client/client.rs new file mode 100644 index 000000000..a3fe8b4f3 --- /dev/null +++ b/src/openhuman/mcp_client/client.rs @@ -0,0 +1,1266 @@ +use crate::openhuman::config::{McpAuthConfig, McpClientIdentityConfig}; +use crate::openhuman::skills::types::ToolResult; +use anyhow::Context; +use base64::Engine; +use parking_lot::Mutex; +use reqwest::header::{HeaderMap, HeaderName, HeaderValue, AUTHORIZATION, CONTENT_TYPE}; +use serde::{Deserialize, Serialize}; +use serde_json::{json, Value}; +use std::collections::HashMap; +use std::sync::atomic::{AtomicI64, Ordering}; +use std::time::Duration; + +const LATEST_PROTOCOL_VERSION: &str = "2025-11-25"; +const SUPPORTED_PROTOCOL_VERSIONS: &[&str] = &[ + "2024-11-05", + "2025-03-26", + "2025-06-18", + LATEST_PROTOCOL_VERSION, +]; +const HEADER_PROTOCOL_VERSION: &str = "MCP-Protocol-Version"; +const HEADER_SESSION_ID: &str = "Mcp-Session-Id"; +const HEADER_METHOD: &str = "Mcp-Method"; +const HEADER_NAME: &str = "Mcp-Name"; + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +pub struct McpRemoteTool { + pub name: String, + #[serde(default)] + pub title: Option, + #[serde(default)] + pub description: Option, + #[serde(default, rename = "inputSchema")] + pub input_schema: Value, +} + +#[derive(Debug, Clone)] +pub struct McpServerToolResult { + pub raw_result: Value, + pub rendered: ToolResult, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub struct McpClientInfo { + pub name: String, + #[serde(default)] + pub title: Option, + pub version: String, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +pub struct McpInitializeResult { + #[serde(rename = "protocolVersion")] + pub protocol_version: String, + #[serde(default)] + pub capabilities: Value, + #[serde(default, rename = "serverInfo")] + pub server_info: Value, + #[serde(default)] + pub instructions: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +pub struct ProtectedResourceMetadata { + pub resource: String, + #[serde(default)] + pub authorization_servers: Vec, + #[serde(default)] + pub scopes_supported: Vec, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +pub struct AuthorizationServerMetadata { + pub issuer: String, + #[serde(default)] + pub authorization_endpoint: Option, + #[serde(default)] + pub token_endpoint: Option, + #[serde(default)] + pub registration_endpoint: Option, + #[serde(default)] + pub response_types_supported: Vec, + #[serde(default)] + pub grant_types_supported: Vec, + #[serde(default)] + pub code_challenge_methods_supported: Vec, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +pub struct McpAuthChallenge { + pub scheme: String, + pub realm: Option, + pub resource_metadata: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +pub struct McpAuthorizationContext { + pub challenge: McpAuthChallenge, + pub protected_resource_metadata: Option, + pub authorization_server_metadata: Vec, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +pub struct McpSseEvent { + pub event: Option, + pub id: Option, + pub data: Option, +} + +#[derive(Debug)] +pub struct McpHttpClient { + endpoint: String, + http: reqwest::Client, + next_id: AtomicI64, + client_info: McpClientInfo, + auth: McpAuthConfig, + state: Mutex, +} + +#[derive(Debug, Default)] +struct SessionState { + initialized: bool, + negotiated_protocol_version: String, + session_id: Option, + initialize: Option, + cached_tools: HashMap, +} + +impl McpHttpClient { + pub fn new(endpoint: String, timeout_secs: u64) -> Self { + Self::with_options( + endpoint, + timeout_secs, + McpAuthConfig::None, + McpClientIdentityConfig::default(), + ) + } + + pub fn with_options( + endpoint: String, + timeout_secs: u64, + auth: McpAuthConfig, + identity: McpClientIdentityConfig, + ) -> Self { + let builder = reqwest::Client::builder() + .timeout(Duration::from_secs(timeout_secs)) + .connect_timeout(Duration::from_secs(10)) + .redirect(reqwest::redirect::Policy::none()); + let builder = + crate::openhuman::config::apply_runtime_proxy_to_builder(builder, "tool.mcp_client"); + let http = builder.build().expect("reqwest client must build"); + Self { + endpoint, + http, + next_id: AtomicI64::new(1), + client_info: McpClientInfo { + name: identity.name, + title: Some(identity.title), + version: identity.version, + }, + auth, + state: Mutex::new(SessionState { + negotiated_protocol_version: LATEST_PROTOCOL_VERSION.to_string(), + ..SessionState::default() + }), + } + } + + pub fn endpoint(&self) -> &str { + &self.endpoint + } + + pub fn initialize_snapshot(&self) -> Option { + self.state.lock().initialize.clone() + } + + pub async fn initialize(&self) -> anyhow::Result { + if let Some(existing) = self.state.lock().initialize.clone() { + return Ok(existing); + } + + let params = json!({ + "protocolVersion": LATEST_PROTOCOL_VERSION, + "capabilities": {}, + "clientInfo": self.client_info, + }); + let id = self.next_id.fetch_add(1, Ordering::Relaxed); + let body = json!({ + "jsonrpc": "2.0", + "id": id, + "method": "initialize", + "params": params, + }); + let request = self + .apply_auth( + self.http + .post(&self.endpoint) + .header(CONTENT_TYPE, "application/json"), + true, + ) + .body(serde_json::to_vec(&body)?); + let response = self.read_response(request.send().await?).await?; + let init: McpInitializeResult = + serde_json::from_value(response.result.clone()).context("parsing initialize result")?; + self.validate_protocol_version(&init.protocol_version)?; + + { + let mut state = self.state.lock(); + state.initialized = true; + state.negotiated_protocol_version = init.protocol_version.clone(); + state.session_id = response.session_id.clone(); + state.initialize = Some(init.clone()); + } + + self.send_notification("notifications/initialized", json!({})) + .await?; + + Ok(init) + } + + pub async fn list_tools(&self) -> anyhow::Result> { + self.initialize().await?; + let result = self + .send_jsonrpc( + "tools/list", + json!({}), + RequestOptions::standard("tools/list", None, None), + ) + .await? + .result; + let tools = serde_json::from_value::>( + result + .get("tools") + .cloned() + .ok_or_else(|| anyhow::anyhow!("MCP tools/list response missing `tools`"))?, + )?; + let mut state = self.state.lock(); + state.cached_tools = tools + .iter() + .cloned() + .map(|tool| (tool.name.clone(), tool)) + .collect(); + Ok(tools) + } + + pub async fn call_tool( + &self, + name: &str, + arguments: Value, + ) -> anyhow::Result { + self.initialize().await?; + let cached_tool = { self.state.lock().cached_tools.get(name).cloned() }; + let tool = if let Some(tool) = cached_tool { + Some(tool) + } else { + self.list_tools() + .await? + .into_iter() + .find(|tool| tool.name == name) + }; + let extra_headers = tool + .as_ref() + .map(|tool| x_mcp_headers_from_schema(tool, &arguments)) + .transpose()? + .unwrap_or_default(); + + let result = self + .send_jsonrpc( + "tools/call", + json!({ + "name": name, + "arguments": arguments, + }), + RequestOptions::standard("tools/call", Some(name), Some(extra_headers)), + ) + .await? + .result; + let rendered = render_tool_result(&result); + Ok(McpServerToolResult { + raw_result: result, + rendered, + }) + } + + pub async fn discover_authorization(&self) -> anyhow::Result> { + let request = self + .http + .post(&self.endpoint) + .header(CONTENT_TYPE, "application/json") + .body(serde_json::to_vec(&json!({ + "jsonrpc": "2.0", + "id": self.next_id.fetch_add(1, Ordering::Relaxed), + "method": "initialize", + "params": { + "protocolVersion": LATEST_PROTOCOL_VERSION, + "capabilities": {}, + "clientInfo": self.client_info, + } + }))?); + let response = self.apply_auth(request, true).send().await?; + if response.status() != reqwest::StatusCode::UNAUTHORIZED { + return Ok(None); + } + + let challenge = parse_www_authenticate_challenge(response.headers()) + .ok_or_else(|| anyhow::anyhow!("401 response missing parseable WWW-Authenticate"))?; + let prm = if let Some(url) = challenge.resource_metadata.as_deref() { + Some(self.fetch_json::(url).await?) + } else { + None + }; + let mut auth_servers = Vec::new(); + if let Some(prm) = prm.as_ref() { + for issuer in &prm.authorization_servers { + if let Ok(metadata) = self.fetch_authorization_server_metadata(issuer).await { + auth_servers.push(metadata); + } + } + } + Ok(Some(McpAuthorizationContext { + challenge, + protected_resource_metadata: prm, + authorization_server_metadata: auth_servers, + })) + } + + pub async fn drain_events( + &self, + last_event_id: Option<&str>, + ) -> anyhow::Result> { + self.initialize().await?; + let protocol_version = self.state.lock().negotiated_protocol_version.clone(); + let session_id = self.state.lock().session_id.clone(); + let mut request = self + .apply_auth(self.http.get(&self.endpoint), false) + .header(HEADER_PROTOCOL_VERSION, protocol_version); + if let Some(session_id) = session_id { + request = request.header(HEADER_SESSION_ID, session_id); + } + if let Some(last_event_id) = last_event_id { + request = request.header("Last-Event-ID", last_event_id); + } + let response = request.send().await?; + let status = response.status(); + let text = response.text().await?; + if !status.is_success() { + anyhow::bail!("MCP events GET {} — {}", status.as_u16(), text); + } + parse_sse_events(&text) + } + + pub async fn close_session(&self) -> anyhow::Result<()> { + let session_id = self.state.lock().session_id.clone(); + let Some(session_id) = session_id else { + return Ok(()); + }; + let response = self + .http + .delete(&self.endpoint) + .header(HEADER_SESSION_ID, session_id) + .send() + .await?; + if !(response.status().is_success() + || response.status() == reqwest::StatusCode::METHOD_NOT_ALLOWED) + { + anyhow::bail!("MCP DELETE failed with {}", response.status()); + } + let mut state = self.state.lock(); + state.initialized = false; + state.session_id = None; + state.initialize = None; + state.cached_tools.clear(); + Ok(()) + } + + async fn send_notification(&self, method: &str, params: Value) -> anyhow::Result<()> { + let body = json!({ + "jsonrpc": "2.0", + "method": method, + "params": params, + }); + let request = self + .http + .post(&self.endpoint) + .header(CONTENT_TYPE, "application/json"); + let request = self.apply_standard_headers(request, false, method, None, &[]); + let response = request.body(serde_json::to_vec(&body)?).send().await?; + let status = response.status(); + if !status.is_success() { + let text = response.text().await.unwrap_or_default(); + anyhow::bail!( + "MCP notification {method} failed with {} — {}", + status, + text + ); + } + Ok(()) + } + + async fn send_jsonrpc( + &self, + method: &str, + params: Value, + options: RequestOptions, + ) -> anyhow::Result { + self.send_jsonrpc_inner(method, params, options, true).await + } + + async fn send_jsonrpc_inner( + &self, + method: &str, + params: Value, + options: RequestOptions, + allow_reinitialize: bool, + ) -> anyhow::Result { + let id = self.next_id.fetch_add(1, Ordering::Relaxed); + let body = json!({ + "jsonrpc": "2.0", + "id": id, + "method": method, + "params": params, + }); + + tracing::debug!( + target: "[mcp_client]", + endpoint = %redact_endpoint(&self.endpoint), + method, + initialize = options.initialize, + "dispatch MCP request" + ); + + let request = self + .http + .post(&self.endpoint) + .header(CONTENT_TYPE, "application/json"); + let request = if options.initialize { + self.apply_auth(request, true) + } else { + self.apply_standard_headers( + request, + false, + options.method_header.unwrap_or(method), + options.name_header.as_deref(), + &options.extra_headers, + ) + }; + let response = request.body(serde_json::to_vec(&body)?).send().await?; + + if response.status() == reqwest::StatusCode::NOT_FOUND + && allow_reinitialize + && self.state.lock().session_id.is_some() + { + tracing::info!( + target: "[mcp_client]", + endpoint = %redact_endpoint(&self.endpoint), + method, + "session expired with 404; reinitializing and retrying once" + ); + self.reset_session(); + self.initialize().await?; + return Box::pin(self.send_jsonrpc_inner( + method, + body["params"].clone(), + options, + false, + )) + .await; + } + + self.read_response(response).await + } + + fn apply_standard_headers( + &self, + request: reqwest::RequestBuilder, + initialize: bool, + method: &str, + name: Option<&str>, + extra_headers: &[(HeaderName, HeaderValue)], + ) -> reqwest::RequestBuilder { + let protocol_version = self.state.lock().negotiated_protocol_version.clone(); + let session_id = self.state.lock().session_id.clone(); + let mut request = self.apply_auth(request, initialize); + request = request.header(HEADER_METHOD, method); + if let Some(name) = name { + request = request.header(HEADER_NAME, name); + } + if !initialize { + request = request.header(HEADER_PROTOCOL_VERSION, protocol_version); + if let Some(session_id) = session_id { + request = request.header(HEADER_SESSION_ID, session_id); + } + } + for (name, value) in extra_headers { + request = request.header(name, value); + } + request + } + + fn apply_auth( + &self, + request: reqwest::RequestBuilder, + _initialize: bool, + ) -> reqwest::RequestBuilder { + match &self.auth { + McpAuthConfig::None => request, + McpAuthConfig::BearerToken { token } => { + request.header(AUTHORIZATION, format!("Bearer {}", token.trim())) + } + McpAuthConfig::Basic { username, password } => { + let encoded = base64::engine::general_purpose::STANDARD + .encode(format!("{username}:{password}")); + request.header(AUTHORIZATION, format!("Basic {encoded}")) + } + McpAuthConfig::Header { name, value } => match ( + HeaderName::try_from(name.as_str()), + HeaderValue::from_str(value), + ) { + (Ok(name), Ok(value)) => request.header(name, value), + _ => request, + }, + McpAuthConfig::QueryParam { name, value } => { + request.query(&[(name.as_str(), value.as_str())]) + } + } + } + + async fn fetch_json(&self, url: &str) -> anyhow::Result + where + T: for<'de> Deserialize<'de>, + { + let response = self.http.get(url).send().await?; + let status = response.status(); + let text = response.text().await?; + if !status.is_success() { + anyhow::bail!("HTTP {} while fetching {} — {}", status.as_u16(), url, text); + } + serde_json::from_str(&text).with_context(|| format!("parsing JSON from {url}")) + } + + async fn fetch_authorization_server_metadata( + &self, + issuer: &str, + ) -> anyhow::Result { + let trimmed = issuer.trim_end_matches('/'); + let oidc = format!("{trimmed}/.well-known/openid-configuration"); + if let Ok(metadata) = self.fetch_json::(&oidc).await { + return Ok(metadata); + } + let oauth = format!("{trimmed}/.well-known/oauth-authorization-server"); + self.fetch_json::(&oauth).await + } + + fn validate_protocol_version(&self, version: &str) -> anyhow::Result<()> { + if SUPPORTED_PROTOCOL_VERSIONS.contains(&version) { + Ok(()) + } else { + anyhow::bail!("unsupported MCP protocol version negotiated by server: {version}"); + } + } + + fn reset_session(&self) { + let mut state = self.state.lock(); + state.initialized = false; + state.session_id = None; + state.initialize = None; + state.cached_tools.clear(); + state.negotiated_protocol_version = LATEST_PROTOCOL_VERSION.to_string(); + } +} + +#[derive(Debug, Clone)] +struct RequestOptions { + initialize: bool, + method_header: Option<&'static str>, + name_header: Option, + extra_headers: Vec<(HeaderName, HeaderValue)>, +} + +impl RequestOptions { + fn standard( + method_header: &'static str, + name_header: Option<&str>, + extra_headers: Option>, + ) -> Self { + Self { + initialize: false, + method_header: Some(method_header), + name_header: name_header.map(str::to_string), + extra_headers: extra_headers.unwrap_or_default(), + } + } +} + +#[derive(Debug, Clone)] +struct ResponseEnvelope { + result: Value, + session_id: Option, +} + +pub fn render_tool_result(result: &Value) -> ToolResult { + let is_error = result + .get("isError") + .and_then(Value::as_bool) + .unwrap_or(false); + + let mut out = String::new(); + if let Some(content) = result.get("content").and_then(Value::as_array) { + for block in content { + if let Some(t) = block.get("text").and_then(Value::as_str) { + if !out.is_empty() { + out.push_str("\n\n"); + } + out.push_str(t); + } + } + } + if out.is_empty() { + out = result.to_string(); + } + + if is_error { + ToolResult::error(out) + } else { + ToolResult::success(out) + } +} + +pub fn redact_endpoint(raw: &str) -> String { + let trimmed = raw.trim(); + let (scheme, rest) = if let Some(r) = trimmed.strip_prefix("https://") { + ("https", r) + } else if let Some(r) = trimmed.strip_prefix("http://") { + ("http", r) + } else { + return "".into(); + }; + let authority = rest.split(['/', '?', '#']).next().unwrap_or(""); + if authority.is_empty() || authority.contains('@') { + return "".into(); + } + format!("{scheme}://{authority}") +} + +fn parse_sse_message(body: &str) -> anyhow::Result { + let events = parse_sse_events(body)?; + let event = events + .into_iter() + .find_map(|event| event.data) + .ok_or_else(|| anyhow::anyhow!("No SSE data frame found in MCP response: {body}"))?; + Ok(event) +} + +fn parse_sse_events(body: &str) -> anyhow::Result> { + let mut events = Vec::new(); + let mut event_type: Option = None; + let mut event_id: Option = None; + let mut data_lines: Vec = Vec::new(); + + let flush = |events: &mut Vec, + event_type: &mut Option, + event_id: &mut Option, + data_lines: &mut Vec| + -> anyhow::Result<()> { + if event_type.is_none() && event_id.is_none() && data_lines.is_empty() { + return Ok(()); + } + let data = if data_lines.is_empty() { + None + } else { + let joined = data_lines.join("\n"); + Some( + serde_json::from_str(&joined) + .with_context(|| format!("Failed to parse SSE data frame JSON: {joined}"))?, + ) + }; + events.push(McpSseEvent { + event: event_type.take(), + id: event_id.take(), + data, + }); + data_lines.clear(); + Ok(()) + }; + + for raw_line in body.lines() { + let line = raw_line.trim_end_matches('\r'); + if line.is_empty() { + flush(&mut events, &mut event_type, &mut event_id, &mut data_lines)?; + continue; + } + if line.starts_with(':') { + continue; + } + if let Some(value) = line.strip_prefix("event:") { + event_type = Some(value.trim_start().to_string()); + } else if let Some(value) = line.strip_prefix("id:") { + event_id = Some(value.trim_start().to_string()); + } else if let Some(value) = line.strip_prefix("data:") { + data_lines.push(value.trim_start().to_string()); + } + } + flush(&mut events, &mut event_type, &mut event_id, &mut data_lines)?; + Ok(events) +} + +fn parse_www_authenticate_challenge(headers: &HeaderMap) -> Option { + let raw = headers.get("WWW-Authenticate")?.to_str().ok()?.trim(); + let mut parts = raw.splitn(2, ' '); + let scheme = parts.next()?.trim().to_string(); + let params = parts.next().unwrap_or("").trim(); + let attrs = parse_auth_attribute_list(params); + Some(McpAuthChallenge { + scheme, + realm: attrs.get("realm").cloned(), + resource_metadata: attrs.get("resource_metadata").cloned(), + }) +} + +fn parse_auth_attribute_list(input: &str) -> HashMap { + let mut attrs = HashMap::new(); + for part in input.split(',') { + let Some((key, value)) = part.split_once('=') else { + continue; + }; + let value = value.trim().trim_matches('"').to_string(); + attrs.insert(key.trim().to_string(), value); + } + attrs +} + +fn header_to_string(headers: &HeaderMap, name: &str) -> Option { + headers.get(name)?.to_str().ok().map(|s| s.to_string()) +} + +fn x_mcp_headers_from_schema( + tool: &McpRemoteTool, + arguments: &Value, +) -> anyhow::Result> { + let mut headers = Vec::new(); + let Some(args) = arguments.as_object() else { + return Ok(headers); + }; + let properties = tool + .input_schema + .get("properties") + .and_then(Value::as_object) + .cloned() + .unwrap_or_default(); + + for (param_name, schema) in properties { + let Some(header_suffix) = schema.get("x-mcp-header").and_then(Value::as_str) else { + continue; + }; + let Some(value) = args.get(¶m_name) else { + continue; + }; + let header_name = + HeaderName::from_bytes(format!("Mcp-Param-{header_suffix}").as_bytes()) + .with_context(|| format!("invalid x-mcp-header name for `{param_name}`"))?; + let header_value = match value { + Value::String(s) => HeaderValue::from_str(s), + other => HeaderValue::from_str(&other.to_string()), + } + .with_context(|| format!("invalid x-mcp-header value for `{param_name}`"))?; + headers.push((header_name, header_value)); + } + + Ok(headers) +} + +impl McpHttpClient { + async fn read_response(&self, response: reqwest::Response) -> anyhow::Result { + let status = response.status(); + let headers = response.headers().clone(); + let content_type = headers + .get(reqwest::header::CONTENT_TYPE) + .and_then(|v| v.to_str().ok()) + .unwrap_or("") + .to_string(); + let text = response.text().await?; + + if status == reqwest::StatusCode::UNAUTHORIZED { + let auth_suffix = if let Some(challenge) = parse_www_authenticate_challenge(&headers) { + match challenge.resource_metadata.as_deref() { + Some(resource_metadata) => { + format!("; resource metadata: {resource_metadata}") + } + None => String::new(), + } + } else { + String::new() + }; + anyhow::bail!( + "MCP unauthorized for `{}` (HTTP 401{})", + redact_endpoint(&self.endpoint), + auth_suffix + ); + } + if !status.is_success() { + anyhow::bail!("MCP HTTP {} — {}", status.as_u16(), text); + } + + let payload: Value = if content_type.starts_with("text/event-stream") { + parse_sse_message(&text)? + } else { + serde_json::from_str(&text).map_err(|e| { + anyhow::anyhow!("Failed to parse MCP JSON response: {e} — body: {text}") + })? + }; + if let Some(err) = payload.get("error") { + anyhow::bail!("MCP error: {err}"); + } + let result = payload + .get("result") + .ok_or_else(|| anyhow::anyhow!("MCP response missing `result`: {payload}"))? + .clone(); + Ok(ResponseEnvelope { + result, + session_id: header_to_string(&headers, HEADER_SESSION_ID), + }) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use axum::{ + extract::State, + http::{HeaderMap as AxumHeaderMap, Method, StatusCode}, + response::{IntoResponse, Response}, + routing::{get, post}, + Json, Router, + }; + use serde_json::Value; + use std::sync::{ + atomic::{AtomicUsize, Ordering as AtomicOrdering}, + Arc, + }; + + #[derive(Clone)] + struct TestState { + init_count: Arc, + call_count: Arc, + } + + async fn mcp_handler( + State(state): State, + headers: AxumHeaderMap, + method: Method, + Json(body): Json, + ) -> Response { + let rpc_method = body.get("method").and_then(Value::as_str).unwrap_or(""); + if method == Method::POST && rpc_method == "initialize" { + state.init_count.fetch_add(1, AtomicOrdering::SeqCst); + return ( + [(HEADER_SESSION_ID, "session-1")], + Json(json!({ + "jsonrpc": "2.0", + "id": body["id"].clone(), + "result": { + "protocolVersion": LATEST_PROTOCOL_VERSION, + "capabilities": { "tools": { "listChanged": true } }, + "serverInfo": { "name": "test-server", "version": "1.0.0" } + } + })), + ) + .into_response(); + } + + if headers.get(HEADER_SESSION_ID).and_then(|v| v.to_str().ok()) != Some("session-1") { + return ( + StatusCode::BAD_REQUEST, + "missing or invalid session".to_string(), + ) + .into_response(); + } + + if headers + .get(HEADER_PROTOCOL_VERSION) + .and_then(|v| v.to_str().ok()) + != Some(LATEST_PROTOCOL_VERSION) + { + return ( + StatusCode::BAD_REQUEST, + "missing protocol version".to_string(), + ) + .into_response(); + } + + match rpc_method { + "notifications/initialized" => StatusCode::NO_CONTENT.into_response(), + "tools/list" => Json(json!({ + "jsonrpc": "2.0", + "id": body["id"].clone(), + "result": { + "tools": [{ + "name": "needs_header", + "description": "needs x-mcp-header", + "inputSchema": { + "type": "object", + "properties": { + "tenant": { + "type": "string", + "x-mcp-header": "tenant" + } + } + } + }] + } + })) + .into_response(), + "tools/call" => { + state.call_count.fetch_add(1, AtomicOrdering::SeqCst); + if headers + .get("Mcp-Param-tenant") + .and_then(|v| v.to_str().ok()) + != Some("acme") + { + return ( + StatusCode::BAD_REQUEST, + "missing mirrored tenant header".to_string(), + ) + .into_response(); + } + Json(json!({ + "jsonrpc": "2.0", + "id": body["id"].clone(), + "result": { + "content": [{ + "type": "text", + "text": "remote result" + }] + } + })) + .into_response() + } + _ => ( + StatusCode::BAD_REQUEST, + format!("unexpected method {rpc_method}"), + ) + .into_response(), + } + } + + async fn events_handler(headers: AxumHeaderMap) -> Response { + if headers.get(HEADER_SESSION_ID).is_none() { + return (StatusCode::BAD_REQUEST, "no session".to_string()).into_response(); + } + ( + [(CONTENT_TYPE.as_str(), "text/event-stream")], + "id: 1\nevent: message\ndata: {\"jsonrpc\":\"2.0\",\"method\":\"notifications/message\",\"params\":{\"ok\":true}}\n\n", + ) + .into_response() + } + + async fn delete_handler() -> Response { + StatusCode::NO_CONTENT.into_response() + } + + async fn bearer_required_handler(headers: AxumHeaderMap, Json(body): Json) -> Response { + if headers.get(AUTHORIZATION).and_then(|v| v.to_str().ok()) != Some("Bearer secret-token") { + return (StatusCode::UNAUTHORIZED, "missing bearer".to_string()).into_response(); + } + Json(json!({ + "jsonrpc": "2.0", + "id": body["id"].clone(), + "result": { + "protocolVersion": LATEST_PROTOCOL_VERSION, + "capabilities": {}, + "serverInfo": { "name": "bearer-server", "version": "1.0.0" } + } + })) + .into_response() + } + + async fn retrying_mcp_handler( + State(state): State, + headers: AxumHeaderMap, + Json(body): Json, + ) -> Response { + let rpc_method = body.get("method").and_then(Value::as_str).unwrap_or(""); + if rpc_method == "initialize" { + state.init_count.fetch_add(1, AtomicOrdering::SeqCst); + return ( + [(HEADER_SESSION_ID, "session-retry")], + Json(json!({ + "jsonrpc": "2.0", + "id": body["id"].clone(), + "result": { + "protocolVersion": LATEST_PROTOCOL_VERSION, + "capabilities": { "tools": {} }, + "serverInfo": { "name": "retry-server", "version": "1.0.0" } + } + })), + ) + .into_response(); + } + if rpc_method == "notifications/initialized" { + return StatusCode::NO_CONTENT.into_response(); + } + if rpc_method == "tools/list" { + let call_number = state.call_count.fetch_add(1, AtomicOrdering::SeqCst); + if call_number == 0 + && headers.get(HEADER_SESSION_ID).and_then(|v| v.to_str().ok()) + == Some("session-retry") + { + return (StatusCode::NOT_FOUND, "expired".to_string()).into_response(); + } + return Json(json!({ + "jsonrpc": "2.0", + "id": body["id"].clone(), + "result": { "tools": [] } + })) + .into_response(); + } + (StatusCode::BAD_REQUEST, "unexpected".to_string()).into_response() + } + + async fn spawn_test_server() -> (String, TestState) { + let state = TestState { + init_count: Arc::new(AtomicUsize::new(0)), + call_count: Arc::new(AtomicUsize::new(0)), + }; + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let app = Router::new() + .route( + "/", + post(mcp_handler).get(events_handler).delete(delete_handler), + ) + .with_state(state.clone()); + tokio::spawn(async move { + axum::serve(listener, app).await.unwrap(); + }); + (format!("http://{addr}/"), state) + } + + async fn spawn_retry_server() -> (String, TestState) { + let state = TestState { + init_count: Arc::new(AtomicUsize::new(0)), + call_count: Arc::new(AtomicUsize::new(0)), + }; + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let app = Router::new() + .route("/", post(retrying_mcp_handler)) + .with_state(state.clone()); + tokio::spawn(async move { + axum::serve(listener, app).await.unwrap(); + }); + (format!("http://{addr}/"), state) + } + + async fn spawn_discovery_server() -> String { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let base = format!("http://{addr}"); + let auth_header = format!( + "Bearer realm=\"mcp\", resource_metadata=\"{base}/.well-known/oauth-protected-resource\"" + ); + let prm_base = base.clone(); + let issuer_base = base.clone(); + let app = Router::new() + .route( + "/", + post(move || { + let auth_header = auth_header.clone(); + async move { + ( + StatusCode::UNAUTHORIZED, + [("WWW-Authenticate", auth_header.as_str())], + "", + ) + .into_response() + } + }), + ) + .route( + "/.well-known/oauth-protected-resource", + get(move || { + let prm_base = prm_base.clone(); + async move { + let resource = format!("{prm_base}/"); + Json(json!({ + "resource": resource, + "authorization_servers": [prm_base], + "scopes_supported": ["mcp:tools"] + })) + } + }), + ) + .route( + "/.well-known/openid-configuration", + get(move || { + let issuer_base = issuer_base.clone(); + async move { + let authorization_endpoint = format!("{}/authorize", issuer_base); + let token_endpoint = format!("{}/token", issuer_base); + Json(json!({ + "issuer": issuer_base, + "authorization_endpoint": authorization_endpoint, + "token_endpoint": token_endpoint, + "grant_types_supported": ["authorization_code"], + "code_challenge_methods_supported": ["S256"] + })) + } + }), + ); + tokio::spawn(async move { + axum::serve(listener, app).await.unwrap(); + }); + format!("http://{addr}/") + } + + #[tokio::test] + async fn initialize_and_list_tools_negotiate_session() { + let (endpoint, state) = spawn_test_server().await; + let client = McpHttpClient::new(endpoint, 5); + let tools = client.list_tools().await.expect("list_tools"); + assert_eq!(tools.len(), 1); + assert_eq!(state.init_count.load(AtomicOrdering::SeqCst), 1); + let snapshot = client.initialize_snapshot().expect("snapshot"); + assert_eq!(snapshot.protocol_version, LATEST_PROTOCOL_VERSION); + } + + #[tokio::test] + async fn call_tool_mirrors_x_mcp_header_parameters() { + let (endpoint, state) = spawn_test_server().await; + let client = McpHttpClient::new(endpoint, 5); + let result = client + .call_tool("needs_header", json!({"tenant": "acme"})) + .await + .expect("call_tool"); + assert_eq!(result.rendered.output(), "remote result"); + assert_eq!(state.call_count.load(AtomicOrdering::SeqCst), 1); + } + + #[tokio::test] + async fn session_404_triggers_reinitialize_and_retry() { + let (endpoint, state) = spawn_retry_server().await; + let client = McpHttpClient::new(endpoint, 5); + let tools = client.list_tools().await.expect("list_tools"); + assert!(tools.is_empty()); + assert_eq!(state.init_count.load(AtomicOrdering::SeqCst), 2); + assert_eq!(state.call_count.load(AtomicOrdering::SeqCst), 2); + } + + #[tokio::test] + async fn drain_events_parses_sse_stream() { + let (endpoint, _) = spawn_test_server().await; + let client = McpHttpClient::new(endpoint, 5); + let events = client.drain_events(None).await.expect("drain events"); + assert_eq!(events.len(), 1); + assert_eq!(events[0].id.as_deref(), Some("1")); + assert_eq!(events[0].event.as_deref(), Some("message")); + assert_eq!(events[0].data.as_ref().unwrap()["params"]["ok"], true); + } + + #[tokio::test] + async fn close_session_sends_delete() { + let (endpoint, _) = spawn_test_server().await; + let client = McpHttpClient::new(endpoint, 5); + client.initialize().await.expect("initialize"); + client.close_session().await.expect("close_session"); + assert!(client.initialize_snapshot().is_none()); + } + + #[test] + fn redact_endpoint_hides_paths_and_credentials() { + assert_eq!( + redact_endpoint("https://example.com/path?x=1"), + "https://example.com" + ); + assert_eq!( + redact_endpoint("https://user:pw@example.com/a"), + "" + ); + } + + #[test] + fn parse_sse_events_handles_multiple_frames() { + let body = "id: 1\nevent: message\ndata: {\"a\":1}\n\ndata: {\"b\":2}\n\n"; + let events = parse_sse_events(body).expect("events"); + assert_eq!(events.len(), 2); + assert_eq!(events[0].id.as_deref(), Some("1")); + assert_eq!(events[1].data.as_ref().unwrap()["b"], 2); + } + + #[test] + fn parse_www_authenticate_extracts_resource_metadata() { + let mut headers = HeaderMap::new(); + headers.insert( + "WWW-Authenticate", + HeaderValue::from_static( + "Bearer realm=\"mcp\", resource_metadata=\"https://example.com/.well-known/oauth-protected-resource\"", + ), + ); + let challenge = parse_www_authenticate_challenge(&headers).expect("challenge"); + assert_eq!(challenge.scheme, "Bearer"); + assert_eq!(challenge.realm.as_deref(), Some("mcp")); + assert_eq!( + challenge.resource_metadata.as_deref(), + Some("https://example.com/.well-known/oauth-protected-resource") + ); + } + + #[tokio::test] + async fn discover_authorization_returns_none_when_not_401() { + let (endpoint, _) = spawn_test_server().await; + let client = McpHttpClient::new(endpoint, 5); + assert!(client.discover_authorization().await.unwrap().is_none()); + } + + #[tokio::test] + async fn discover_authorization_fetches_metadata() { + let endpoint = spawn_discovery_server().await; + let client = McpHttpClient::new(endpoint, 2); + let ctx = client + .discover_authorization() + .await + .expect("discover") + .expect("some"); + assert_eq!(ctx.challenge.scheme, "Bearer"); + assert_eq!( + ctx.protected_resource_metadata + .as_ref() + .unwrap() + .scopes_supported, + vec!["mcp:tools"] + ); + assert_eq!(ctx.authorization_server_metadata.len(), 1); + let expected_authorization_endpoint = format!( + "{}/authorize", + ctx.protected_resource_metadata + .as_ref() + .unwrap() + .authorization_servers[0] + ); + assert_eq!( + ctx.authorization_server_metadata[0] + .authorization_endpoint + .as_deref(), + Some(expected_authorization_endpoint.as_str()) + ); + } + + #[tokio::test] + async fn bearer_auth_is_attached_to_initialize() { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let app = Router::new().route("/", post(bearer_required_handler)); + tokio::spawn(async move { + axum::serve(listener, app).await.unwrap(); + }); + let client = McpHttpClient::with_options( + format!("http://{addr}/"), + 2, + McpAuthConfig::BearerToken { + token: "secret-token".into(), + }, + McpClientIdentityConfig::default(), + ); + let init = client.initialize().await.expect("initialize"); + assert_eq!(init.server_info["name"], "bearer-server"); + } +} diff --git a/src/openhuman/mcp_client/mod.rs b/src/openhuman/mcp_client/mod.rs new file mode 100644 index 000000000..33db8dd4c --- /dev/null +++ b/src/openhuman/mcp_client/mod.rs @@ -0,0 +1,17 @@ +//! Shared MCP client + registry for remote MCP servers exposed to agents. +//! +//! Supports Streamable HTTP and stdio transports. HTTP transport carries +//! session + auth lifecycle; stdio launches a subprocess and exchanges +//! newline-delimited JSON-RPC messages over stdin/stdout per the MCP spec. + +mod client; +mod registry; +mod stdio; + +pub use client::{ + redact_endpoint, AuthorizationServerMetadata, McpAuthChallenge, McpAuthorizationContext, + McpHttpClient, McpInitializeResult, McpRemoteTool, McpServerToolResult, McpSseEvent, + ProtectedResourceMetadata, +}; +pub use registry::{McpRegistrySource, McpServerDefinition, McpServerRegistry, McpTransportClient}; +pub use stdio::McpStdioClient; diff --git a/src/openhuman/mcp_client/registry.rs b/src/openhuman/mcp_client/registry.rs new file mode 100644 index 000000000..8373e5733 --- /dev/null +++ b/src/openhuman/mcp_client/registry.rs @@ -0,0 +1,281 @@ +use super::client::{ + McpAuthorizationContext, McpHttpClient, McpInitializeResult, McpRemoteTool, McpServerToolResult, +}; +use super::stdio::McpStdioClient; +use crate::openhuman::config::{Config, McpAuthConfig, McpClientIdentityConfig, McpServerConfig}; +use serde::{Deserialize, Serialize}; +use serde_json::Value; +use std::collections::HashMap; +use std::path::PathBuf; +use std::sync::Arc; + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum McpRegistrySource { + Config, + LegacyGitbooks, +} + +#[derive(Debug, Clone)] +pub struct McpServerDefinition { + pub name: String, + pub endpoint: String, + pub command: Option, + pub description: Option, + pub timeout_secs: u64, + pub auth: McpAuthConfig, + pub source: McpRegistrySource, + client: Arc, +} + +#[derive(Debug)] +pub enum McpTransportClient { + Http(McpHttpClient), + Stdio(McpStdioClient), +} + +#[derive(Debug, Default, Clone)] +pub struct McpServerRegistry { + by_name: HashMap, + order: Vec, +} + +impl McpServerRegistry { + pub fn from_config(config: &Config) -> Self { + let mut registry = Self::default(); + if !config.mcp_client.enabled { + return registry; + } + + for server in &config.mcp_client.servers { + registry.register_config_server( + server, + &config.mcp_client.client_identity, + McpRegistrySource::Config, + ); + } + + if config.gitbooks.enabled && registry.get("gitbooks").is_none() { + registry.insert(McpServerDefinition { + name: "gitbooks".into(), + endpoint: config.gitbooks.endpoint.clone(), + command: None, + description: Some("OpenHuman GitBook documentation MCP server.".into()), + timeout_secs: config.gitbooks.timeout_secs, + auth: McpAuthConfig::None, + source: McpRegistrySource::LegacyGitbooks, + client: Arc::new(McpTransportClient::Http(McpHttpClient::with_options( + config.gitbooks.endpoint.clone(), + config.gitbooks.timeout_secs, + McpAuthConfig::None, + config.mcp_client.client_identity.clone(), + ))), + }); + } + + registry + } + + pub fn is_empty(&self) -> bool { + self.order.is_empty() + } + + pub fn list(&self) -> Vec<&McpServerDefinition> { + self.order + .iter() + .filter_map(|name| self.by_name.get(name)) + .collect() + } + + pub fn get(&self, name: &str) -> Option<&McpServerDefinition> { + self.by_name.get(name) + } + + pub async fn list_tools(&self, server: &str) -> anyhow::Result> { + let server = self + .get(server) + .ok_or_else(|| anyhow::anyhow!("unknown MCP server `{server}`"))?; + server.client.list_tools().await + } + + pub async fn call_tool( + &self, + server: &str, + tool: &str, + arguments: Value, + ) -> anyhow::Result { + let server = self + .get(server) + .ok_or_else(|| anyhow::anyhow!("unknown MCP server `{server}`"))?; + server.client.call_tool(tool, arguments).await + } + + pub async fn initialize(&self, server: &str) -> anyhow::Result { + let server = self + .get(server) + .ok_or_else(|| anyhow::anyhow!("unknown MCP server `{server}`"))?; + server.client.initialize().await + } + + pub async fn discover_authorization( + &self, + server: &str, + ) -> anyhow::Result> { + let server = self + .get(server) + .ok_or_else(|| anyhow::anyhow!("unknown MCP server `{server}`"))?; + server.client.discover_authorization().await + } + + fn register_config_server( + &mut self, + server: &McpServerConfig, + identity: &McpClientIdentityConfig, + source: McpRegistrySource, + ) { + if !server.enabled { + return; + } + let name = server.name.trim(); + let endpoint = server.endpoint.trim(); + let command = server.command.trim(); + if name.is_empty() || (endpoint.is_empty() && command.is_empty()) { + tracing::warn!( + name = server.name, + endpoint = server.endpoint, + command = server.command, + "[mcp_client] skipping malformed MCP server config entry" + ); + return; + } + self.insert(McpServerDefinition { + name: name.to_string(), + endpoint: endpoint.to_string(), + command: transport_command(server), + description: server.description.clone(), + timeout_secs: server.timeout_secs, + auth: server.auth.clone(), + source, + client: Arc::new(build_transport_client(server, identity)), + }); + } + + fn insert(&mut self, def: McpServerDefinition) { + let name = def.name.clone(); + if self.by_name.insert(name.clone(), def).is_none() { + self.order.push(name); + } + } +} + +impl McpTransportClient { + pub async fn initialize(&self) -> anyhow::Result { + match self { + Self::Http(client) => client.initialize().await, + Self::Stdio(client) => client.initialize().await, + } + } + + pub async fn list_tools(&self) -> anyhow::Result> { + match self { + Self::Http(client) => client.list_tools().await, + Self::Stdio(client) => client.list_tools().await, + } + } + + pub async fn call_tool( + &self, + tool: &str, + arguments: Value, + ) -> anyhow::Result { + match self { + Self::Http(client) => client.call_tool(tool, arguments).await, + Self::Stdio(client) => client.call_tool(tool, arguments).await, + } + } + + pub async fn discover_authorization(&self) -> anyhow::Result> { + match self { + Self::Http(client) => client.discover_authorization().await, + Self::Stdio(_) => Ok(None), + } + } +} + +fn build_transport_client( + server: &McpServerConfig, + identity: &McpClientIdentityConfig, +) -> McpTransportClient { + if !server.command.trim().is_empty() { + let env = server + .env + .iter() + .map(|(k, v)| (k.clone(), v.clone())) + .collect::>(); + McpTransportClient::Stdio(McpStdioClient::new( + server.command.trim().to_string(), + server.args.clone(), + env, + server.cwd.as_ref().map(PathBuf::from), + identity.clone(), + )) + } else { + McpTransportClient::Http(McpHttpClient::with_options( + server.endpoint.trim().to_string(), + server.timeout_secs, + server.auth.clone(), + identity.clone(), + )) + } +} + +fn transport_command(server: &McpServerConfig) -> Option { + let command = server.command.trim(); + if command.is_empty() { + None + } else { + Some(command.to_string()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn from_config_seeds_legacy_gitbooks_when_enabled() { + let config = Config::default(); + let registry = McpServerRegistry::from_config(&config); + let gitbooks = registry.get("gitbooks").expect("gitbooks"); + assert_eq!(gitbooks.source, McpRegistrySource::LegacyGitbooks); + } + + #[test] + fn explicit_server_overrides_legacy_name() { + let mut config = Config::default(); + config.mcp_client.servers.push(McpServerConfig { + name: "gitbooks".into(), + endpoint: "https://example.com/mcp".into(), + command: String::new(), + args: Vec::new(), + env: HashMap::new(), + cwd: None, + description: Some("Custom docs".into()), + enabled: true, + timeout_secs: 9, + auth: crate::openhuman::config::McpAuthConfig::None, + }); + let registry = McpServerRegistry::from_config(&config); + let gitbooks = registry.get("gitbooks").expect("gitbooks"); + assert_eq!(gitbooks.source, McpRegistrySource::Config); + assert_eq!(gitbooks.endpoint, "https://example.com/mcp"); + } + + #[test] + fn disabled_config_short_circuits_registry() { + let mut config = Config::default(); + config.mcp_client.enabled = false; + let registry = McpServerRegistry::from_config(&config); + assert!(registry.is_empty()); + } +} diff --git a/src/openhuman/mcp_client/stdio.rs b/src/openhuman/mcp_client/stdio.rs new file mode 100644 index 000000000..67b192dc2 --- /dev/null +++ b/src/openhuman/mcp_client/stdio.rs @@ -0,0 +1,279 @@ +use super::client::{render_tool_result, McpInitializeResult, McpRemoteTool, McpServerToolResult}; +use crate::openhuman::config::McpClientIdentityConfig; +use anyhow::Context; +use serde_json::{json, Value}; +use std::path::PathBuf; +use std::process::Stdio; +use std::sync::atomic::{AtomicI64, Ordering}; +use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; +use tokio::process::{Child, ChildStdin, ChildStdout, Command}; +use tokio::sync::Mutex; + +const LATEST_PROTOCOL_VERSION: &str = "2025-11-25"; + +#[derive(Debug)] +pub struct McpStdioClient { + command: String, + args: Vec, + env: Vec<(String, String)>, + cwd: Option, + next_id: AtomicI64, + client_name: String, + client_title: String, + client_version: String, + state: Mutex>, +} + +#[derive(Debug)] +struct StdioSession { + child: Child, + stdin: ChildStdin, + stdout: BufReader, + initialize: McpInitializeResult, +} + +impl McpStdioClient { + pub fn new( + command: String, + args: Vec, + env: Vec<(String, String)>, + cwd: Option, + identity: McpClientIdentityConfig, + ) -> Self { + Self { + command, + args, + env, + cwd, + next_id: AtomicI64::new(1), + client_name: identity.name, + client_title: identity.title, + client_version: identity.version, + state: Mutex::new(None), + } + } + + pub async fn initialize(&self) -> anyhow::Result { + let mut state = self.state.lock().await; + if let Some(session) = state.as_ref() { + return Ok(session.initialize.clone()); + } + + let mut command = Command::new(&self.command); + command + .args(&self.args) + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::null()); + if let Some(cwd) = self.cwd.as_ref() { + command.current_dir(cwd); + } + for (key, value) in &self.env { + command.env(key, value); + } + + let mut child = command + .spawn() + .with_context(|| format!("spawning MCP stdio server `{}`", self.command))?; + let stdin = child + .stdin + .take() + .ok_or_else(|| anyhow::anyhow!("stdio server missing stdin"))?; + let stdout = child + .stdout + .take() + .ok_or_else(|| anyhow::anyhow!("stdio server missing stdout"))?; + let mut session = StdioSession { + child, + stdin, + stdout: BufReader::new(stdout), + initialize: McpInitializeResult { + protocol_version: LATEST_PROTOCOL_VERSION.into(), + capabilities: json!({}), + server_info: json!({}), + instructions: None, + }, + }; + + let response = self + .send_request_on_session( + &mut session, + "initialize", + json!({ + "protocolVersion": LATEST_PROTOCOL_VERSION, + "capabilities": {}, + "clientInfo": { + "name": self.client_name, + "title": self.client_title, + "version": self.client_version, + } + }), + ) + .await?; + let init: McpInitializeResult = + serde_json::from_value(response).context("parsing stdio initialize result")?; + self.send_notification_on_session(&mut session, "notifications/initialized", json!({})) + .await?; + session.initialize = init.clone(); + *state = Some(session); + Ok(init) + } + + pub async fn list_tools(&self) -> anyhow::Result> { + self.initialize().await?; + let mut state = self.state.lock().await; + let session = state + .as_mut() + .ok_or_else(|| anyhow::anyhow!("stdio MCP session not initialized"))?; + let response = self + .send_request_on_session(session, "tools/list", json!({})) + .await?; + serde_json::from_value( + response + .get("tools") + .cloned() + .ok_or_else(|| anyhow::anyhow!("stdio tools/list response missing `tools`"))?, + ) + .context("parsing stdio tools/list payload") + } + + pub async fn call_tool( + &self, + name: &str, + arguments: Value, + ) -> anyhow::Result { + self.initialize().await?; + let mut state = self.state.lock().await; + let session = state + .as_mut() + .ok_or_else(|| anyhow::anyhow!("stdio MCP session not initialized"))?; + let result = self + .send_request_on_session( + session, + "tools/call", + json!({ + "name": name, + "arguments": arguments, + }), + ) + .await?; + let rendered = render_tool_result(&result); + Ok(McpServerToolResult { + raw_result: result, + rendered, + }) + } + + pub async fn close_session(&self) -> anyhow::Result<()> { + let mut state = self.state.lock().await; + if let Some(mut session) = state.take() { + let _ = session.child.start_kill(); + let _ = session.child.wait().await; + } + Ok(()) + } + + async fn send_request_on_session( + &self, + session: &mut StdioSession, + method: &str, + params: Value, + ) -> anyhow::Result { + let id = self.next_id.fetch_add(1, Ordering::Relaxed); + let line = serde_json::to_string(&json!({ + "jsonrpc": "2.0", + "id": id, + "method": method, + "params": params, + }))?; + session.stdin.write_all(line.as_bytes()).await?; + session.stdin.write_all(b"\n").await?; + session.stdin.flush().await?; + + loop { + let mut response = String::new(); + let read = session.stdout.read_line(&mut response).await?; + if read == 0 { + anyhow::bail!("stdio MCP server closed stdout while waiting for `{method}`"); + } + let trimmed = response.trim(); + if trimmed.is_empty() { + continue; + } + if !trimmed.starts_with('{') && !trimmed.starts_with('[') { + tracing::debug!( + target: "[mcp_client::stdio]", + command = %self.command, + line = %trimmed, + "ignoring non-JSON stdout line from stdio MCP server" + ); + continue; + } + let payload: Value = serde_json::from_str(trimmed) + .with_context(|| format!("parsing stdio MCP response: {trimmed}"))?; + if let Some(err) = payload.get("error") { + anyhow::bail!("MCP stdio error: {err}"); + } + return payload + .get("result") + .cloned() + .ok_or_else(|| anyhow::anyhow!("stdio MCP response missing `result`: {payload}")); + } + } + + async fn send_notification_on_session( + &self, + session: &mut StdioSession, + method: &str, + params: Value, + ) -> anyhow::Result<()> { + let line = serde_json::to_string(&json!({ + "jsonrpc": "2.0", + "method": method, + "params": params, + }))?; + session.stdin.write_all(line.as_bytes()).await?; + session.stdin.write_all(b"\n").await?; + session.stdin.flush().await?; + Ok(()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::path::PathBuf; + + fn openhuman_core_bin() -> PathBuf { + let status = std::process::Command::new("cargo") + .args(["build", "--quiet", "--bin", "openhuman-core"]) + .status() + .expect("spawn cargo build"); + assert!(status.success(), "cargo build --bin openhuman-core failed"); + + let exe = std::env::current_exe().expect("current_exe"); + let debug_dir = exe + .parent() + .and_then(|p| p.parent()) + .expect("target/debug dir"); + let bin = debug_dir.join("openhuman-core"); + assert!(bin.exists(), "expected openhuman-core at {}", bin.display()); + bin + } + + #[tokio::test] + async fn stdio_client_talks_to_openhuman_mcp_server() { + let client = McpStdioClient::new( + openhuman_core_bin().display().to_string(), + vec!["mcp".into()], + Vec::new(), + Some(std::env::current_dir().unwrap()), + McpClientIdentityConfig::default(), + ); + let init = client.initialize().await.expect("initialize"); + assert_eq!(init.protocol_version, LATEST_PROTOCOL_VERSION); + let tools = client.list_tools().await.expect("list_tools"); + assert!(tools.iter().any(|tool| tool.name == "memory.search")); + client.close_session().await.expect("close"); + } +} diff --git a/src/openhuman/mcp_server/protocol.rs b/src/openhuman/mcp_server/protocol.rs index 3ae2288cc..a0eaa69d3 100644 --- a/src/openhuman/mcp_server/protocol.rs +++ b/src/openhuman/mcp_server/protocol.rs @@ -227,7 +227,7 @@ fn initialize_result(params: Value) -> Value { "name": "openhuman-core", "version": env!("CARGO_PKG_VERSION") }, - "instructions": "OpenHuman MCP exposes a small read-only memory surface. Use memory.search or memory.recall first, then tree.read_chunk for source text." + "instructions": "OpenHuman MCP exposes first-level core integration: inspect the live tool catalog with core.list_tools or core.tool_instructions, inspect subagents with agent.list_subagents, run a standalone subagent with agent.run_subagent, and use memory.search or memory.recall plus tree.read_chunk for local memory reads." }) } @@ -324,7 +324,7 @@ mod tests { } #[tokio::test] - async fn tools_list_returns_curated_tools() { + async fn tools_list_returns_first_level_core_tools() { let response = request(json!({ "jsonrpc": "2.0", "id": 2, @@ -340,7 +340,15 @@ mod tests { .collect::>(); assert_eq!( names, - vec!["memory.search", "memory.recall", "tree.read_chunk"] + vec![ + "core.list_tools", + "core.tool_instructions", + "agent.list_subagents", + "agent.run_subagent", + "memory.search", + "memory.recall", + "tree.read_chunk" + ] ); } diff --git a/src/openhuman/mcp_server/stdio.rs b/src/openhuman/mcp_server/stdio.rs index da5523790..aa42d7f24 100644 --- a/src/openhuman/mcp_server/stdio.rs +++ b/src/openhuman/mcp_server/stdio.rs @@ -76,7 +76,12 @@ fn print_help() { eprintln!("Usage: openhuman-core mcp [-v|--verbose]"); eprintln!(); eprintln!("Start an opt-in stdio Model Context Protocol server."); - eprintln!("The server exposes a curated read-only memory surface:"); + eprintln!("The server exposes first-level core MCP tools:"); + eprintln!(" core.list_tools"); + eprintln!(" core.tool_instructions"); + eprintln!(" agent.list_subagents"); + eprintln!(" agent.run_subagent"); + eprintln!("And the read-only memory surface:"); eprintln!(" memory.search"); eprintln!(" memory.recall"); eprintln!(" tree.read_chunk"); diff --git a/src/openhuman/mcp_server/tools.rs b/src/openhuman/mcp_server/tools.rs index 6f8616d8d..40fefae77 100644 --- a/src/openhuman/mcp_server/tools.rs +++ b/src/openhuman/mcp_server/tools.rs @@ -1,20 +1,24 @@ use serde_json::{json, Map, Value}; use crate::core::all; +use crate::openhuman::agent::harness::AgentDefinitionRegistry; +use crate::openhuman::agent::Agent; use crate::openhuman::config::rpc as config_rpc; +use crate::openhuman::providers::traits::build_tool_instructions_text; use crate::openhuman::security::{SecurityPolicy, ToolOperation}; const DEFAULT_LIMIT: u64 = 10; const MAX_LIMIT: u64 = 50; const QUERY_ARGUMENTS: &[&str] = &["query", "k"]; const TREE_READ_CHUNK_ARGUMENTS: &[&str] = &["chunk_id"]; +const SUBAGENT_RUN_ARGUMENTS: &[&str] = &["agent_id", "prompt"]; #[derive(Debug, Clone)] pub struct McpToolSpec { pub name: &'static str, pub title: &'static str, pub description: &'static str, - pub rpc_method: &'static str, + pub rpc_method: Option<&'static str>, pub input_schema: Value, } @@ -57,25 +61,67 @@ impl ToolCallError { pub fn tool_specs() -> Vec { vec![ + McpToolSpec { + name: "core.list_tools", + title: "List Core Tools", + description: "List the live core agent tool catalog that OpenHuman exposes to its orchestrator session.", + rpc_method: None, + input_schema: no_args_schema(), + }, + McpToolSpec { + name: "core.tool_instructions", + title: "Get Tool Instructions", + description: "Emit the markdown tool-use instructions block that OpenHuman injects into prompt-guided agents.", + rpc_method: None, + input_schema: no_args_schema(), + }, + McpToolSpec { + name: "agent.list_subagents", + title: "List Subagents", + description: "List registered sub-agent definitions that the core can dispatch for specialized work.", + rpc_method: None, + input_schema: no_args_schema(), + }, + McpToolSpec { + name: "agent.run_subagent", + title: "Run Subagent", + description: "Run a registered OpenHuman sub-agent directly from the core and return its final response.", + rpc_method: None, + input_schema: json!({ + "type": "object", + "properties": { + "agent_id": { + "type": "string", + "description": "Registered sub-agent id (for example `researcher`, `planner`, `code_executor`)." + }, + "prompt": { + "type": "string", + "description": "Task prompt for the sub-agent. Include the context it needs because this is a fresh session." + } + }, + "required": ["agent_id", "prompt"], + "additionalProperties": false + }), + }, McpToolSpec { name: "memory.search", title: "Search Memory", description: "Keyword-search OpenHuman's local memory tree and return matching chunks ordered by recency.", - rpc_method: "openhuman.memory_tree_search", + rpc_method: Some("openhuman.memory_tree_search"), input_schema: query_schema("Substring to match against stored memory chunks."), }, McpToolSpec { name: "memory.recall", title: "Recall Memory", description: "Semantically recall local memory-tree chunks relevant to a natural-language query.", - rpc_method: "openhuman.memory_tree_recall", + rpc_method: Some("openhuman.memory_tree_recall"), input_schema: query_schema("Natural-language query to embed and rerank against memory summaries."), }, McpToolSpec { name: "tree.read_chunk", title: "Read Memory Chunk", description: "Read one memory-tree chunk by id. Use this to inspect the source text behind search or recall results.", - rpc_method: "openhuman.memory_tree_get_chunk", + rpc_method: Some("openhuman.memory_tree_get_chunk"), input_schema: json!({ "type": "object", "properties": { @@ -113,17 +159,47 @@ pub async fn call_tool(name: &str, arguments: Value) -> Result { + reject_unexpected_arguments(¶ms, &[])?; + enforce_read_policy(spec.name).await?; + return list_core_tools().await; + } + "core.tool_instructions" => { + reject_unexpected_arguments(¶ms, &[])?; + enforce_read_policy(spec.name).await?; + return core_tool_instructions().await; + } + "agent.list_subagents" => { + reject_unexpected_arguments(¶ms, &[])?; + enforce_read_policy(spec.name).await?; + return list_subagents().await; + } + "agent.run_subagent" => { + enforce_act_policy(spec.name).await?; + return run_subagent_tool(¶ms).await; + } + _ => {} + } + validate_controller_params(&spec, ¶ms)?; enforce_read_policy(spec.name).await?; + let rpc_method = spec.rpc_method.ok_or_else(|| { + ToolCallError::Internal(format!( + "MCP tool `{}` is missing its RPC mapping", + spec.name + )) + })?; + log::debug!( "[mcp_server] tools/call dispatch tool={} rpc_method={} arg_keys={:?}", spec.name, - spec.rpc_method, + rpc_method, params.keys().collect::>() ); - match all::try_invoke_registered_rpc(spec.rpc_method, params).await { + match all::try_invoke_registered_rpc(rpc_method, params).await { Some(Ok(value)) => { log::debug!("[mcp_server] tools/call success tool={}", spec.name); Ok(tool_success(value)) @@ -140,16 +216,24 @@ pub async fn call_tool(name: &str, arguments: Value) -> Result Value { + json!({ + "type": "object", + "properties": {}, + "additionalProperties": false + }) +} + fn query_schema(query_description: &str) -> Value { json!({ "type": "object", @@ -177,6 +261,19 @@ fn build_rpc_params( ) -> Result, ToolCallError> { let args = object_arguments(arguments)?; match tool_name { + "core.list_tools" | "core.tool_instructions" | "agent.list_subagents" => { + reject_unexpected_arguments(&args, &[])?; + Ok(Map::new()) + } + "agent.run_subagent" => { + reject_unexpected_arguments(&args, SUBAGENT_RUN_ARGUMENTS)?; + let agent_id = required_non_empty_string(&args, "agent_id")?; + let prompt = required_non_empty_string(&args, "prompt")?; + Ok(Map::from_iter([ + ("agent_id".to_string(), Value::String(agent_id)), + ("prompt".to_string(), Value::String(prompt)), + ])) + } "memory.search" | "memory.recall" => { reject_unexpected_arguments(&args, QUERY_ARGUMENTS)?; let query = required_non_empty_string(&args, "query")?; @@ -276,10 +373,16 @@ fn validate_controller_params( spec: &McpToolSpec, params: &Map, ) -> Result<(), ToolCallError> { - let schema = all::schema_for_rpc_method(spec.rpc_method).ok_or_else(|| { + let rpc_method = spec.rpc_method.ok_or_else(|| { + ToolCallError::Internal(format!( + "MCP tool `{}` does not dispatch through RPC validation", + spec.name + )) + })?; + let schema = all::schema_for_rpc_method(rpc_method).ok_or_else(|| { ToolCallError::InvalidParams(format!( "mapped RPC method `{}` is not registered", - spec.rpc_method + rpc_method )) })?; all::validate_params(&schema, params).map_err(ToolCallError::InvalidParams) @@ -309,6 +412,156 @@ async fn enforce_read_policy(tool_name: &str) -> Result<(), ToolCallError> { .map_err(ToolCallError::InvalidParams) } +async fn enforce_act_policy(tool_name: &str) -> Result<(), ToolCallError> { + let config = match config_rpc::load_config_with_timeout().await { + Ok(config) => config, + Err(err) => { + log::warn!( + "[mcp_server] enforce_act_policy config load failed tool={tool_name} error={err}" + ); + return Err(ToolCallError::Internal(format!( + "failed to load config: {err}" + ))); + } + }; + let policy = SecurityPolicy::from_config(&config.autonomy, &config.workspace_dir); + policy + .enforce_tool_operation(ToolOperation::Act, tool_name) + .map_err(ToolCallError::InvalidParams) +} + +async fn load_config_and_init_registry() -> Result +{ + let config = config_rpc::load_config_with_timeout() + .await + .map_err(|err| ToolCallError::Internal(format!("failed to load config: {err}")))?; + AgentDefinitionRegistry::init_global(&config.workspace_dir).map_err(|err| { + ToolCallError::Internal(format!( + "failed to initialise AgentDefinitionRegistry: {err}" + )) + })?; + Ok(config) +} + +async fn build_orchestrator_agent() -> Result { + let config = load_config_and_init_registry().await?; + let mut agent = Agent::from_config_for_agent(&config, "orchestrator").map_err(|err| { + ToolCallError::Internal(format!("failed to build orchestrator agent: {err}")) + })?; + agent.fetch_connected_integrations().await; + let _ = agent.refresh_delegation_tools(); + Ok(agent) +} + +async fn list_core_tools() -> Result { + let agent = build_orchestrator_agent().await?; + let tools = agent + .tool_specs() + .iter() + .map(|spec| { + json!({ + "name": spec.name, + "description": spec.description, + "parameters": spec.parameters, + }) + }) + .collect::>(); + Ok(tool_success(json!({ "tools": tools }))) +} + +async fn core_tool_instructions() -> Result { + let agent = build_orchestrator_agent().await?; + Ok(tool_text_success(build_tool_instructions_text( + agent.tool_specs(), + ))) +} + +async fn list_subagents() -> Result { + let config = load_config_and_init_registry().await?; + let registry = AgentDefinitionRegistry::global().ok_or_else(|| { + ToolCallError::Internal("AgentDefinitionRegistry missing after init".to_string()) + })?; + + let definitions = registry + .list() + .into_iter() + .map(|def| { + json!({ + "id": def.id, + "display_name": def.display_name(), + "when_to_use": def.when_to_use, + "temperature": def.temperature, + "max_iterations": def.max_iterations, + "sandbox_mode": def.sandbox_mode, + "tool_scope": def.tools, + "subagents": def.subagents, + "source": def.source, + }) + }) + .collect::>(); + + let summary = format!( + "# OpenHuman Subagents\n\nWorkspace: `{}`\n\n{}", + config.workspace_dir.display(), + definitions + .iter() + .map(|def| { + let id = def.get("id").and_then(Value::as_str).unwrap_or(""); + let when = def.get("when_to_use").and_then(Value::as_str).unwrap_or(""); + format!("- **{id}**: {when}") + }) + .collect::>() + .join("\n") + ); + + Ok(json!({ + "content": [{ + "type": "text", + "text": summary, + }], + "structuredContent": { + "definitions": definitions, + } + })) +} + +async fn run_subagent_tool(params: &Map) -> Result { + let agent_id = required_non_empty_string(params, "agent_id")?; + let prompt = required_non_empty_string(params, "prompt")?; + if agent_id == "integrations_agent" { + return Err(ToolCallError::InvalidParams( + "agent.run_subagent does not yet support `integrations_agent`; first-level MCP support is currently limited to standalone agents that do not require toolkit binding".to_string(), + )); + } + + let config = load_config_and_init_registry().await?; + let mut agent = Agent::from_config_for_agent(&config, &agent_id).map_err(|err| { + ToolCallError::InvalidParams(format!("failed to build agent `{agent_id}`: {err}")) + })?; + agent.set_event_context( + format!("mcp:{}:{}", agent_id, uuid::Uuid::new_v4()), + "mcp_server", + ); + agent.fetch_connected_integrations().await; + let _ = agent.refresh_delegation_tools(); + + let response = agent + .run_single(&prompt) + .await + .map_err(|err| ToolCallError::Internal(format!("subagent `{agent_id}` failed: {err}")))?; + + Ok(json!({ + "content": [{ + "type": "text", + "text": response, + }], + "structuredContent": { + "agent_id": agent_id, + "response": response, + } + })) +} + fn tool_success(value: Value) -> Value { json!({ "content": [{ @@ -318,6 +571,15 @@ fn tool_success(value: Value) -> Value { }) } +fn tool_text_success(text: String) -> Value { + json!({ + "content": [{ + "type": "text", + "text": text, + }] + }) +} + fn tool_error(message: String) -> Value { json!({ "content": [{ @@ -344,7 +606,7 @@ mod tests { use super::*; #[test] - fn list_tools_exposes_curated_read_only_surface() { + fn list_tools_exposes_first_level_mcp_surface() { let result = list_tools_result(); let names = result["tools"] .as_array() @@ -355,22 +617,70 @@ mod tests { assert_eq!( names, - vec!["memory.search", "memory.recall", "tree.read_chunk"] + vec![ + "core.list_tools", + "core.tool_instructions", + "agent.list_subagents", + "agent.run_subagent", + "memory.search", + "memory.recall", + "tree.read_chunk" + ] ); } #[test] fn mapped_rpc_methods_are_registered() { for spec in tool_specs() { - assert!( - all::schema_for_rpc_method(spec.rpc_method).is_some(), - "missing registered RPC method for {} -> {}", - spec.name, - spec.rpc_method - ); + if let Some(rpc_method) = spec.rpc_method { + assert!( + all::schema_for_rpc_method(rpc_method).is_some(), + "missing registered RPC method for {} -> {}", + spec.name, + rpc_method + ); + } } } + #[test] + fn build_rpc_params_parses_run_subagent_arguments() { + let params = build_rpc_params( + "agent.run_subagent", + json!({ + "agent_id": "researcher", + "prompt": "Find the root cause." + }), + ) + .expect("params should parse"); + + assert_eq!( + params.get("agent_id").and_then(Value::as_str), + Some("researcher") + ); + assert_eq!( + params.get("prompt").and_then(Value::as_str), + Some("Find the root cause.") + ); + } + + #[test] + fn build_rpc_params_rejects_extra_run_subagent_fields() { + let err = build_rpc_params( + "agent.run_subagent", + json!({ + "agent_id": "researcher", + "prompt": "Find the root cause.", + "toolkit": "gmail" + }), + ) + .expect_err("unexpected field should be rejected"); + + assert!( + matches!(err, ToolCallError::InvalidParams(message) if message.contains("unexpected argument")) + ); + } + #[test] fn memory_search_params_trim_query_and_use_default_k() { let params = build_rpc_params( diff --git a/src/openhuman/mod.rs b/src/openhuman/mod.rs index b96079364..bdae0fc19 100644 --- a/src/openhuman/mod.rs +++ b/src/openhuman/mod.rs @@ -41,6 +41,7 @@ pub mod integrations; pub mod javascript; pub mod learning; pub mod local_ai; +pub mod mcp_client; pub mod mcp_server; pub mod meet; pub mod meet_agent; diff --git a/src/openhuman/tools/impl/network/gitbooks.rs b/src/openhuman/tools/impl/network/gitbooks.rs index 45ac08d30..99583d31f 100644 --- a/src/openhuman/tools/impl/network/gitbooks.rs +++ b/src/openhuman/tools/impl/network/gitbooks.rs @@ -1,176 +1,11 @@ //! `gitbooks` — answer questions about OpenHuman by talking to the -//! GitBook MCP server. -//! -//! GitBook hosts a stateless Streamable-HTTP MCP server that exposes -//! exactly two tools: -//! -//! - `searchDocumentation { query }` — returns excerpts + page links -//! - `getPage { url }` — returns the full markdown of a page -//! -//! We mirror them as `gitbooks_search` and `gitbooks_get_page`. The -//! server returns each JSON-RPC response in a single -//! `event: message\ndata: {…}` SSE frame, so we do a tiny inline -//! parse — no need to pull in a full SSE/MCP client crate yet. +//! GitBook MCP server through the shared `openhuman::mcp_client` path. +use crate::openhuman::mcp_client::{redact_endpoint, McpHttpClient}; use crate::openhuman::tools::traits::{PermissionLevel, Tool, ToolResult}; use async_trait::async_trait; use serde_json::{json, Value}; -use std::sync::atomic::{AtomicI64, Ordering}; use std::sync::Arc; -use std::time::Duration; - -/// Minimal MCP-over-HTTP client for stateless servers (no session id). -struct McpHttpClient { - endpoint: String, - timeout_secs: u64, - next_id: AtomicI64, -} - -impl McpHttpClient { - fn new(endpoint: String, timeout_secs: u64) -> Self { - Self { - endpoint, - timeout_secs, - next_id: AtomicI64::new(1), - } - } - - async fn call_tool(&self, name: &str, arguments: Value) -> anyhow::Result { - let id = self.next_id.fetch_add(1, Ordering::Relaxed); - let body = json!({ - "jsonrpc": "2.0", - "id": id, - "method": "tools/call", - "params": { "name": name, "arguments": arguments } - }); - - let builder = reqwest::Client::builder() - .timeout(Duration::from_secs(self.timeout_secs)) - .connect_timeout(Duration::from_secs(10)) - .redirect(reqwest::redirect::Policy::none()); - let builder = - crate::openhuman::config::apply_runtime_proxy_to_builder(builder, "tool.gitbooks"); - let client = builder.build()?; - - // Log only the redacted host so query strings / path params / - // any future bearer-token-bearing endpoints don't leak into - // logs aggregated for triage. - tracing::debug!( - target: "[gitbooks]", - endpoint = %redact_endpoint(&self.endpoint), - tool = %name, - "MCP tools/call" - ); - - let resp = client - .post(&self.endpoint) - .header("Content-Type", "application/json") - .header("Accept", "application/json, text/event-stream") - .body(serde_json::to_vec(&body)?) - .send() - .await?; - - let status = resp.status(); - let content_type = resp - .headers() - .get(reqwest::header::CONTENT_TYPE) - .and_then(|v| v.to_str().ok()) - .unwrap_or("") - .to_string(); - let text = resp.text().await?; - - if !status.is_success() { - anyhow::bail!("MCP HTTP {} — {}", status.as_u16(), text); - } - - let payload: Value = if content_type.starts_with("text/event-stream") { - parse_sse_message(&text)? - } else { - serde_json::from_str(&text).map_err(|e| { - anyhow::anyhow!("Failed to parse MCP JSON response: {e} — body: {text}") - })? - }; - - if let Some(err) = payload.get("error") { - anyhow::bail!("MCP error: {err}"); - } - let result = payload - .get("result") - .ok_or_else(|| anyhow::anyhow!("MCP response missing 'result': {payload}"))? - .clone(); - Ok(result) - } -} - -/// Reduce a configured endpoint URL to `scheme://host[:port]` for -/// safe logging. Anything that doesn't parse as a recognisable -/// http(s) URL is reported as `` rather than echoed. -fn redact_endpoint(raw: &str) -> String { - let trimmed = raw.trim(); - let (scheme, rest) = if let Some(r) = trimmed.strip_prefix("https://") { - ("https", r) - } else if let Some(r) = trimmed.strip_prefix("http://") { - ("http", r) - } else { - return "".into(); - }; - let authority = rest.split(['/', '?', '#']).next().unwrap_or(""); - if authority.is_empty() || authority.contains('@') { - return "".into(); - } - format!("{scheme}://{authority}") -} - -/// Parse the first `data: {…}` line from a Streamable-HTTP SSE -/// response. The GitBook server emits exactly one frame per JSON-RPC -/// request, so we do not need a full SSE state machine. -fn parse_sse_message(body: &str) -> anyhow::Result { - for line in body.lines() { - let line = line.trim_end_matches('\r'); - if let Some(data) = line.strip_prefix("data:") { - let data = data.trim_start(); - if data.is_empty() { - continue; - } - return serde_json::from_str(data).map_err(|e| { - anyhow::anyhow!("Failed to parse SSE data frame: {e} — line: {data}") - }); - } - } - anyhow::bail!("No SSE data frame found in MCP response: {body}") -} - -/// Render an MCP `tools/call` result into a single string for the -/// agent. MCP returns `{ content: [{ type, text }, …], isError? }`. -fn render_tool_result(result: &Value) -> ToolResult { - let is_error = result - .get("isError") - .and_then(Value::as_bool) - .unwrap_or(false); - - let mut out = String::new(); - if let Some(content) = result.get("content").and_then(Value::as_array) { - for block in content { - if let Some(t) = block.get("text").and_then(Value::as_str) { - if !out.is_empty() { - out.push_str("\n\n"); - } - out.push_str(t); - } - } - } - if out.is_empty() { - out = result.to_string(); - } - - if is_error { - ToolResult::error(out) - } else { - ToolResult::success(out) - } -} - -// ── Search ───────────────────────────────────────────────────────── pub struct GitbooksSearchTool { client: Arc, @@ -223,19 +58,24 @@ impl Tool for GitbooksSearchTool { return Ok(ToolResult::error("query cannot be empty")); } + tracing::debug!( + target: "[gitbooks]", + endpoint = %redact_endpoint(self.client.endpoint()), + tool = "searchDocumentation", + "dispatching via shared MCP client" + ); + match self .client .call_tool("searchDocumentation", json!({ "query": query })) .await { - Ok(result) => Ok(render_tool_result(&result)), + Ok(result) => Ok(result.rendered), Err(e) => Ok(ToolResult::error(format!("gitbooks_search failed: {e}"))), } } } -// ── Get page ────────────────────────────────────────────────────── - pub struct GitbooksGetPageTool { client: Arc, } @@ -286,12 +126,19 @@ impl Tool for GitbooksGetPageTool { return Ok(ToolResult::error("url cannot be empty")); } + tracing::debug!( + target: "[gitbooks]", + endpoint = %redact_endpoint(self.client.endpoint()), + tool = "getPage", + "dispatching via shared MCP client" + ); + match self .client .call_tool("getPage", json!({ "url": url })) .await { - Ok(result) => Ok(render_tool_result(&result)), + Ok(result) => Ok(result.rendered), Err(e) => Ok(ToolResult::error(format!("gitbooks_get_page failed: {e}"))), } } @@ -313,73 +160,6 @@ mod tests { ); } - #[test] - fn redact_endpoint_rejects_userinfo_and_unknown_schemes() { - assert_eq!( - redact_endpoint("https://user:pass@example.com/x"), - "" - ); - assert_eq!(redact_endpoint("ftp://example.com"), ""); - assert_eq!(redact_endpoint(""), ""); - } - - #[test] - fn parse_sse_extracts_data_frame() { - let body = "event: message\ndata: {\"jsonrpc\":\"2.0\",\"id\":1,\"result\":{\"x\":1}}\n\n"; - let v = parse_sse_message(body).unwrap(); - assert_eq!(v["result"]["x"], 1); - } - - #[test] - fn parse_sse_handles_crlf() { - let body = "event: message\r\ndata: {\"jsonrpc\":\"2.0\",\"id\":1,\"result\":{}}\r\n\r\n"; - assert!(parse_sse_message(body).is_ok()); - } - - #[test] - fn parse_sse_errors_when_no_data_line() { - let body = "event: ping\n\n"; - assert!(parse_sse_message(body).is_err()); - } - - #[test] - fn parse_sse_errors_on_invalid_json() { - let body = "data: not-json\n\n"; - assert!(parse_sse_message(body).is_err()); - } - - #[test] - fn render_tool_result_concatenates_text_blocks() { - let r = json!({ - "content": [ - {"type": "text", "text": "first"}, - {"type": "text", "text": "second"} - ] - }); - let out = render_tool_result(&r); - assert!(!out.is_error); - assert!(out.output().contains("first")); - assert!(out.output().contains("second")); - } - - #[test] - fn render_tool_result_marks_errors() { - let r = json!({ - "content": [{"type": "text", "text": "boom"}], - "isError": true - }); - let out = render_tool_result(&r); - assert!(out.is_error); - assert!(out.output().contains("boom")); - } - - #[test] - fn render_tool_result_falls_back_to_raw_json() { - let r = json!({"weird": "shape"}); - let out = render_tool_result(&r); - assert!(out.output().contains("weird")); - } - #[tokio::test] async fn search_rejects_empty_query() { let t = GitbooksSearchTool::new("https://example.com/mcp".into(), 5); @@ -395,9 +175,6 @@ mod tests { assert!(result.is_error); } - /// Live integration test against the real GitBook MCP endpoint. - /// Gated behind `OPENHUMAN_GITBOOKS_LIVE_TEST=1` so CI / offline - /// builds don't depend on the public network. #[tokio::test] async fn live_search_smoke() { if std::env::var("OPENHUMAN_GITBOOKS_LIVE_TEST") diff --git a/src/openhuman/tools/impl/network/mcp.rs b/src/openhuman/tools/impl/network/mcp.rs new file mode 100644 index 000000000..f3349285c --- /dev/null +++ b/src/openhuman/tools/impl/network/mcp.rs @@ -0,0 +1,313 @@ +use crate::openhuman::mcp_client::{McpRegistrySource, McpServerRegistry}; +use crate::openhuman::security::{SecurityPolicy, ToolOperation}; +use crate::openhuman::tools::traits::{PermissionLevel, Tool, ToolCallOptions, ToolResult}; +use async_trait::async_trait; +use serde_json::{json, Value}; +use std::sync::Arc; + +pub struct McpListServersTool { + registry: Arc, +} + +impl McpListServersTool { + pub fn new(registry: Arc) -> Self { + Self { registry } + } +} + +#[async_trait] +impl Tool for McpListServersTool { + fn name(&self) -> &str { + "mcp_list_servers" + } + + fn description(&self) -> &str { + "List named remote MCP servers registered in OpenHuman core. Use this before browsing tools on a specific MCP server." + } + + fn parameters_schema(&self) -> Value { + json!({ + "type": "object", + "properties": {}, + "additionalProperties": false + }) + } + + fn permission_level(&self) -> PermissionLevel { + PermissionLevel::ReadOnly + } + + fn supports_markdown(&self) -> bool { + true + } + + async fn execute(&self, _args: Value) -> anyhow::Result { + let servers = self + .registry + .list() + .into_iter() + .map(|server| { + json!({ + "name": server.name, + "endpoint": server.endpoint, + "description": server.description, + "timeout_secs": server.timeout_secs, + "auth": server.auth, + "source": server.source, + }) + }) + .collect::>(); + + let markdown = if servers.is_empty() { + "# MCP Servers\n\nNo remote MCP servers are registered.".to_string() + } else { + let mut md = String::from("# MCP Servers\n"); + for server in self.registry.list() { + let source = match server.source { + McpRegistrySource::Config => "config", + McpRegistrySource::LegacyGitbooks => "legacy_gitbooks", + }; + md.push_str(&format!( + "\n- **{}** ({source})\n - endpoint: `{}`\n - auth: `{}`", + server.name, + server.endpoint, + match &server.auth { + crate::openhuman::config::McpAuthConfig::None => "none", + crate::openhuman::config::McpAuthConfig::BearerToken { .. } => + "bearer_token", + crate::openhuman::config::McpAuthConfig::Basic { .. } => "basic", + crate::openhuman::config::McpAuthConfig::Header { .. } => "header", + crate::openhuman::config::McpAuthConfig::QueryParam { .. } => "query_param", + } + )); + if let Some(description) = server.description.as_deref() { + md.push_str(&format!("\n - {description}")); + } + } + md + }; + + Ok(ToolResult::success_with_markdown( + json!({ "servers": servers }), + markdown, + )) + } +} + +pub struct McpListToolsTool { + registry: Arc, +} + +impl McpListToolsTool { + pub fn new(registry: Arc) -> Self { + Self { registry } + } +} + +#[async_trait] +impl Tool for McpListToolsTool { + fn name(&self) -> &str { + "mcp_list_tools" + } + + fn description(&self) -> &str { + "List tools exposed by a named remote MCP server. Use this before calling `mcp_call_tool`." + } + + fn parameters_schema(&self) -> Value { + json!({ + "type": "object", + "properties": { + "server": { + "type": "string", + "description": "Registered MCP server name from `mcp_list_servers`." + } + }, + "required": ["server"], + "additionalProperties": false + }) + } + + fn permission_level(&self) -> PermissionLevel { + PermissionLevel::ReadOnly + } + + fn supports_markdown(&self) -> bool { + true + } + + async fn execute(&self, args: Value) -> anyhow::Result { + let server = required_string_arg(&args, "server")?; + let tools = match self.registry.list_tools(&server).await { + Ok(tools) => tools, + Err(err) => return Ok(ToolResult::error(format!("mcp_list_tools failed: {err}"))), + }; + + let payload = tools + .iter() + .map(|tool| { + json!({ + "name": tool.name, + "title": tool.title, + "description": tool.description, + "input_schema": tool.input_schema, + }) + }) + .collect::>(); + + let mut markdown = format!("# MCP Tools: `{server}`\n"); + if tools.is_empty() { + markdown.push_str("\nNo tools were returned by the remote server."); + } else { + for tool in &tools { + markdown.push_str(&format!( + "\n- **{}**: {}\n - schema: `{}`", + tool.name, + tool.description.as_deref().unwrap_or("No description."), + serde_json::to_string(&tool.input_schema).unwrap_or_else(|_| "{}".into()) + )); + } + } + + Ok(ToolResult::success_with_markdown( + json!({ "server": server, "tools": payload }), + markdown, + )) + } +} + +pub struct McpCallTool { + registry: Arc, + security: Arc, +} + +impl McpCallTool { + pub fn new(registry: Arc, security: Arc) -> Self { + Self { registry, security } + } +} + +#[async_trait] +impl Tool for McpCallTool { + fn name(&self) -> &str { + "mcp_call_tool" + } + + fn description(&self) -> &str { + "Call a tool on a named remote MCP server. First inspect available tools with `mcp_list_tools`, then pass the remote tool name and its JSON arguments here." + } + + fn parameters_schema(&self) -> Value { + json!({ + "type": "object", + "properties": { + "server": { + "type": "string", + "description": "Registered MCP server name from `mcp_list_servers`." + }, + "tool": { + "type": "string", + "description": "Remote MCP tool name from `mcp_list_tools`." + }, + "arguments": { + "type": "object", + "description": "Arguments object passed through to the remote MCP tool." + } + }, + "required": ["server", "tool", "arguments"], + "additionalProperties": false + }) + } + + fn permission_level(&self) -> PermissionLevel { + PermissionLevel::Execute + } + + fn supports_markdown(&self) -> bool { + true + } + + async fn execute_with_options( + &self, + args: Value, + options: ToolCallOptions, + ) -> anyhow::Result { + self.security + .enforce_tool_operation(ToolOperation::Act, self.name()) + .map_err(|err| anyhow::anyhow!(err))?; + + let server = required_string_arg(&args, "server")?; + let tool = required_string_arg(&args, "tool")?; + let arguments = args + .get("arguments") + .cloned() + .ok_or_else(|| anyhow::anyhow!("missing required `arguments` object"))?; + if !arguments.is_object() { + return Ok(ToolResult::error("`arguments` must be an object")); + } + + let mut result = match self.registry.call_tool(&server, &tool, arguments).await { + Ok(result) => result.rendered, + Err(err) => return Ok(ToolResult::error(format!("mcp_call_tool failed: {err}"))), + }; + + if options.prefer_markdown && result.markdown_formatted.is_none() { + result.markdown_formatted = Some(result.output()); + } + Ok(result) + } + + async fn execute(&self, args: Value) -> anyhow::Result { + self.execute_with_options(args, ToolCallOptions::default()) + .await + } +} + +fn required_string_arg(args: &Value, key: &str) -> anyhow::Result { + let value = args + .get(key) + .and_then(Value::as_str) + .map(str::trim) + .filter(|s| !s.is_empty()) + .ok_or_else(|| anyhow::anyhow!("missing required `{key}`"))?; + Ok(value.to_string()) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::openhuman::config::{Config, McpServerConfig}; + + fn test_registry() -> Arc { + let mut config = Config::default(); + config.gitbooks.enabled = false; + config.mcp_client.servers.push(McpServerConfig { + name: "docs".into(), + endpoint: "https://example.com/mcp".into(), + command: String::new(), + args: Vec::new(), + env: std::collections::HashMap::new(), + cwd: None, + description: Some("Docs MCP".into()), + enabled: true, + timeout_secs: 30, + auth: crate::openhuman::config::McpAuthConfig::None, + }); + Arc::new(McpServerRegistry::from_config(&config)) + } + + #[tokio::test] + async fn list_servers_renders_registry_entries() { + let tool = McpListServersTool::new(test_registry()); + let result = tool.execute(json!({})).await.expect("execute"); + assert!(result.output().contains("docs")); + assert!(result.markdown_formatted.is_some()); + } + + #[tokio::test] + async fn list_tools_requires_server() { + let tool = McpListToolsTool::new(test_registry()); + let result = tool.execute(json!({})).await; + assert!(result.is_err()); + } +} diff --git a/src/openhuman/tools/impl/network/mod.rs b/src/openhuman/tools/impl/network/mod.rs index 9865a0a9a..3d9386ef8 100644 --- a/src/openhuman/tools/impl/network/mod.rs +++ b/src/openhuman/tools/impl/network/mod.rs @@ -3,6 +3,7 @@ mod curl; mod gitbooks; mod gmail_unsubscribe; mod http_request; +mod mcp; mod url_guard; mod web_fetch; mod web_search; @@ -12,5 +13,6 @@ pub use curl::CurlTool; pub use gitbooks::{GitbooksGetPageTool, GitbooksSearchTool}; pub use gmail_unsubscribe::GmailUnsubscribeTool; pub use http_request::HttpRequestTool; +pub use mcp::{McpCallTool, McpListServersTool, McpListToolsTool}; pub use web_fetch::WebFetchTool; pub use web_search::WebSearchTool; diff --git a/src/openhuman/tools/ops.rs b/src/openhuman/tools/ops.rs index 53ed828fb..0e656d3b3 100644 --- a/src/openhuman/tools/ops.rs +++ b/src/openhuman/tools/ops.rs @@ -242,6 +242,26 @@ pub fn all_tools_with_runtime( tracing::debug!("[gitbooks] registered gitbooks_search + gitbooks_get_page"); } + // Generic remote MCP bridge tools. These let the agent enumerate + // named MCP servers and forward `tools/call` through the core + // instead of hardcoding one bespoke MCP integration per server. + let mcp_registry = + Arc::new(crate::openhuman::mcp_client::McpServerRegistry::from_config(root_config)); + if !mcp_registry.is_empty() { + tools.push(Box::new(McpListServersTool::new(Arc::clone(&mcp_registry)))); + tools.push(Box::new(McpListToolsTool::new(Arc::clone(&mcp_registry)))); + tools.push(Box::new(McpCallTool::new( + Arc::clone(&mcp_registry), + security.clone(), + ))); + tracing::debug!( + count = mcp_registry.list().len(), + "[mcp_client] registered generic MCP bridge tools" + ); + } else { + tracing::debug!("[mcp_client] no MCP servers registered — bridge tools skipped"); + } + // Web search — always registered. Result/timeout budget // knobs still come from `config.web_search`, but there is no // enable flag: every session needs research as a baseline diff --git a/src/openhuman/tools/ops_tests.rs b/src/openhuman/tools/ops_tests.rs index a44aa8cf6..ade1b0f19 100644 --- a/src/openhuman/tools/ops_tests.rs +++ b/src/openhuman/tools/ops_tests.rs @@ -239,6 +239,34 @@ fn all_tools_registers_gitbooks_when_enabled() { ); } +#[test] +fn all_tools_registers_generic_mcp_bridge_tools_when_servers_exist() { + let tmp = TempDir::new().unwrap(); + let mut cfg = test_config(&tmp); + cfg.gitbooks.enabled = false; + cfg.mcp_client + .servers + .push(crate::openhuman::config::McpServerConfig { + name: "docs".into(), + endpoint: "https://example.com/mcp".into(), + command: String::new(), + args: Vec::new(), + env: std::collections::HashMap::new(), + cwd: None, + description: Some("Example docs MCP".into()), + enabled: true, + timeout_secs: 30, + auth: crate::openhuman::config::McpAuthConfig::None, + }); + + let tools = integration_tools_for_config(&tmp, &cfg); + let names = tool_names(&tools); + assert_contains_all( + &names, + &["mcp_list_servers", "mcp_list_tools", "mcp_call_tool"], + ); +} + #[test] fn all_tools_skips_gitbooks_when_disabled() { let tmp = TempDir::new().unwrap();