feat: allow failable service creation in streamable HTTP tower service (#244)

* feat: allow failable service creation in streamable HTTP tower service

In our setup we need to lookup some info that may fail before we serve
the request.

* Fix examples build

Author: howardjohn

crates/rmcp/src/transport/streamable_http_server/tower.rs

@@ -47,7 +47,7 @@ impl Default for StreamableHttpServerConfig {
 pub struct StreamableHttpService<S, M = super::session::local::LocalSessionManager> {
     pub config: StreamableHttpServerConfig,
     session_manager: Arc<M>,
-    service_factory: Arc<dyn Fn() -> S + Send + Sync>,
+    service_factory: Arc<dyn Fn() -> Result<S, std::io::Error> + Send + Sync>,
 }
 
 impl<S, M> Clone for StreamableHttpService<S, M> {
@@ -92,7 +92,7 @@ where
     M: SessionManager,
 {
     pub fn new(
-        service_factory: impl Fn() -> S + Send + Sync + 'static,
+        service_factory: impl Fn() -> Result<S, std::io::Error> + Send + Sync + 'static,
         session_manager: Arc<M>,
         config: StreamableHttpServerConfig,
     ) -> Self {
@@ -102,7 +102,7 @@ where
             service_factory: Arc::new(service_factory),
         }
     }
-    fn get_service(&self) -> S {
+    fn get_service(&self) -> Result<S, std::io::Error> {
         (self.service_factory)()
     }
     pub async fn handle<B>(&self, request: Request<B>) -> Response<UnsyncBoxBody<Bytes, Infallible>>
@@ -318,7 +318,9 @@ where
                     .create_session()
                     .await
                     .map_err(internal_error_response("create session"))?;
-                let service = self.get_service();
+                let service = self
+                    .get_service()
+                    .map_err(internal_error_response("get service"))?;
                 // spawn a task to serve the session
                 tokio::spawn({
                     let session_manager = self.session_manager.clone();
@@ -372,7 +374,9 @@ where
                 Ok(response)
             }
         } else {
-            let service = self.get_service();
+            let service = self
+                .get_service()
+                .map_err(internal_error_response("get service"))?;
             match message {
                 ClientJsonRpcMessage::Request(request) => {
                     let (transport, receiver) =

crates/rmcp/tests/test_with_js.rs

@@ -96,7 +96,7 @@ async fn test_with_js_streamable_http_client() -> anyhow::Result<()> {
 
     let service: StreamableHttpService<Calculator, LocalSessionManager> =
         StreamableHttpService::new(
-            Calculator::default,
+            || Ok(Calculator),
             Default::default(),
             StreamableHttpServerConfig {
                 stateful_mode: true,

examples/servers/src/counter_hyper_streamable_http.rs

@@ -12,7 +12,7 @@ use rmcp::transport::streamable_http_server::{
 #[tokio::main]
 async fn main() -> anyhow::Result<()> {
     let service = TowerToHyperService::new(StreamableHttpService::new(
-        Counter::new,
+        || Ok(Counter::new()),
         LocalSessionManager::default().into(),
         Default::default(),
     ));

examples/servers/src/counter_streamhttp.rs

@@ -22,7 +22,7 @@ async fn main() -> anyhow::Result<()> {
         .init();
 
     let service = StreamableHttpService::new(
-        Counter::new,
+        || Ok(Counter::new()),
         LocalSessionManager::default().into(),
         Default::default(),
     );