Merge pull request #88 from nikomatsakis/main

feat(sacp)!: merge pending_tasks into JrResponder and remove 'scope lifetime

Author: nikomatsakis

src/elizacp/src/lib.rs

@@ -117,15 +117,15 @@ impl ElizaAgent {
     ) -> Result<(), sacp::Error> {
         let session_id = request.session_id.clone();
 
+        // Extract text from the prompt
+        let input_text = extract_text_from_prompt(&request.prompt);
+
         tracing::debug!(
-            "Processing prompt in session {}: {} content blocks",
+            "Processing prompt in session {}: {input_text:?} over {} content blocks",
             session_id,
             request.prompt.len()
         );
 
-        // Extract text from the prompt
-        let input_text = extract_text_from_prompt(&request.prompt);
-
         // Check for MCP commands first before invoking Eliza
         let final_response = if let Some(server_name) = parse_list_tools_command(&input_text) {
             // List tools from a specific server

src/sacp-conductor/src/conductor.rs

@@ -191,7 +191,10 @@ impl Conductor {
         self
     }
 
-    pub fn into_connection_builder(self) -> JrConnectionBuilder<ConductorMessageHandler> {
+    pub fn into_connection_builder(
+        self,
+    ) -> JrConnectionBuilder<ConductorMessageHandler, impl sacp::JrResponder<ConductorToClient>>
+    {
         let (mut conductor_tx, mut conductor_rx) = mpsc::channel(128 /* chosen arbitrarily */);
 
         let mut state = ConductorHandlerState {

src/sacp-conductor/tests/mcp_integration/proxy.rs

@@ -34,7 +34,6 @@ impl Component for ProxyComponent {
                         result: format!("Echo: {}", params.message),
                     })
                 },
-                sacp::tool_fn!(),
             )
             .build();
 

src/sacp-conductor/tests/mcp_server_handler_chain.rs

@@ -82,7 +82,6 @@ impl Component for ProxyWithMcpAndHandler {
                         result: format!("Echo: {}", params.message),
                     })
                 },
-                sacp::tool_fn!(),
             )
             .build();
 

src/sacp-conductor/tests/scoped_mcp_server.rs

