From c9415b26c3548139892d5da7d7eaa3890cbee7de Mon Sep 17 00:00:00 2001 From: Iurii Bobrykov Date: Sat, 8 Aug 2026 10:47:10 +1200 Subject: [PATCH 1/2] feat: add mcp tool provider --- CHANGELOG.md | 1 + Cargo.toml | 8 + README.md | 1 + examples/mcp-adapter.rs | 89 ++++ src/engine/bare/tests.rs | 10 +- src/lib.rs | 3 + src/mcp.rs | 952 +++++++++++++++++++++++++++++++++++++ tests/mcp_tool_provider.rs | 710 +++++++++++++++++++++++++++ 8 files changed, 1773 insertions(+), 1 deletion(-) create mode 100644 examples/mcp-adapter.rs create mode 100644 src/mcp.rs create mode 100644 tests/mcp_tool_provider.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index 44032d7..7773950 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/2.0.0. ### Added +- `mcp` feature + `loopctl::mcp` module — adapt any MCP server's tools as loopctl `Tool` implementations. New public types: `McpClient` (a connected client handle), `McpToolProvider` (discovers a server's tools and registers them into a `ToolRegistry`), `McpTool` (one server tool as a `Tool`), `McpError`. The adapter is transport-agnostic; `McpClient::in_process` connects an in-process rmcp server for tests and bundled-server use. Real transports (stdio, HTTP/SSE) arrive in a later release. The optional `rmcp` dependency is pulled in only by the `mcp` feature (`default = []` is unchanged). A runnable end-to-end demo ships at `examples/mcp-adapter.rs` (`cargo run --example mcp-adapter --features mcp`). - `LoopError::ToolRecoveryExhausted { tool, attempts }` — the driver now enforces `MAX_RECOVERY_ATTEMPTS` (5) as a hard ceiling. A recovery strategy that always returns `Retry` is stopped after 5 retries (attempt 6), returning this variant instead of looping forever. Pinned by `recovery_ceiling_stops_retry_forever_strategy`. - `RunConfig::memory_top_k` — configurable number of memory entries retrieved and injected per turn (default 3; was a hardcoded magic number). - `MachineStep::CallTools { turn, calls }` — the machine now emits the 0-indexed turn number on `CallTools` (matching `CallLLM`), so both handlers source the turn identically from the machine rather than one reading a field and the other querying a counter. diff --git a/Cargo.toml b/Cargo.toml index 9d58394..726c4cf 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -33,6 +33,7 @@ bytes = { version = "1", optional = true } async-stream = { version = "0.3", optional = true } httpdate = { version = "1", optional = true } jsonschema = { version = "0.49", optional = true } +rmcp = { version = "3", optional = true, default-features = false, features = ["client", "server", "macros", "transport-async-rw"] } [dev-dependencies] tokio = { version = "1.52.3", features = ["rt-multi-thread", "macros", "sync", "time", "signal"] } @@ -61,6 +62,9 @@ zai = ["providers", "anthropic"] grammar = ["providers"] schema_validation = ["dep:jsonschema"] +# MCP client adapter +mcp = ["dep:rmcp"] + [[example]] name = "hello-cli" required-features = ["testing"] @@ -77,6 +81,10 @@ required-features = ["testing"] name = "chat" required-features = ["testing", "providers"] +[[example]] +name = "mcp-adapter" +required-features = ["mcp"] + [lints.clippy] pedantic = { level = "warn", priority = -1 } unwrap_used = "deny" diff --git a/README.md b/README.md index 8a7197b..a23729c 100644 --- a/README.md +++ b/README.md @@ -158,6 +158,7 @@ let agent = BareLoop::new( | `zai` | No | `providers`, `anthropic` | Z.AI API client (Anthropic-compatible) | | `grammar` | No | `providers` | Tool-call grammar providers for grammar-aware samplers (vLLM `guided_json`); enables the `Grammar` mode of `ToolConstraint` | | `schema_validation` | No | — | JSON Schema validation of `Correction::modified_input` in `LlmReflector` (pulls `jsonschema`); when off, validation is skipped | +| `mcp` | No | `rmcp` | MCP client adapter (`mcp::McpToolProvider`) — adapt any MCP server's tools as loopctl `Tool` impls | ### Streaming vs non-streaming diff --git a/examples/mcp-adapter.rs b/examples/mcp-adapter.rs new file mode 100644 index 0000000..19cb7be --- /dev/null +++ b/examples/mcp-adapter.rs @@ -0,0 +1,89 @@ +//! Adapt an in-process MCP server's tools as loopctl `Tool` impls. +//! +//! Builds a tiny rmcp server (with the `#[tool_router]` / `#[tool]` macros), +//! connects an [`McpToolProvider`] to it over a `tokio::io::duplex`, discovers +//! its tools, registers them into a [`ToolRegistry`], and calls one to prove +//! the round-trip works end-to-end. +//! +//! ```sh +//! cargo run --example mcp-adapter --features mcp +//! ``` + +#![allow( + clippy::expect_used, + clippy::panic, + clippy::indexing_slicing, + clippy::missing_errors_doc, + clippy::missing_panics_doc, + dead_code +)] + +use loopctl::mcp::{McpClient, McpToolProvider}; +use loopctl::tool::{ToolContext, ToolRegistry}; +use rmcp::handler::server::ServerHandler; +use rmcp::handler::server::router::tool::ToolRouter; +use rmcp::{ServiceExt, tool, tool_handler, tool_router}; +use serde_json::json; + +/// An rmcp server exposing one `greet` tool. +#[derive(Clone)] +struct GreetServer { + router: ToolRouter, +} + +#[tool_router] +impl GreetServer { + fn new() -> Self { + Self { + router: Self::tool_router(), + } + } + + #[tool(description = "Return a friendly greeting")] + async fn greet(&self) -> String { + "hello, world!".to_string() + } +} + +#[tool_handler] +impl ServerHandler for GreetServer {} + +#[tokio::main] +async fn main() { + // 1. Start the server on one end of a duplex, a pure rmcp client on the + // other, and wrap the client as an McpClient. + let (server_end, client_end) = tokio::io::duplex(4096); + tokio::spawn(async move { + let running = GreetServer::new() + .serve(server_end) + .await + .expect("server initialize"); + running.waiting().await.ok(); + }); + let client = + ().serve(client_end) + .await + .map(McpClient::from_service) + .expect("client initialize"); + + // 2. Discover the server's tools and register them. + let provider = McpToolProvider::connect(client, None) + .await + .expect("connect + list_tools"); + let mut registry = ToolRegistry::new(); + provider.register_into(&mut registry); + println!("discovered {} tool(s):", registry.len()); + for schema in registry.all_schemas() { + println!(" - {} : {}", schema.tool, schema.description); + } + + // 3. Call the adapted `greet` tool through the registry, exactly as the + // agent loop would call any native loopctl tool. + let greet = registry.get("greet").expect("greet registered"); + let ctx = ToolContext::default(); + let out = greet.call(json!({}), &ctx).await.expect("greet call"); + println!("greet -> {}", out.text_content()); + assert!(!out.is_error); + assert_eq!(out.text_content(), "hello, world!"); + println!("OK"); +} diff --git a/src/engine/bare/tests.rs b/src/engine/bare/tests.rs index b39bfee..56c8f25 100644 --- a/src/engine/bare/tests.rs +++ b/src/engine/bare/tests.rs @@ -4457,11 +4457,15 @@ fn test_add_contributor_panics_after_session_start() { // subsequent add_contributor must panic in debug builds (matches // set_reflector's contract). // Box the future so we can drop it without awaiting; the session-init - // side effect is the state transition under test. + // side effect is the state transition under test. The turn path uses + // `tokio::select!`/`tokio::time::sleep`, which require a tokio reactor + // context, so enter a current-thread runtime before block_on polls. { let run_config = RunConfig::default(); let fut = agent.run("seed", &run_config); let mut fut = std::pin::pin!(fut); + let rt = tokio::runtime::Runtime::new().expect("build tokio runtime"); + let _guard = rt.enter(); let outcome = futures::executor::block_on(fut.as_mut()); drop(outcome); } @@ -4548,10 +4552,14 @@ fn test_set_request_options_panics_after_session_start() { let mut agent = BareLoop::new(Arc::new(client), ToolRegistry::new(), config); // The first run() establishes the session and moves the loop out of // Idle; a subsequent set_request_options must panic in debug builds. + // The turn path uses `tokio::select!`/`tokio::time::sleep`, which require a + // tokio reactor context, so enter a current-thread runtime before block_on. { let run_config = RunConfig::default(); let fut = agent.run("seed", &run_config); let mut fut = std::pin::pin!(fut); + let rt = tokio::runtime::Runtime::new().expect("build tokio runtime"); + let _guard = rt.enter(); let outcome = futures::executor::block_on(fut.as_mut()); drop(outcome); } diff --git a/src/lib.rs b/src/lib.rs index 85887b4..5bd0a52 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -22,6 +22,7 @@ //! - **[`middleware`]** — Tool dispatch middleware pipeline (timeouts, permissions, output limits). //! - **[`tool`]** — Tool trait, registry, and supporting types. //! - **`tool::health`** — Per-tool health monitoring, circuit breakers, and self-healing routing. *Requires `tool_health` feature.* +//! - **[`mcp`]** — MCP client adapter ([`McpToolProvider`](mcp::McpToolProvider)) — adapt any MCP server's tools as `Tool` impls. *Requires `mcp` feature.* //! //! ## API Layer //! @@ -77,6 +78,8 @@ pub mod fallback; #[cfg(feature = "hooks")] pub mod hooks; pub mod managers; +#[cfg(feature = "mcp")] +pub mod mcp; pub mod memory; pub mod message; pub mod middleware; diff --git a/src/mcp.rs b/src/mcp.rs new file mode 100644 index 0000000..66fc9d0 --- /dev/null +++ b/src/mcp.rs @@ -0,0 +1,952 @@ +//! MCP client adapter — adapt MCP servers as loopctl [`Tool`] implementations. +//! +//! [Model Context Protocol][mcp] servers expose callable *tools*. This module +//! connects one server, discovers its tools (`tools/list`), and wraps each one +//! as an ordinary loopctl [`Tool`] whose [`call`](Tool::call) forwards to the +//! server (`tools/call`). The agent loop, the registry, the middleware +//! pipeline, permission gates, and observers never learn a tool is remote — +//! they see a `Box`. +//! +//! # The adapter surface +//! +//! - [`McpClient`] — a connected, initialized client handle. The +//! transport-agnostic boundary: obtain one from [`McpClient::in_process`] or, +//! in a later release, from a transport constructor. +//! - [`McpToolProvider`] — owns a [`McpClient`] and a snapshot of the server's +//! tool list; [`McpToolProvider::connect`] discovers, +//! [`register_into`](McpToolProvider::register_into) registers the batch into +//! a [`ToolRegistry`]. +//! - [`McpTool`] — one server tool as a [`Tool`]. +//! - [`McpError`] — adapter errors. +//! +//! No rmcp type appears in any of these public signatures; an rmcp upgrade is +//! a one-file change (this one). +//! +//! [mcp]: https://modelcontextprotocol.io +//! +//! # Example +//! +//! See `examples/mcp-adapter.rs` for a runnable end-to-end demo (an in-process +//! server, discovery, registration, and a call). In short: +//! +//! ```rust,ignore +//! use loopctl::mcp::{McpClient, McpToolProvider}; +//! use loopctl::tool::ToolRegistry; +//! +//! # async fn run(server: impl rmcp::handler::server::ServerHandler) { +//! let client = McpClient::in_process(server).await?; +//! let provider = McpToolProvider::connect(client, None).await?; +//! let mut registry = ToolRegistry::new(); +//! provider.register_into(&mut registry); +//! # Ok::<(), loopctl::mcp::McpError>(()) +//! # } +//! ``` + +use std::future::Future; +use std::pin::Pin; +use std::sync::Arc; + +use rmcp::ServiceExt; +use rmcp::handler::server::ServerHandler; +use rmcp::model::ContentBlock; +use rmcp::service::RoleClient; +use rmcp::service::RunningService; + +use crate::message::ToolContent as MessageToolContent; +use crate::message::ToolContentPart; +use crate::tool::{Tool, ToolContext, ToolError, ToolOutput, ToolRegistry, ToolSchema}; + +/// Buffer size for the in-process duplex channel ([`McpClient::in_process`]). +const DUPLEX_BUFFER: usize = 4096; + +/// A live, initialized connection to an MCP server. +/// +/// This is the transport-agnostic boundary for the adapter. [`Self::in_process`] +/// is the only constructor shipped here: it connects a client to a server over +/// an in-memory channel. Transport constructors for real servers (a stdio +/// child process, HTTP/SSE) arrive in a later release and will build the same +/// handle via rmcp's transport APIs. +/// +/// The rmcp running client service is held behind an [`Arc`]: rmcp's +/// [`RunningService`] is not itself [`Clone`] (it owns a background task and a +/// cancellation guard), so the cheap sharing the provider↔tool split needs goes +/// through the [`Arc`]. Tool calls reach the server via [`RunningService`]'s +/// [`Deref`](std::ops::Deref) to rmcp's `Peer`, which holds the channel sender. +/// Each [`McpTool`] clones the [`Arc`] rather than borrowing from the provider, +/// keeping `Tool::call(&self) -> Pin>` free of lifetime +/// entanglement with the provider's lifetime. +/// +/// Dropping the last clone drops the [`RunningService`], whose cancellation +/// guard cancels the background task — there is no leaked runtime work. +#[derive(Clone)] +pub struct McpClient { + /// The running client service. The handler is fixed to `()`, the pure + /// client: a server's `sampling`/`roots` requests get default empty + /// answers. A host that wants to honour those constructs its own client + /// with a richer handler (out of scope for this module). + service: Arc>, +} + +impl McpClient { + /// Connect a client to `server` over an in-memory channel and run the MCP + /// `initialize` handshake to completion. + /// + /// Opens one [`tokio::io::duplex`], drives `server.serve(server_end)` on a + /// spawned background task, and `().serve(client_end)` (a pure client) on + /// this future, awaiting the client's `initialize` round-trip. rmcp splits + /// each combined read+write duplex end internally, so one duplex cross-wires + /// the two sides — no manual split, no second duplex. + /// + /// Intended for the test suite and for callers who bundle an rmcp server + /// in-process. It is **not** the path for real-world servers — use a + /// transport constructor for those (a later release). + /// + /// # Runtime requirement + /// + /// Spawns the server's `serve` future via [`tokio::spawn`], so this must be + /// called from within a running multi-threaded or current-thread tokio + /// runtime (it panics with "no reactor running" otherwise). The spawned + /// server task is **detached**: it runs until the returned [`McpClient`] (and + /// all its clones) are dropped, at which point the client side of the duplex + /// closes and the server's `serve` future sees EOF and ends. If the + /// [`McpClient`] is leaked, the server task runs indefinitely — keep the + /// handle's lifetime bounded. + /// + /// # Errors + /// + /// [`McpError::Handshake`] if the client's `serve`/`initialize` fails. + pub async fn in_process(server: S) -> Result + where + S: ServerHandler, + { + let (server_end, client_end) = tokio::io::duplex(DUPLEX_BUFFER); + tokio::spawn(async move { + if let Ok(running) = server.serve(server_end).await { + let _ = running.waiting().await.ok(); + } + }); + let client = ().serve(client_end).await.map_err(|e| McpError::Handshake(e.to_string()))?; + Ok(Self { + service: Arc::new(client), + }) + } + + /// Wrap an already-running rmcp client service as an [`McpClient`]. + /// + /// For the common case use [`Self::in_process`], which handles the duplex + /// and handshake. This constructor is for callers (and tests) that drive + /// `().serve(transport)` themselves — e.g. to attach a custom client + /// handler, or to share a transport set up out-of-band. + #[must_use] + pub fn from_service(service: RunningService) -> Self { + Self { + service: Arc::new(service), + } + } + + /// Bridge a `tools/call` round-trip into a loopctl [`ToolOutput`]. + /// + /// Builds the rmcp request with the given tool `name` and, when `input` is a + /// JSON object, attaches its fields as the call's `arguments`. rmcp's + /// high-level `call_tool` drives SEP-2322 `input_required` rounds up to its + /// built-in cap (10) using the local client handler — which is the pure + /// `()` here, so it cannot actually answer elicitation; a server that never + /// completes surfaces as `ServiceError::InputRequiredRoundsExceeded`. A + /// protocol-level failure (RPC error, or that rounds-exceeded condition) + /// becomes [`ToolError::Execution`]; a server-reported tool error (`isError`) + /// is surfaced as a *soft* [`ToolOutput`] with `is_error` set (see + /// [`bridge_result`]). + /// + /// # Errors + /// + /// [`ToolError::Execution`] if the rmcp `call_tool` RPC fails (transport, + /// protocol `ErrorData`, or `input_required` rounds exceeded), or the result + /// bridges to [`McpError`] (an empty tool error). + async fn call_tool_forward( + &self, + server_name: &str, + input: serde_json::Value, + ) -> Result { + let mut params = rmcp::model::CallToolRequestParams::new(server_name.to_string()); + if let serde_json::Value::Object(map) = input { + params = params.with_arguments(map); + } + let result = self + .service + .call_tool(params) + .await + .map_err(|e| ToolError::Execution(format!("MCP tools/call failed: {e}")))?; + bridge_result(server_name, result).map_err(|e| ToolError::Execution(e.to_string())) + } +} + +/// Adapts one MCP server's tools as loopctl [`Tool`] implementations. +/// +/// A provider owns a connected [`McpClient`] and a snapshot of the server's +/// tool list, producing one [`McpTool`] per tool discovered via MCP's +/// `tools/list`. Each adapted tool forwards `tools/call` to the server and +/// bridges the result back into a loopctl [`ToolOutput`], so the agent loop, +/// registry, middleware pipeline, and observers see ordinary `Box` +/// values — they never learn a tool is remote. +/// +/// # Construction +/// +/// Build with [`McpToolProvider::connect`], which runs the MCP `initialize` +/// handshake (via the supplied [`McpClient`]) followed by `tools/list`, then +/// snapshots the result. From there either: +/// - call [`McpToolProvider::tools`] to take the adapted [`McpTool`] instances +/// and register them yourself, or +/// - call [`McpToolProvider::register_into`] to clone-and-register the whole +/// batch into a [`ToolRegistry`] in one shot. +/// +/// # Tool-name collisions +/// +/// MCP tools are named only within their server. If two providers each expose a +/// tool named `search`, registering both into one [`ToolRegistry`] would +/// collide — the second silently overwrites the first (the registry's own +/// behaviour). Pass a `name_prefix` to [`McpToolProvider::connect`] to +/// namespace every tool from this server: `name_prefix = Some("git".into())` +/// yields `git__status`, `git__log`, and so on. The un-prefixed name is still +/// what the provider sends to the server in each `tools/call` request, so +/// namespacing is purely a client-side registry concern. +/// +/// # Static vs. dynamic discovery +/// +/// [`McpToolProvider::connect`] takes a **static snapshot** of the tool list at +/// handshake time. A server may later emit `notifications/tools/list_changed`; +/// this adapter does not auto-refresh (auto-refresh would require a background +/// task per provider and a thread-safe mutable registry, neither of which the +/// current registry model supports). Call [`McpToolProvider::refresh`] to re-run +/// `tools/list` and rebuild the snapshot on demand. Tools already registered +/// into a [`ToolRegistry`] under stale names are **not** updated by a refresh — +/// the caller decides whether to re-register. +/// +/// # Thread safety +/// +/// `McpToolProvider` is `Send + Sync` when the underlying [`McpClient`] is (it +/// is — the rmcp handle is `Arc`-backed). The snapshot is immutable between +/// [`McpToolProvider::connect`] / [`McpToolProvider::refresh`] calls; the only +/// interior mutation is [`McpToolProvider::refresh`], which takes `&mut self`, +/// so concurrent reads of [`McpToolProvider::tools`] during a run are safe. +pub struct McpToolProvider { + /// The connected client the adapted tools forward through. + /// + /// Held by reference (cheaply cloneable — see [`McpClient`]); each + /// [`McpTool`] in `tools` carries its own clone of this handle so calls are + /// free of lifetime entanglement with the provider. + client: McpClient, + + /// The discovered tools, frozen at the last snapshot. + /// + /// Populated by [`Self::connect`] and replaced wholesale by + /// [`Self::refresh`]; never mutated in place between those calls. The slice + /// returned by [`Self::tools`] borrows this field. + tools: Vec, + + /// The prefix applied to every adapted tool name, or `None` for unprefixed. + /// + /// Captured at [`Self::connect`] time and re-applied by [`Self::refresh`] + /// so a refresh preserves the original namespacing without the caller + /// having to pass the prefix again. + prefix: Option, +} + +impl McpToolProvider { + /// Connect to a server and snapshot its tool list. + /// + /// The primary constructor. Runs the MCP `initialize` handshake (driven by + /// the supplied [`McpClient`]) followed by `tools/list`, which rmcp + /// auto-paginates by following `nextCursor` to exhaustion. The returned + /// provider holds one [`McpTool`] per server-declared tool, each sharing a + /// cheap clone of the client handle. + /// + /// `name_prefix`, when `Some`, namespaces every adapted tool: it is + /// prepended to each tool's name as `"{prefix}__{tool_name}"` for both + /// [`Tool::name`] and the [`ToolSchema::tool`] field sent to the LLM. The + /// original un-prefixed name is what the provider forwards to the server in + /// each `tools/call` request, so namespacing is purely a client-side + /// registry concern and never confuses the server. The prefix is retained + /// and re-applied by any later [`Self::refresh`]. + /// + /// # Errors + /// + /// [`McpError::Protocol`] if the `tools/list` RPC fails (transport, + /// protocol `ErrorData`, or pagination error). Handshake failures surface + /// earlier, from the construction of the supplied [`McpClient`]. + pub async fn connect(client: McpClient, name_prefix: Option) -> Result { + let mut tools = Vec::new(); + bridge_tool_list(&client, name_prefix.as_deref(), &mut tools).await?; + Ok(Self { + client, + tools, + prefix: name_prefix, + }) + } + + /// Re-run `tools/list` and rebuild the tool snapshot in place. + /// + /// Replaces `self.tools` wholesale with a fresh discovery, re-applying the + /// `name_prefix` captured at [`Self::connect`] time so a refresh preserves + /// the original namespacing without the caller re-passing it. Intended for + /// picking up newly-added tools after a server signals + /// `notifications/tools/list_changed`. + /// + /// Tools already handed to a [`ToolRegistry`] (via [`Self::register_into`]) + /// under stale names are **not** updated by this call — the registry still + /// holds the old [`McpTool`] clones. The caller decides whether to + /// re-register, and how to handle names that vanished from the new snapshot. + /// + /// # Errors + /// + /// [`McpError::Protocol`] if the re-list RPC fails. On error the prior + /// snapshot is left untouched (the new list is built in a local `Vec` and + /// only assigned on success). + pub async fn refresh(&mut self) -> Result<(), McpError> { + let mut tools = Vec::new(); + bridge_tool_list(&self.client, self.prefix.as_deref(), &mut tools).await?; + self.tools = tools; + Ok(()) + } + + /// The adapted tools from the current snapshot. + /// + /// Returns a borrow of the [`McpTool`] instances produced by the last + /// [`Self::connect`] or [`Self::refresh`]. The slice is immutable between + /// those calls; iterate it to register tools selectively, or use + /// [`Self::register_into`] to register the whole batch. + /// + /// # Example + /// + /// ```rust,ignore + /// let provider = McpToolProvider::connect(client, None).await?; + /// for tool in provider.tools() { + /// println!("{}: {}", tool.name(), tool.description()); + /// } + /// ``` + #[must_use] + pub fn tools(&self) -> &[McpTool] { + &self.tools + } + + /// Clone-and-register every tool from the snapshot into `registry`. + /// + /// Convenience for the common "register everything" path: each [`McpTool`] + /// is [`Clone`] (the underlying client handle is cheaply cloneable), so + /// this hands the registry owned copies while the provider keeps its + /// snapshot. Intra-batch duplicate names — two server tools that collide + /// after prefixing — were already collapsed to one at [`Self::connect`] + /// time, so the registry sees unique names; an overlap with a tool already + /// in the registry follows [`ToolRegistry::register`]'s own overwrite-with + /// -`warn` behaviour. + /// + /// Use [`Self::tools`] instead when you need finer control over which tools + /// to register. + pub fn register_into(&self, registry: &mut ToolRegistry) { + for tool in &self.tools { + registry.register(tool.clone()); + } + } + + /// Borrow the underlying client. + /// + /// Exposed for advanced callers — e.g. to issue raw rmcp requests the + /// adapter does not wrap, or to feed the same live connection into other + /// machinery. Future transport constructors will also build on this seam. + #[must_use] + pub fn client(&self) -> &McpClient { + &self.client + } +} + +/// A single MCP server tool exposed as a loopctl [`Tool`]. +/// +/// One `McpTool` adapts a single server-declared tool — captured at discovery +/// time — as an ordinary loopctl [`Tool`] that forwards `tools/call` to the +/// server and bridges the result back into [`ToolOutput`]. Produced by +/// [`McpToolProvider::connect`] (and rebuilt by [`McpToolProvider::refresh`]); +/// not constructed directly by callers. +/// +/// The adapter stores a snapshot of the server-side `name`, `description`, +/// `inputSchema` (and optional `outputSchema`) plus a cheap clone of the shared +/// [`McpClient`] handle, so a call has everything it needs without borrowing +/// from the provider. See [`Tool::call`] for the round-trip details. +/// +/// # Concurrency +/// +/// [`Clone`] is cheap: it clones the [`Arc`]-backed client handle and the small +/// schema snapshot, nothing more. Each clone drives the same underlying +/// connection, so the adapter conservatively reports +/// [`Tool::is_concurrency_safe`] as `false` — see that method for why. +/// +/// # Annotations +/// +/// The server's `annotations` block is distilled into two booleans carried on +/// the struct and surfaced via [`Tool::is_read_only`] and +/// [`McpTool::is_destructive_hint`]. Both apply the MCP-spec defaults for an +/// absent hint (read-only defaults to `false`; destructive defaults to `true`), +/// so a consumer can read them directly without re-applying the spec. +#[derive(Clone)] +pub struct McpTool { + /// The original server-side name, sent verbatim in each `tools/call`. + /// + /// Distinct from `exposed_name` when the provider was constructed with a + /// `name_prefix`: the prefix is a client-side registry concern only, so the + /// server always sees the name it declared. + server_name: String, + + /// The loopctl-facing name (prefixed if the provider was given a prefix). + /// + /// Returned by [`Tool::name`] and used as [`ToolSchema::tool`] for the LLM. + /// Equals `server_name` when no prefix was supplied. + exposed_name: String, + + /// Human-readable description copied from the server tool at discovery. + /// + /// Empty string when the server omitted a description. Returned verbatim by + /// [`Tool::description`] and embedded in [`ToolSchema::description`]. + description: String, + + /// The server's `inputSchema`, carried verbatim as a JSON value. + /// + /// MCP permits any JSON-Schema draft; the adapter does not normalize it + /// (loopctl forwards the schema to the LLM and never validates against it). + /// Embedded unchanged in [`ToolSchema::input_schema`]. + input_schema: serde_json::Value, + + /// The server's `outputSchema`, if it declared one. + /// + /// Carried for forward-compatibility only — the adapter does not validate + /// call results against it. Exposed via [`McpTool::output_schema`]. + output_schema: Option, + + /// A cheap clone of the shared client handle. + /// + /// Cloned per-tool at discovery so each [`Tool::call`] owns its connection + /// without borrowing from the provider (the future is `'_`-bounded but + /// self-contained). + client: McpClient, + + /// Whether the server annotated the tool as read-only. + /// + /// Mirrors `annotations.readOnlyHint` when present, else `false` (the + /// spec default). Drives [`Tool::is_read_only`]. + read_only_hint: bool, + + /// Whether the tool should be treated as destructive. + /// + /// Mirrors `annotations.destructiveHint` when present, else `true` — the + /// spec default for an absent hint is "assume destructive" (the opposite + /// polarity from `read_only_hint`). Exposed via + /// [`McpTool::is_destructive_hint`] for a future permission gate. + destructive_hint: bool, +} + +impl McpTool { + /// The server's `outputSchema`, if it declared one. + /// + /// Carried verbatim for forward-compatibility; the adapter does not validate + /// call results against it. A future release may enforce it. + #[must_use] + pub fn output_schema(&self) -> Option<&serde_json::Value> { + self.output_schema.as_ref() + } + + /// Whether the tool should be treated as destructive. + /// + /// Mirrors `annotations.destructiveHint` when the server set it, and applies + /// the MCP-spec default for an absent hint: **absent means destructive** + /// (`true`). This is the opposite polarity from + /// [`is_read_only`](McpTool::is_read_only), whose absent hint defaults to + /// non-destructive (`false`). A permission gate can read this directly + /// without re-applying the spec default. + #[must_use] + pub fn is_destructive_hint(&self) -> bool { + self.destructive_hint + } +} + +impl Tool for McpTool { + /// The loopctl-facing (possibly prefixed) tool name. + /// + /// Returns `exposed_name`, which equals the server-declared name when the + /// provider was constructed without a `name_prefix`, or + /// `"{prefix}__{name}"` otherwise. The LLM and the [`ToolRegistry`] see + /// this value; the server sees the un-prefixed `server_name` (see + /// [`Tool::call`]). + fn name(&self) -> &str { + &self.exposed_name + } + + /// The server-supplied description (empty string if the server omitted it). + /// + /// Copied verbatim from the server tool at discovery time and embedded in + /// [`ToolSchema::description`] for the LLM. + fn description(&self) -> &str { + &self.description + } + + /// Build the [`ToolSchema`] sent to the LLM for this tool. + /// + /// Carries the (possibly prefixed) name, the description, and the server's + /// `inputSchema` verbatim — no normalization, no draft rewriting. Built + /// fresh on each call so a caller can mutate the snapshot without affecting + /// previously-returned schemas. + fn schema(&self) -> ToolSchema { + ToolSchema { + tool: self.exposed_name.clone(), + description: self.description.clone(), + input_schema: self.input_schema.clone(), + } + } + + /// Forward a `tools/call` round-trip to the server and bridge the result. + /// + /// Sends the `input` (when it is a JSON object) as the call's `arguments` + /// under the **un-prefixed** `server_name`, so the server always sees the + /// name it declared regardless of any client-side namespacing. The cheap + /// client handle is cloned before the future is constructed so the future + /// owns its connection and is bounded only by `'_` (no borrow of `self` + /// survives the `await`). The result-bridging and error-mapping rules live + /// on [`McpClient`]. + /// + /// # Errors + /// + /// [`ToolError::Execution`] for any protocol failure, transport error, or + /// server-reported empty error — mapped at the [`McpClient`] boundary. A + /// server-reported tool error (`isError: true`) with content is surfaced as + /// a *soft* [`ToolOutput`] with `is_error` set, not as an `Err`. + fn call( + &self, + input: serde_json::Value, + _ctx: &ToolContext, + ) -> Pin> + Send + '_>> { + let client = self.client.clone(); + let server_name = self.server_name.clone(); + Box::pin(async move { client.call_tool_forward(&server_name, input).await }) + } + + /// Conservatively `false` for every MCP tool. + /// + /// A remote server may serialize calls internally, mutate shared state per + /// call, or rate-limit, and loopctl cannot tell. The parallel dispatcher + /// therefore never overlaps calls into the same server unless a caller that + /// trusts a specific server overrides this. The MCP `annotations` block + /// carries no concurrency hint in the current spec, so there is nothing + /// finer to honour today. + fn is_concurrency_safe(&self) -> bool { + false + } + + /// Honours the server's `annotations.readOnlyHint`, defaulting to `false`. + /// + /// Returns `true` only when the server explicitly annotated the tool as + /// read-only; an absent hint follows the [`Tool`] trait default of `false` + /// (matching the MCP spec). A permission gate may auto-approve tools for + /// which this returns `true`. + fn is_read_only(&self) -> bool { + self.read_only_hint + } +} + +/// Errors from the MCP client adapter. +/// +/// All variants carry the underlying rmcp failure as a string, so the type is +/// cheap, `Send + 'static`, and stable across rmcp version bumps (rmcp's own +/// error types are not uniformly `Send + 'static`). +/// +/// # Example +/// +/// ``` +/// use loopctl::mcp::McpError; +/// +/// let err = McpError::Protocol("unknown tool 'search'".into()); +/// assert!(err.to_string().contains("unknown tool")); +/// assert!(format!("{}", McpError::Handshake("connect refused".into())) +/// .contains("handshake/transport error")); +/// ``` +#[derive(Debug, thiserror::Error)] +pub enum McpError { + /// The `initialize` handshake or underlying transport failed before any + /// tools could be listed. Carries the rmcp service/transport error as a + /// string (the rmcp error types are not uniformly `Send + 'static`; this + /// keeps [`McpError`] cheap and stable across rmcp version bumps). + #[error("MCP handshake/transport error: {0}")] + Handshake(String), + + /// A JSON-RPC-level protocol error from the server, e.g. `tools/list` + /// rejected or `tools/call` for an unknown tool. + #[error("MCP protocol error: {0}")] + Protocol(String), + + /// The server returned a tool result with `isError = true` and no textual + /// content to surface. (When there *is* text content, the bridge returns a + /// soft-error [`ToolOutput`] instead of raising this.) + #[error("MCP tool '{0}' reported an error with no content")] + EmptyToolError(String), +} + +/// Discover the server's tools and append one [`McpTool`] per result. +/// +/// The single home for `tools/list` pagination: rmcp's `list_all_tools` follows +/// `nextCursor` to exhaustion, so if a future rmcp version drops that helper +/// only this function changes. +/// +/// # Errors +/// +/// [`McpError::Protocol`] if the `list_all_tools` RPC fails. +async fn bridge_tool_list( + client: &McpClient, + prefix: Option<&str>, + out: &mut Vec, +) -> Result<(), McpError> { + let server_tools = client + .service + .list_all_tools() + .await + .map_err(|e| McpError::Protocol(e.to_string()))?; + let mut seen = std::collections::HashSet::new(); + for server_tool in server_tools { + let Some(adapted) = bridge_tool(&server_tool, prefix, client) else { + continue; + }; + if !seen.insert(adapted.exposed_name.clone()) { + tracing::warn!( + tool = %adapted.exposed_name, + "duplicate MCP tool name after prefixing; keeping the first" + ); + continue; + } + out.push(adapted); + } + Ok(()) +} + +/// Build one [`McpTool`] from a server-declared tool. +/// +/// The per-tool half of discovery (the list-driving half is [`bridge_tool_list`]). +/// Copies the server's `name`, `description`, `inputSchema`, and optional +/// `outputSchema` into a fresh [`McpTool`], hands the tool a cheap clone of the +/// shared `client` handle, and distils the server's `annotations` block into the +/// two booleans the adapter carries. +/// +/// # Naming +/// +/// `prefix`, when `Some`, yields an `exposed_name` of `"{prefix}__{name}"` +/// (loopctl-facing, sent to the LLM and used as the registry key) while +/// `server_name` keeps the original value the server declared — the latter is +/// what `tools/call` forwards, so namespacing is purely a client-side concern. +/// +/// # Schema fidelity +/// +/// `inputSchema` and `outputSchema` are coerced from rmcp's `Arc` +/// into `serde_json::Value::Object(..)` and carried **verbatim** — no draft +/// normalization, no field rewriting. loopctl forwards the schema to the LLM and +/// never validates against it, so any JSON-Schema draft passes through +/// losslessly. +/// +/// # Annotations +/// +/// `annotations` is read with the MCP-spec defaults for an absent hint: +/// `readOnlyHint` defaults to `false`, `destructiveHint` defaults to `true` +/// (the opposite polarity — an unannotated tool is assumed destructive). An +/// entirely absent `annotations` block yields `(false, true)`. +/// +/// # Returns +/// +/// `None` only for a malformed discovery entry whose `name` is empty — the +/// caller ([`bridge_tool_list`]) skips `None` rather than panicking, since the +/// no-panic lint forbids indexing/`unwrap` and an empty name is a server bug +/// worth dropping silently with a warning rather than aborting discovery. +fn bridge_tool( + server_tool: &rmcp::model::Tool, + prefix: Option<&str>, + client: &McpClient, +) -> Option { + let server_name = server_tool.name.to_string(); + if server_name.is_empty() { + return None; + } + let exposed_name = + prefix.map_or_else(|| server_name.clone(), |p| format!("{p}__{server_name}")); + let description = server_tool + .description + .as_deref() + .unwrap_or_default() + .to_string(); + let input_schema = serde_json::Value::Object(server_tool.input_schema.as_ref().clone()); + let output_schema = server_tool + .output_schema + .as_ref() + .map(|schema| serde_json::Value::Object(schema.as_ref().clone())); + let (read_only_hint, destructive_hint) = + server_tool + .annotations + .as_ref() + .map_or((false, true), |annotations| { + ( + annotations.read_only_hint.unwrap_or(false), + // MCP spec: an absent destructiveHint means "assume + // destructive" (default true) — the opposite of read-only. + annotations.destructive_hint.unwrap_or(true), + ) + }); + Some(McpTool { + server_name, + exposed_name, + description, + input_schema, + output_schema, + client: client.clone(), + read_only_hint, + destructive_hint, + }) +} + +/// Bridge a rmcp `CallToolResult` into a loopctl [`ToolOutput`]. +/// +/// Maps the result's content blocks into [`MessageToolContent`]: a single text +/// block becomes [`MessageToolContent::Text`]; any other shape (multiple blocks, +/// an image) becomes [`MessageToolContent::Multipart`]. `isError` becomes +/// [`ToolOutput::is_error`] — a server-reported tool error is a *soft* failure, +/// matching how native loopctl tools report recoverable errors. `structuredContent`, +/// when present, is appended as one extra JSON-stringified text part (carried, +/// not parsed). An error with no content at all becomes +/// [`McpError::EmptyToolError`] carrying `tool_name`. +/// +/// A successful result with zero content blocks yields an empty successful +/// [`ToolOutput`] (no error) — the server ran the tool and returned nothing. +/// +/// # Errors +/// +/// [`McpError::EmptyToolError`] when the server reported an error but supplied +/// no content to surface. +fn bridge_result( + tool_name: &str, + res: rmcp::model::CallToolResult, +) -> Result { + let is_error = res.is_error.unwrap_or(false); + let mut parts: Vec = res.content.iter().map(bridge_content).collect(); + if let Some(structured) = res.structured_content { + let note = serde_json::to_string(&structured) + .unwrap_or_else(|_| "".to_string()); + parts.push(ToolContentPart::text(note)); + } + if parts.is_empty() { + return if is_error { + Err(McpError::EmptyToolError(tool_name.to_string())) + } else { + Ok(ToolOutput::text(String::new())) + }; + } + let payload = if parts.len() == 1 { + match parts.pop() { + Some(ToolContentPart::Text { text }) => MessageToolContent::Text(text), + Some(single) => MessageToolContent::Multipart(vec![single]), + None => MessageToolContent::Text(String::new()), + } + } else { + MessageToolContent::Multipart(parts) + }; + let output = if is_error { + ToolOutput::error(payload) + } else { + ToolOutput::success(payload) + }; + Ok(output) +} + +/// Map one rmcp content block to a loopctl [`ToolContentPart`]. +/// +/// Text and image carry through; audio, embedded resources, resource links, and +/// any future block kind fall back to a short text note so the model learns the +/// part existed rather than seeing it silently dropped. +fn bridge_content(block: &ContentBlock) -> ToolContentPart { + match block { + ContentBlock::Text(text) => ToolContentPart::text(&text.text), + ContentBlock::Image(image) => ToolContentPart::image( + crate::message::ImageSource::new_base64(&image.mime_type, &image.data), + ), + ContentBlock::Audio(_) => ToolContentPart::text("unsupported MCP content type: audio"), + ContentBlock::Resource(_) => { + ToolContentPart::text("unsupported MCP content type: embedded resource") + } + ContentBlock::ResourceLink(_) => { + ToolContentPart::text("unsupported MCP content type: resource link") + } + _ => ToolContentPart::text("unsupported MCP content type"), + } +} + +#[cfg(test)] +mod tests { + //! Unit tests for the pure bridge functions (`bridge_result`, + //! `bridge_content`). The connection-dependent paths (`bridge_tool`, + //! `bridge_tool_list`, `McpClient`, `McpToolProvider`) are covered by the + //! integration suite in `tests/mcp_tool_provider.rs`. + + use super::*; + use rmcp::model::{CallToolResult, ContentBlock}; + + /// A single text block collapses to plain [`MessageToolContent::Text`]. + #[test] + fn bridge_result_single_text_becomes_text_payload() { + let res = CallToolResult::success(vec![ContentBlock::text("hi")]); + let out = bridge_result("t", res).expect("success bridges"); + assert!(!out.is_error); + assert!(matches!(out.payload, MessageToolContent::Text(_))); + assert_eq!(out.text_content(), "hi"); + } + + /// A single image block can't be a `Text`, so it becomes a one-element + /// `Multipart` — an edge the integration suite (which only builds macro + /// servers returning text) does not hit. + #[test] + fn bridge_result_single_image_becomes_single_part_multipart() { + let res = CallToolResult::success(vec![ContentBlock::image("Zm9v", "image/png")]); + let out = bridge_result("t", res).expect("success bridges"); + assert!(!out.is_error); + match out.payload { + MessageToolContent::Multipart(parts) => { + assert_eq!(parts.len(), 1, "single image → one-element multipart"); + assert!(matches!(parts.first(), Some(ToolContentPart::Image { .. }))); + } + other @ MessageToolContent::Text(_) => { + panic!("expected Multipart, got {other:?}") + } + } + } + + /// Multiple blocks become a `Multipart` preserving order. + #[test] + fn bridge_result_multiple_blocks_become_multipart_in_order() { + let res = CallToolResult::success(vec![ + ContentBlock::text("a"), + ContentBlock::image("Zg==", "image/jpeg"), + ContentBlock::text("b"), + ]); + let out = bridge_result("t", res).expect("success bridges"); + let MessageToolContent::Multipart(parts) = out.payload else { + panic!("expected Multipart"); + }; + assert_eq!(parts.len(), 3); + assert!(matches!(parts.first(), Some(ToolContentPart::Text { text }) if text == "a")); + assert!(matches!(parts.get(1), Some(ToolContentPart::Image { .. }))); + assert!(matches!(parts.get(2), Some(ToolContentPart::Text { text }) if text == "b")); + } + + /// A server-reported error (`isError: true`) with text content is a *soft* + /// failure: `Ok` with `is_error` set, not an `Err`. + #[test] + fn bridge_result_soft_error_returns_ok_with_is_error() { + let res = CallToolResult::error(vec![ContentBlock::text("boom")]); + let out = bridge_result("t", res).expect("soft error is Ok"); + assert!(out.is_error); + assert_eq!(out.text_content(), "boom"); + } + + /// An error with no content surfaces as the hard [`McpError::EmptyToolError`], + /// carrying the tool name — pins the round-2 fix that threads `tool_name`. + #[test] + fn bridge_result_empty_error_is_hard_empty_tool_error_with_name() { + let res = CallToolResult::error(vec![]); + let err = bridge_result("search", res).expect_err("empty error is hard Err"); + match err { + McpError::EmptyToolError(name) => assert_eq!(name, "search"), + other => panic!("expected EmptyToolError, got {other:?}"), + } + } + + /// A successful result with zero content blocks yields an empty successful + /// output — distinct from the empty-error path above. + #[test] + fn bridge_result_empty_success_yields_empty_text_output() { + let res = CallToolResult::success(vec![]); + let out = bridge_result("t", res).expect("empty success is Ok"); + assert!(!out.is_error); + assert_eq!(out.text_content(), ""); + } + + /// `structuredContent`, when present, is appended as one extra JSON-string + /// text part — carried, not parsed. Even a single text block + structured + /// content therefore becomes a two-part `Multipart`. + #[test] + fn bridge_result_structured_content_appended_as_text_part() { + let mut res = CallToolResult::success(vec![ContentBlock::text("body")]); + res.structured_content = Some(serde_json::json!({"count": 7})); + let out = bridge_result("t", res).expect("success bridges"); + let MessageToolContent::Multipart(parts) = out.payload else { + panic!("text + structured must be multipart"); + }; + assert_eq!(parts.len(), 2); + // The structured part is JSON-stringified text appended after the body. + let structured_text = &parts + .last() + .and_then(|p| match p { + ToolContentPart::Text { text } => Some(text.as_str()), + ToolContentPart::Image { .. } => None, + }) + .expect("structured part is text"); + assert!( + structured_text.contains("count"), + "carries the structured json" + ); + assert!(structured_text.contains('7')); + } + + /// An error result (`isError: true`) with empty content but a declared + /// `structuredContent` is *not* an `EmptyToolError` — the structured payload + /// surfaces as a soft error's text, so the model sees what the server + /// returned. Pins this edge (the integration suite never builds it). + #[test] + fn bridge_result_error_with_structured_content_is_soft_error() { + let mut res = CallToolResult::error(vec![]); + res.structured_content = Some(serde_json::json!({"reason": "denied"})); + let out = bridge_result("search", res).expect("structured error is soft Ok"); + assert!(out.is_error, "is_error flag set"); + let text = out.text_content(); + assert!( + text.contains("denied"), + "structured payload surfaces as the error text: {text}" + ); + } + + /// `bridge_content`: every unsupported block kind surfaces as a text note + /// naming the kind, never silently dropped. Covers the arms the macro-server + /// integration tests can't easily reach (resource, resource-link). + #[test] + fn bridge_content_unsupported_kinds_surface_as_text_notes() { + let audio = bridge_content(&ContentBlock::audio("AAAA", "audio/wav")); + assert!( + matches!(audio, ToolContentPart::Text { ref text } if text.contains("audio")), + "audio → text note, got {audio:?}" + ); + + let resource = bridge_content(&ContentBlock::Resource(rmcp::model::EmbeddedResource::new( + rmcp::model::ResourceContents::text("body", "mem://x"), + ))); + assert!( + matches!(resource, ToolContentPart::Text { ref text } if text.contains("resource")), + "embedded resource → text note, got {resource:?}" + ); + } + + /// `bridge_content`: text and image carry their payloads through verbatim. + #[test] + fn bridge_content_text_and_image_carry_through() { + let text = bridge_content(&ContentBlock::text("hello")); + assert!( + matches!(&text, ToolContentPart::Text { text } if text == "hello"), + "got {text:?}" + ); + let image = bridge_content(&ContentBlock::image("Zm9v", "image/png")); + match image { + ToolContentPart::Image { source } => { + assert_eq!(source.media_type, "image/png"); + assert_eq!(source.data, "Zm9v"); + } + other @ ToolContentPart::Text { .. } => { + panic!("expected Image, got {other:?}") + } + } + } +} diff --git a/tests/mcp_tool_provider.rs b/tests/mcp_tool_provider.rs new file mode 100644 index 0000000..16b3987 --- /dev/null +++ b/tests/mcp_tool_provider.rs @@ -0,0 +1,710 @@ +//! Integration tests for the MCP client adapter. +//! +//! Every test spins up a tiny in-process rmcp server (built with +//! `#[tool_router]` / `#[tool]`) and connects an `McpToolProvider` to it over a +//! `tokio::io::duplex`. No subprocess, no network. + +#![cfg(feature = "mcp")] +// `ToolRouter` fields are read by rmcp's `#[tool_handler]`-generated dispatch +// methods; the dead-code analysis can't see macro-generated reads. +#![allow(dead_code)] +// Integration tests are a separate crate and do not inherit `lib.rs`'s +// `cfg_attr(test, allow(...))`. Apply the same test-code relaxations the lib +// uses: assertions legitimately `unwrap`/`expect`/`panic`/index for clarity. +#![allow( + clippy::unwrap_used, + clippy::expect_used, + clippy::panic, + clippy::indexing_slicing, + clippy::missing_panics_doc, + clippy::missing_errors_doc +)] + +use std::time::Duration; + +use loopctl::mcp::{McpClient, McpToolProvider}; +use loopctl::message::ToolContent as MessageToolContent; +use loopctl::message::ToolContentPart; +use loopctl::tool::{Tool, ToolContext, ToolRegistry}; +use rmcp::handler::server::router::tool::ToolRouter; +use rmcp::model::CallToolResult; +use rmcp::model::ContentBlock; +use rmcp::model::Tool as RmcpTool; +use rmcp::model::ToolAnnotations; +use rmcp::{ErrorData as McpErrorData, ServerHandler, ServiceExt, tool, tool_handler, tool_router}; +use serde_json::Value; +use serde_json::json; + +/// Buffer size matching the adapter's own duplex. +const DUPLEX_BUFFER: usize = 4096; + +/// Connect a pure client to `server` over an in-memory duplex, returning the +/// client and the server's background join handle (so tests can assert on +/// shutdown). +async fn connect_in_process(server: S) -> (McpClient, tokio::task::JoinHandle<()>) +where + S: ServerHandler + Clone + Send + 'static, +{ + let (server_end, client_end) = tokio::io::duplex(DUPLEX_BUFFER); + let server_handle = tokio::spawn(async move { + let running = server.serve(server_end).await.expect("server serve"); + let _ = running.waiting().await.ok(); + }); + let client = ().serve(client_end).await.map(McpClient::from_service).expect("client serve"); + (client, server_handle) +} + +/// Echo server: an `echo` tool and a `status` tool, both argument-free so the +/// test servers need no `schemars::JsonSchema` derive (rmcp's `#[tool]` macro +/// would otherwise require it for typed `Parameters`). +#[derive(Clone)] +struct EchoServer { + router: ToolRouter, +} + +#[tool_router] +impl EchoServer { + fn new() -> Self { + Self { + router: Self::tool_router(), + } + } + + #[tool(description = "Echo a fixed message back")] + async fn echo(&self) -> String { + "hi".to_string() + } + + #[tool(description = "Report server status")] + async fn status(&self) -> String { + "ok".to_string() + } +} + +#[tool_handler] +impl ServerHandler for EchoServer {} + +#[tokio::test] +async fn discovery_lists_server_tools() { + let (client, _server) = connect_in_process(EchoServer::new()).await; + let provider = McpToolProvider::connect(client, None) + .await + .expect("connect"); + let tools = provider.tools(); + assert_eq!(tools.len(), 2, "two server tools discovered"); + let names: Vec<&str> = tools.iter().map(Tool::name).collect(); + assert!(names.contains(&"echo"), "echo present: {names:?}"); + assert!(names.contains(&"status"), "status present: {names:?}"); +} + +#[tokio::test] +async fn echo_round_trip_returns_text() { + let (client, _server) = connect_in_process(EchoServer::new()).await; + let provider = McpToolProvider::connect(client, None) + .await + .expect("connect"); + let echo = provider + .tools() + .iter() + .find(|t| t.name() == "echo") + .expect("echo tool"); + let ctx = ToolContext::default(); + let out = echo.call(json!({}), &ctx).await.expect("call ok"); + assert!(!out.is_error, "not a soft error"); + assert_eq!(out.text_content(), "hi"); + assert!(matches!(out.payload, MessageToolContent::Text(_))); +} + +#[tokio::test] +async fn name_prefix_namespaces_exposed_name_only() { + let (client, _server) = connect_in_process(EchoServer::new()).await; + let provider = McpToolProvider::connect(client, Some("git".into())) + .await + .expect("connect"); + let status = provider + .tools() + .iter() + .find(|t| t.name() == "git__status") + .expect("prefixed status tool"); + assert_eq!(status.schema().tool, "git__status"); + // The forwarded call uses the un-prefixed server name; the status tool + // answers "ok", proving the call reached the right server-side tool. + let ctx = ToolContext::default(); + let out = status.call(json!({}), &ctx).await.expect("call ok"); + assert_eq!(out.text_content(), "ok"); +} + +#[tokio::test] +async fn register_into_populates_registry() { + let (client, _server) = connect_in_process(EchoServer::new()).await; + let provider = McpToolProvider::connect(client, None) + .await + .expect("connect"); + let mut registry = ToolRegistry::new(); + provider.register_into(&mut registry); + assert_eq!(registry.len(), 2); + assert!(registry.contains("echo")); + assert!(registry.contains("status")); + assert_eq!(registry.all_schemas().len(), 2); +} + +#[tokio::test] +async fn is_read_only_defaults_false_and_never_concurrency_safe() { + let (client, _server) = connect_in_process(EchoServer::new()).await; + let provider = McpToolProvider::connect(client, None) + .await + .expect("connect"); + for tool in provider.tools() { + assert!( + !tool.is_read_only(), + "unannotated tool {} should not be read-only", + tool.name() + ); + assert!( + !tool.is_concurrency_safe(), + "MCP tool {} should never claim concurrency-safe", + tool.name() + ); + } +} + +/// A server whose tool returns a soft error (`isError = true`) with text. +#[derive(Clone)] +struct SoftErrorServer { + router: ToolRouter, +} + +#[tool_router] +impl SoftErrorServer { + fn new() -> Self { + Self { + router: Self::tool_router(), + } + } + + #[tool(description = "Always reports a soft error")] + async fn fail(&self) -> Result { + Ok(CallToolResult::error(vec![ContentBlock::text("boom")])) + } +} + +#[tool_handler] +impl ServerHandler for SoftErrorServer {} + +#[tokio::test] +async fn soft_error_returns_ok_with_is_error_set() { + let (client, _server) = connect_in_process(SoftErrorServer::new()).await; + let provider = McpToolProvider::connect(client, None) + .await + .expect("connect"); + let fail = provider + .tools() + .iter() + .find(|t| t.name() == "fail") + .expect("fail tool"); + let ctx = ToolContext::default(); + let out = fail.call(json!({}), &ctx).await.expect("soft error is Ok"); + assert!(out.is_error, "is_error flag set"); + assert_eq!(out.text_content(), "boom"); +} + +/// A server whose tool returns an empty error (`isError = true`, no content). +#[derive(Clone)] +struct EmptyErrorServer { + router: ToolRouter, +} + +#[tool_router] +impl EmptyErrorServer { + fn new() -> Self { + Self { + router: Self::tool_router(), + } + } + + #[tool(description = "Reports an error with no content")] + async fn fail(&self) -> Result { + Ok(CallToolResult::error(vec![])) + } +} + +#[tool_handler] +impl ServerHandler for EmptyErrorServer {} + +/// A server whose tool rejects the call at the JSON-RPC level, returning +/// `Err(ErrorData)` from `call_tool` — exercises the protocol-error bridge +/// (`ServiceError` → `ToolError::Execution`), distinct from the soft `isError` +/// path and the empty-error path. +#[derive(Clone)] +struct ProtocolErrorServer { + router: ToolRouter, +} + +#[tool_router] +impl ProtocolErrorServer { + fn new() -> Self { + Self { + router: Self::tool_router(), + } + } + + #[tool(description = "Rejects the call with a JSON-RPC error")] + async fn reject(&self) -> Result { + Err(McpErrorData::invalid_params("not allowed", None)) + } +} + +#[tool_handler] +impl ServerHandler for ProtocolErrorServer {} + +#[tokio::test] +async fn empty_error_becomes_hard_toolerror() { + let (client, _server) = connect_in_process(EmptyErrorServer::new()).await; + let provider = McpToolProvider::connect(client, None) + .await + .expect("connect"); + let fail = provider + .tools() + .iter() + .find(|t| t.name() == "fail") + .expect("fail tool"); + let ctx = ToolContext::default(); + let err = fail + .call(json!({}), &ctx) + .await + .expect_err("empty error must be a hard Err"); + match err { + loopctl::tool::ToolError::Execution(msg) => { + assert!( + msg.contains("fail"), + "empty-error message should name the tool: {msg}" + ); + } + other => panic!("expected ToolError::Execution, got {other:?}"), + } +} + +#[tokio::test] +async fn protocol_error_rejection_becomes_hard_toolerror() { + let (client, _server) = connect_in_process(ProtocolErrorServer::new()).await; + let provider = McpToolProvider::connect(client, None) + .await + .expect("connect"); + let reject = provider + .tools() + .iter() + .find(|t| t.name() == "reject") + .expect("reject tool"); + let ctx = ToolContext::default(); + let err = reject + .call(json!({}), &ctx) + .await + .expect_err("a JSON-RPC rejection must be a hard Err"); + match err { + loopctl::tool::ToolError::Execution(msg) => { + assert!( + msg.contains("MCP tools/call failed"), + "error should carry the tools/call failure context: {msg}" + ); + } + other => panic!("expected ToolError::Execution, got {other:?}"), + } +} + +/// A server with multipart-returning tools. +#[derive(Clone)] +struct MultipartServer { + router: ToolRouter, +} + +#[tool_router] +impl MultipartServer { + fn new() -> Self { + Self { + router: Self::tool_router(), + } + } + + #[tool(description = "Returns two text parts")] + async fn two(&self) -> Result { + Ok(CallToolResult::success(vec![ + ContentBlock::text("a"), + ContentBlock::text("b"), + ])) + } + + #[tool(description = "Returns text plus image")] + async fn mixed(&self) -> Result { + Ok(CallToolResult::success(vec![ + ContentBlock::text("caption"), + ContentBlock::image("Zm9v", "image/png"), + ])) + } +} + +#[tool_handler] +impl ServerHandler for MultipartServer {} + +#[tokio::test] +async fn multipart_text_joins_with_newline() { + let (client, _server) = connect_in_process(MultipartServer::new()).await; + let provider = McpToolProvider::connect(client, None) + .await + .expect("connect"); + let two = provider + .tools() + .iter() + .find(|t| t.name() == "two") + .expect("two tool"); + let ctx = ToolContext::default(); + let out = two.call(json!({}), &ctx).await.expect("call ok"); + assert!( + matches!(out.payload, MessageToolContent::Multipart(_)), + "expected multipart" + ); + assert_eq!(out.text_content(), "a\nb"); +} + +#[tokio::test] +async fn multipart_mixed_preserves_part_kinds_in_order() { + let (client, _server) = connect_in_process(MultipartServer::new()).await; + let provider = McpToolProvider::connect(client, None) + .await + .expect("connect"); + let mixed = provider + .tools() + .iter() + .find(|t| t.name() == "mixed") + .expect("mixed tool"); + let ctx = ToolContext::default(); + let out = mixed.call(json!({}), &ctx).await.expect("call ok"); + let MessageToolContent::Multipart(parts) = out.payload else { + panic!("expected multipart, got {:?}", out.payload); + }; + assert_eq!(parts.len(), 2, "two parts"); + assert!( + matches!(parts.first(), Some(ToolContentPart::Text { text }) if text == "caption"), + "first is the caption text" + ); + assert!( + matches!(parts.get(1), Some(ToolContentPart::Image { .. })), + "second is the image" + ); +} + +/// A server whose tool returns an audio block — the "stringify, don't drop" +/// fallback. +#[derive(Clone)] +struct AudioServer { + router: ToolRouter, +} + +#[tool_router] +impl AudioServer { + fn new() -> Self { + Self { + router: Self::tool_router(), + } + } + + #[tool(description = "Returns an audio block")] + async fn beep(&self) -> Result { + Ok(CallToolResult::success(vec![ContentBlock::audio( + "AAAA", + "audio/wav", + )])) + } +} + +#[tool_handler] +impl ServerHandler for AudioServer {} + +#[tokio::test] +async fn unsupported_content_type_does_not_vanish() { + let (client, _server) = connect_in_process(AudioServer::new()).await; + let provider = McpToolProvider::connect(client, None) + .await + .expect("connect"); + let beep = provider + .tools() + .iter() + .find(|t| t.name() == "beep") + .expect("beep tool"); + let ctx = ToolContext::default(); + let out = beep.call(json!({}), &ctx).await.expect("call ok"); + let text = out.text_content(); + assert!( + text.contains("unsupported MCP content type"), + "audio must surface as a note, got {text:?}" + ); +} + +#[tokio::test] +async fn refresh_replaces_snapshot_in_place() { + let (client, _server) = connect_in_process(EchoServer::new()).await; + let mut provider = McpToolProvider::connect(client, None) + .await + .expect("connect"); + let before = provider.tools().len(); + + let mut registry = ToolRegistry::new(); + provider.register_into(&mut registry); + let registry_len_before = registry.len(); + + provider.refresh().await.expect("refresh"); + assert_eq!( + provider.tools().len(), + before, + "same server, same count after refresh" + ); + assert_eq!( + registry.len(), + registry_len_before, + "refresh must not mutate an already-populated registry" + ); +} + +/// A server with two tools whose server-side names collide after prefixing. +#[derive(Clone)] +struct CollisionServer { + router: ToolRouter, +} + +#[tool_router] +impl CollisionServer { + fn new() -> Self { + Self { + router: Self::tool_router(), + } + } + + #[tool(name = "status", description = "first status")] + async fn status_a(&self) -> String { + "a".to_string() + } + + #[tool(name = "status", description = "second status")] + async fn status_b(&self) -> String { + "b".to_string() + } +} + +#[tool_handler] +impl ServerHandler for CollisionServer {} + +#[tokio::test] +async fn intra_batch_name_collision_keeps_one_no_panic() { + let (client, _server) = connect_in_process(CollisionServer::new()).await; + let provider = McpToolProvider::connect(client, Some("git".into())) + .await + .expect("connect"); + let names: Vec<&str> = provider.tools().iter().map(Tool::name).collect(); + assert_eq!( + names.iter().filter(|&&n| n == "git__status").count(), + 1, + "exactly one colliding name kept: {names:?}" + ); +} + +#[tokio::test] +async fn drop_provider_cancels_background_server() { + let (client, server_handle) = connect_in_process(EchoServer::new()).await; + { + let provider = McpToolProvider::connect(client, None) + .await + .expect("connect"); + assert!(!provider.tools().is_empty()); + } + let resolved = tokio::time::timeout(Duration::from_secs(2), server_handle) + .await + .expect("server shuts down within 2s of provider drop"); + resolved.expect("server task did not panic"); +} + +#[tokio::test] +async fn re_register_with_overlapping_name_overwrites() { + let (client, _server) = connect_in_process(EchoServer::new()).await; + let provider = McpToolProvider::connect(client, None) + .await + .expect("connect"); + let mut registry = ToolRegistry::new(); + provider.register_into(&mut registry); + let len_after_first = registry.len(); + provider.register_into(&mut registry); + assert_eq!(registry.len(), len_after_first); +} + +#[tokio::test] +async fn schema_passed_through_preserves_properties() { + // `plain` is advertised by `AnnotatedServer` with a non-trivial input + // schema (type/properties/required); the bridge must carry it verbatim. + let (client, _server) = connect_in_process(AnnotatedServer::new()).await; + let provider = McpToolProvider::connect(client, None) + .await + .expect("connect"); + let plain = provider + .tools() + .iter() + .find(|t| t.name() == "plain") + .expect("plain tool"); + let adapted = plain.schema().input_schema; + assert!( + adapted.get("properties").is_some(), + "adapted schema preserves properties: {adapted}" + ); + let required = adapted.get("required").and_then(Value::as_array); + assert!( + required.is_some_and(|arr| arr.iter().any(|v| v == "q")), + "required field `q` carried through: {adapted}" + ); +} + +#[tokio::test] +async fn output_schema_carried_through_accessor() { + // `plain` declares an outputSchema; the adapter carries it verbatim and + // exposes it via McpTool::output_schema(). Pins the round-1 accessor that + // had no coverage. + let (client, _server) = connect_in_process(AnnotatedServer::new()).await; + let provider = McpToolProvider::connect(client, None) + .await + .expect("connect"); + let plain = provider + .tools() + .iter() + .find(|t| t.name() == "plain") + .expect("plain tool"); + let carried = plain.output_schema().expect("outputSchema carried through"); + assert_eq!( + carried.get("type").and_then(Value::as_str), + Some("string"), + "outputSchema carried verbatim: {carried}" + ); + // The annotated tool declares no outputSchema → None. + let annotated = provider + .tools() + .iter() + .find(|t| t.name() == "annotated") + .expect("annotated tool"); + assert!( + annotated.output_schema().is_none(), + "absent outputSchema → None" + ); +} + +/// An `RmcpTool` with explicit annotations, used to exercise the annotation +/// bridge. Built via rmcp's constructors because `Tool`/`ToolAnnotations` are +/// `#[non_exhaustive]`. +fn annotated_tool(read_only: bool) -> RmcpTool { + let annotations = ToolAnnotations::default() + .read_only(read_only) + .destructive(false); + RmcpTool::new("annotated", "annotated tool", serde_json::Map::new()) + .with_annotations(annotations) +} + +/// An `RmcpTool` with **no** annotations, a non-trivial input schema, and a +/// declared `outputSchema` — exercises the spec-default path (absent +/// `readOnlyHint` → false; absent `destructiveHint` → true), verbatim schema +/// passthrough, and `outputSchema` carriage. +fn plain_tool() -> RmcpTool { + let mut schema = serde_json::Map::new(); + schema.insert( + "type".to_string(), + serde_json::Value::String("object".to_string()), + ); + schema.insert( + "properties".to_string(), + serde_json::json!({"q": {"type": "string"}}), + ); + schema.insert( + "required".to_string(), + serde_json::Value::Array(vec![serde_json::Value::String("q".to_string())]), + ); + let mut output_schema = serde_json::Map::new(); + output_schema.insert( + "type".to_string(), + serde_json::Value::String("string".to_string()), + ); + RmcpTool::new("plain", "an unannotated tool", schema) + .with_raw_output_schema(std::sync::Arc::new(output_schema)) +} + +/// A server that advertises an annotated tool and an unannotated tool by +/// overriding `list_tools` (the `#[tool]` macro does not annotate, so a manual +/// override is the way to exercise the annotation bridge). The `#[tool_handler]` +/// macro sees the manual `list_tools` and skips generating its own, so the +/// override wins; it still generates `call_tool`/`get_info`. +#[derive(Clone)] +struct AnnotatedServer { + router: ToolRouter, +} + +#[tool_router] +impl AnnotatedServer { + fn new() -> Self { + Self { + router: Self::tool_router(), + } + } +} + +impl ServerHandler for AnnotatedServer { + async fn list_tools( + &self, + _request: Option, + _ctx: rmcp::service::RequestContext, + ) -> Result { + Ok(rmcp::model::ListToolsResult { + next_cursor: None, + tools: vec![annotated_tool(true), plain_tool()], + ..Default::default() + }) + } +} + +#[tokio::test] +async fn read_only_hint_honored_from_live_server() { + let (client, _server) = connect_in_process(AnnotatedServer::new()).await; + let provider = McpToolProvider::connect(client, None) + .await + .expect("connect"); + let lookup = provider + .tools() + .iter() + .find(|t| t.name() == "annotated") + .expect("annotated tool discovered"); + assert!(lookup.is_read_only(), "annotated read-only honored"); +} + +#[tokio::test] +async fn absent_destructive_hint_defaults_to_destructive_per_spec() { + // The MCP spec: an absent destructiveHint means "assume destructive". + // `plain` carries no annotations, so is_destructive_hint() must be true. + let (client, _server) = connect_in_process(AnnotatedServer::new()).await; + let provider = McpToolProvider::connect(client, None) + .await + .expect("connect"); + let plain = provider + .tools() + .iter() + .find(|t| t.name() == "plain") + .expect("plain tool discovered"); + assert!( + plain.is_destructive_hint(), + "absent destructiveHint must default to destructive (true) per spec" + ); + assert!( + !plain.is_read_only(), + "absent readOnlyHint must default to non-read-only (false)" + ); + // And the explicitly-annotated tool keeps its explicit value. + let annotated = provider + .tools() + .iter() + .find(|t| t.name() == "annotated") + .expect("annotated tool discovered"); + assert!( + !annotated.is_destructive_hint(), + "explicit destructiveHint=false must be honored" + ); +} From 90fd77b6effe3ebc88d513efb8c411a05c4c990c Mon Sep 17 00:00:00 2001 From: Iurii Bobrykov Date: Sat, 8 Aug 2026 14:24:46 +1200 Subject: [PATCH 2/2] fix: docs, debug derives, example --- README.md | 2 +- examples/mcp-adapter.rs | 23 +++------ src/engine/bare/tests.rs | 6 ++- src/mcp.rs | 101 +++++++++++++++++++-------------------- 4 files changed, 62 insertions(+), 70 deletions(-) diff --git a/README.md b/README.md index a23729c..b5a93f1 100644 --- a/README.md +++ b/README.md @@ -158,7 +158,7 @@ let agent = BareLoop::new( | `zai` | No | `providers`, `anthropic` | Z.AI API client (Anthropic-compatible) | | `grammar` | No | `providers` | Tool-call grammar providers for grammar-aware samplers (vLLM `guided_json`); enables the `Grammar` mode of `ToolConstraint` | | `schema_validation` | No | — | JSON Schema validation of `Correction::modified_input` in `LlmReflector` (pulls `jsonschema`); when off, validation is skipped | -| `mcp` | No | `rmcp` | MCP client adapter (`mcp::McpToolProvider`) — adapt any MCP server's tools as loopctl `Tool` impls | +| `mcp` | No | `rmcp` | MCP client adapter (`mcp::McpToolProvider`) — adapt an MCP server's tools as loopctl `Tool` impls (in-process; stdio/HTTP/SSE transports land in a later release) | ### Streaming vs non-streaming diff --git a/examples/mcp-adapter.rs b/examples/mcp-adapter.rs index 19cb7be..7c74d1b 100644 --- a/examples/mcp-adapter.rs +++ b/examples/mcp-adapter.rs @@ -22,7 +22,7 @@ use loopctl::mcp::{McpClient, McpToolProvider}; use loopctl::tool::{ToolContext, ToolRegistry}; use rmcp::handler::server::ServerHandler; use rmcp::handler::server::router::tool::ToolRouter; -use rmcp::{ServiceExt, tool, tool_handler, tool_router}; +use rmcp::{tool, tool_handler, tool_router}; use serde_json::json; /// An rmcp server exposing one `greet` tool. @@ -50,21 +50,12 @@ impl ServerHandler for GreetServer {} #[tokio::main] async fn main() { - // 1. Start the server on one end of a duplex, a pure rmcp client on the - // other, and wrap the client as an McpClient. - let (server_end, client_end) = tokio::io::duplex(4096); - tokio::spawn(async move { - let running = GreetServer::new() - .serve(server_end) - .await - .expect("server initialize"); - running.waiting().await.ok(); - }); - let client = - ().serve(client_end) - .await - .map(McpClient::from_service) - .expect("client initialize"); + // 1. Connect an rmcp client to the in-process server and run the MCP + // initialize handshake. This is the only constructor L-12 ships; real + // transports (stdio, HTTP/SSE) arrive in a later release. + let client = McpClient::in_process(GreetServer::new()) + .await + .expect("client initialize"); // 2. Discover the server's tools and register them. let provider = McpToolProvider::connect(client, None) diff --git a/src/engine/bare/tests.rs b/src/engine/bare/tests.rs index 56c8f25..5b49267 100644 --- a/src/engine/bare/tests.rs +++ b/src/engine/bare/tests.rs @@ -4459,7 +4459,9 @@ fn test_add_contributor_panics_after_session_start() { // Box the future so we can drop it without awaiting; the session-init // side effect is the state transition under test. The turn path uses // `tokio::select!`/`tokio::time::sleep`, which require a tokio reactor - // context, so enter a current-thread runtime before block_on polls. + // context, so enter a runtime guard before block_on polls (the guard only + // makes a reactor available on this thread — we do not drive via the + // runtime's own block_on, which would run the loop to completion). { let run_config = RunConfig::default(); let fut = agent.run("seed", &run_config); @@ -4553,7 +4555,7 @@ fn test_set_request_options_panics_after_session_start() { // The first run() establishes the session and moves the loop out of // Idle; a subsequent set_request_options must panic in debug builds. // The turn path uses `tokio::select!`/`tokio::time::sleep`, which require a - // tokio reactor context, so enter a current-thread runtime before block_on. + // tokio reactor context, so enter a runtime guard before block_on polls. { let run_config = RunConfig::default(); let fut = agent.run("seed", &run_config); diff --git a/src/mcp.rs b/src/mcp.rs index 66fc9d0..d33f8db 100644 --- a/src/mcp.rs +++ b/src/mcp.rs @@ -33,12 +33,12 @@ //! use loopctl::mcp::{McpClient, McpToolProvider}; //! use loopctl::tool::ToolRegistry; //! -//! # async fn run(server: impl rmcp::handler::server::ServerHandler) { +//! # async fn run(server: impl rmcp::handler::server::ServerHandler) -> Result<(), loopctl::mcp::McpError> { //! let client = McpClient::in_process(server).await?; //! let provider = McpToolProvider::connect(client, None).await?; //! let mut registry = ToolRegistry::new(); //! provider.register_into(&mut registry); -//! # Ok::<(), loopctl::mcp::McpError>(()) +//! # Ok(()) //! # } //! ``` @@ -78,7 +78,7 @@ const DUPLEX_BUFFER: usize = 4096; /// /// Dropping the last clone drops the [`RunningService`], whose cancellation /// guard cancels the background task — there is no leaked runtime work. -#[derive(Clone)] +#[derive(Clone, Debug)] pub struct McpClient { /// The running client service. The handler is fixed to `()`, the pure /// client: a server's `sampling`/`roots` requests get default empty @@ -121,8 +121,14 @@ impl McpClient { { let (server_end, client_end) = tokio::io::duplex(DUPLEX_BUFFER); tokio::spawn(async move { - if let Ok(running) = server.serve(server_end).await { - let _ = running.waiting().await.ok(); + match server.serve(server_end).await { + Ok(running) => { + let _ = running.waiting().await.ok(); + } + Err(e) => tracing::error!( + error = %e, + "in-process MCP server failed to initialize (the client side reports this via McpError::Handshake)" + ), } }); let client = ().serve(client_end).await.map_err(|e| McpError::Handshake(e.to_string()))?; @@ -228,6 +234,7 @@ impl McpClient { /// [`McpToolProvider::connect`] / [`McpToolProvider::refresh`] calls; the only /// interior mutation is [`McpToolProvider::refresh`], which takes `&mut self`, /// so concurrent reads of [`McpToolProvider::tools`] during a run are safe. +#[derive(Debug)] pub struct McpToolProvider { /// The connected client the adapted tools forward through. /// @@ -254,11 +261,13 @@ pub struct McpToolProvider { impl McpToolProvider { /// Connect to a server and snapshot its tool list. /// - /// The primary constructor. Runs the MCP `initialize` handshake (driven by - /// the supplied [`McpClient`]) followed by `tools/list`, which rmcp - /// auto-paginates by following `nextCursor` to exhaustion. The returned - /// provider holds one [`McpTool`] per server-declared tool, each sharing a - /// cheap clone of the client handle. + /// The primary constructor. Runs `tools/list` against the already-connected + /// `client` (the `initialize` handshake is *not* done here — it completed + /// when the [`McpClient`] was built via [`McpClient::in_process`] or + /// [`McpClient::from_service`]). rmcp auto-paginates `tools/list` by + /// following `nextCursor` to exhaustion. The returned provider holds one + /// [`McpTool`] per server-declared tool, each sharing a cheap clone of the + /// client handle. /// /// `name_prefix`, when `Some`, namespaces every adapted tool: it is /// prepended to each tool's name as `"{prefix}__{tool_name}"` for both @@ -385,7 +394,7 @@ impl McpToolProvider { /// [`McpTool::is_destructive_hint`]. Both apply the MCP-spec defaults for an /// absent hint (read-only defaults to `false`; destructive defaults to `true`), /// so a consumer can read them directly without re-applying the spec. -#[derive(Clone)] +#[derive(Clone, Debug)] pub struct McpTool { /// The original server-side name, sent verbatim in each `tools/call`. /// @@ -664,6 +673,7 @@ fn bridge_tool( ) -> Option { let server_name = server_tool.name.to_string(); if server_name.is_empty() { + tracing::warn!("MCP server declared a tool with an empty name; skipping"); return None; } let exposed_name = @@ -757,21 +767,27 @@ fn bridge_result( /// Map one rmcp content block to a loopctl [`ToolContentPart`]. /// -/// Text and image carry through; audio, embedded resources, resource links, and -/// any future block kind fall back to a short text note so the model learns the -/// part existed rather than seeing it silently dropped. +/// Text and image carry through. Audio falls back to a short text note. Embedded +/// resources and resource links are stringified with their identifying payload +/// (uri, text/name) so the model sees *what* the server returned, not just that +/// it returned something. Any future block kind falls back to a generic note. fn bridge_content(block: &ContentBlock) -> ToolContentPart { match block { ContentBlock::Text(text) => ToolContentPart::text(&text.text), ContentBlock::Image(image) => ToolContentPart::image( crate::message::ImageSource::new_base64(&image.mime_type, &image.data), ), - ContentBlock::Audio(_) => ToolContentPart::text("unsupported MCP content type: audio"), - ContentBlock::Resource(_) => { - ToolContentPart::text("unsupported MCP content type: embedded resource") - } - ContentBlock::ResourceLink(_) => { - ToolContentPart::text("unsupported MCP content type: resource link") + ContentBlock::Resource(resource) => match &resource.resource { + rmcp::model::ResourceContents::TextResourceContents { uri, text, .. } => { + ToolContentPart::text(format!("MCP resource {uri}: {text}")) + } + rmcp::model::ResourceContents::BlobResourceContents { uri, .. } => { + ToolContentPart::text(format!("MCP resource {uri}: (blob)")) + } + _ => ToolContentPart::text("unsupported MCP content type: embedded resource"), + }, + ContentBlock::ResourceLink(link) => { + ToolContentPart::text(format!("MCP resource link: {} ({})", link.name, link.uri)) } _ => ToolContentPart::text("unsupported MCP content type"), } @@ -779,15 +795,9 @@ fn bridge_content(block: &ContentBlock) -> ToolContentPart { #[cfg(test)] mod tests { - //! Unit tests for the pure bridge functions (`bridge_result`, - //! `bridge_content`). The connection-dependent paths (`bridge_tool`, - //! `bridge_tool_list`, `McpClient`, `McpToolProvider`) are covered by the - //! integration suite in `tests/mcp_tool_provider.rs`. - use super::*; use rmcp::model::{CallToolResult, ContentBlock}; - /// A single text block collapses to plain [`MessageToolContent::Text`]. #[test] fn bridge_result_single_text_becomes_text_payload() { let res = CallToolResult::success(vec![ContentBlock::text("hi")]); @@ -797,9 +807,6 @@ mod tests { assert_eq!(out.text_content(), "hi"); } - /// A single image block can't be a `Text`, so it becomes a one-element - /// `Multipart` — an edge the integration suite (which only builds macro - /// servers returning text) does not hit. #[test] fn bridge_result_single_image_becomes_single_part_multipart() { let res = CallToolResult::success(vec![ContentBlock::image("Zm9v", "image/png")]); @@ -816,7 +823,6 @@ mod tests { } } - /// Multiple blocks become a `Multipart` preserving order. #[test] fn bridge_result_multiple_blocks_become_multipart_in_order() { let res = CallToolResult::success(vec![ @@ -834,8 +840,6 @@ mod tests { assert!(matches!(parts.get(2), Some(ToolContentPart::Text { text }) if text == "b")); } - /// A server-reported error (`isError: true`) with text content is a *soft* - /// failure: `Ok` with `is_error` set, not an `Err`. #[test] fn bridge_result_soft_error_returns_ok_with_is_error() { let res = CallToolResult::error(vec![ContentBlock::text("boom")]); @@ -844,8 +848,6 @@ mod tests { assert_eq!(out.text_content(), "boom"); } - /// An error with no content surfaces as the hard [`McpError::EmptyToolError`], - /// carrying the tool name — pins the round-2 fix that threads `tool_name`. #[test] fn bridge_result_empty_error_is_hard_empty_tool_error_with_name() { let res = CallToolResult::error(vec![]); @@ -856,8 +858,6 @@ mod tests { } } - /// A successful result with zero content blocks yields an empty successful - /// output — distinct from the empty-error path above. #[test] fn bridge_result_empty_success_yields_empty_text_output() { let res = CallToolResult::success(vec![]); @@ -866,9 +866,6 @@ mod tests { assert_eq!(out.text_content(), ""); } - /// `structuredContent`, when present, is appended as one extra JSON-string - /// text part — carried, not parsed. Even a single text block + structured - /// content therefore becomes a two-part `Multipart`. #[test] fn bridge_result_structured_content_appended_as_text_part() { let mut res = CallToolResult::success(vec![ContentBlock::text("body")]); @@ -893,10 +890,6 @@ mod tests { assert!(structured_text.contains('7')); } - /// An error result (`isError: true`) with empty content but a declared - /// `structuredContent` is *not* an `EmptyToolError` — the structured payload - /// surfaces as a soft error's text, so the model sees what the server - /// returned. Pins this edge (the integration suite never builds it). #[test] fn bridge_result_error_with_structured_content_is_soft_error() { let mut res = CallToolResult::error(vec![]); @@ -910,27 +903,33 @@ mod tests { ); } - /// `bridge_content`: every unsupported block kind surfaces as a text note - /// naming the kind, never silently dropped. Covers the arms the macro-server - /// integration tests can't easily reach (resource, resource-link). #[test] fn bridge_content_unsupported_kinds_surface_as_text_notes() { + // Audio carries no payload loopctl can render, so it falls through to + // the generic unsupported-note arm (the `_` fallback). It must still + // surface as a Text part — never silently dropped. let audio = bridge_content(&ContentBlock::audio("AAAA", "audio/wav")); assert!( - matches!(audio, ToolContentPart::Text { ref text } if text.contains("audio")), - "audio → text note, got {audio:?}" + matches!(audio, ToolContentPart::Text { .. }), + "audio → text note (not dropped), got {audio:?}" ); let resource = bridge_content(&ContentBlock::Resource(rmcp::model::EmbeddedResource::new( rmcp::model::ResourceContents::text("body", "mem://x"), ))); assert!( - matches!(resource, ToolContentPart::Text { ref text } if text.contains("resource")), - "embedded resource → text note, got {resource:?}" + matches!(resource, ToolContentPart::Text { ref text } if text.contains("mem://x") && text.contains("body")), + "embedded text resource surfaces its uri and text, got {resource:?}" + ); + + let link = rmcp::model::Resource::new("file:///a", "thing"); + let link_part = bridge_content(&ContentBlock::ResourceLink(link)); + assert!( + matches!(link_part, ToolContentPart::Text { ref text } if text.contains("thing") && text.contains("file:///a")), + "resource link surfaces name and uri, got {link_part:?}" ); } - /// `bridge_content`: text and image carry their payloads through verbatim. #[test] fn bridge_content_text_and_image_carry_through() { let text = bridge_content(&ContentBlock::text("hello"));