tools_kit/mcp.rs
1//! MCP client construction and independently owned server lifecycle.
2//!
3//! [`McpClientConfig`](crate::mcp::McpClientConfig) selects stdio or Streamable HTTP plus naming and interaction
4//! policy. Construction is passive; [`crate::Toolset::mcp`] connects. The optional
5//! `mcp-server` feature adds a server whose consuming shutdown settles serving
6//! work while the caller retains separate toolset shutdown authority.
7//!
8//! # Invoke an explicitly configured service
9//!
10//! This example requires a running Streamable HTTP server at the configured URL,
11//! with an `echo` tool accepting `{"text":"hello"}` and returning that text.
12//! MCP invocation arguments must be JSON objects. Supply any
13//! required authentication through the explicit header map. Namespaces affect public
14//! selection; they do not rename the native tool. Never infer a server endpoint from
15//! a Toolset. A request timeout bounds calls, not a whole application scenario.
16//!
17//! ```no_run
18//! use std::time::Duration;
19//! use tools_kit::{Toolset, ToolsetBackend, ToolHandlerBackend, ToolError};
20//! use tools_kit::declaration::ToolName;
21//! use tools_kit::execution::{ToolContent, ToolHandlerInput, ToolHandlerOutput, ToolInvocation};
22//! use tools_kit::mcp::{McpClientConfig, McpConnectionConfig};
23//! # #[tokio::main(flavor = "current_thread")]
24//! # async fn main() -> Result<(), ToolError> {
25//! let config = McpClientConfig::new(McpConnectionConfig::StreamableHttp {
26//! endpoint: "http://127.0.0.1:8000/mcp".parse().unwrap(),
27//! headers: Default::default(),
28//! }).with_request_timeout(Duration::from_secs(30))?;
29//! let tools = Toolset::mcp(config).await?;
30//! let work = async {
31//! let name = ToolName::new("echo")?;
32//! let inventory = tools.list_tools().await?;
33//! assert!(inventory.iter().any(|info| info.name == name));
34//! let mut handler = tools.tool_handler(name).await?;
35//! let output = handler.step(ToolHandlerInput::Invoke(ToolInvocation {
36//! call_id: Some("owned-call".into()), arguments: serde_json::json!({"text":"hello"}),
37//! })).await?;
38//! let ToolHandlerOutput::Completed(result) = output else {
39//! return Err(ToolError::Protocol { message: "configured echo unexpectedly interacted".into() });
40//! };
41//! assert!(!result.is_error);
42//! assert!(matches!(result.content.as_slice(), [ToolContent::Text { text, .. }] if text == "hello"));
43//! Ok::<(), ToolError>(())
44//! }.await;
45//! let cleanup = tools.shutdown().await;
46//! work?;
47//! cleanup?;
48//! # Ok(())
49//! # }
50//! ```
51//!
52//! With `mcp-server`, the `McpServer` type's complete example starts a separately
53//! owned loopback server, invokes it through a client, then settles the client,
54//! server, and underlying Toolset in that order. Without that feature only client
55//! construction is available. A server does not own the caller's underlying source.
56
57#[cfg(feature = "mcp-server")]
58use std::net::SocketAddr;
59use std::{collections::BTreeMap, fmt, path::PathBuf, time::Duration};
60
61use serde::{Deserialize, Serialize};
62use url::Url;
63
64use crate::{ToolError, name::ToolName};
65
66#[cfg(feature = "mcp-server")]
67pub use crate::sources::mcp::server::McpServer;
68
69/// Connection to one MCP server.
70#[derive(Clone, PartialEq, Eq, Serialize, Deserialize)]
71#[serde(tag = "transport", rename_all = "snake_case", deny_unknown_fields)]
72pub enum McpConnectionConfig {
73 /// Spawns an MCP server and communicates over standard input/output.
74 Stdio {
75 /// Executable to spawn.
76 program: String,
77 /// Arguments passed to the executable.
78 args: Vec<String>,
79 /// Optional working directory for the child process.
80 cwd: Option<PathBuf>,
81 /// Explicit environment entries supplied to the child process.
82 env: BTreeMap<String, String>,
83 },
84 /// Connects to a Streamable HTTP MCP endpoint.
85 StreamableHttp {
86 /// Absolute MCP endpoint URL.
87 endpoint: Url,
88 /// Request headers, including caller-owned authentication when needed.
89 headers: BTreeMap<String, String>,
90 },
91}
92
93impl fmt::Debug for McpConnectionConfig {
94 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
95 match self {
96 Self::Stdio {
97 program,
98 args,
99 cwd,
100 env,
101 } => formatter
102 .debug_struct("McpConnectionConfig::Stdio")
103 .field("program", program)
104 .field("argument_count", &args.len())
105 .field("has_working_directory", &cwd.is_some())
106 .field("environment_count", &env.len())
107 .finish(),
108 Self::StreamableHttp { endpoint, headers } => formatter
109 .debug_struct("McpConnectionConfig::StreamableHttp")
110 .field("endpoint_scheme", &endpoint.scheme())
111 .field("endpoint_host", &endpoint.host_str())
112 .field("endpoint_port", &endpoint.port_or_known_default())
113 .field("header_count", &headers.len())
114 .finish(),
115 }
116 }
117}
118
119/// Validated MCP client Toolset configuration.
120/// Configuration construction does not connect or spawn a process; [`crate::Toolset::mcp`]
121/// performs that work. Discovered names may be namespaced without exposing native routing.
122///
123/// # Examples
124///
125/// ```
126/// use tools_kit::mcp::{McpClientConfig, McpConnectionConfig};
127/// let connection = McpConnectionConfig::StreamableHttp { endpoint: "http://127.0.0.1:8000/mcp".parse().unwrap(), headers: Default::default() };
128/// let config = McpClientConfig::new(connection);
129/// let value = serde_json::to_value(&config).unwrap();
130/// assert_eq!(value["request_timeout_secs"], 30);
131/// assert_eq!(value["form_interactions"], false);
132/// ```
133///
134#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
135#[serde(deny_unknown_fields)]
136pub struct McpClientConfig {
137 connection: McpConnectionConfig,
138 namespace: Option<ToolName>,
139 request_timeout_secs: u64,
140 form_interactions: bool,
141 validate_forms: bool,
142}
143
144impl McpClientConfig {
145 /// Creates a client configuration with a 30-second request timeout.
146 ///
147 /// # Examples
148 ///
149 /// ```
150 /// use tools_kit::mcp::{McpClientConfig, McpConnectionConfig};
151 /// let connection = McpConnectionConfig::StreamableHttp { endpoint: "http://127.0.0.1:8000/mcp".parse().unwrap(), headers: Default::default() };
152 /// let config = McpClientConfig::new(connection);
153 /// let value = serde_json::to_value(&config).unwrap();
154 /// assert_eq!(value["request_timeout_secs"], 30);
155 /// assert_eq!(value["form_interactions"], false);
156 /// ```
157 ///
158 #[must_use]
159 pub fn new(connection: McpConnectionConfig) -> Self {
160 Self {
161 connection,
162 namespace: None,
163 request_timeout_secs: 30,
164 form_interactions: false,
165 validate_forms: false,
166 }
167 }
168
169 /// Applies a caller-visible namespace to discovered names.
170 ///
171 /// # Errors
172 ///
173 /// Returns [`ToolError::InvalidConfiguration`] if the namespace is not a valid [`ToolName`].
174 ///
175 /// # Examples
176 ///
177 /// ```
178 /// use tools_kit::mcp::{McpClientConfig, McpConnectionConfig};
179 /// let connection = McpConnectionConfig::StreamableHttp { endpoint: "http://127.0.0.1:8000/mcp".parse().unwrap(), headers: Default::default() };
180 /// let config = McpClientConfig::new(connection);
181 /// let config = config.with_namespace("search")?;
182 /// assert_eq!(serde_json::to_value(config).unwrap()["namespace"], "search");
183 ///
184 /// # Ok::<(), tools_kit::ToolError>(())
185 /// ```
186 ///
187 pub fn with_namespace<N>(mut self, namespace: N) -> Result<Self, ToolError>
188 where
189 N: TryInto<ToolName>,
190 N::Error: fmt::Display,
191 {
192 self.namespace =
193 Some(
194 namespace
195 .try_into()
196 .map_err(|error| ToolError::InvalidConfiguration {
197 message: error.to_string(),
198 })?,
199 );
200 Ok(self)
201 }
202
203 /// Sets the nonzero per-request timeout.
204 ///
205 /// # Errors
206 ///
207 /// Returns [`ToolError::InvalidConfiguration`] for zero. Positive subsecond values become one second; other values use whole seconds.
208 ///
209 /// # Examples
210 ///
211 /// ```
212 /// use tools_kit::mcp::{McpClientConfig, McpConnectionConfig};
213 /// let connection = McpConnectionConfig::StreamableHttp { endpoint: "http://127.0.0.1:8000/mcp".parse().unwrap(), headers: Default::default() };
214 /// let config = McpClientConfig::new(connection);
215 /// assert!(config.clone().with_request_timeout(std::time::Duration::ZERO).is_err());
216 /// let config = config.with_request_timeout(std::time::Duration::from_secs(5))?;
217 /// assert_eq!(serde_json::to_value(config).unwrap()["request_timeout_secs"], 5);
218 ///
219 /// # Ok::<(), tools_kit::ToolError>(())
220 /// ```
221 ///
222 pub fn with_request_timeout(mut self, timeout: Duration) -> Result<Self, ToolError> {
223 if timeout.is_zero() {
224 return Err(ToolError::InvalidConfiguration {
225 message: "MCP request timeout must be greater than zero".to_owned(),
226 });
227 }
228 self.request_timeout_secs = timeout.as_secs().max(1);
229 Ok(self)
230 }
231
232 /// Enables form interactions and selects response schema validation.
233 /// Passing false still enables forms. Replies must remain JSON objects; only schema
234 /// constraints are relaxed.
235 ///
236 /// # Examples
237 ///
238 /// ```
239 /// use tools_kit::mcp::{McpClientConfig, McpConnectionConfig};
240 /// let connection = McpConnectionConfig::StreamableHttp { endpoint: "http://127.0.0.1:8000/mcp".parse().unwrap(), headers: Default::default() };
241 /// let config = McpClientConfig::new(connection);
242 /// let config = config.with_form_validation(false);
243 /// let value = serde_json::to_value(config).unwrap();
244 /// assert_eq!(value["form_interactions"], true);
245 /// assert_eq!(value["validate_forms"], false);
246 /// ```
247 ///
248 #[must_use]
249 pub fn with_form_validation(mut self, enabled: bool) -> Self {
250 self.form_interactions = true;
251 self.validate_forms = enabled;
252 self
253 }
254
255 pub(crate) fn connection(&self) -> &McpConnectionConfig {
256 &self.connection
257 }
258
259 pub(crate) fn namespace(&self) -> Option<&ToolName> {
260 self.namespace.as_ref()
261 }
262
263 pub(crate) fn request_timeout(&self) -> Duration {
264 Duration::from_secs(self.request_timeout_secs)
265 }
266
267 pub(crate) fn validates_forms(&self) -> bool {
268 self.validate_forms
269 }
270
271 pub(crate) fn supports_forms(&self) -> bool {
272 self.form_interactions
273 }
274}
275
276/// Policy for resolving interactions while serving a Toolset over MCP.
277#[cfg(feature = "mcp-server")]
278#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
279#[serde(rename_all = "snake_case")]
280pub enum McpServerInteractionPolicy {
281 /// Use MCP elicitation for form requests when the client advertises support.
282 ElicitFormIfClientSupports,
283 /// Reject an interaction-requesting call with an MCP request error.
284 #[default]
285 RejectWithToolError,
286}
287
288/// Validated independent MCP server configuration.
289/// Available with the `mcp-server` feature. Constructing this value does not bind a socket.
290///
291/// # Examples
292///
293/// ```
294/// use tools_kit::mcp::McpServerConfig;
295/// let config = McpServerConfig::new("127.0.0.1:8000".parse().unwrap(), "/mcp", "page")?;
296/// assert!(serde_json::to_value(&config).unwrap().is_object());
297///
298/// # Ok::<(), tools_kit::ToolError>(())
299/// ```
300///
301#[cfg(feature = "mcp-server")]
302#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
303#[serde(deny_unknown_fields)]
304pub struct McpServerConfig {
305 bind: SocketAddr,
306 endpoint_path: String,
307 cursor_prefix: String,
308 interaction_policy: McpServerInteractionPolicy,
309 shutdown_timeout_secs: u64,
310}
311
312#[cfg(feature = "mcp-server")]
313impl McpServerConfig {
314 /// Creates a validated loopback server configuration.
315 ///
316 /// # Errors
317 ///
318 /// Returns [`ToolError::InvalidConfiguration`] unless the address is loopback with a nonzero port, the endpoint starts with `/` and contains no query or fragment, and the cursor prefix is nonblank.
319 ///
320 /// # Examples
321 ///
322 /// ```
323 /// use tools_kit::mcp::McpServerConfig;
324 /// let config = McpServerConfig::new("127.0.0.1:8000".parse().unwrap(), "/mcp", "page")?;
325 /// let value = serde_json::to_value(&config).unwrap();
326 /// assert_eq!(value["shutdown_timeout_secs"], 5);
327 /// assert_eq!(value["interaction_policy"], "reject_with_tool_error");
328 ///
329 /// # Ok::<(), tools_kit::ToolError>(())
330 /// ```
331 ///
332 pub fn new(
333 bind: SocketAddr,
334 endpoint_path: impl Into<String>,
335 cursor_prefix: impl Into<String>,
336 ) -> Result<Self, ToolError> {
337 let config = Self {
338 bind,
339 endpoint_path: endpoint_path.into(),
340 cursor_prefix: cursor_prefix.into(),
341 interaction_policy: McpServerInteractionPolicy::default(),
342 shutdown_timeout_secs: 5,
343 };
344 config.validate()?;
345 Ok(config)
346 }
347
348 /// Selects how the server handles tool-requested caller interactions.
349 ///
350 /// # Examples
351 ///
352 /// ```
353 /// use tools_kit::mcp::McpServerConfig;
354 /// let config = McpServerConfig::new("127.0.0.1:8000".parse().unwrap(), "/mcp", "page")?;
355 /// let config = config.with_interaction_policy(tools_kit::mcp::McpServerInteractionPolicy::ElicitFormIfClientSupports);
356 /// let value = serde_json::to_value(&config).unwrap();
357 /// assert_eq!(value["shutdown_timeout_secs"], 5);
358 /// assert_eq!(value["interaction_policy"], "elicit_form_if_client_supports");
359 ///
360 /// # Ok::<(), tools_kit::ToolError>(())
361 /// ```
362 ///
363 #[must_use]
364 pub fn with_interaction_policy(mut self, policy: McpServerInteractionPolicy) -> Self {
365 self.interaction_policy = policy;
366 self
367 }
368
369 /// Sets the nonzero deadline for joining server shutdown.
370 ///
371 /// # Errors
372 ///
373 /// Returns [`ToolError::InvalidConfiguration`] for zero. Positive subsecond values become one second; other values use whole seconds. This bounds observation and never discards retained cleanup.
374 ///
375 /// # Examples
376 ///
377 /// ```
378 /// use tools_kit::mcp::McpServerConfig;
379 /// let config = McpServerConfig::new("127.0.0.1:8000".parse().unwrap(), "/mcp", "page")?;
380 /// assert!(config.clone().with_shutdown_timeout(std::time::Duration::ZERO).is_err());
381 /// let config = config.with_shutdown_timeout(std::time::Duration::from_secs(10))?;
382 /// let value = serde_json::to_value(&config).unwrap();
383 /// assert_eq!(value["shutdown_timeout_secs"], 10);
384 /// assert_eq!(value["interaction_policy"], "reject_with_tool_error");
385 ///
386 /// # Ok::<(), tools_kit::ToolError>(())
387 /// ```
388 ///
389 pub fn with_shutdown_timeout(mut self, timeout: Duration) -> Result<Self, ToolError> {
390 if timeout.is_zero() {
391 return Err(ToolError::InvalidConfiguration {
392 message: "MCP server shutdown timeout must be greater than zero".to_owned(),
393 });
394 }
395 self.shutdown_timeout_secs = timeout.as_secs().max(1);
396 Ok(self)
397 }
398
399 pub(crate) fn validate(&self) -> Result<(), ToolError> {
400 if !self.bind.ip().is_loopback() || self.bind.port() == 0 {
401 return Err(ToolError::InvalidConfiguration {
402 message: "MCP server requires a loopback address with a nonzero port".to_owned(),
403 });
404 }
405 if self.endpoint_path.is_empty()
406 || !self.endpoint_path.starts_with('/')
407 || self.endpoint_path.contains('?')
408 || self.endpoint_path.contains('#')
409 {
410 return Err(ToolError::InvalidConfiguration {
411 message: "MCP endpoint path must be absolute and contain no query or fragment"
412 .to_owned(),
413 });
414 }
415 if self.cursor_prefix.trim().is_empty() {
416 return Err(ToolError::InvalidConfiguration {
417 message: "MCP pagination cursor prefix must not be empty".to_owned(),
418 });
419 }
420 Ok(())
421 }
422
423 pub(crate) fn bind(&self) -> SocketAddr {
424 self.bind
425 }
426
427 pub(crate) fn endpoint_path(&self) -> &str {
428 &self.endpoint_path
429 }
430
431 pub(crate) fn cursor_prefix(&self) -> &str {
432 &self.cursor_prefix
433 }
434
435 pub(crate) fn interaction_policy(&self) -> McpServerInteractionPolicy {
436 self.interaction_policy
437 }
438
439 pub(crate) fn shutdown_timeout(&self) -> Duration {
440 Duration::from_secs(self.shutdown_timeout_secs)
441 }
442}