refactor(fast-time): use rmcp SDK for Streamable HTTP transport and tools

Replace the hand-rolled MCP layer with the official Rust SDK (rmcp):

- /mcp Streamable HTTP transport is now served by rmcp's
  StreamableHttpService + LocalSessionManager, removing the bespoke
  session map and JSON-RPC dispatch (streamable_http.rs). A lightweight
  gate caps concurrent sessions at MAX_ACTIVE_SESSIONS (rmcp's session
  manager has no built-in cap); existing sessions are cleaned up by rmcp
  on disconnect/DELETE.
- Tools are declared once via #[tool] macros with schemas derived from
  their parameter types, removing the hand-written tool schemas and
  dispatch (mcp.rs). The schema_error/schema_success fixtures retain
  their declared outputSchema.
- The legacy HTTP+SSE transport remains a hand-rolled shim (documented
  exception) because rmcp does not reproduce the bespoke ContextForge
  legacy-SSE semantics (endpoint event, sessionId/session_id aliases,
  202/404/410 codes). It delegates tools/list and tools/call back to the
  rmcp server so schemas and logic never drift.
- serverInfo is built explicitly so /mcp reports the server identity
  rather than rmcp's own crate name.

Behavior parity (DST conversion, echo delay limit, protocol-version echo,
output schemas) is covered by unit tests; both transports verified
end-to-end.

Signed-off-by: lucarlig <luca.carlig@ibm.com>

Author: lucarlig

Cargo.lock

@@ -16,6 +16,10 @@ tokio-stream = "0.1.17"
 # Web framework
 axum.workspace = true
 
+# MCP SDK (official Rust SDK) — drives the /mcp Streamable HTTP transport and
+# tool schemas. The legacy SSE transport is a hand-rolled shim (see transports/sse.rs).
+rmcp = { workspace = true, features = ["server", "macros", "transport-streamable-http-server"] }
+
 # Serialization
 serde.workspace = true
 serde_json.workspace = true

mcp-servers/rust/fast-time-server/Cargo.toml

@@ -2,16 +2,27 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use axum::Router;
+use axum::extract::{Request, State};
+use axum::http::StatusCode;
+use axum::middleware::{Next, from_fn_with_state};
+use axum::response::{IntoResponse, Response};
 use axum::serve::ListenerExt;
+use rmcp::transport::streamable_http_server::StreamableHttpService;
+use rmcp::transport::streamable_http_server::session::local::LocalSessionManager;
 use serde_json::json;
 use std::env;
+use std::sync::Arc;
 use tracing::info;
 use tracing::trace;
 use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
 
-use crate::config::{APP_NAME, APP_VERSION, DEFAULT_BIND_ADDRESS, MCP_PROTOCOL_VERSION};
+use crate::config::{
+    APP_NAME, APP_VERSION, DEFAULT_BIND_ADDRESS, MAX_ACTIVE_SESSIONS, MCP_PROTOCOL_VERSION,
+    SESSION_HEADER,
+};
 use crate::rest;
