diff --git a/CHANGELOG.md b/CHANGELOG.md index 7773950..dccee1a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,7 +9,9 @@ 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`). +- `McpToolProvider::with_call_timeout(Duration)` — a per-call budget for every adapted tool's `tools/call` round-trip (default 60s). A call that exceeds it resolves to a *soft* error result naming the tool and the budget, so a wedged MCP server costs one tool result instead of hanging the agent loop indefinitely. Updates already-discovered tools and applies to later refreshes. Pinned by `call_timeout_cuts_a_slow_tool_with_a_soft_error` / `default_timeout_lets_a_quick_tool_through` in `tests/mcp_tool_provider.rs`. +- MCP transports (stdio + Streamable HTTP/SSE) — `McpClient::stdio(command)` spawns an MCP server as a child process over stdio; `McpClient::http_sse(endpoint)` and `McpClient::http_sse_with_client(endpoint, reqwest::Client)` connect via the Streamable HTTP transport (rmcp handles `Mcp-Session-Id`, JSON-vs-SSE response splitting, and DELETE-on-close). `McpClient::reconnect(&StreamRetryConfig)` re-establishes a dropped connection using the crate's existing backoff strategy (the one retry strategy for the crate, not a second one). New public `CommandSpec` describes what to spawn. A stdio example server ships at `examples/mcp-stdio-server.rs` for transport testing. The `mcp` feature now enables `streaming` (for `StreamRetryConfig`) and `reqwest` (for the HTTP client); `default = []` is unchanged. +- `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`. Real-server transports ship alongside (stdio child processes and Streamable HTTP/SSE — see the transports entry above); `McpClient::in_process` connects an in-process rmcp server for tests and bundled-server use. 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. @@ -17,6 +19,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/2.0.0. ### Changed +- Provider HTTP clients now set a **read timeout** (maximum gap between response bytes) instead of a total request timeout, and `with_timeout` configures that read timeout. A total HTTP-layer cap aborted every SSE stream longer than the configured duration (default 2 minutes) — pre-empting the `StreamHandler`'s per-event/total-stream deadlines and the engine's turn timeout, which own generation-length budgets. Long healthy streams now run as long as they keep producing bytes; a server silent for the configured gap (default 120s) is still aborted. Behavior change: generations longer than the old total cap no longer fail at the HTTP layer. + - **Breaking:** Machine turn indices are now 0-indexed (`CallLLM { turn: 0 }` for the first turn). Previously 1-indexed (`turn: 1`). `AwaitingModel { turn }` and `AwaitingTools { turn }` follow the same convention. Callers matching on these variants in tests or drivers must adjust. - **Breaking:** `LoopError` gains `ToolRecoveryExhausted` variant. Exhaustive matches on `LoopError` must add this arm. - **Breaking:** `RunConfig` gains `memory_top_k` field. Struct-literal construction must add it (use `..Default::default()` or the builders). @@ -37,6 +41,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/2.0.0. ### Fixed +- The non-streaming fallback forwards the turn's `RequestOptions` and is bounded by the total-stream deadline (`StreamHandler::fallback_non_streaming`): it previously called `create_message` (dropping any configured `response_format`/`tool_constraint`) and raced only the cancel signal, so a hanging fallback request hung the turn. Pinned by `fallback_non_streaming_forwards_request_options`, `fallback_non_streaming_honors_the_total_deadline`, and `completed_fallback_response_racing_the_deadline_is_accepted` (the deadline bounds waiting, not completion — a resolved response is accepted even past expiry, while a hanging one is still cut). +- MCP `tools/call` round-trips are bounded by the new per-call timeout (see `McpToolProvider::with_call_timeout` above) — previously a wedged server hung the agent loop with cancellation as the only exit. - Tool-result parts in a `CallTools` turn preserve **model request order** across preresolved (unknown-tool) and dispatched calls. Previously the turn's results were assembled as `[all preresolved, then all dispatched]`, which reordered the parts the model saw relative to the calls it made. Provider-safe in practice (providers match by `tool_call_id`, not position), but order-non-preserving and surprising to hosts that assume positional alignment. Pinned by `test_mixed_known_unknown_tools_preserve_request_order`. - The `run()` `Done` arm now matches every `MachineOutcome` variant explicitly (`Completed`, `MaxTurnsExceeded`, `Cancelled`, `Failed`) instead of using a wildcard `other => ... unwrap_or(Cancelled)` fallback. `MachineOutcome` is `#[non_exhaustive]` but defined in this crate, so the compiler proves this exhaustive — a future variant forces a compile error here rather than being silently mislabelled as `Cancelled`. - `handle_call_tools` doc corrected to state where cancellation is actually honored (the in-flight tool call is raced against the cancel signal in `execute_tool_call`'s `select!`; the sequential path checks the signal between calls), instead of claiming a `select!` that does not exist in the function itself. diff --git a/Cargo.toml b/Cargo.toml index 726c4cf..d71cd98 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -33,7 +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"] } +rmcp = { version = "3", optional = true, default-features = false, features = ["client", "server", "macros", "transport-async-rw", "transport-child-process", "transport-streamable-http-client-reqwest", "transport-io"] } [dev-dependencies] tokio = { version = "1.52.3", features = ["rt-multi-thread", "macros", "sync", "time", "signal"] } @@ -62,8 +62,10 @@ zai = ["providers", "anthropic"] grammar = ["providers"] schema_validation = ["dep:jsonschema"] -# MCP client adapter -mcp = ["dep:rmcp"] +# MCP client adapter + transports (stdio, Streamable HTTP/SSE). Enables +# `streaming` for `StreamRetryConfig` (reconnect backoff) and `reqwest` for the +# HTTP transport's caller-supplied client. +mcp = ["dep:rmcp", "streaming", "dep:reqwest"] [[example]] name = "hello-cli" @@ -85,6 +87,10 @@ required-features = ["testing", "providers"] name = "mcp-adapter" required-features = ["mcp"] +[[example]] +name = "mcp-stdio-server" +required-features = ["mcp"] + [lints.clippy] pedantic = { level = "warn", priority = -1 } unwrap_used = "deny" diff --git a/README.md b/README.md index b5a93f1..6f1b065 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 an MCP server's tools as loopctl `Tool` impls (in-process; stdio/HTTP/SSE transports land in a later release) | +| `mcp` | No | `rmcp`, `reqwest`, `async-stream` | MCP client adapter (`mcp::McpToolProvider`) — adapt an MCP server's tools as loopctl `Tool` impls over stdio, Streamable HTTP/SSE, or in-process (`McpClient::stdio`/`http_sse`/`in_process`) | ### Streaming vs non-streaming diff --git a/examples/mcp-stdio-server.rs b/examples/mcp-stdio-server.rs new file mode 100644 index 0000000..ae7742a --- /dev/null +++ b/examples/mcp-stdio-server.rs @@ -0,0 +1,58 @@ +//! A minimal MCP server over stdio — the subprocess target for the L-13 stdio +//! transport tests. +//! +//! Built with rmcp's `#[tool_router]` / `#[tool]` macros, exposing one `greet` +//! tool. Runs the server on stdin/stdout (the MCP stdio transport); stderr is +//! inherited so server logs surface during tests. Exits when the client +//! disconnects. +//! +//! ```sh +//! cargo run --example mcp-stdio-server --features mcp +//! ``` + +#![allow( + clippy::expect_used, + clippy::panic, + clippy::missing_errors_doc, + clippy::missing_panics_doc, + dead_code +)] + +use rmcp::ServiceExt; +use rmcp::handler::server::ServerHandler; +use rmcp::handler::server::router::tool::ToolRouter; +use rmcp::{tool, tool_handler, tool_router}; + +/// A server exposing one `greet` tool. +#[derive(Clone)] +struct StdioServer { + router: ToolRouter, +} + +#[tool_router] +impl StdioServer { + fn new() -> Self { + Self { + router: Self::tool_router(), + } + } + + #[tool(description = "Return a friendly greeting")] + async fn greet(&self) -> String { + "hello from stdio".to_string() + } +} + +#[tool_handler] +impl ServerHandler for StdioServer {} + +#[tokio::main] +async fn main() { + let server = StdioServer::new(); + let transport = rmcp::transport::io::stdio(); + let running = server + .serve(transport) + .await + .expect("stdio server initialize"); + running.waiting().await.ok(); +} diff --git a/src/engine/bare/config.rs b/src/engine/bare/config.rs index 5ec9e0e..94bc17a 100644 --- a/src/engine/bare/config.rs +++ b/src/engine/bare/config.rs @@ -284,16 +284,30 @@ impl BareLoop { /// automatically from `self.tools` so that schema generation and dispatch /// always share the same underlying registry: /// + /// The pipeline is also the injection point for per-dispatch host state: + /// a middleware may augment `ctx.tool_context` — set extensions, `cwd`, + /// `is_non_interactive` — before the tool runs. See + /// [`ToolContext`](crate::tool::ToolContext) ("Passing host state to + /// tools") and the `host-state` example for the full pattern. + /// /// # Example /// - /// ```rust,ignore - /// use loopctl::engine::middleware::{ToolPipeline, TimeoutMiddleware}; + /// ```rust,no_run + /// # use std::sync::Arc; + /// # use loopctl::config::SessionConfig; + /// # use loopctl::engine::BareLoop; + /// # use loopctl::testing::MockApiClient; + /// # use loopctl::tool::ToolRegistry; + /// use loopctl::middleware::{ToolPipeline, TimeoutMiddleware}; /// /// let builder = ToolPipeline::builder() /// .with_middleware(TimeoutMiddleware::from_secs(30)); /// - /// let mut agent = BareLoop::new(client, registry, config); - /// agent.set_pipeline(builder)?; + /// # let client = MockApiClient::new("demo"); + /// # let registry = ToolRegistry::new(); + /// # let config = SessionConfig::default(); + /// let mut agent = BareLoop::new(Arc::new(client), registry, config); + /// agent.set_pipeline(builder).expect("static composition is valid"); /// ``` /// /// # Errors diff --git a/src/managers.rs b/src/managers.rs index 2892d74..060036c 100644 --- a/src/managers.rs +++ b/src/managers.rs @@ -357,7 +357,10 @@ impl LoopManagers { /// Set the middleware pipeline for tool dispatch. /// /// Non-consuming variant of [`with_pipeline`](Self::with_pipeline) for - /// cases where the managers is already constructed. + /// cases where a `LoopManagers` is already constructed. If you are installing + /// onto a [`BareLoop`](crate::engine::BareLoop), prefer + /// [`BareLoop::set_pipeline`](crate::engine::BareLoop::set_pipeline), which + /// shares the loop's tool registry with the pipeline core automatically. pub fn set_pipeline(&mut self, pipeline: ToolPipeline) { self.tool_pipeline = Some(pipeline); } diff --git a/src/mcp.rs b/src/mcp.rs index d33f8db..a47df9b 100644 --- a/src/mcp.rs +++ b/src/mcp.rs @@ -10,8 +10,8 @@ //! # 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. +//! transport-agnostic boundary: obtain one from [`McpClient::stdio`], +//! [`McpClient::http_sse`], or [`McpClient::in_process`]. //! - [`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 @@ -22,6 +22,24 @@ //! No rmcp type appears in any of these public signatures; an rmcp upgrade is //! a one-file change (this one). //! +//! # Transports +//! +//! Three ways to obtain an [`McpClient`], all yielding the same type so +//! [`McpToolProvider`] is indifferent to how the client was built: +//! +//! - **stdio** — [`McpClient::stdio`] spawns an MCP server as a child process +//! (NDJSON JSON-RPC over the child's stdin/stdout). The common local +//! deployment shape (Claude Desktop, Cursor, the reference clients). +//! - **Streamable HTTP/SSE** — [`McpClient::http_sse`] (rmcp's default HTTP +//! client) or [`McpClient::http_sse_with_client`] (caller-supplied +//! `reqwest::Client`) connect to a remote server. rmcp handles +//! `Mcp-Session-Id`, JSON-vs-SSE response splitting, and DELETE-on-close. +//! - **in-process** — [`McpClient::in_process`] for tests and bundled servers. +//! +//! A dropped connection can be re-established via [`McpClient::reconnect`], +//! which reuses loopctl's [`StreamRetryConfig`](crate::stream::handler::StreamRetryConfig) +//! backoff — the one retry strategy for the crate, not a second one for MCP. +//! //! [mcp]: https://modelcontextprotocol.io //! //! # Example @@ -45,12 +63,25 @@ use std::future::Future; use std::pin::Pin; use std::sync::Arc; +use std::time::Duration; + +/// Default per-call budget for a forwarded `tools/call` round-trip. +/// +/// Generous enough for interactive server tools (searches, reads, short +/// builds) while still bounding a wedged server so the agent loop cannot +/// hang indefinitely. Override per provider with +/// [`McpToolProvider::with_call_timeout`](McpToolProvider::with_call_timeout). +const DEFAULT_MCP_CALL_TIMEOUT: Duration = Duration::from_mins(1); use rmcp::ServiceExt; use rmcp::handler::server::ServerHandler; use rmcp::model::ContentBlock; use rmcp::service::RoleClient; use rmcp::service::RunningService; +use rmcp::transport::IntoTransport; +use rmcp::transport::StreamableHttpClientTransport; +use rmcp::transport::TokioChildProcess; +use rmcp::transport::streamable_http_client::StreamableHttpClientTransportConfig; use crate::message::ToolContent as MessageToolContent; use crate::message::ToolContentPart; @@ -80,11 +111,29 @@ const DUPLEX_BUFFER: usize = 4096; /// guard cancels the background task — there is no leaked runtime work. #[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 - /// answers. A host that wants to honour those constructs its own client - /// with a richer handler (out of scope for this module). + /// The running rmcp client service backing this connection. + /// + /// Held behind an [`Arc`] so every [`McpTool`](crate::mcp::McpTool) clone + /// shares one connection cheaply; calls reach the server via + /// [`RunningService`]'s `Deref` to rmcp's `Peer`. The handler type is fixed + /// to `()` (the pure client), so 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). Dropping + /// the last clone drops the [`RunningService`], whose cancellation guard + /// ends the background task. service: Arc>, + + /// How to rebuild this connection on [`Self::reconnect`], or `None`. + /// + /// Set by [`Self::stdio`] (a [`CommandSpec`]) and [`Self::http_sse`] (an + /// endpoint), so a dropped connection can re-spawn the child or re-connect + /// to the server. `None` for [`Self::in_process`] and + /// [`Self::from_service`]: those clients have no way to reconstruct their + /// transport, so [`Self::reconnect`] returns [`McpError::Handshake`] for + /// them. Private because the concrete [`ReconnectSpec`] enum is an + /// implementation detail; callers drive reconnect through the method, not + /// the field. + reconnect_spec: Option, } impl McpClient { @@ -134,6 +183,7 @@ impl McpClient { let client = ().serve(client_end).await.map_err(|e| McpError::Handshake(e.to_string()))?; Ok(Self { service: Arc::new(client), + reconnect_spec: None, }) } @@ -142,11 +192,28 @@ impl 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. + /// handler, or to share a transport set up out-of-band. The returned client + /// has no [`Self::reconnect`] spec (returns [`McpError::Handshake`]). #[must_use] pub fn from_service(service: RunningService) -> Self { Self { service: Arc::new(service), + reconnect_spec: None, + } + } + + /// Build an [`McpClient`] from a running service + the spec to rebuild it. + /// + /// Shared tail of the three constructors: [`Self::stdio`] and + /// [`Self::http_sse`] (via [`Self::http_connect`]) pass `Some(spec)` so + /// [`Self::reconnect`] can rebuild the transport; [`Self::in_process`] and + /// [`Self::from_service`] pass `None` (no rebuildable transport). Wraps the + /// service in an [`Arc`] so each [`McpTool`](crate::mcp::McpTool) clone + /// shares one connection cheaply. + fn wrap(service: RunningService, spec: Option) -> Self { + Self { + service: Arc::new(service), + reconnect_spec: spec, } } @@ -184,6 +251,300 @@ impl McpClient { .map_err(|e| ToolError::Execution(format!("MCP tools/call failed: {e}")))?; bridge_result(server_name, result).map_err(|e| ToolError::Execution(e.to_string())) } + + /// Connect to an MCP server running as a child process over stdio. + /// + /// `command` is spawned via rmcp's [`TokioChildProcess`] + /// (`transport-child-process` cargo-feature): it pipes the child's + /// stdin/stdout (NDJSON JSON-RPC) and **inherits stderr** so server logs + /// surface during development. The transport implements rmcp's `Transport` + /// directly, so `().serve(transport)` (a pure client, handler `()`) drives + /// the MCP `initialize` handshake before returning. + /// + /// **Lifecycle:** rmcp kills the child on drop via its `ChildWithCleanup` + /// (force-kill after a 3s graceful timeout). Callers do **not** set + /// `kill_on_drop` themselves — it is redundant with rmcp's cleanup. + /// Dropping the returned [`McpClient`] (and all its clones) terminates the + /// server process. + /// + /// **Runtime:** spawns the child and drives the handshake from the caller's + /// async context, so this must be called from within a running tokio + /// runtime (it panics with "no reactor running" otherwise). + /// + /// The [`CommandSpec`] is retained so [`Self::reconnect`] can re-spawn the + /// same server. + /// + /// # Errors + /// + /// [`McpError::Handshake`] if the child fails to spawn (`io::Error` from + /// [`TokioChildProcess::new`]), exits before handshake, or the `initialize` + /// round-trip fails (`ClientInitializeError`). + pub async fn stdio(command: CommandSpec) -> Result { + let transport = TokioChildProcess::new(command.as_tokio_command()) + .map_err(|e| McpError::Handshake(e.to_string()))?; + let service = ().serve(transport).await.map_err(|e| McpError::Handshake(e.to_string()))?; + Ok(Self::wrap(service, Some(ReconnectSpec::Stdio(command)))) + } + + /// Connect to a remote MCP server via Streamable HTTP, using rmcp's default + /// HTTP client. + /// + /// JSON-RPC 2.0 over a single endpoint: `POST` for requests (the server + /// replies `application/json` or `text/event-stream`), optional `GET` for a + /// long-lived SSE notification stream, `DELETE` for session termination. + /// Session identity is the `Mcp-Session-Id` response header, echoed back. + /// + /// rmcp's [`StreamableHttpClientTransport`] handles **all** of + /// `Mcp-Session-Id` capture/echo, JSON-vs-SSE response splitting, the + /// optional GET-opened SSE stream, DELETE-on-close, and transparent session + /// re-init on HTTP 404 — loopctl does none of it. This constructor builds + /// the transport via `from_uri` and runs `().serve(transport)` to handshake. + /// + /// The endpoint is retained so [`Self::reconnect`] can re-connect. rmcp's + /// default client deliberately disables connection pooling + /// (`pool_max_idle_per_host(0)`) to avoid ~40ms TCP Delayed-ACK stalls; for + /// pooling/TLS/timeouts use [`Self::http_sse_with_client`]. + /// + /// **Runtime:** rmcp's Streamable HTTP transport spawns a background worker + /// task, so this must be called from within a running tokio runtime. + /// + /// # Errors + /// + /// [`McpError::Handshake`] if the HTTP connection cannot be established or + /// `initialize` fails (`ClientInitializeError`). + pub async fn http_sse(endpoint: impl Into>) -> Result { + let endpoint = endpoint.into(); + let transport = StreamableHttpClientTransport::from_uri(Arc::clone(&endpoint)); + Self::http_connect(transport, endpoint, None).await + } + + /// Connect via Streamable HTTP with a caller-supplied [`reqwest::Client`]. + /// + /// Like [`Self::http_sse`], but the caller provides its own HTTP client — + /// for connection pooling, custom TLS, or timeouts. Built via + /// [`StreamableHttpClientTransport::with_client`]. Prefer the plain + /// [`Self::http_sse`] unless you need pooling (rmcp's default disables it to + /// avoid TCP Delayed-ACK stalls). + /// + /// Both the endpoint and the supplied client are retained for + /// [`Self::reconnect`]; a reconnect re-connects with this same client, + /// preserving its pooling, TLS, and timeout configuration. + /// + /// # Errors + /// + /// [`McpError::Handshake`] if the HTTP connection cannot be established or + /// `initialize` fails. + pub async fn http_sse_with_client( + endpoint: impl Into>, + client: reqwest::Client, + ) -> Result { + let endpoint = endpoint.into(); + let transport = StreamableHttpClientTransport::with_client( + client.clone(), + StreamableHttpClientTransportConfig::with_uri(Arc::clone(&endpoint)), + ); + Self::http_connect(transport, endpoint, Some(client)).await + } + + /// Drive `().serve(transport)` for an HTTP transport and wrap the result. + /// + /// # Errors + /// + /// [`McpError::Handshake`] if the rmcp `serve`/`initialize` fails + /// (`ClientInitializeError`). + async fn http_connect( + transport: T, + endpoint: Arc, + client: Option, + ) -> Result + where + T: IntoTransport, + E: std::error::Error + Send + Sync + 'static, + { + let service = ().serve(transport).await.map_err(|e| McpError::Handshake(e.to_string()))?; + Ok(Self::wrap( + service, + Some(ReconnectSpec::HttpSse { endpoint, client }), + )) + } + + /// Re-establish a dropped connection using [`StreamRetryConfig`](crate::stream::handler::StreamRetryConfig) + /// backoff. + /// + /// After a transient failure (child exited, HTTP 5xx, broken pipe), call + /// this on the dead [`McpClient`]: it backs off per `retry` and re-runs the + /// constructor that originally built this client (a spec stored on the + /// client at construction time), returning a **new** [`McpClient`]. The old + /// client's transport is dead; the caller re-issues `tools/list` via a fresh + /// [`McpToolProvider`] on the returned client. + /// + /// Returns [`McpError::Handshake`] if this client is an [`Self::in_process`] + /// client (no reconnect spec — an in-process server cannot be rebuilt) or + /// after the retry config's max retries. + /// + /// **Reuse, don't reinvent:** loopctl already ships exponential-backoff- + /// with-jitter retry ([`StreamRetryConfig`](crate::stream::handler::StreamRetryConfig)). + /// This method consumes it directly — there is one retry strategy for the + /// whole crate, not a second one for MCP. (rmcp ships no transport-level + /// reconnect, only an in-stream SSE resume policy — verified.) + /// + /// # Errors + /// + /// [`McpError::Handshake`] for `in_process` clients, or after + /// `retry.max_retries` failed attempts (the last error is carried). The + /// loop always runs at least one attempt before giving up, so a returned + /// `Handshake` carries a real connection failure, not a placeholder. + pub async fn reconnect( + &self, + retry: &crate::stream::handler::StreamRetryConfig, + ) -> Result { + let Some(spec) = self.reconnect_spec.clone() else { + return Err(McpError::Handshake( + "this McpClient cannot be reconnected (in-process)".to_string(), + )); + }; + let mut last_err: Option = None; + for attempt in 0..=retry.max_retries { + if attempt > 0 { + let delay = retry.jittered_base_delay(attempt.saturating_sub(1)); + tokio::time::sleep(delay).await; + } + match spec.connect().await { + Ok(client) => return Ok(client), + Err(e) => last_err = Some(e), + } + } + Err(last_err.unwrap_or_else(|| { + McpError::Handshake("reconnect made no attempts (max_retries overflow)".to_string()) + })) + } +} + +/// What to spawn for [`McpClient::stdio`]. [`Clone`] so [`McpClient::reconnect`] +/// can re-establish the child. +/// +/// Build with struct-literal syntax or [`CommandSpec::default`] then set the +/// fields you need; only [`program`](Self::program) is required for a working +/// spawn. +/// +/// # Example +/// +/// ``` +/// use loopctl::mcp::CommandSpec; +/// +/// let spec = CommandSpec { +/// program: "npx".into(), +/// args: vec!["-y".into(), "@modelcontextprotocol/server-everything".into()], +/// ..Default::default() +/// }; +/// assert_eq!(spec.program, "npx"); +/// assert_eq!(spec.args.len(), 2); +/// ``` +#[derive(Clone, Debug, Default)] +pub struct CommandSpec { + /// Executable path or name resolvable on `PATH`. + /// + /// Passed verbatim to [`tokio::process::Command::new`]; resolution follows + /// the platform's usual `PATH` search. Required for a working spawn. + pub program: String, + + /// Arguments after [`program`](Self::program). + /// + /// Forwarded to the child in order via `Command::args`. Empty by default. + pub args: Vec, + + /// Extra environment variables for the child. + /// + /// Each `(key, value)` pair is added via `Command::env` on top of the + /// parent's environment; existing keys are overwritten. Empty by default. + pub env: Vec<(String, String)>, + + /// Working directory, or inherit the parent's. + /// + /// `None` (the default) inherits the calling process's cwd; `Some(path)` + /// sets it via `Command::current_dir`. + pub cwd: Option, +} + +impl CommandSpec { + /// Build a [`tokio::process::Command`] from this spec. + /// + /// Maps the four fields onto the equivalent `tokio::process::Command` + /// calls. Does **not** set `kill_on_drop`: rmcp's `TokioChildProcess` kills + /// the child on drop via its own cleanup guard, so a caller-set + /// `kill_on_drop` would be redundant. + fn as_tokio_command(&self) -> tokio::process::Command { + let mut cmd = tokio::process::Command::new(&self.program); + cmd.args(&self.args); + for (key, value) in &self.env { + cmd.env(key, value); + } + if let Some(cwd) = &self.cwd { + cmd.current_dir(cwd); + } + cmd + } +} + +/// How to rebuild a dropped connection for [`McpClient::reconnect`]. +/// +/// Stored on [`McpClient`] at construction time by each transport +/// constructor, so a reconnect is a faithful re-run of the original +/// connection rather than a re-derivation. In-process clients carry no +/// spec — their server lives in memory and cannot be re-established — so +/// [`McpClient::in_process`] stores `None` and reconnecting one is an +/// error. +#[derive(Clone, Debug)] +enum ReconnectSpec { + /// Re-spawn this command as a stdio child process. + /// + /// Carries the full [`CommandSpec`] so the reconnect reproduces the + /// original program, arguments, environment, and working directory of + /// the server it replaces. + Stdio(CommandSpec), + + /// Re-connect to this Streamable HTTP endpoint. + /// + /// The endpoint is reused verbatim. The client mirrors the original + /// constructor: when the caller supplied a [`reqwest::Client`] (via + /// [`McpClient::http_sse_with_client`], stored as `Some`), the reconnect + /// reuses it and preserves its pooling, TLS, and timeout configuration; + /// `None` (from [`McpClient::http_sse`]) reconnects with rmcp's default + /// client, exactly as the first connection did. + HttpSse { + /// The endpoint URL to re-connect to. + /// + /// The same `Arc` the original constructor received, kept + /// verbatim so the reconnect targets the identical server. + endpoint: Arc, + + /// The HTTP client to reconnect with, when the caller supplied one. + /// + /// `Some` retains the [`reqwest::Client`] handed to + /// [`McpClient::http_sse_with_client`]; `None` selects rmcp's + /// default client, matching [`McpClient::http_sse`]. + client: Option, + }, +} + +impl ReconnectSpec { + /// Re-run the constructor that originally built the client. + /// + /// # Errors + /// + /// Propagates [`McpError::Handshake`] from the underlying constructor + /// (`stdio`/`http_sse`). + async fn connect(&self) -> Result { + match self { + Self::Stdio(command) => McpClient::stdio(command.clone()).await, + Self::HttpSse { endpoint, client } => match client { + Some(client) => { + McpClient::http_sse_with_client(Arc::clone(endpoint), client.clone()).await + } + None => McpClient::http_sse(Arc::clone(endpoint)).await, + }, + } + } } /// Adapts one MCP server's tools as loopctl [`Tool`] implementations. @@ -256,6 +617,16 @@ pub struct McpToolProvider { /// so a refresh preserves the original namespacing without the caller /// having to pass the prefix again. prefix: Option, + + /// The per-call timeout applied to every adapted tool's `tools/call`. + /// + /// Bounds each forwarded call so a wedged server cannot hang the agent + /// loop indefinitely — a call that exceeds it resolves to a *soft* error + /// result the model can see and adapt to. Defaults to + /// [`DEFAULT_MCP_CALL_TIMEOUT`]; overridden via + /// [`with_call_timeout`](Self::with_call_timeout), which also updates + /// already-discovered tools and applies to later refreshes. + call_timeout: Duration, } impl McpToolProvider { @@ -284,14 +655,37 @@ impl McpToolProvider { /// 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?; + bridge_tool_list( + &client, + name_prefix.as_deref(), + DEFAULT_MCP_CALL_TIMEOUT, + &mut tools, + ) + .await?; Ok(Self { client, tools, prefix: name_prefix, + call_timeout: DEFAULT_MCP_CALL_TIMEOUT, }) } + /// Set the per-call timeout for every adapted tool's `tools/call`. + /// + /// Updates both already-discovered tools and the value applied to later + /// [`refresh`](Self::refresh) snapshots, so the knob takes effect + /// immediately regardless of when it is called. A call that exceeds the + /// timeout resolves to a soft error result naming the tool and the + /// budget — the run continues and the model decides how to adapt. + #[must_use] + pub fn with_call_timeout(mut self, timeout: Duration) -> Self { + self.call_timeout = timeout; + for tool in &mut self.tools { + tool.call_timeout = timeout; + } + self + } + /// Re-run `tools/list` and rebuild the tool snapshot in place. /// /// Replaces `self.tools` wholesale with a fresh discovery, re-applying the @@ -312,7 +706,13 @@ impl McpToolProvider { /// 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?; + bridge_tool_list( + &self.client, + self.prefix.as_deref(), + self.call_timeout, + &mut tools, + ) + .await?; self.tools = tools; Ok(()) } @@ -448,6 +848,14 @@ pub struct McpTool { /// polarity from `read_only_hint`). Exposed via /// [`McpTool::is_destructive_hint`] for a future permission gate. destructive_hint: bool, + + /// The per-call timeout for this tool's `tools/call` round-trip. + /// + /// Copied from the provider at discovery (or updated by + /// [`with_call_timeout`](McpToolProvider::with_call_timeout)). A call that + /// exceeds it resolves to a soft error result rather than hanging the + /// agent loop on a wedged server. + call_timeout: Duration, } impl McpTool { @@ -523,7 +931,10 @@ impl Tool for McpTool { /// [`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`. + /// a *soft* [`ToolOutput`] with `is_error` set, not as an `Err`. A call + /// that exceeds the tool's [`call_timeout`](McpTool) likewise resolves to + /// a soft error result naming the tool and the budget, so a wedged server + /// costs one tool result instead of the whole run. fn call( &self, input: serde_json::Value, @@ -531,7 +942,18 @@ impl Tool for McpTool { ) -> 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 }) + let exposed_name = self.exposed_name.clone(); + let call_timeout = self.call_timeout; + Box::pin(async move { + match tokio::time::timeout(call_timeout, client.call_tool_forward(&server_name, input)) + .await + { + Ok(result) => result, + Err(_) => Ok(ToolOutput::error_text(format!( + "MCP tool '{exposed_name}' timed out after {call_timeout:?} without a response" + ))), + } + }) } /// Conservatively `false` for every MCP tool. @@ -606,6 +1028,7 @@ pub enum McpError { async fn bridge_tool_list( client: &McpClient, prefix: Option<&str>, + call_timeout: Duration, out: &mut Vec, ) -> Result<(), McpError> { let server_tools = client @@ -615,7 +1038,7 @@ async fn bridge_tool_list( .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 { + let Some(adapted) = bridge_tool(&server_tool, prefix, client, call_timeout) else { continue; }; if !seen.insert(adapted.exposed_name.clone()) { @@ -670,6 +1093,7 @@ fn bridge_tool( server_tool: &rmcp::model::Tool, prefix: Option<&str>, client: &McpClient, + call_timeout: Duration, ) -> Option { let server_name = server_tool.name.to_string(); if server_name.is_empty() { @@ -709,6 +1133,7 @@ fn bridge_tool( client: client.clone(), read_only_hint, destructive_hint, + call_timeout, }) } diff --git a/src/provider.rs b/src/provider.rs index 27ecce1..b9a0714 100644 --- a/src/provider.rs +++ b/src/provider.rs @@ -137,11 +137,15 @@ pub(super) async fn read_bounded_body(resp: reqwest::Response) -> Result Self { @@ -297,8 +303,11 @@ impl HttpClientConfig { /// /// If an external client was supplied via /// [`with_http_client`](Self::with_http_client), it is returned verbatim. Otherwise - /// a new client is constructed with the configured timeouts, pool knobs, - /// and `tcp_nodelay(true)`. + /// a new client is constructed with a connect timeout, a read (idle-gap) + /// timeout, pool knobs, and `tcp_nodelay(true)`. No total request + /// timeout is set at this layer: generation-length budgets belong to the + /// `StreamHandler` and the engine's turn timeout, and a total HTTP cap + /// would abort healthy long streams. /// /// # Errors /// @@ -307,7 +316,7 @@ impl HttpClientConfig { match self.http { Some(shared) => Ok(shared), None => reqwest::Client::builder() - .timeout(self.timeout) + .read_timeout(self.timeout) .connect_timeout(self.connect_timeout) .tcp_nodelay(self.tcp_nodelay) .maybe_pool_max_idle_per_host(self.pool_max_idle_per_host) diff --git a/src/stream/handler.rs b/src/stream/handler.rs index 5b36eed..30dffdc 100644 --- a/src/stream/handler.rs +++ b/src/stream/handler.rs @@ -1737,7 +1737,9 @@ impl StreamHandler { .fallback_non_streaming( client, request, + &options, cancel, + total_deadline, outcome, ) .await?; @@ -2049,20 +2051,35 @@ impl StreamHandler { /// /// Called when streaming fails (timeout, retries exhausted) and /// `fallback_to_non_streaming` is enabled. Uses - /// [`ApiClient::create_message`] to get a complete typed response — the - /// message, stop reason, and token usage are returned directly, with no - /// JSON parsing at this layer. + /// [`ApiClient::create_message_with_options`] with the turn's + /// [`RequestOptions`](crate::structured::RequestOptions) — a configured + /// `response_format` or `tool_constraint` applies to the fallback exactly + /// as it did to the streaming attempt — to get a complete typed response: + /// the message, stop reason, and token usage are returned directly, with + /// no JSON parsing at this layer. + /// + /// The request is raced against the turn's `total_deadline` (the same one + /// that bounded the streaming attempts), so a hanging non-streaming call + /// cannot outlive the budget the stream already spent. The deadline + /// bounds *waiting*, not completion: a response that has resolved by the + /// time the select is polled is accepted even when it lands at or past + /// the deadline — the answer exists and its tokens are already spent, so + /// discarding it would trade finished work for wall-clock bookkeeping. + /// `None` means no deadline is configured and the call is bounded only + /// by cancellation and any client-level limits. /// /// # Errors /// - /// Returns [`StreamHandlerError::FallbackFailed`] if the fallback - /// request also fails, or [`StreamHandlerError::Cancelled`] if the - /// cancel signal fires. + /// Returns [`StreamHandlerError::FallbackFailed`] if the fallback request + /// also fails or the total deadline expires before it completes, or + /// [`StreamHandlerError::Cancelled`] if the cancel signal fires. async fn fallback_non_streaming( &self, client: &C, request: &crate::api::StreamRequest, + options: &crate::structured::RequestOptions, cancel: &Arc, + total_deadline: Option, stream_outcome: Option, ) -> Result<(Message, StreamStopReason, Option), StreamHandlerError> { if cancel.is_cancelled() { @@ -2070,10 +2087,22 @@ impl StreamHandler { } let result = tokio::select! { - res = client.create_message(request) => res, + biased; + () = cancel.notified() => { return Err(StreamHandlerError::Cancelled); } + res = client.create_message_with_options(request, options.clone()) => res, + () = deadline_future(total_deadline) => { + return Err(StreamHandlerError::FallbackFailed { + stream_outcome: stream_outcome.unwrap_or(StreamOutcome::InitFailed { + attempts: 0, + last_error: "unknown".to_string(), + }), + fallback_error: "fallback request exceeded the total stream deadline" + .to_string(), + }); + } }; match result { @@ -2123,8 +2152,9 @@ pub enum HandlerEvent { /// /// Carries the final message, stop reason, and token usage from the /// non-streaming - /// [`create_message`](crate::api::ApiClient::create_message) typed - /// response. The engine should stop accumulating and use these directly — + /// [`create_message_with_options`](crate::api::ApiClient::create_message_with_options) + /// typed response, bounded by the turn's total deadline. The engine + /// should stop accumulating and use these directly — /// the streaming accumulator's partial state from failed attempts is /// irrelevant on this path. /// @@ -2805,7 +2835,9 @@ mod tests { .fallback_non_streaming( &client, &crate::api::StreamRequest::new(vec![]), + &crate::structured::RequestOptions::default(), &cancel, + None, Some(StreamOutcome::InitFailed { last_error: "stream failed".to_string(), attempts: 3, @@ -2845,8 +2877,10 @@ mod tests { .fallback_non_streaming( &client, &crate::api::StreamRequest::new(vec![]), + &crate::structured::RequestOptions::default(), &cancel, None, + None, ) .await .expect_err("should fail on cancellation"); @@ -2857,6 +2891,207 @@ mod tests { ); } + /// Mock recording every `create_message_with_options` invocation, so the + /// fallback's options forwarding is observable. + struct OptionsRecordingMock { + seen: std::sync::Mutex>, + } + + impl ApiClient for OptionsRecordingMock { + fn model(&self) -> String { + "test-model".to_string() + } + + fn stream_messages( + &self, + _request: &crate::api::StreamRequest, + ) -> std::pin::Pin< + Box> + Send + 'static>, + > { + Box::pin(futures::stream::empty()) + } + + fn create_message( + &self, + _request: &crate::api::StreamRequest, + ) -> std::pin::Pin< + Box< + dyn std::future::Future> + + Send + + '_, + >, + > { + Box::pin(async { + Ok(crate::api::NonStreamingResponse { + message: Message::assistant("unused"), + stop_reason: crate::stream::StreamStopReason::EndTurn, + usage: Some(crate::stream::Usage::default()), + }) + }) + } + + fn create_message_with_options( + &self, + _request: &crate::api::StreamRequest, + options: crate::structured::RequestOptions, + ) -> std::pin::Pin< + Box< + dyn std::future::Future> + + Send + + '_, + >, + > { + self.seen.lock().unwrap().push(options); + Box::pin(async { + Ok(crate::api::NonStreamingResponse { + message: Message::assistant("fallback works"), + stop_reason: crate::stream::StreamStopReason::EndTurn, + usage: Some(crate::stream::Usage::default()), + }) + }) + } + } + + #[tokio::test] + async fn fallback_non_streaming_forwards_request_options() { + let handler = StreamHandler::new().with_timeout_config(StreamTimeoutConfig { + fallback_to_non_streaming: true, + ..Default::default() + }); + let client = OptionsRecordingMock { + seen: std::sync::Mutex::new(Vec::new()), + }; + let cancel = Arc::new(CancelSignal::new()); + + let mut options = crate::structured::RequestOptions::default(); + options.response_format = Some(crate::structured::ResponseFormat::new( + "probe", + serde_json::json!({"type": "object"}), + )); + + let (message, _stop, _usage) = handler + .fallback_non_streaming( + &client, + &crate::api::StreamRequest::new(vec![]), + &options, + &cancel, + None, + Some(StreamOutcome::InitFailed { + last_error: "stream failed".to_string(), + attempts: 1, + }), + ) + .await + .expect("fallback should succeed"); + + assert!( + message.text_content().contains("fallback works"), + "the options-aware response is the one used" + ); + let seen = client.seen.lock().unwrap(); + assert_eq!(seen.len(), 1, "exactly one options-aware call"); + assert!( + seen[0] + .response_format + .as_ref() + .is_some_and(|format| format.name == "probe"), + "the fallback must receive the turn's RequestOptions verbatim" + ); + } + + /// Mock whose non-streaming call never resolves — the deadline arm must + /// win. + struct HangingFallbackMock; + + impl ApiClient for HangingFallbackMock { + fn model(&self) -> String { + "test-model".to_string() + } + + fn stream_messages( + &self, + _request: &crate::api::StreamRequest, + ) -> std::pin::Pin< + Box> + Send + 'static>, + > { + Box::pin(futures::stream::empty()) + } + + fn create_message( + &self, + _request: &crate::api::StreamRequest, + ) -> std::pin::Pin< + Box< + dyn std::future::Future> + + Send + + '_, + >, + > { + Box::pin(std::future::pending()) + } + } + + #[tokio::test] + async fn fallback_non_streaming_honors_the_total_deadline() { + let handler = StreamHandler::new().with_timeout_config(StreamTimeoutConfig { + fallback_to_non_streaming: true, + ..Default::default() + }); + let cancel = Arc::new(CancelSignal::new()); + let deadline = Instant::now() + Duration::from_millis(10); + + let err = handler + .fallback_non_streaming( + &HangingFallbackMock, + &crate::api::StreamRequest::new(vec![]), + &crate::structured::RequestOptions::default(), + &cancel, + Some(deadline), + None, + ) + .await + .expect_err("a hanging fallback must be cut by the deadline"); + + match err { + StreamHandlerError::FallbackFailed { fallback_error, .. } => { + assert!( + fallback_error.contains("deadline"), + "the deadline arm must be the failure cause: {fallback_error}" + ); + } + other => panic!("expected FallbackFailed, got: {other}"), + } + } + + #[tokio::test] + async fn completed_fallback_response_racing_the_deadline_is_accepted() { + let handler = StreamHandler::new().with_timeout_config(StreamTimeoutConfig { + fallback_to_non_streaming: true, + ..Default::default() + }); + let client = HandlerMock::new().with_text_response("worth keeping"); + let cancel = Arc::new(CancelSignal::new()); + let deadline = Instant::now() + .checked_sub(Duration::from_millis(1)) + .expect("a past instant"); + + let (message, _stop_reason, _usage) = handler + .fallback_non_streaming( + &client, + &crate::api::StreamRequest::new(vec![]), + &crate::structured::RequestOptions::default(), + &cancel, + Some(deadline), + None, + ) + .await + .expect("a completed response outranks the expired deadline"); + assert!( + message.text_content().contains("worth keeping"), + "the completed response is returned, not discarded" + ); + } + #[tokio::test] async fn fallback_non_streaming_error() { let handler = StreamHandler::new().with_timeout_config(StreamTimeoutConfig { @@ -2870,7 +3105,9 @@ mod tests { .fallback_non_streaming( &client, &crate::api::StreamRequest::new(vec![]), + &crate::structured::RequestOptions::default(), &cancel, + None, Some(StreamOutcome::InitFailed { last_error: "stream timeout".to_string(), attempts: 2, diff --git a/src/tool.rs b/src/tool.rs index 0f24230..4a22d68 100644 --- a/src/tool.rs +++ b/src/tool.rs @@ -1008,17 +1008,103 @@ impl ToolError { /// [`ToolContext::get_extension`] to store and retrieve values keyed by /// their Rust type. /// -/// # Example +/// # Passing host state to tools +/// +/// Host state (a working directory, configuration, channels — anything the +/// embedding application owns that a tool needs but the model does not send) +/// reaches a tool through one of two supported options, depending on who +/// dispatches the tool. +/// +/// **Option A — the engine dispatches (`BareLoop`): a middleware injector.** +/// Each dispatch's `ToolContext` is built fresh by the engine with only +/// `session_id` set — `cwd`, `is_non_interactive`, and `extensions` all start +/// at their defaults, and host code never holds that value. The one place it +/// can be augmented is a [`ToolMiddleware`], which receives `&mut` +/// [`ToolDispatchContext`] — whose public `tool_context` field exists for +/// exactly this — before the pipeline core invokes the tool. Install with +/// [`BareLoop::set_pipeline`] (which also shares the loop's own tool registry +/// with the pipeline core), registering the injector first so later +/// middlewares see the enriched context. Without a pipeline installed, +/// engine-dispatched tools observe no host state at all. +/// +/// **Option B — the host dispatches: build the context yourself.** +/// When your code calls [`Tool::call`] directly (tests, scripts, simple +/// integrations), construct the `ToolContext`, set fields and extensions, and +/// pass it in — you own the value end to end, so nothing else is required. +/// +/// Also possible, but discouraged: registering a wrapper tool that clones the +/// incoming context, installs the extension, and delegates (works because +/// `ToolContext` is `Clone`, but costs per-tool wiring plus a `Tool`-trait +/// forwarding layer), and capturing state in the tool struct at construction +/// (which bypasses the context entirely). Prefer the middleware — it is the +/// same idea at the engine's single sanctioned interception point. The +/// `host-state` example in the repository's `examples/` directory demonstrates +/// both options end to end. +/// +/// [`BareLoop`]: crate::engine::BareLoop +/// [`BareLoop::set_pipeline`]: crate::engine::BareLoop::set_pipeline +/// [`Tool::call`]: crate::tool::Tool::call +/// [`ToolDispatchContext`]: crate::middleware::ToolDispatchContext +/// [`ToolDispatchContext::tool_context`]: crate::middleware::ToolDispatchContext::tool_context +/// [`ToolMiddleware`]: crate::middleware::ToolMiddleware +/// +/// # Example — Option B: host-built context (runnable) +/// +/// ``` +/// use loopctl::tool::ToolContext; +/// +/// #[derive(Clone)] +/// struct MyConfig { +/// verbose: bool, +/// } /// -/// ```rust,ignore /// let mut ctx = ToolContext::default(); -/// ctx.cwd = "/tmp/workspace".into(); /// ctx.set_extension(MyConfig { verbose: true }); +/// assert!(ctx.get_extension::().expect("just set").verbose); +/// ``` /// -/// // Inside a tool: -/// if let Some(cfg) = ctx.get_extension::() { -/// println!("verbose={}", cfg.verbose); +/// # Example — Option A: middleware injector (compiles, does not run) +/// +/// ```rust,no_run +/// use std::future::Future; +/// use std::pin::Pin; +/// use std::sync::Arc; +/// use loopctl::config::SessionConfig; +/// use loopctl::engine::BareLoop; +/// use loopctl::middleware::ToolDispatchContext; +/// use loopctl::middleware::ToolDispatchResult; +/// use loopctl::middleware::ToolMiddleware; +/// use loopctl::middleware::ToolPipeline; +/// # use loopctl::testing::MockApiClient; +/// # use loopctl::tool::ToolRegistry; +/// +/// #[derive(Clone)] +/// struct MyConfig { +/// verbose: bool, /// } +/// +/// struct Injector; +/// +/// impl ToolMiddleware for Injector { +/// fn name(&self) -> &str { +/// "my-injector" +/// } +/// fn dispatch<'a>( +/// &'a self, +/// ctx: &'a mut ToolDispatchContext, +/// next: &'a ToolPipeline, +/// ) -> Pin + Send + 'a>> { +/// ctx.tool_context.set_extension(MyConfig { verbose: true }); +/// Box::pin(async move { next.dispatch(ctx).await }) +/// } +/// } +/// +/// let client = MockApiClient::new("demo"); +/// let mut agent = +/// BareLoop::new(Arc::new(client), ToolRegistry::new(), SessionConfig::default()); +/// agent +/// .set_pipeline(ToolPipeline::builder().with_middleware(Injector)) +/// .expect("static pipeline composition"); /// ``` #[derive(Clone)] pub struct ToolContext { diff --git a/tests/mcp_tool_provider.rs b/tests/mcp_tool_provider.rs index 16b3987..acd1801 100644 --- a/tests/mcp_tool_provider.rs +++ b/tests/mcp_tool_provider.rs @@ -708,3 +708,99 @@ async fn absent_destructive_hint_defaults_to_destructive_per_spec() { "explicit destructiveHint=false must be honored" ); } + +/// A server whose only tool sleeps before answering — long relative to the +/// test's shortened call timeout, short relative to the default. +#[derive(Clone)] +struct SlowServer { + router: ToolRouter, +} + +#[tool_router] +impl SlowServer { + fn new() -> Self { + Self { + router: Self::tool_router(), + } + } + + #[tool(description = "Sleeps before answering")] + async fn slow(&self) -> String { + tokio::time::sleep(Duration::from_millis(150)).await; + "finally".to_string() + } +} + +#[tool_handler] +impl ServerHandler for SlowServer {} + +#[tokio::test] +async fn call_timeout_cuts_a_slow_tool_with_a_soft_error() { + let (client, _server) = connect_in_process(SlowServer::new()).await; + let provider = McpToolProvider::connect(client, None) + .await + .expect("connect") + .with_call_timeout(Duration::from_millis(10)); + let slow = provider + .tools() + .iter() + .find(|t| t.name() == "slow") + .expect("slow tool"); + let ctx = ToolContext::default(); + + let out = slow + .call(json!({}), &ctx) + .await + .expect("timeout is a soft error"); + assert!(out.is_error, "the timeout must surface as is_error"); + let text = out.text_content(); + assert!( + text.contains("slow") && text.contains("timed out"), + "the soft error must name the tool and the timeout: {text:?}" + ); +} + +#[tokio::test] +async fn default_timeout_lets_a_quick_tool_through() { + let (client, _server) = connect_in_process(SlowServer::new()).await; + let provider = McpToolProvider::connect(client, None) + .await + .expect("connect"); + let slow = provider + .tools() + .iter() + .find(|t| t.name() == "slow") + .expect("slow tool"); + let ctx = ToolContext::default(); + + let out = slow + .call(json!({}), &ctx) + .await + .expect("call ok under the default timeout"); + assert!(!out.is_error, "a 150ms call must survive the 60s default"); + assert_eq!(out.text_content(), "finally"); +} + +#[tokio::test] +async fn refresh_preserves_the_overridden_call_timeout() { + let (client, _server) = connect_in_process(SlowServer::new()).await; + let mut provider = McpToolProvider::connect(client, None) + .await + .expect("connect") + .with_call_timeout(Duration::from_millis(10)); + + provider.refresh().await.expect("refresh"); + let slow = provider + .tools() + .iter() + .find(|t| t.name() == "slow") + .expect("slow tool after refresh"); + let ctx = ToolContext::default(); + + let out = slow.call(json!({}), &ctx).await.expect("call resolves"); + assert!( + out.is_error && out.text_content().contains("timed out"), + "the refreshed tool must still be bounded by the override: {}", + out.text_content() + ); +} diff --git a/tests/mcp_transports.rs b/tests/mcp_transports.rs new file mode 100644 index 0000000..169940a --- /dev/null +++ b/tests/mcp_transports.rs @@ -0,0 +1,275 @@ +//! Integration tests for the MCP transports (stdio + Streamable HTTP/SSE). +//! +//! **stdio** tests are real-subprocess: they spawn the loopctl-shipped +//! `examples/mcp-stdio-server` binary (an rmcp `#[tool_router]` server over +//! stdio), so they run in CI with no external runtime dependency. +//! +//! **HTTP** tests are `#[ignore]` + `LOOPCTL_MCP_E2E=1`: they spawn an official +//! Python/TypeScript SDK streamable-http server, which CI lacks. A developer +//! opts in with `LOOPCTL_MCP_E2E=1 cargo test --features mcp -- --ignored`. + +#![cfg(feature = "mcp")] +#![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::{CommandSpec, McpClient, McpError, McpToolProvider}; +use loopctl::stream::handler::StreamRetryConfig; + +/// Path to the loopctl-shipped stdio example server binary. +/// +/// Derived from the test binary's own location: `cargo test` places the test +/// binary in `target//deps/` and example binaries in +/// `target//examples/`, so walking two levels up from +/// `current_exe()` lands on the profile directory regardless of `--release` +/// or a custom `CARGO_TARGET_DIR`. +fn stdio_server_bin() -> String { + let exe = std::env::current_exe().expect("test binary path"); + let profile_dir = exe + .parent() + .and_then(|deps| deps.parent()) + .expect("profile directory above the test binary's deps directory"); + let name = if cfg!(windows) { + "mcp-stdio-server.exe" + } else { + "mcp-stdio-server" + }; + profile_dir + .join("examples") + .join(name) + .to_string_lossy() + .to_string() +} + +/// A `CommandSpec` pointing at the loopctl stdio example server. +fn stdio_server_spec() -> CommandSpec { + CommandSpec { + program: stdio_server_bin(), + args: vec![], + env: vec![], + cwd: None, + } +} + +/// Discover tools via the public `McpToolProvider` API. +async fn discover(client: McpClient) -> Vec { + let provider = McpToolProvider::connect(client, None) + .await + .expect("connect + list_tools"); + provider + .tools() + .iter() + .map(loopctl::tool::Tool::name) + .map(str::to_owned) + .collect() +} + +/// A `StreamRetryConfig` tightened for fast tests: 2 retries, ~1ms base. +fn fast_retry() -> StreamRetryConfig { + StreamRetryConfig { + max_retries: 2, + base_delay_ms: 1, + max_delay_ms: 5, + jitter_factor: 0.0, + } +} + +#[tokio::test] +async fn stdio_discovers_tools_from_child() { + let client = McpClient::stdio(stdio_server_spec()) + .await + .expect("stdio connect"); + let names = discover(client).await; + assert_eq!(names, vec!["greet"], "the example server exposes one tool"); +} + +#[tokio::test] +async fn stdio_spawn_failure_is_handshake_error() { + // A nonexistent program: TokioChildProcess::new returns io::Error. + let spec = CommandSpec { + program: "/nonexistent/mcp-server-binary-xyz".into(), + args: vec![], + env: vec![], + cwd: None, + }; + let err = McpClient::stdio(spec).await.expect_err("spawn must fail"); + assert!(matches!(err, McpError::Handshake(_)), "got {err:?}"); +} + +#[tokio::test] +async fn stdio_command_spec_env_is_applied() { + // CommandSpec carries env vars to the child. The example server ignores + // unknown env, so we only assert the spawn + handshake succeeds with an + // extra var set — proving env was forwarded without breaking the child. + let spec = CommandSpec { + program: stdio_server_bin(), + args: vec![], + env: vec![("LOOPCTL_TEST_MARKER".into(), "present".into())], + cwd: None, + }; + let client = McpClient::stdio(spec).await.expect("connect with env"); + let names = discover(client).await; + assert_eq!(names, vec!["greet"]); +} + +#[tokio::test] +async fn stdio_drop_does_not_hang() { + // rmcp's TokioChildProcess kills the child on drop. This test exercises the + // drop path end-to-end and confirms it completes promptly (no hang waiting + // on a leaked child). The reconnect test below proves the child is gone. + let client = McpClient::stdio(stdio_server_spec()) + .await + .expect("connect"); + tokio::time::timeout(Duration::from_secs(2), async move { drop(client) }) + .await + .expect("drop completes within 2s"); +} + +#[tokio::test] +async fn reconnect_in_process_client_is_error() { + // in_process clients have no reconnect_spec — reconnect must reject. + use rmcp::handler::server::ServerHandler; + use rmcp::{ServiceExt, tool, tool_handler, tool_router}; + #[derive(Clone)] + struct S { + router: rmcp::handler::server::router::tool::ToolRouter, + } + #[tool_router] + impl S { + fn new() -> Self { + Self { + router: Self::tool_router(), + } + } + #[tool(description = "noop")] + async fn noop(&self) -> String { + "ok".into() + } + } + #[tool_handler] + impl ServerHandler for S {} + + let (server_end, client_end) = tokio::io::duplex(4096); + tokio::spawn(async move { + if let Ok(r) = S::new().serve(server_end).await { + let _ = r.waiting().await.ok(); + } + }); + let client = ().serve(client_end).await.map(McpClient::from_service).expect("connect"); + let err = client + .reconnect(&fast_retry()) + .await + .expect_err("in_process cannot reconnect"); + assert!(matches!(err, McpError::Handshake(_)), "got {err:?}"); +} + +#[tokio::test] +async fn reconnect_stdio_re_establishes_after_drop() { + // Build a client (child spawned), drop it (child dies per rmcp cleanup), + // then reconnect on the dead client's retained spec — the spec re-spawns + // the same binary and rediscovers the tool. + // + // This covers reconnect's happy path (spec read, constructor re-run, + // success returned). The give-up-after-max-retries path is structurally + // the same loop with a failing constructor; it is not tested directly + // because `ReconnectSpec` is private and a guaranteed-failing spec cannot + // be attached to a client via the public API (the constructor must + // succeed first to produce a client, and a failing constructor produces + // no client to call `reconnect` on). The loop bound `0..=max_retries` is + // simple enough to read; the per-attempt logic is exercised here. + let spec = stdio_server_spec(); + let live = McpClient::stdio(spec).await.expect("first connect"); + let dead = live.clone(); + drop(live); + // Give rmcp's async child-kill a moment to complete. + tokio::time::sleep(Duration::from_millis(100)).await; + + let reconnected = dead + .reconnect(&fast_retry()) + .await + .expect("reconnect re-spawns the child"); + let names = discover(reconnected).await; + assert_eq!( + names, + vec!["greet"], + "reconnected client rediscovers the tool" + ); +} + +fn e2e_enabled() -> bool { + std::env::var("LOOPCTL_MCP_E2E").is_ok_and(|v| v == "1") +} + +#[tokio::test] +#[ignore = "requires LOOPCTL_MCP_E2E=1 and an official SDK streamable-http server on 127.0.0.1:3001"] +async fn http_sse_round_trip_against_local_sdk_server() { + if !e2e_enabled() { + eprintln!("skipped: set LOOPCTL_MCP_E2E=1 to run the live HTTP smoke"); + return; + } + let client = McpClient::http_sse("http://127.0.0.1:3001/mcp") + .await + .expect("http connect"); + let names = discover(client).await; + assert!(!names.is_empty(), "server advertised tools"); +} + +#[tokio::test] +#[ignore = "requires LOOPCTL_MCP_E2E=1 and an official SDK streamable-http server on 127.0.0.1:3001"] +async fn http_sse_with_client_round_trips() { + if !e2e_enabled() { + eprintln!("skipped: set LOOPCTL_MCP_E2E=1 to run the live HTTP smoke"); + return; + } + let req_client = reqwest::Client::builder().build().expect("reqwest client"); + let client = McpClient::http_sse_with_client("http://127.0.0.1:3001/mcp", req_client) + .await + .expect("http connect"); + let names = discover(client).await; + assert!(!names.is_empty(), "server advertised tools"); +} + +/// An endpoint whose single connection is accepted and then severed. +/// +/// Binds a loopback listener on an OS-assigned port, accepts exactly one +/// connection on a background thread, and closes it — so a client dialing +/// the endpoint deterministically sees the connection drop mid-handshake, +/// with no dependence on how any well-known port behaves on this host. +fn severed_endpoint() -> String { + let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind loopback"); + let addr = listener.local_addr().expect("assigned port"); + std::thread::spawn(move || { + if let Ok((conn, _)) = listener.accept() { + drop(conn); + } + }); + format!("http://{addr}/mcp") +} + +#[tokio::test] +async fn http_sse_connect_refused_is_handshake_error() { + let err = McpClient::http_sse(severed_endpoint()) + .await + .expect_err("severed connection must fail the handshake"); + assert!(matches!(err, McpError::Handshake(_)), "got {err:?}"); +} + +#[tokio::test] +async fn http_sse_with_client_connect_refused_is_handshake_error() { + let req_client = reqwest::Client::builder().build().expect("reqwest client"); + let err = McpClient::http_sse_with_client(severed_endpoint(), req_client) + .await + .expect_err("severed connection must fail the handshake"); + assert!(matches!(err, McpError::Handshake(_)), "got {err:?}"); +}