Streamable HTTP error: fetch failed

[alexiokay]

I don't really know why but today I started my app after 2 weeks of little coding holidays, and the mcp client doesn't work anymore via stdio, before it was successfully connecting and now it doesn't. The same npx command (which I use to connect to mcp server) used directly in CMD, works well.

2025-04-30T01:10:02.120Z [Runner] Connecting to server: {"id":"@smithery-ai/server-sequential-thinking","connectionTypes":["stdio","stdio","http","ws"]}
2025-04-30T01:10:02.121Z [Runner] Connecting to Streamable HTTP endpoint: https://server.smithery.ai/@smithery-ai/server-sequential-thinking/mcp
2025-04-30T01:10:02.122Z [Runner] Streamable HTTP connection initiated
2025-04-30T01:10:02.122Z [Runner] Streamable HTTP connection established
2025-04-30T01:10:02.152Z [Runner] Streamable HTTP error: fetch failed
2025-04-30T01:10:02.152Z Failed to send message: fetch failed
2025-04-30T01:10:32.122Z [Runner] Sending heartbeat ping...

[4t145]

Streamable http are not supported yet, but I'am working on this

[alexiokay]

Hmm, but it worked a few weeks ago, lol, the same command, I didn't change anything haha.

But ok. I'm waiting maybe then I won't be forced to use npx

[hwillson]

👋 @4t145 - thanks for tackling streamable HTTP. How far along are you? Is there anything external contributors can help with?

[4t145]

@hwillson You can review the code in #152 now

[kkharji]

Thanks @4t145 that awesome, came at the right time. I was playing around with it. I needed to pass headers and query params to with_service context. I wonder if this something other users will benefit from or maybe done in a better way:

examples/servers/src/axum_streamable_http.rs

