Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 7 additions & 1 deletion CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -9,14 +9,18 @@ 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.
- `parallel_hard_error_discards_sibling_results` test — pins the documented contract that a hard error in a parallel wave aborts the batch and discards already-completed sibling results.

### 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).
Expand All @@ -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.
Expand Down
12 changes: 9 additions & 3 deletions Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -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"] }
Expand Down Expand Up @@ -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"
Expand All @@ -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"
Expand Down
2 changes: 1 addition & 1 deletion README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down
58 changes: 58 additions & 0 deletions examples/mcp-stdio-server.rs
Original file line number Diff line number Diff line change
@@ -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<Self>,
}

#[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();
}
22 changes: 18 additions & 4 deletions src/engine/bare/config.rs
Original file line number Diff line number Diff line change
Expand Up @@ -284,16 +284,30 @@ impl<C: ApiClient> BareLoop<C> {
/// 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
Expand Down
5 changes: 4 additions & 1 deletion src/managers.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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);
}
Expand Down
Loading