feat: add Streamable HTTP Support to MCP Server

README.md

@@ -23,15 +23,26 @@ By default, it uses the **2025-06-18** version, but earlier versions can be enab
 
 
 
-This project currently supports following transports:
-- **stdio** (Standard Input/Output)
-- **sse** (Server-Sent Events).
-
+This project supports following transports:
+- **Stdio** (Standard Input/Output)
+- **SSE** (Server-Sent Events).
+- **Streamable HTTP**.
 
 
 🚀 The **rust-mcp-sdk** includes a lightweight [Axum](https://github.com/tokio-rs/axum) based server that handles all core functionality seamlessly. Switching between `stdio` and `sse` is straightforward, requiring minimal code changes. The server is designed to efficiently handle multiple concurrent client connections and offers built-in support for SSL.
 
-**⚠️** **Streamable HTTP** transport and authentication still in progress and not yet available. Project is currently under development and should be used at your own risk.
+
+
+**MCP Streamable HTTP Support**
+- [x] Streamable HTTP Support for MCP Servers
+- [x] DNS Rebinding Protection
+- [x] Batch Messages
+- [x] Streaming & non-streaming JSON responses
+- [ ] Streamable HTTP Support for MCP Clients
+- [ ] Resumability
+- [ ] Authentication / OAuth
+
+**⚠️** Project is currently under development and should be used at your own risk.
 
 ## Table of Contents
 - [Usage Examples](#usage-examples)

crates/rust-mcp-macros/Cargo.toml

@@ -21,6 +21,7 @@ syn = "2.0"
 quote = "1.0"
 proc-macro2 = "1.0"
 
+
 [dev-dependencies]
 rust-mcp-schema = { workspace = true, default-features = false }
 

crates/rust-mcp-sdk/README.md

@@ -23,15 +23,26 @@ By default, it uses the **2025-06-18** version, but earlier versions can be enab
 
 
 
-This project currently supports following transports:
-- **stdio** (Standard Input/Output)
-- **sse** (Server-Sent Events).
-
+This project supports following transports:
+- **Stdio** (Standard Input/Output)
+- **SSE** (Server-Sent Events).
+- **Streamable HTTP**.
 
 
 🚀 The **rust-mcp-sdk** includes a lightweight [Axum](https://github.com/tokio-rs/axum) based server that handles all core functionality seamlessly. Switching between `stdio` and `sse` is straightforward, requiring minimal code changes. The server is designed to efficiently handle multiple concurrent client connections and offers built-in support for SSL.
 
-**⚠️** **Streamable HTTP** transport and authentication still in progress and not yet available. Project is currently under development and should be used at your own risk.
+
+
+**MCP Streamable HTTP Support**
+- [x] Streamable HTTP Support for MCP Servers
+- [x] DNS Rebinding Protection
+- [x] Batch Messages
+- [x] Streaming & non-streaming JSON responses
+- [ ] Streamable HTTP Support for MCP Clients
+- [ ] Resumability
+- [ ] Authentication / OAuth
+
+**⚠️** Project is currently under development and should be used at your own risk.
 
 ## Table of Contents
 - [Usage Examples](#usage-examples)

crates/rust-mcp-sdk/src/error.rs

@@ -1,6 +1,8 @@
-use crate::schema::RpcError;
+use crate::schema::{ParseProtocolVersionError, RpcError};
+
 use rust_mcp_transport::error::TransportError;
 use thiserror::Error;
+use tokio::task::JoinError;
 
 #[cfg(feature = "hyper-server")]
 use crate::hyper_servers::error::TransportServerError;
@@ -16,14 +18,18 @@ pub enum McpSdkError {
     #[error("{0}")]
     TransportError(#[from] TransportError),
     #[error("{0}")]
+    JoinError(#[from] JoinError),
+    #[error("{0}")]
     AnyError(Box<(dyn std::error::Error + Send + Sync)>),
     #[error("{0}")]
     SdkError(#[from] crate::schema::schema_utils::SdkError),
     #[cfg(feature = "hyper-server")]
     #[error("{0}")]
     TransportServerError(#[from] TransportServerError),
-    #[error("Incompatible mcp protocol version: client:{0} server:{1}")]
+    #[error("Incompatible mcp protocol version: requested:{0} current:{1}")]
     IncompatibleProtocolVersion(String, String),
+    #[error("{0}")]
+    ParseProtocolVersionError(#[from] ParseProtocolVersionError),
 }
 
 impl McpSdkError {

crates/rust-mcp-sdk/src/hyper_servers.rs

@@ -1,5 +1,6 @@
 mod app_state;
 pub mod error;
+pub mod hyper_runtime;
 pub mod hyper_server;
 pub mod hyper_server_core;
 mod middlewares;

crates/rust-mcp-sdk/src/hyper_servers/app_state.rs

@@ -19,5 +19,23 @@ pub struct AppState {
     pub handler: Arc<dyn McpServerHandler>,
     pub ping_interval: Duration,
     pub sse_message_endpoint: String,
+    pub http_streamable_endpoint: String,
     pub transport_options: Arc<TransportOptions>,
+    pub enable_json_response: bool,
+    /// List of allowed host header values for DNS rebinding protection.
+    /// If not specified, host validation is disabled.
+    pub allowed_hosts: Option<Vec<String>>,
+    /// List of allowed origin header values for DNS rebinding protection.
+    /// If not specified, origin validation is disabled.
+    pub allowed_origins: Option<Vec<String>>,
+    /// Enable DNS rebinding protection (requires allowedHosts and/or allowedOrigins to be configured).
+    /// Default is false for backwards compatibility.
+    pub dns_rebinding_protection: bool,
+}
+
+impl AppState {
+    pub fn needs_dns_protection(&self) -> bool {
+        self.dns_rebinding_protection
+            && (self.allowed_hosts.is_some() || self.allowed_origins.is_some())
+    }
 }

crates/rust-mcp-sdk/src/hyper_servers/error.rs

@@ -21,6 +21,8 @@ pub enum TransportServerError {
     InvalidServerOptions(String),
     #[error("{0}")]
     SslCertError(String),
+    #[error("{0}")]
+    TransportError(String),
 }
 
 impl IntoResponse for TransportServerError {

crates/rust-mcp-sdk/src/hyper_servers/hyper_runtime.rs

@@ -0,0 +1,198 @@
+use std::{sync::Arc, time::Duration};
+
+use crate::{
+    mcp_server::HyperServer,
+    schema::{
+        schema_utils::{NotificationFromServer, RequestFromServer, ResultFromClient},
+        CreateMessageRequestParams, CreateMessageResult, LoggingMessageNotificationParams,
+        PromptListChangedNotificationParams, ResourceListChangedNotificationParams,
+        ResourceUpdatedNotificationParams, ToolListChangedNotificationParams,
+    },
+    McpServer,
+};
+
+use axum_server::Handle;
+use rust_mcp_transport::SessionId;
+use tokio::{sync::Mutex, task::JoinHandle};
+
+use crate::{
+    error::SdkResult,
+    hyper_servers::app_state::AppState,
+    mcp_server::{
+        error::{TransportServerError, TransportServerResult},
+        ServerRuntime,
+    },
+};
+
+pub struct HyperRuntime {
+    pub(crate) state: Arc<AppState>,
+    pub(crate) server_task: JoinHandle<Result<(), TransportServerError>>,
+    pub(crate) server_handle: Handle,
+}
+
+impl HyperRuntime {
+    pub async fn create(server: HyperServer) -> SdkResult<Self> {
+        let addr = server.options.resolve_server_address().await?;
+        let state = server.state();
+
+        let server_handle = server.server_handle();
+
+        let server_task = tokio::spawn(async move {
+            #[cfg(feature = "ssl")]
+            if server.options.enable_ssl {
+                server.start_ssl(addr).await
+            } else {
+                server.start_http(addr).await
+            }
+
+            #[cfg(not(feature = "ssl"))]
+            if server.options.enable_ssl {
+                panic!("SSL requested but the 'ssl' feature is not enabled");
+            } else {
+                server.start_http(addr).await
+            }
+        });
+
+        Ok(Self {
+            state,
+            server_task,
+            server_handle,
+        })
+    }
+
+    pub fn graceful_shutdown(&self, timeout: Option<Duration>) {
+        self.server_handle.graceful_shutdown(timeout);
+    }
+
+    pub async fn await_server(self) -> SdkResult<()> {
+        let result = self.server_task.await?;
+        result.map_err(|err| err.into())
+    }
+
+    pub async fn runtime_by_session(
+        &self,
+        session_id: &SessionId,
+    ) -> TransportServerResult<Arc<Mutex<Arc<ServerRuntime>>>> {
+        self.state.session_store.get(session_id).await.ok_or(
+            TransportServerError::SessionIdInvalid(session_id.to_string()),
+        )
+    }
+
+    pub async fn send_request(
+        &self,
+        session_id: &SessionId,
+        request: RequestFromServer,
+        timeout: Option<Duration>,
+    ) -> SdkResult<ResultFromClient> {
+        let runtime = self.runtime_by_session(session_id).await?;
+        let runtime = runtime.lock().await.to_owned();
+        runtime.request(request, timeout).await
+    }
+
+    pub async fn send_notification(
+        &self,
+        session_id: &SessionId,
+        notification: NotificationFromServer,
+    ) -> SdkResult<()> {
+        let runtime = self.runtime_by_session(session_id).await?;
+        let runtime = runtime.lock().await.to_owned();
+        runtime.send_notification(notification).await
+    }
+
+    pub async fn send_logging_message(
+        &self,
+        session_id: &SessionId,
+        params: LoggingMessageNotificationParams,
+    ) -> SdkResult<()> {
+        let runtime = self.runtime_by_session(session_id).await?;
+        let runtime = runtime.lock().await.to_owned();
+        runtime.send_logging_message(params).await
+    }
+
+    /// An optional notification from the server to the client, informing it that
+    /// the list of prompts it offers has changed.
+    /// This may be issued by servers without any previous subscription from the client.
+    pub async fn send_prompt_list_changed(
+        &self,
+        session_id: &SessionId,
+        params: Option<PromptListChangedNotificationParams>,
+    ) -> SdkResult<()> {
+        let runtime = self.runtime_by_session(session_id).await?;
+        let runtime = runtime.lock().await.to_owned();
+        runtime.send_prompt_list_changed(params).await
+    }
+
+    /// An optional notification from the server to the client,
+    /// informing it that the list of resources it can read from has changed.
+    /// This may be issued by servers without any previous subscription from the client.
+    pub async fn send_resource_list_changed(
+        &self,
+        session_id: &SessionId,
+        params: Option<ResourceListChangedNotificationParams>,
+    ) -> SdkResult<()> {
+        let runtime = self.runtime_by_session(session_id).await?;
+        let runtime = runtime.lock().await.to_owned();
+        runtime.send_resource_list_changed(params).await
+    }
+
+    /// A notification from the server to the client, informing it that
+    /// a resource has changed and may need to be read again.
+    ///  This should only be sent if the client previously sent a resources/subscribe request.
+    pub async fn send_resource_updated(
+        &self,
+        session_id: &SessionId,
+        params: ResourceUpdatedNotificationParams,
+    ) -> SdkResult<()> {
+        let runtime = self.runtime_by_session(session_id).await?;
+        let runtime = runtime.lock().await.to_owned();
+        runtime.send_resource_updated(params).await
+    }
+
+    /// An optional notification from the server to the client, informing it that
+    /// the list of tools it offers has changed.
+    /// This may be issued by servers without any previous subscription from the client.
+    pub async fn send_tool_list_changed(
+        &self,
+        session_id: &SessionId,
+        params: Option<ToolListChangedNotificationParams>,
+    ) -> SdkResult<()> {
+        let runtime = self.runtime_by_session(session_id).await?;
+        let runtime = runtime.lock().await.to_owned();
+        runtime.send_tool_list_changed(params).await
+    }
+
+    /// A ping request to check that the other party is still alive.
+    /// The receiver must promptly respond, or else may be disconnected.
+    ///
+    /// This function creates a `PingRequest` with no specific parameters, sends the request and awaits the response
+    /// Once the response is received, it attempts to convert it into the expected
+    /// result type.
+    ///
+    /// # Returns
+    /// A `SdkResult` containing the `rust_mcp_schema::Result` if the request is successful.
+    /// If the request or conversion fails, an error is returned.
+    pub async fn ping(
+        &self,
+        session_id: &SessionId,
+        timeout: Option<Duration>,
+    ) -> SdkResult<crate::schema::Result> {
+        let runtime = self.runtime_by_session(session_id).await?;
+        let runtime = runtime.lock().await.to_owned();
+        runtime.ping(timeout).await
+    }
+
+    /// A request from the server to sample an LLM via the client.
+    /// The client has full discretion over which model to select.
+    /// The client should also inform the user before beginning sampling,
+    /// to allow them to inspect the request (human in the loop)
+    /// and decide whether to approve it.
+    pub async fn create_message(
+        &self,
+        session_id: &SessionId,
+        params: CreateMessageRequestParams,
+    ) -> SdkResult<CreateMessageResult> {
+        let runtime = self.runtime_by_session(session_id).await?;
+        let runtime = runtime.lock().await.to_owned();
+        runtime.create_message(params).await
+    }
+}

crates/rust-mcp-sdk/src/hyper_servers/middlewares.rs

@@ -1 +1,2 @@
+pub(crate) mod protect_dns_rebinding;
 pub(crate) mod session_id_gen;