-use crate::transports::{sse, streamable_http};
+use crate::server::FastTimeServer;
+use crate::transports::sse;
 
 pub async fn run() -> anyhow::Result<()> {
     init_logging();
@@ -67,18 +78,50 @@ fn init_logging() {
 }
 
 fn router() -> Router {
+    // Modern Streamable HTTP transport, served by the rmcp SDK. We hold a handle
+    // to the session manager so a lightweight gate can cap concurrent sessions —
+    // rmcp's LocalSessionManager has no built-in cap. Existing sessions are
+    // cleaned up by rmcp on client disconnect / HTTP DELETE.
+    let session_manager: Arc<LocalSessionManager> = Arc::new(LocalSessionManager::default());
+    let mcp = StreamableHttpService::new(
+        || Ok(FastTimeServer::new()),
+        session_manager.clone(),
+        Default::default(),
+    );
+    let mcp = Router::new()
+        .fallback_service(mcp)
+        .layer(from_fn_with_state(session_manager, session_cap_gate));
+
     Router::new()
         .route("/health", axum::routing::get(health_handler))
         .route("/version", axum::routing::get(version_handler))
         .route("/api/echo", axum::routing::post(rest::echo_handler))
         .route("/api/time", axum::routing::get(rest::time_handler))
-        .route(
-            "/mcp",
-            axum::routing::post(streamable_http::handler).delete(streamable_http::delete_handler),
-        )
+        // Legacy HTTP+SSE transport — hand-rolled shim (see transports/sse.rs).
         .route("/sse", axum::routing::get(sse::handler))
         .route("/messages", axum::routing::post(sse::message_handler))
         .route("/message", axum::routing::post(sse::message_handler))
+        .nest("/mcp", mcp)
+}
+
+/// Cap concurrent Streamable HTTP sessions at `MAX_ACTIVE_SESSIONS`. Requests
+/// carrying an existing `mcp-session-id` always pass so in-flight sessions keep
+/// working; only session-creating requests (no session header) are rejected
+/// with `503` once the cap is reached.
+async fn session_cap_gate(
+    State(session_manager): State<Arc<LocalSessionManager>>,
+    request: Request,
+    next: Next,
+) -> Response {
+    let creates_new_session = !request.headers().contains_key(SESSION_HEADER);
+    if creates_new_session && session_manager.sessions.read().await.len() >= MAX_ACTIVE_SESSIONS {
+        return (
+            StatusCode::SERVICE_UNAVAILABLE,
+            "Maximum active sessions reached",
+        )
+            .into_response();
+    }
+    next.run(request).await
 }
 
 async fn health_handler() -> axum::Json<serde_json::Value> {

mcp-servers/rust/fast-time-server/src/app.rs

@@ -1,23 +1,25 @@
 // Copyright 2026
 // SPDX-License-Identifier: Apache-2.0
 
-use std::time::Duration;
-
 pub(crate) const DEFAULT_BIND_ADDRESS: &str = "0.0.0.0:9080";
 pub(crate) const APP_NAME: &str = "fast-time-server";
 pub(crate) const APP_VERSION: &str = env!("CARGO_PKG_VERSION");
+/// Latest protocol version advertised by the legacy SSE shim and `/version`.
+/// The `/mcp` Streamable HTTP transport negotiates independently inside rmcp.
 pub(crate) const MCP_PROTOCOL_VERSION: &str = "2025-11-25";
-/// Protocol versions echoed back during initialize negotiation; anything else
-/// falls back to `MCP_PROTOCOL_VERSION`.
+/// Protocol versions echoed back during legacy SSE initialize negotiation;
+/// anything else falls back to `MCP_PROTOCOL_VERSION`.
 pub(crate) const SUPPORTED_PROTOCOL_VERSIONS: &[&str] = &[
     "2024-11-05",
     "2025-03-26",
     "2025-06-18",
     MCP_PROTOCOL_VERSION,
 ];
+/// Session id header for the `/mcp` Streamable HTTP transport; used by the
+/// session-cap gate to tell session-creating requests from existing ones.
 pub(crate) const SESSION_HEADER: &str = "mcp-session-id";
+/// Cap on concurrent sessions, applied to both the legacy SSE transport and the
+/// `/mcp` Streamable HTTP transport.
 pub(crate) const MAX_ACTIVE_SESSIONS: usize = 10_000;
-/// Idle lifetime for streamable-HTTP sessions before they are evicted.
-pub(crate) const SESSION_IDLE_TTL: Duration = Duration::from_secs(300);
 pub(crate) const SSE_CHANNEL_CAPACITY: usize = 64;
 pub(crate) const MAX_DELAY_MS: u64 = 60_000;

mcp-servers/rust/fast-time-server/src/config.rs

@@ -6,8 +6,8 @@
 mod app;
 mod config;
 mod delay;
-mod mcp;
 mod rest;
+mod server;
 mod time;
 mod transports;