feat: add Streamable HTTP server

Signed-off-by: mrizzi <mrizzi@redhat.com>

Author: mrizzi

Cargo.lock

@@ -11,12 +11,17 @@ path = "src/bin/stdio.rs"
 name = "sse"
 path = "src/bin/sse.rs"
 
+[[bin]]
+name = "streamable"
+path = "src/bin/streamhttp.rs"
+
 [dependencies]
 anyhow = "1.0"
+axum = { version = "0.8", features = ["macros"] }
 chrono = "0.4"
 openid = "0.17"
 reqwest = { version = "0.12", features = ["json", "blocking"] }
-rmcp = { version = "0.2.1", features = ["server", "transport-sse-server", "transport-io"] }
+rmcp = { version = "0.2.1", features = ["server", "transport-sse-server", "transport-io", "transport-streamable-http-server"] }
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
 tokio = { version = "1", features = ["macros", "rt", "rt-multi-thread", "io-std", "signal"] }

Cargo.toml

@@ -0,0 +1,36 @@
+use rmcp::transport::streamable_http_server::{
+    StreamableHttpService, session::local::LocalSessionManager,
+};
+use tracing_subscriber::{
+    layer::SubscriberExt,
+    util::SubscriberInitExt,
+    {self},
+};
+mod common;
+use common::trustify::Trustify;
+
+const BIND_ADDRESS: &str = "[::]:8000";
+
+#[tokio::main]
+async fn main() -> anyhow::Result<()> {
+    tracing_subscriber::registry()
+        .with(
+            tracing_subscriber::EnvFilter::try_from_default_env()
+                .unwrap_or_else(|_| "debug".to_string().into()),
+        )
+        .with(tracing_subscriber::fmt::layer())
+        .init();
+
+    let service = StreamableHttpService::new(
+        || Ok(Trustify::new()),
+        LocalSessionManager::default().into(),
+        Default::default(),
+    );
+
+    let router = axum::Router::new().nest_service("/mcp", service);
+    let tcp_listener = tokio::net::TcpListener::bind(BIND_ADDRESS).await?;
+    let _ = axum::serve(tcp_listener, router)
+        .with_graceful_shutdown(async { tokio::signal::ctrl_c().await.unwrap() })
+        .await;
+    Ok(())
+}

src/bin/streamhttp.rs

@@ -1,31 +1,10 @@
 use anyhow::Error;
 use rmcp::{ServiceExt, transport::TokioChildProcess};
-use serde_json::json;
+use serde_json::Value;
 use std::{env, process::Command};
 use trustify_test_context::subset::ContainsSubset;
 
-#[test]
-fn tools_list_mcp_inspector() {
-    let inspector_commmand = format!(
-        "npx @modelcontextprotocol/inspector --cli {} --method tools/list",
-        env!("CARGO_BIN_EXE_stdio")
-    );
-    log::debug!("inspector command: {}", inspector_commmand);
-    let output = Command::new("sh")
-        .arg("-c")
-        .arg(inspector_commmand)
-        .env("API_URL", "")
-        .env("OPENID_ISSUER_URL", "")
-        .env("OPENID_CLIENT_ID", "")
-        .env("OPENID_CLIENT_SECRET", "")
-        .output()
-        .expect("failed to execute process");
-
-    let result = serde_json::from_str(str::from_utf8(&output.stdout).unwrap_or_default())
-        .unwrap_or_default();
-    log::debug!("{:#?}", result);
-    log::debug!("{:#?}", str::from_utf8(&output.stderr).unwrap_or_default());
-    let expected_result = json!({
+const EXPECTED_TOOLS_LIST_RESPONSE: &str = r#"{
       "tools": [
         {
           "name": "trustify_vulnerabilities_list",
@@ -261,10 +240,47 @@ fn tools_list_mcp_inspector() {
           }
         }
       ]
-    });
+    }"#;
+
+#[test]
+fn tools_list_mcp_inspector_stdio() {
+    let inspector_commmand = format!(
+        "npx @modelcontextprotocol/inspector --cli {} --method tools/list",
+        env!("CARGO_BIN_EXE_stdio")
+    );
+    log::debug!("inspector command: {}", inspector_commmand);
+    let output = Command::new("sh")
+        .arg("-c")
+        .arg(inspector_commmand)
+        .env("API_URL", "")
+        .env("OPENID_ISSUER_URL", "")
+        .env("OPENID_CLIENT_ID", "")
+        .env("OPENID_CLIENT_SECRET", "")
+        .output()
+        .expect("failed to execute process");
+
+    let result = serde_json::from_str(str::from_utf8(&output.stdout).unwrap_or_default())
+        .unwrap_or_default();
+    log::debug!("{:#?}", result);
+    log::debug!("{:#?}", str::from_utf8(&output.stderr).unwrap_or_default());
+    let expected_result: Value =
+        serde_json::from_str(EXPECTED_TOOLS_LIST_RESPONSE).unwrap_or_default();
     assert!(expected_result.contains_subset(result));
 }
 
+#[test]
+fn tools_list_mcp_inspector_sse() -> Result<(), Error> {
+    run_server_test(env!("CARGO_BIN_EXE_sse"), "http://localhost:8081/sse")
+}
+
+#[test]
+fn tools_list_mcp_inspector_streamable_http() -> Result<(), Error> {
+    run_server_test(
+        env!("CARGO_BIN_EXE_streamable"),
+        "http://localhost:8000/mcp  --transport http",
+    )
+}
+
 #[tokio::test]
 async fn tools_list_mcp_client() -> Result<(), Error> {
     let mut command = tokio::process::Command::new(env!("CARGO_BIN_EXE_stdio"));
@@ -288,3 +304,33 @@ async fn tools_list_mcp_client() -> Result<(), Error> {
 
     Ok(())
 }
+
+fn run_server_test(server_command: &str, inspector_cli_parameter: &str) -> Result<(), Error> {
+    let mut server = Command::new("sh")
+        .arg("-c")
+        .arg(server_command)
+        .env("API_URL", "")
+        .env("OPENID_ISSUER_URL", "")
+        .env("OPENID_CLIENT_ID", "")
+        .env("OPENID_CLIENT_SECRET", "")
+        .spawn()?;
+
+    let inspector_commmand = format!(
+        "npx @modelcontextprotocol/inspector --cli {} --method tools/list",
+        inspector_cli_parameter
+    );
+    log::debug!("inspector command: {}", inspector_commmand);
+    let output = Command::new("sh")
+        .arg("-c")
+        .arg(inspector_commmand)
+        .output()?;
+
+    let result: Value = serde_json::from_str(str::from_utf8(&output.stdout)?)?;
+    log::debug!("{:#?}", result);
+
+    let expected_result: Value = serde_json::from_str(EXPECTED_TOOLS_LIST_RESPONSE)?;
+    assert!(expected_result.contains_subset(result));
+
+    server.kill()?;
+    Ok(())
+}