@@ -0,0 +1,130 @@
+//! Test that MCP servers can reference stack-local data.
+//!
+//! This test demonstrates the new scoped lifetime feature where an MCP tool
+//! can capture references to stack-local data (like a Vec) and push to it
+//! when the tool is invoked.
+
+use elizacp::ElizaAgent;
+use sacp::mcp_server::McpServer;
+use sacp::{Agent, ClientToAgent, Component, DynComponent, HasEndpoint, JrRole, ProxyToConductor};
+use sacp_conductor::{Conductor, McpBridgeMode};
+use schemars::JsonSchema;
+use serde::{Deserialize, Serialize};
+use std::sync::Mutex;
+
+/// Test that an MCP tool can push to a stack-local vector.
+///
+/// This validates the scoped lifetime feature - the tool closure captures
+/// a reference to `collected_values` which lives on the stack.
+#[tokio::test]
+async fn test_scoped_mcp_server_through_proxy() -> Result<(), sacp::Error> {
+    let conductor = Conductor::new(
+        "conductor".to_string(),
+        vec![
+            DynComponent::new(ScopedProxy),
+            DynComponent::new(ElizaAgent::new()),
+        ],
+        Default::default(),
+    );
+
+    let result = yopo::prompt(
+        conductor,
+        r#"Use tool test::push with {"elements": ["Hello", "world"]}"#,
+    )
+    .await?;
+
+    expect_test::expect![[r#"
+        "OK: CallToolResult { content: [Annotated { raw: Text(RawTextContent { text: \"2\", meta: None }), annotations: None }], structured_content: Some(Number(2)), is_error: Some(false), meta: None }"
+    "#]].assert_debug_eq(&result);
+
+    Ok(())
+}
+
+/// Test that an MCP tool can push to a stack-local vector through a session.
+///
+/// This validates the scoped lifetime feature with session-scoped MCP servers.
+/// The MCP server captures a reference to stack-local data that lives for
+/// the duration of the session.
+#[tokio::test]
+async fn test_scoped_mcp_server_through_session() -> Result<(), sacp::Error> {
+    ClientToAgent::builder()
+        .connect_to(Conductor::new("conductor".to_string(), vec![ElizaAgent::new()], McpBridgeMode::default()))?
+        .with_client(async |cx| {
+            // Initialize first
+            cx.send_request(sacp::schema::InitializeRequest {
+                protocol_version: Default::default(),
+                client_capabilities: Default::default(),
+                client_info: None,
+                meta: None,
+            })
+            .block_task()
+            .await?;
+
+            let collected_values = Mutex::new(Vec::new());
+            let result = cx
+                .build_session(".")
+                .with_mcp_server(make_mcp_server(&collected_values))?
+                .run_session(async |mut active_session| {
+                    active_session
+                        .send_prompt(r#"Use tool test::push with {"elements": ["Hello", "world"]}"#)?;
+                    active_session.read_to_string().await
+                })
+                .await?;
+
+            expect_test::expect![[r#"
+                "OK: CallToolResult { content: [Annotated { raw: Text(RawTextContent { text: \"2\", meta: None }), annotations: None }], structured_content: Some(Number(2)), is_error: Some(false), meta: None }"
+            "#]].assert_debug_eq(&result);
+
+            Ok(())
+        }).await?;
+
+    Ok(())
+}
+
+struct ScopedProxy;
+
+fn make_mcp_server<Role: JrRole>(
+    values: &Mutex<Vec<String>>,
+) -> McpServer<Role, impl sacp::JrResponder<Role> + use<'_, Role>>
+where
+    Role: HasEndpoint<Agent>,
+{
+    #[derive(Serialize, Deserialize, JsonSchema)]
+    struct PushInput {
+        elements: Vec<String>,
+    }
+
+    McpServer::builder("test".to_string())
+        .instructions("A test MCP server with scoped tool")
+        .tool_fn(
+            "push",
+            "Push a value to the collected values",
+            async |input: PushInput, _cx| {
+                let mut values = values.lock().expect("not poisoned");
+                values.extend(input.elements);
+                Ok(values.len())
+            },
+        )
+        .tool_fn("get", "Get the collected values", async |(): (), _cx| {
+            let values = values.lock().expect("not poisoned");
+            Ok(values.clone())
+        })
+        .build()
+}
+
+impl Component for ScopedProxy {
+    async fn serve(self, client: impl Component) -> Result<(), sacp::Error> {
+        // Stack-local data that the MCP tool will push to
+        let values: Mutex<Vec<String>> = Mutex::new(Vec::new());
+
+        // Build the MCP server that captures a reference to collected_values
+        let mcp_server = make_mcp_server(&values);
+
+        ProxyToConductor::builder()
+            .name("scoped-mcp-server")
+            .with_mcp_server(mcp_server)
+            .connect_to(client)?
+            .serve()
+            .await
+    }
+}

src/sacp-conductor/tests/test_session_id_in_mcp_tools.rs

@@ -40,23 +40,22 @@ fn create_echo_proxy() -> Result<sacp::DynComponent, sacp::Error> {
                     acp_url: context.acp_url(),
                 })
             },
-            sacp::tool_fn!(),
         )
         .build();
 
     // Create proxy component
     Ok(sacp::DynComponent::new(ProxyWithEchoServer { mcp_server }))
 }
 
-struct ProxyWithEchoServer {
-    mcp_server: McpServer<ProxyToConductor>,
+struct ProxyWithEchoServer<R: sacp::JrResponder<ProxyToConductor>> {
+    mcp_server: McpServer<ProxyToConductor, R>,
 }
 
-impl Component for ProxyWithEchoServer {
+impl<R: sacp::JrResponder<ProxyToConductor> + 'static> Component for ProxyWithEchoServer<R> {
     async fn serve(self, client: impl Component) -> Result<(), sacp::Error> {
         ProxyToConductor::builder()
             .name("echo-proxy")
-            .with_handler(self.mcp_server)
+            .with_mcp_server(self.mcp_server)
             .serve(client)
             .await
     }

src/sacp-rmcp/examples/with_mcp_server.rs

@@ -92,7 +92,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     ProxyToConductor::builder()
         .name("mcp-server-proxy")
         // Register the MCP server as a handler
-        .with_handler(mcp_server)
+        .with_mcp_server(mcp_server)
         // Start serving
         .connect_to(sacp::ByteStreams::new(
             tokio::io::stdout().compat_write(),

src/sacp-rmcp/src/lib.rs

@@ -22,7 +22,7 @@
 
 use rmcp::ServiceExt;
 use sacp::mcp_server::{McpContext, McpServer, McpServerConnect};
-use sacp::{Agent, ByteStreams, Component, DynComponent, HasEndpoint, JrRole};
+use sacp::{Agent, ByteStreams, Component, DynComponent, HasEndpoint, JrRole, NullResponder};
 use tokio_util::compat::{TokioAsyncReadCompatExt, TokioAsyncWriteCompatExt};
 
 pub trait McpServerExt<Role: JrRole>
@@ -37,7 +37,7 @@ where
     fn from_rmcp<S>(
         name: impl ToString,
         new_fn: impl Fn() -> S + Send + Sync + 'static,
-    ) -> McpServer<Role>
+    ) -> McpServer<Role, NullResponder>
     where
         S: rmcp::Service<rmcp::RoleServer>,
     {
@@ -62,10 +62,13 @@ where
             }
         }
 
-        McpServer::new(RmcpServer {
-            name: name.to_string(),
-            new_fn,
-        })
+        McpServer::new(
+            RmcpServer {
+                name: name.to_string(),
+                new_fn,
+            },
+            NullResponder,
+        )
     }
 }