diff --git a/crates/rmcp-macros/src/lib.rs b/crates/rmcp-macros/src/lib.rs index 573fdd28c..44bab6f4d 100644 --- a/crates/rmcp-macros/src/lib.rs +++ b/crates/rmcp-macros/src/lib.rs @@ -33,6 +33,47 @@ mod tool_router; /// // handling tool request /// } /// ``` +/// +/// ## Async tool bodies must be `Send` +/// +/// An `async fn` is rewritten into +/// `fn(..) -> Pin + Send + '_>>`, because the router keeps +/// every handler as a `Send` boxed future. Everything the tool awaits therefore has to +/// be `Send`. +/// +/// One shape fails that check even though the code is sound. A closure that takes a +/// reference and returns an `async move` block, fed into a `futures` combinator that +/// buffers the futures, does not compile: +/// +/// ```rust,ignore +/// stream::iter(items.iter()) +/// .map(|item| async move { work(item).await }) +/// .buffer_unordered(4) +/// ``` +/// +/// ```text +/// error: implementation of `FnOnce` is not general enough +/// = note: closure with signature `fn(&'0 String) -> {async block}` +/// must implement `FnOnce<(&'1 String,)>`, for any two lifetimes `'0` and `'1`... +/// = note: ...but it actually implements `FnOnce<(&String,)>` +/// ``` +/// +/// The span points at `#[tool]`, which is misleading. This is a rustc limitation around +/// `Send` checking of async blocks, not something the macro introduces: the same code +/// fails under `tokio::spawn`, under `-> impl Send`, or under any `Send`-bounded generic, +/// with no macro involved. See [rust-lang/rust#104382] and [rust-lang/rust#126551]. +/// +/// Ways round it, least disruptive first: +/// +/// - add `.boxed()` before `.buffer_unordered(..)` or `.buffered(..)`; +/// - own the items instead of borrowing them (`items.into_iter()`); +/// - if the futures do not need to run at the same time, use `.then(..)` instead, which +/// runs them one at a time; +/// - if the server does not have to be `Send`, the `local` crate feature drops the +/// `+ Send` bound from the generated signature. +/// +/// [rust-lang/rust#104382]: https://github.com/rust-lang/rust/issues/104382 +/// [rust-lang/rust#126551]: https://github.com/rust-lang/rust/issues/126551 #[proc_macro_attribute] pub fn tool(attr: TokenStream, input: TokenStream) -> TokenStream { tool::tool(attr.into(), input.into()) @@ -251,6 +292,16 @@ pub fn tool_handler(attr: TokenStream, input: TokenStream) -> TokenStream { /// // Generate prompt messages based on arguments /// } /// ``` +/// +/// ## Async prompt bodies must be `Send` +/// +/// An `async fn` is rewritten into +/// `fn(..) -> Pin + Send + '_>>`, the same as [`macro@tool`], +/// so the same `Send` limitation applies: a closure taking a reference and returning an +/// `async move` block, fed into `.buffer_unordered(..)` or `.buffered(..)`, fails to +/// compile with "implementation of `FnOnce` is not general enough" even though the code +/// is sound. See the matching section on [`macro@tool`] for the error text and the ways +/// round it. #[proc_macro_attribute] pub fn prompt(attr: TokenStream, input: TokenStream) -> TokenStream { prompt::prompt(attr.into(), input.into()) diff --git a/crates/rmcp/tests/test_prompt_macros.rs b/crates/rmcp/tests/test_prompt_macros.rs index 1ee864549..83964ba25 100644 --- a/crates/rmcp/tests/test_prompt_macros.rs +++ b/crates/rmcp/tests/test_prompt_macros.rs @@ -387,3 +387,54 @@ async fn test_optional_i64_field_with_null_input() -> anyhow::Result<()> { server_handle.await??; Ok(()) } + +/// Regression cover for . +/// +/// `#[prompt]` rewrites an `async fn` into `-> Pin + Send + '_>>` +/// exactly as `#[tool]` does, so it carries the same `Send` limitation: a closure taking a +/// reference and returning an `async move` block, buffered by a `futures` combinator, does +/// not compile without boxing the stream first (rust-lang/rust#104382). +#[derive(Debug, Clone)] +struct BufferedStreamPromptServer; + +impl BufferedStreamPromptServer { + /// Fan out over borrowed items, boxing the stream so the closure drops out of the + /// `Send` computation. + async fn fan_out_borrowed(&self, items: &[String]) -> Vec { + use futures::stream::{self, StreamExt}; + stream::iter(items.iter()) + .map(|item| async move { item.to_uppercase() }) + .boxed() + .buffer_unordered(4) + .collect() + .await + } +} + +#[prompt_router] +impl BufferedStreamPromptServer { + /// Build a message from a fixed list run through a boxed, buffered stream. + #[prompt(name = "buffered_stream", description = "Fan out over borrowed items")] + async fn buffered_stream(&self) -> Vec { + let items = vec!["a".to_string(), "b".to_string()]; + let mut out = self.fan_out_borrowed(&items).await; + out.sort(); + vec![PromptMessage::new_text(Role::User, out.join(","))] + } +} + +#[prompt_handler] +impl ServerHandler for BufferedStreamPromptServer {} + +#[tokio::test] +async fn test_async_prompt_with_buffered_stream() { + let server = BufferedStreamPromptServer; + let messages = server.buffered_stream().await; + match &messages.first().unwrap().content { + ContentBlock::Text(text) => assert_eq!(text.text, "A,B"), + _ => panic!("Expected text content"), + } + + let router = BufferedStreamPromptServer::prompt_router(); + assert!(router.has_route("buffered_stream")); +} diff --git a/crates/rmcp/tests/test_tool_macros.rs b/crates/rmcp/tests/test_tool_macros.rs index 6f9aac634..4862a7be4 100644 --- a/crates/rmcp/tests/test_tool_macros.rs +++ b/crates/rmcp/tests/test_tool_macros.rs @@ -634,3 +634,73 @@ impl EmptyRouterServer {} fn test_allow_empty_builds_a_router_without_tools() { assert!(EmptyRouterServer::tool_router().list_all().is_empty()); } + +/// Regression cover for . +/// +/// An `async fn` tool is rewritten into `-> Pin + Send + '_>>`, +/// so the whole call tree has to pass the `Send` check. A closure taking a reference and +/// returning an `async move` block, buffered by a `futures` combinator, fails that check +/// on current rustc (rust-lang/rust#104382). Both shapes below are the documented ways +/// round it; this test exists so they keep compiling. +#[derive(Debug, Clone)] +struct BufferedStreamServer; + +impl BufferedStreamServer { + /// Fan out over borrowed items, boxing the stream so the closure drops out of the + /// `Send` computation. + async fn fan_out_borrowed(&self, items: &[String]) -> Vec { + use futures::stream::{self, StreamExt}; + stream::iter(items.iter()) + .map(|item| async move { item.to_uppercase() }) + .boxed() + .buffer_unordered(4) + .collect() + .await + } + + /// Fan out over owned items, so the closure argument is not a reference at all. + async fn fan_out_owned(&self, items: Vec) -> Vec { + use futures::stream::{self, StreamExt}; + stream::iter(items) + .map(|item| async move { item.to_uppercase() }) + .buffer_unordered(4) + .collect() + .await + } +} + +#[tool_router] +impl BufferedStreamServer { + /// Uppercase a fixed list through a boxed, buffered stream over borrowed items. + #[tool(description = "Fan out over borrowed items")] + async fn borrowed(&self) -> String { + let items = vec!["a".to_string(), "b".to_string()]; + let mut out = self.fan_out_borrowed(&items).await; + out.sort(); + out.join(",") + } + + /// Uppercase a fixed list through a buffered stream over owned items. + #[tool(description = "Fan out over owned items")] + async fn owned(&self) -> String { + let mut out = self + .fan_out_owned(vec!["c".to_string(), "d".to_string()]) + .await; + out.sort(); + out.join(",") + } +} + +#[tool_handler] +impl ServerHandler for BufferedStreamServer {} + +#[tokio::test] +async fn test_async_tool_with_buffered_stream() { + let server = BufferedStreamServer; + assert_eq!(server.borrowed().await, "A,B"); + assert_eq!(server.owned().await, "C,D"); + + let router = BufferedStreamServer::tool_router(); + assert!(router.has_route("borrowed")); + assert!(router.has_route("owned")); +}