diff --git a/examples/servers/src/axum_streamable_http.rs b/examples/servers/src/axum_streamable_http.rs
index e65be6b..4ce70ac 100644
--- a/examples/servers/src/axum_streamable_http.rs
+++ b/examples/servers/src/axum_streamable_http.rs
@@ -14,14 +14,21 @@ async fn main() -> anyhow::Result<()> {
     tracing_subscriber::registry()
         .with(
             tracing_subscriber::EnvFilter::try_from_default_env()
-                .unwrap_or_else(|_| "debug".to_string().into()),
+                .unwrap_or_else(|_| "info,rmcp=debug".to_string().into()), // Adjusted default filter
         )
         .with(tracing_subscriber::fmt::layer())
         .init();
 
     let ct = StreamableHttpServer::serve(BIND_ADDRESS.parse()?)
         .await?
-        .with_service(Counter::new);
+        .with_service(|init_ctx| {
+            tracing::info!(
+                "New service instance created with headers: {:?} and query_params: {:?}",
+                init_ctx.headers,
+                init_ctx.query_params
+            );
+            Counter::new()
+        });
 
     tokio::signal::ctrl_c().await?;
     ct.cancel();
diff --git a/crates/rmcp/src/transport/streamable_http_server/axum.rs b/crates/rmcp/src/transport/streamable_http_server/axum.rs
index f01ecec..21ea482 100644
--- a/crates/rmcp/src/transport/streamable_http_server/axum.rs
+++ b/crates/rmcp/src/transport/streamable_http_server/axum.rs
@@ -2,12 +2,9 @@ use std::{collections::HashMap, io, net::SocketAddr, sync::Arc, time::Duration};
 
 use axum::{
     Json, Router,
-    extract::State,
+    extract::{Query, State},
     http::{HeaderMap, HeaderValue, StatusCode},
-    response::{
-        IntoResponse, Response,
-        sse::{Event, KeepAlive, Sse},
-    },
+    response::{IntoResponse, Response, Sse, sse::Event, sse::KeepAlive},
     routing::get,
 };
 use futures::Stream;
@@ -16,14 +13,15 @@ use tokio_util::sync::CancellationToken;
 use tracing::Instrument;
 
 use super::session::{
-    EventId, HEADER_LAST_EVENT_ID, Session, SessionTransport, StreamableHttpMessageReceiver,
+    EventId, HEADER_LAST_EVENT_ID, HEADER_SESSION_ID, Session, SessionId, SessionTransport,
+    StreamableHttpMessageReceiver,
 };
 use crate::{
     RoleServer, Service,
     model::ClientJsonRpcMessage,
     transport::{
-        common::axum::{DEFAULT_AUTO_PING_INTERVAL, SessionId, session_id},
-        streamable_http_server::session::HEADER_SESSION_ID,
+        common::axum::{DEFAULT_AUTO_PING_INTERVAL, session_id},
+        streamable_http_server::session::InitializationContext,
     },
 };
 type SessionManager = Arc<tokio::sync::RwLock<HashMap<SessionId, Session>>>;
@@ -68,12 +66,13 @@ fn receiver_as_stream(
 
 async fn post_handler(
     State(app): State<App>,
+    Query(query_params): Query<HashMap<String, String>>,
     header_map: HeaderMap,
     Json(message): Json<ClientJsonRpcMessage>,
 ) -> Result<Response, Response> {
     use futures::StreamExt;
-    if let Some(session_id) = header_map.get(HEADER_SESSION_ID) {
-        let session_id = session_id
+    if let Some(session_id_val) = header_map.get(HEADER_SESSION_ID) {
+        let session_id = session_id_val
             .to_str()
             .map_err(|e| (StatusCode::BAD_REQUEST, e.to_string()).into_response())?;
         tracing::debug!(session_id, ?message, "new client message");
@@ -128,10 +127,14 @@ async fn post_handler(
     } else {
         // expect initialize message
         let session_id = session_id();
+        let init_ctx = InitializationContext {
+            headers: header_map,
+            query_params,
+        };
         let (session, transport) =
-            super::session::create_session(session_id.clone(), Default::default());
+            super::session::create_session(session_id.clone(), Default::default(), init_ctx);
         let Ok(_) = app.transport_tx.send(transport) else {
-            return Err((StatusCode::GONE, "session terminated").into_response());
+            return Err((StatusCode::GONE, "session terminated by server").into_response());
         };
 
         let response = session.handle().initialize(message).await.map_err(|e| {
@@ -307,20 +310,30 @@ impl StreamableHttpServer {
     pub fn with_service<S, F>(mut self, service_provider: F) -> CancellationToken
     where
         S: Service<RoleServer>,
-        F: Fn() -> S + Send + 'static,
+        F: Fn(Arc<InitializationContext>) -> S + Send + Sync + 'static,
     {
         use crate::service::ServiceExt;
         let ct = self.config.ct.clone();
         tokio::spawn(async move {
             while let Some(transport) = self.next_transport().await {
-                let service = service_provider();
+                let service = service_provider(transport.initialization_context.clone());
                 let ct = self.config.ct.child_token();
                 tokio::spawn(async move {
-                    let server = service.serve_with_ct(transport, ct).await?;
-                    server.waiting().await?;
-                    tokio::io::Result::Ok(())
+                    match service.serve_with_ct(transport, ct).await {
+                        Ok(running_service) => {
+                            if let Err(join_err) = running_service.waiting().await {
+                                tracing::error!("Service waiting error: {:?}", join_err);
+                            }
+                        }
+                        Err(e) => {
+                            tracing::error!("Failed to serve service: {:?}", e);
+                        }
+                    }
                 });
             }
+            tracing::info!(
+                "StreamableHttpServer's transport channel closed, stopping service provisioning."
+            );
         });
         ct
     }
diff --git a/crates/rmcp/src/transport/streamable_http_server/session.rs b/crates/rmcp/src/transport/streamable_http_server/session.rs
index 574ad17..caf0b18 100644
--- a/crates/rmcp/src/transport/streamable_http_server/session.rs
+++ b/crates/rmcp/src/transport/streamable_http_server/session.rs
@@ -5,6 +5,7 @@ use std::{
     sync::Arc,
 };
 
+use axum::http::HeaderMap;
 use futures::{Sink, SinkExt, Stream};
 use thiserror::Error;
 use tokio::sync::{
@@ -27,6 +28,14 @@ use crate::{
 
 pub const HEADER_SESSION_ID: &str = "Mcp-Session-Id";
 pub const HEADER_LAST_EVENT_ID: &str = "Last-Event-Id";
+
+/// Holds the initial HTTP headers and URL query parameters from the request that established the session.
+#[derive(Debug, Clone)]
+pub struct InitializationContext {
+    pub headers: HeaderMap,
+    pub query_params: HashMap<String, String>,
+}
+
 #[derive(Debug, Clone)]
 pub struct ServerSessionMessage {
     pub event_id: EventId,
@@ -701,6 +710,7 @@ impl SessionHandle {
 pub struct SessionTransport {
     session_handle: SessionHandle,
     to_service_rx: Receiver<ClientJsonRpcMessage>,
+    pub initialization_context: Arc<InitializationContext>,
 }
 
 impl IntoTransport<RoleServer, SessionError, ()> for SessionTransport {
@@ -735,7 +745,11 @@ impl Default for SessionConfig {
     }
 }
 
-pub fn create_session(id: SessionId, config: SessionConfig) -> (Session, SessionTransport) {
+pub fn create_session(
+    id: SessionId,
+    config: SessionConfig,
+    init_context: InitializationContext,
+) -> (Session, SessionTransport) {
     let (to_service_tx, to_service_rx) = tokio::sync::mpsc::channel(config.channel_capacity);
     let (event_tx, event_rx) = tokio::sync::mpsc::channel(config.channel_capacity);
     let (common_tx, _) = tokio::sync::mpsc::channel(config.channel_capacity);
@@ -762,6 +776,7 @@ pub fn create_session(id: SessionId, config: SessionConfig) -> (Session, Session
     let session_transport = SessionTransport {
         to_service_rx,
         session_handle: handle,
+        initialization_context: Arc::new(init_context),
     };
     (session, session_transport)
 }

[4t145]

@kkharji Thank you for remind me of this #153 (comment)

We can store the initialize request's Parts in rmcp request extension, so you can get it when handling initialze request, sounds good to you?

update: I made a PR: #163

[kkharji]

Awesome, Thanks @4t145 for looking into this swiftly. Okay, that seems to handle exposing it, DX side, how can you use it? In my case, I need a header value to configure the service (or at least how it is constructed).

[4t145]

@kkharji I added a example to show how to get the request parts from context https://github.com/modelcontextprotocol/rust-sdk/pull/163/files#diff-e8a7b3352088986e9a3c5b83cfea597a505d62b2881bb6f0b6dbc07ca6518e9d

[kkharji]

AMAZING. Thanks so much @4t145. Yeah, having it as an example will be beneficial. I will try to finish what I have in front of me and migrate to that PR or main if its merge and let you know.

[alexiokay]

I still don't know how to make It work, I'm still having the same errors on windows.

2025-05-19T12:44:51.948Z Failed to send message: fetch failed
2025-05-19T12:44:52.210Z [Runner] Connecting to server: {"id":"@smithery-ai/server-sequential-thinking","connectionTypes":["stdio","stdio","http"]}
2025-05-19T12:44:52.211Z [Runner] Connecting to Streamable HTTP endpoint: https://server.smithery.ai/@smithery-ai/server-sequential-thinking/mcp
2025-05-19T12:44:52.211Z [Runner] Streamable HTTP connection initiated
2025-05-19T12:44:52.211Z [Runner] Streamable HTTP connection established
2025-05-19T12:44:52.215Z [Runner] Streamable HTTP error: fetch failed
2025-05-19T12:44:52.215Z Failed to send message: fetch failed

These are my configs:

{
  "models": {
    "model": [
      {
        "name": "gpt-3.5",
        "url": "https://api.openai.com/v1/chat/completions",
        "apiKey": "example"
      }
    ]
  },
  "mcp": {
    "server": [
      {
        "name": "server-sequential-thinking",
        "protocol": "stdio",
        "command": "cmd",
        "args": [
          "/c",
          "npx",
          "-y",
          "@smithery/cli@latest",
          "run",
          "@smithery-ai/server-sequential-thinking",
          "--key",
          "example"
        ],
        "envs": {
          "PATH": "C:/Program Files/nodejs;%PATH%"
        }
      },
      {
        "name": "duckduckgo-mcp-server",
        "protocol": "stdio",
        "command": "cmd",
        "args": [
          "/c",
          "npx",
          "-y",
          "@smithery/cli@latest",
          "run",
          "@nickclyde/duckduckgo-mcp-server",
          "--key",
          "example"
        ],
        "envs": {
          "PATH": "C:/Program Files/nodejs;%PATH%"
        }
      },
    
    ]
  }
}

[4t145]

@alexiokay Sorry I don't get it, it seems you are connecting non-rust server from a non-rust client through stdio, what does it to do with streamable http and this sdk?

[alexiokay]

Sorry I'm new into the concept of MCP I don't really understand how it works in details.

You're right, I'll use different transport implementations. Thanks for making http-streamable working 😊