feat(sacp)!: add scoped lifetime support for MCP servers

MCP servers can now capture references to stack-local data, enabling
tools that interact with data scoped to a session or proxy lifetime.

Key changes:

- McpServer<'scope, Role>: MCP servers now have a lifetime parameter
  allowing tool closures to capture non-'static references

- SessionBuilder<'scope, Role>: Sessions can now use scoped MCP servers
  via the new run_session() method which races the session operation
  against the MCP server's tool loop future

- SessionBuilder::with_mcp_server now takes ownership of McpServer
  instead of a reference, accumulating tool futures

- New run_session() API provides scoped session execution where MCP
  server futures are automatically cancelled when the session ends

- send_request() still available for 'static MCP servers, spawning
  the session in a background task

- Added ActiveSession::read_to_string() convenience method

- Added run_until() and both() future combinators for cleaner async code

- Added test demonstrating scoped MCP servers capturing stack-local
  Mutex<Vec<String>>

BREAKING CHANGE: SessionBuilder::with_mcp_server signature changed from
taking &McpServer<Role> to taking McpServer<'scope, Role> by value.
McpServer now requires a lifetime parameter.

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/tests/scoped_mcp_server.rs

@@ -0,0 +1,134 @@
+//! 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>
+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())
+            },
+            sacp::tool_fn!(),
+        )
+        .tool_fn(
+            "get",
+            "Get the collected values",
+            async |(): (), _cx| {
+                let values = values.lock().expect("not poisoned");
+                Ok(values.clone())
+            },
+            sacp::tool_fn!(),
+        )
+        .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

@@ -49,14 +49,14 @@ fn create_echo_proxy() -> Result<sacp::DynComponent, sacp::Error> {
 }
 
 struct ProxyWithEchoServer {
-    mcp_server: McpServer<ProxyToConductor>,
+    mcp_server: McpServer<'static, ProxyToConductor>,
 }
 
 impl Component for ProxyWithEchoServer {
     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

@@ -37,7 +37,7 @@ where
     fn from_rmcp<S>(
         name: impl ToString,
         new_fn: impl Fn() -> S + Send + Sync + 'static,
-    ) -> McpServer<Role>
+    ) -> McpServer<'static, Role>
     where
         S: rmcp::Service<rmcp::RoleServer>,
     {
@@ -62,14 +62,17 @@ where
             }
         }
 
-        McpServer::new(RmcpServer {
-            name: name.to_string(),
-            new_fn,
-        })
+        McpServer::new(
+            RmcpServer {
+                name: name.to_string(),
+                new_fn,
+            },
+            Box::pin(futures::future::ready(Ok(()))),
+        )
     }
 }
 
-impl<Role: JrRole> McpServerExt<Role> for McpServer<Role> where Role: HasEndpoint<Agent> {}
+impl<Role: JrRole> McpServerExt<Role> for McpServer<'_, Role> where Role: HasEndpoint<Agent> {}
 
 /// Component wrapper for rmcp services.
 struct RmcpServerComponent<S> {

src/sacp/src/jsonrpc.rs

@@ -30,7 +30,7 @@ pub use crate::jsonrpc::handlers::NullHandler;
 use crate::jsonrpc::handlers::{MessageHandler, NotificationHandler, RequestHandler};
 use crate::jsonrpc::outgoing_actor::{OutgoingMessageTx, send_raw_message};
 use crate::jsonrpc::task_actor::{PendingTask, Task, TaskTx};
-use crate::mcp_server::McpServer;
+use crate::mcp_server::{McpMessageHandler, McpServer};
 use crate::role::{HasDefaultEndpoint, HasEndpoint, JrEndpoint, JrRole};
 use crate::{Agent, Client, Component};
 
@@ -917,13 +917,15 @@ impl<'scope, H: JrMessageHandler> JrConnectionBuilder<'scope, H> {
     /// ```
     pub fn with_mcp_server<Role: JrRole>(
         self,
-        registry: McpServer<Role>,
-    ) -> JrConnectionBuilder<'scope, ChainedHandler<H, McpServer<Role>>>
+        server: McpServer<'scope, Role>,
+    ) -> JrConnectionBuilder<'scope, ChainedHandler<H, McpMessageHandler<Role>>>
     where
         H: JrMessageHandler<Role = Role>,
         Role: HasEndpoint<Client> + HasEndpoint<Agent>,
     {
-        self.with_handler(registry)
+        let (message_handler, future) = server.into_handler_and_future();
+        self.with_handler(message_handler)
+            .with_spawned(move |_cx| future)
     }
 
     /// Connect these handlers to a transport layer.

src/sacp/src/mcp.rs

@@ -34,7 +34,7 @@ impl HasEndpoint<McpServerEnd> for McpClientToServer {
 
 impl McpClientToServer {
     /// Create a connection builder for an MCP client talking to an MCP server.
-    pub fn builder() -> JrConnectionBuilder<'static, NullHandler<McpClientToServer>> {
+    pub fn builder<'any>() -> JrConnectionBuilder<'any, NullHandler<McpClientToServer>> {
         JrConnectionBuilder::new(McpClientToServer)
     }
 }
@@ -58,7 +58,7 @@ impl HasEndpoint<McpClient> for McpServerToClient {
 
 impl McpServerToClient {
     /// Create a connection builder for an MCP server talking to an MCP client.
-    pub fn builder() -> JrConnectionBuilder<'static, NullHandler<McpServerToClient>> {
+    pub fn builder<'any>() -> JrConnectionBuilder<'any, NullHandler<McpServerToClient>> {
         JrConnectionBuilder::new(McpServerToClient)
     }
 }