From 60d664d269f4ca662921b1cec55a8a48bc634e8d Mon Sep 17 00:00:00 2001 From: Jonathan Springer Date: Wed, 29 Jul 2026 15:52:50 +0100 Subject: [PATCH 1/3] Reimplement fast-time-server on the official rmcp SDK Replace the hand-rolled JSON-RPC/session/protocol-negotiation layer with rmcp 3.0.0 (StreamableHttpService + tool macros). The server now runs dual-era only: legacy 2025-11-25 sessionful and modern 2026-07-28 stateless traffic are served concurrently by the SDK transport, and the --strict/--protocol flags are removed. All seven existing tools keep their names, schemas, and behavior, joined by a new verify-protocol tool that reports the protocol version active for the current request (per-request _meta on the modern era, negotiated session version on the legacy era). REST benchmark routes, /health, and /version are unchanged. rust-version bumps to 1.88 for the SDK MSRV. Co-authored-by: Sisyphus Signed-off-by: Jonathan Springer --- mcp-servers/rust/fast-time-server/Cargo.lock | 316 ++- mcp-servers/rust/fast-time-server/Cargo.toml | 15 +- mcp-servers/rust/fast-time-server/src/main.rs | 2082 +++++++---------- 3 files changed, 1124 insertions(+), 1289 deletions(-) diff --git a/mcp-servers/rust/fast-time-server/Cargo.lock b/mcp-servers/rust/fast-time-server/Cargo.lock index 12cf01e..e7e2d31 100644 --- a/mcp-servers/rust/fast-time-server/Cargo.lock +++ b/mcp-servers/rust/fast-time-server/Cargo.lock @@ -27,25 +27,14 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2a4385e2e34eb35d6b3efe798b9eb88096925d87726c0798709bf56d9ed84af3" [[package]] -name = "async-stream" -version = "0.3.6" +name = "async-trait" +version = "0.1.91" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0b5a71a6f37880a80d1d7f19efd781e4b5de42c88f0722cc13bcb6cc2cfe8476" -dependencies = [ - "async-stream-impl", - "futures-core", - "pin-project-lite", -] - -[[package]] -name = "async-stream-impl" -version = "0.3.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c7c24de15d275a1ecfd47a380fb4d5ec9bfe0933f309ed5e705b775596a3574d" +checksum = "ae36dc4177970ef04fde5178d3e2429882def40e57a451f919c098f72baa6cec" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 3.0.3", ] [[package]] @@ -112,6 +101,12 @@ dependencies = [ "tracing", ] +[[package]] +name = "base64" +version = "0.23.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b25655df2c3cdd83c5e5b293b88acd880332b2ddadd7c30ac43144fdc0033da9" + [[package]] name = "bitflags" version = "2.13.0" @@ -196,6 +191,46 @@ dependencies = [ "libc", ] +[[package]] +name = "darling" +version = "0.23.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "25ae13da2f202d56bd7f91c25fba009e7717a1e4a1cc98a76d844b65ae912e9d" +dependencies = [ + "darling_core", + "darling_macro", +] + +[[package]] +name = "darling_core" +version = "0.23.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9865a50f7c335f53564bb694ef660825eb8610e0a53d3e11bf1b0d3df31e03b0" +dependencies = [ + "ident_case", + "proc-macro2", + "quote", + "strsim", + "syn 2.0.118", +] + +[[package]] +name = "darling_macro" +version = "0.23.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ac3984ec7bd6cfa798e62b4a642426a5be0e68f9401cfc2a01e3fa9ea2fcdb8d" +dependencies = [ + "darling_core", + "quote", + "syn 2.0.118", +] + +[[package]] +name = "dyn-clone" +version = "1.0.20" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d0881ea181b1df73ff77ffaaf9c7544ecc11e82fba9b5f27b262a3c73a332555" + [[package]] name = "errno" version = "0.3.14" @@ -211,19 +246,19 @@ name = "fast-time-server" version = "0.1.0" dependencies = [ "anyhow", - "async-stream", "axum", "chrono", "chrono-tz", - "futures-util", "rand", "rand_distr", + "rmcp", "serde", "serde_json", "tokio", + "tokio-util", + "tower", "tracing", "tracing-subscriber", - "uuid", ] [[package]] @@ -241,6 +276,21 @@ dependencies = [ "percent-encoding", ] +[[package]] +name = "futures" +version = "0.3.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8b147ee9d1f6d097cef9ce628cd2ee62288d963e16fb287bd9286455b241382d" +dependencies = [ + "futures-channel", + "futures-core", + "futures-executor", + "futures-io", + "futures-sink", + "futures-task", + "futures-util", +] + [[package]] name = "futures-channel" version = "0.3.32" @@ -248,6 +298,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "07bbe89c50d7a535e539b8c17bc0b49bdb77747034daa8087407d655f3f7cc1d" dependencies = [ "futures-core", + "futures-sink", ] [[package]] @@ -256,6 +307,23 @@ version = "0.3.32" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7e3450815272ef58cec6d564423f6e755e25379b217b0bc688e295ba24df6b1d" +[[package]] +name = "futures-executor" +version = "0.3.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "baf29c38818342a3b26b5b923639e7b1f4a61fc5e76102d4b1981c6dc7a7579d" +dependencies = [ + "futures-core", + "futures-task", + "futures-util", +] + +[[package]] +name = "futures-io" +version = "0.3.33" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4577ecaa3c4f96589d473f679a71b596316f6641bc350038b962a5daf0085d7a" + [[package]] name = "futures-macro" version = "0.3.32" @@ -264,9 +332,15 @@ checksum = "e835b70203e41293343137df5c0664546da5745f82ec9b84d40be8336958447b" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] +[[package]] +name = "futures-sink" +version = "0.3.33" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e34418ac499d6305c2fb5ad0ed2f6ac998c5f8ca209b4510f7f94242c647e307" + [[package]] name = "futures-task" version = "0.3.32" @@ -279,9 +353,13 @@ version = "0.3.32" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "389ca41296e6190b48053de0321d02a77f32f8a5d2461dd38762c0593805c6d6" dependencies = [ + "futures-channel", "futures-core", + "futures-io", "futures-macro", + "futures-sink", "futures-task", + "memchr", "pin-project-lite", "slab", ] @@ -402,6 +480,12 @@ dependencies = [ "cc", ] +[[package]] +name = "ident_case" +version = "1.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b9e0384b61958566e926dc50660321d12159025e767c18e043daf26b70104c39" + [[package]] name = "itoa" version = "1.0.18" @@ -538,6 +622,12 @@ dependencies = [ "windows-link", ] +[[package]] +name = "pastey" +version = "0.2.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2ee67f1008b1ba2321834326597b8e186293b049a023cdef258527550b9935b4" + [[package]] name = "percent-encoding" version = "2.3.2" @@ -628,6 +718,26 @@ dependencies = [ "bitflags", ] +[[package]] +name = "ref-cast" +version = "1.0.26" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "216e8f773d7923bcba9ceb86a86c93cabb3903a11872fc3f138c49630e50b96d" +dependencies = [ + "ref-cast-impl", +] + +[[package]] +name = "ref-cast-impl" +version = "1.0.26" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2c9283685feec7d69af75fb0e858d5e7378f33fe4fc699383b2916ab9273e03c" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.3", +] + [[package]] name = "regex-automata" version = "0.4.14" @@ -645,6 +755,50 @@ version = "0.8.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d6f6ff9a378485b298a5286656da665ba74413d36db0979633275d2e708145d4" +[[package]] +name = "rmcp" +version = "3.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fcd2b6dd3b18129368955f32661a7718969e8c152c7d8866434c09cf15a512e0" +dependencies = [ + "async-trait", + "base64", + "bytes", + "chrono", + "futures", + "http", + "http-body", + "http-body-util", + "pastey", + "pin-project-lite", + "rand", + "rmcp-macros", + "schemars", + "serde", + "serde_json", + "sse-stream", + "thiserror", + "tokio", + "tokio-stream", + "tokio-util", + "tower-service", + "tracing", + "uuid", +] + +[[package]] +name = "rmcp-macros" +version = "3.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e1aa4b9345795260a43fc23d6d05e096407c8b953903f673af7c4404b49fb2d6" +dependencies = [ + "darling", + "proc-macro2", + "quote", + "serde_json", + "syn 2.0.118", +] + [[package]] name = "rustversion" version = "1.0.22" @@ -657,6 +811,32 @@ version = "1.0.23" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9774ba4a74de5f7b1c1451ed6cd5285a32eddb5cccb8cc655a4e50009e06477f" +[[package]] +name = "schemars" +version = "1.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "687274d293b6cdc6e73e0fee520bf2049650090d7164f87672d212a3c530cf4a" +dependencies = [ + "chrono", + "dyn-clone", + "ref-cast", + "schemars_derive", + "serde", + "serde_json", +] + +[[package]] +name = "schemars_derive" +version = "1.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d98c67716b46af2f0b8cf752abc930f6f9aecfbf671ecfb531db8a31dbe4e2ba" +dependencies = [ + "proc-macro2", + "quote", + "serde_derive_internals", + "syn 3.0.3", +] + [[package]] name = "scopeguard" version = "1.2.0" @@ -690,7 +870,18 @@ checksum = "d540f220d3187173da220f885ab66608367b6574e925011a9353e4badda91d79" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", +] + +[[package]] +name = "serde_derive_internals" +version = "0.30.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f852137cce035d6a4df67ccce505ff6b3e9fd3a10e3e52b24dc71e650bb1a9bd" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.3", ] [[package]] @@ -782,6 +973,25 @@ dependencies = [ "windows-sys", ] +[[package]] +name = "sse-stream" +version = "0.2.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c123f296ade4ec4b8b0f6162116e6629f5146922ca5ab40ca9d3c2e73ab4761e" +dependencies = [ + "bytes", + "futures-util", + "http-body", + "http-body-util", + "pin-project-lite", +] + +[[package]] +name = "strsim" +version = "0.11.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7da8b5736845d9f2fcb837ea5d9e2628564b3b043a70948a3f0b778838c5fb4f" + [[package]] name = "syn" version = "2.0.118" @@ -793,12 +1003,43 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "syn" +version = "3.0.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "53e9bae58849f64dfa4f5d5ae372c8341f7305f82a3868709269343628b659a3" +dependencies = [ + "proc-macro2", + "quote", + "unicode-ident", +] + [[package]] name = "sync_wrapper" version = "1.0.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0bf256ce5efdfa370213c1dabab5935a12e49f2c58d15e9eac2870d3b4f27263" +[[package]] +name = "thiserror" +version = "2.0.19" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "09a43598840e33d5b0331f38c5e30d13bb11c11210a4b58f0d9b18a5a5eefcd9" +dependencies = [ + "thiserror-impl", +] + +[[package]] +name = "thiserror-impl" +version = "2.0.19" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "43cbfe0cf76104d42a574802844187e84a305e531ed54455f11fbde0f10541cd" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.3", +] + [[package]] name = "thread_local" version = "1.1.9" @@ -833,7 +1074,32 @@ checksum = "385a6cb71ab9ab790c5fe8d67f1645e6c450a7ce006a33de03daa956cf70a496" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", +] + +[[package]] +name = "tokio-stream" +version = "0.1.19" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a3d06f0b082ba57c26b79407372e57cf2a1e28124f78e9479fe80322cf53420b" +dependencies = [ + "futures-core", + "pin-project-lite", + "tokio", +] + +[[package]] +name = "tokio-util" +version = "0.7.19" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "494815d09bf52b5548659851081238f0ca39ff638363907596da739561c62c52" +dependencies = [ + "bytes", + "futures-core", + "futures-sink", + "libc", + "pin-project-lite", + "tokio", ] [[package]] @@ -884,7 +1150,7 @@ checksum = "7490cfa5ec963746568740651ac6781f701c9c5ea257c58e057f3ba8cf69e8da" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -987,7 +1253,7 @@ dependencies = [ "bumpalo", "proc-macro2", "quote", - "syn", + "syn 2.0.118", "wasm-bindgen-shared", ] @@ -1021,7 +1287,7 @@ checksum = "053e2e040ab57b9dc951b72c264860db7eb3b0200ba345b4e4c3b14f67855ddf" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] @@ -1032,7 +1298,7 @@ checksum = "3f316c4a2570ba26bbec722032c4099d8c8bc095efccdc15688708623367e358" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.118", ] [[package]] diff --git a/mcp-servers/rust/fast-time-server/Cargo.toml b/mcp-servers/rust/fast-time-server/Cargo.toml index 693aa88..b1b5380 100644 --- a/mcp-servers/rust/fast-time-server/Cargo.toml +++ b/mcp-servers/rust/fast-time-server/Cargo.toml @@ -5,7 +5,7 @@ name = "fast-time-server" version = "0.1.0" edition = "2024" -rust-version = "1.85" +rust-version = "1.88" description = "Ultra-fast MCP test server with echo and time tools" license = "Apache-2.0" authors = ["ContextForge Contributors"] @@ -36,11 +36,16 @@ tracing-subscriber = { version = "0.3.22", features = ["env-filter"] } # Time handling chrono = { version = "0.4.45", features = ["serde"] } chrono-tz = "0.10.4" -uuid = { version = "1.23.0", features = ["v4"] } -# SSE support -async-stream = "0.3" -futures-util = "0.3" +# Official MCP Rust SDK +rmcp = { version = "3.0", features = ["server", "macros", "transport-streamable-http-server"] } + +# Cancellation token for graceful shutdown of SSE streams +tokio-util = "0.7" + +[dev-dependencies] +# tower::ServiceExt::oneshot for in-process integration tests +tower = { version = "0.5", features = ["util"] } [lints.clippy] multiple_crate_versions = "allow" diff --git a/mcp-servers/rust/fast-time-server/src/main.rs b/mcp-servers/rust/fast-time-server/src/main.rs index 7b20b54..634ab59 100644 --- a/mcp-servers/rust/fast-time-server/src/main.rs +++ b/mcp-servers/rust/fast-time-server/src/main.rs @@ -5,13 +5,21 @@ // // This server provides minimal, blazing-fast tools for load testing: // - echo: Echoes back whatever you send it +// - flaky: Fails N times per key before succeeding (retry testing) // - get_system_time: Returns current time in specified timezone +// - convert_time: Converts a time between IANA timezones +// - schema_error / schema_success: Output-schema validation fixtures +// - get_stats: Returns server statistics +// - verify-protocol: Reports the MCP protocol version of the current request // -// Transport: Streamable HTTP (no auth) +// Transport: Streamable HTTP (no auth) via the official MCP Rust SDK (rmcp). +// Dual-era by default: legacy 2025-11-25 (initialize handshake + sessions) +// and modern 2026-07-28 (stateless, per-request _meta) are served +// simultaneously on POST/DELETE /mcp. // Default: http://127.0.0.1:9080/mcp use axum::Router; -use axum::http::{HeaderMap, HeaderValue, StatusCode, header}; +use axum::http::{StatusCode, header}; use axum::response::{IntoResponse, Response}; use axum::serve::ListenerExt; #[cfg(test)] @@ -20,43 +28,38 @@ use chrono::{DateTime, FixedOffset, SecondsFormat, TimeZone, Utc}; use chrono_tz::Tz; use rand_distr::Distribution; use rand_distr::Normal; +use rmcp::handler::server::router::tool::ToolRouter; +use rmcp::handler::server::wrapper::Parameters; +use rmcp::model::{ + CallToolResult, ContentBlock, Implementation, ProtocolVersion, ServerCapabilities, ServerInfo, +}; +use rmcp::service::RequestContext; +use rmcp::transport::streamable_http_server::{ + StreamableHttpServerConfig, StreamableHttpService, session::local::LocalSessionManager, +}; +use rmcp::{ErrorData as McpError, Json, RoleServer, ServerHandler, schemars}; +use rmcp::{tool, tool_handler, tool_router}; use serde_json::json; -use std::collections::{HashMap, HashSet}; +use std::borrow::Cow; +use std::collections::HashMap; use std::env; use std::sync::atomic::{AtomicU64, Ordering}; -use std::sync::{Arc, LazyLock, Mutex, RwLock}; +use std::sync::{Arc, Mutex}; use tracing::info; use tracing::trace; use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt}; -use uuid::Uuid; const DEFAULT_BIND_ADDRESS: &str = "0.0.0.0:9080"; const APP_NAME: &str = "fast-time-server"; const APP_VERSION: &str = env!("CARGO_PKG_VERSION"); -// Legacy revisions (2025-11-25 and earlier) negotiate via the `initialize` -// handshake; modern revisions (2026-07-28 and later) declare the version -// per-request in `_meta`. The legacy revision is always served; the modern -// revision is enabled with `--protocol`. +const MAX_DELAY_MS: u64 = 60_000; +/// The legacy revision negotiates via the `initialize` handshake and uses +/// `mcp-session-id` sessions; the modern revision declares the version +/// per-request in `_meta` and is served statelessly. Both are served at once. const MCP_PROTOCOL_VERSION: &str = "2025-11-25"; const MCP_PROTOCOL_VERSION_MODERN: &str = "2026-07-28"; -const SESSION_HEADER: &str = "mcp-session-id"; -const PROTOCOL_VERSION_HEADER: &str = "mcp-protocol-version"; -const PROTOCOL_VERSION_META_KEY: &str = "io.modelcontextprotocol/protocolVersion"; -const SERVER_INFO_META_KEY: &str = "io.modelcontextprotocol/serverInfo"; -const ERR_UNSUPPORTED_PROTOCOL_VERSION: i32 = -32022; -const ERR_HEADER_MISMATCH: i32 = -32020; -const MAX_ACTIVE_SESSIONS: usize = 10_000; -const MAX_DELAY_MS: u64 = 60_000; -static DIRECT_REQUEST_COUNT: AtomicU64 = AtomicU64::new(0); -static ACTIVE_SESSIONS: LazyLock>> = - LazyLock::new(|| RwLock::new(HashSet::new())); - -/// Per-key attempt counter for the `flaky` test tool. Keyed by the caller- -/// supplied `key` argument so back-to-back test sequences stay isolated; the -/// gateway re-sends identical arguments on each retry, so all attempts of one -/// logical call share a key and increment the same counter. -static FLAKY_STATE: LazyLock>> = - LazyLock::new(|| Mutex::new(HashMap::new())); +const SUPPORTED_PROTOCOL_VERSIONS: &[ProtocolVersion] = + &[ProtocolVersion::V_2025_11_25, ProtocolVersion::V_2026_07_28]; // ============================================================================ // Delay Helpers @@ -192,95 +195,294 @@ fn parse_offset(s: &str) -> Result { } // ============================================================================ -// Server Configuration & CLI +// MCP Server (official rmcp SDK) // ============================================================================ -#[derive(Debug, Clone, PartialEq, Eq)] -struct ServerConfig { - modern: bool, - legacy: bool, - strict: bool, +/// Shared state, Arc-cloned into the single handler the service factory hands +/// to every session and stateless request. +#[derive(Default)] +struct SharedState { + request_count: AtomicU64, + /// Per-key attempt counter for the `flaky` test tool. Keyed by the caller- + /// supplied `key` argument so back-to-back test sequences stay isolated; + /// the gateway re-sends identical arguments on each retry, so all attempts + /// of one logical call share a key and increment the same counter. + flaky: Mutex>, +} + +struct FastTimeServer { + state: Arc, + tool_router: ToolRouter, +} + +#[derive(Debug, serde::Deserialize, schemars::JsonSchema)] +struct EchoRequest { + message: String, + #[schemars(range(min = 0, max = 60000))] + delay: Option, + #[schemars(range(min = 0))] + delay_stddev: Option, +} + +#[derive(Debug, serde::Deserialize, schemars::JsonSchema)] +struct FlakyRequest { + /// Unique key to track attempt count across retries + key: String, + /// Number of times to return isError=true before succeeding (default 0) + #[schemars(range(min = 0))] + fail_times: Option, +} + +#[derive(Debug, serde::Deserialize, schemars::JsonSchema)] +struct GetSystemTimeRequest { + timezone: Option, +} + +#[derive(Debug, serde::Deserialize, schemars::JsonSchema)] +struct ConvertTimeRequest { + time: String, + source_timezone: String, + target_timezone: String, } -impl Default for ServerConfig { - fn default() -> Self { +#[derive(Debug, serde::Serialize, schemars::JsonSchema)] +#[serde(rename_all = "camelCase")] +struct RecognitionResult { + recognition_id: String, + #[serde(skip_serializing_if = "Option::is_none")] + message: Option, +} + +#[derive(Debug, serde::Serialize, schemars::JsonSchema)] +#[serde(rename_all = "camelCase")] +struct VerifyProtocolResult { + protocol_version: String, + transport: String, +} + +/// Resolve the protocol version active for one request: the per-request +/// `_meta` version wins (modern, stateless era); otherwise fall back to the +/// version the session negotiated at `initialize` (legacy era). +fn protocol_report( + meta_version: Option, + negotiated: Option, +) -> VerifyProtocolResult { + if let Some(version) = meta_version { + return VerifyProtocolResult { + protocol_version: version.to_string(), + transport: "stateless".to_string(), + }; + } + VerifyProtocolResult { + protocol_version: negotiated + .map(|version| version.to_string()) + .unwrap_or_else(|| "unknown".to_string()), + transport: "session".to_string(), + } +} + +#[tool_router] +impl FastTimeServer { + fn new() -> Self { Self { - modern: false, - legacy: true, - strict: false, + state: Arc::new(SharedState::default()), + tool_router: Self::tool_router(), } } -} -impl ServerConfig { - fn supported_versions(&self) -> Vec<&'static str> { - let mut versions = Vec::new(); - if self.modern { - versions.push(MCP_PROTOCOL_VERSION_MODERN); + #[tool(description = "Echo back the provided message.")] + async fn echo( + &self, + Parameters(request): Parameters, + ) -> Result { + let delay = validate_delay(request.delay) + .map_err(|message| McpError::invalid_params(message, None))?; + + self.state.request_count.fetch_add(1, Ordering::Relaxed); + if let Some(ms) = delay + && ms > 0 + { + let actual_ms = compute_delay(ms, request.delay_stddev); + tokio::time::sleep(std::time::Duration::from_millis(actual_ms)).await; } - if self.legacy { - versions.push(MCP_PROTOCOL_VERSION); + Ok(CallToolResult::success(vec![ContentBlock::text( + request.message, + )])) + } + + #[tool( + description = "Return isError=true for the first fail_times calls per key, then succeed (retry testing)." + )] + fn flaky( + &self, + Parameters(request): Parameters, + ) -> Result { + let fail_times = request.fail_times.unwrap_or(0); + + self.state.request_count.fetch_add(1, Ordering::Relaxed); + let mut state = self.state.flaky.lock().unwrap(); + let attempt = { + let counter = state.entry(request.key.clone()).or_insert(0); + *counter += 1; + *counter + }; + if attempt <= fail_times { + Ok(CallToolResult::error(vec![ContentBlock::text(format!( + "flaky transient failure (attempt {attempt}/{fail_times})" + ))])) + } else { + state.remove(&request.key); + Ok(CallToolResult::success(vec![ContentBlock::text(format!( + "flaky recovered after {attempt} attempt(s)" + ))])) } - versions } -} -fn usage() -> String { - format!( - "Usage: {APP_NAME} [--protocol ]... [--strict]\n\ - \n\ - Options:\n\ - \x20 --protocol Also serve the given MCP protocol revision.\n\ - \x20 Supported: {MCP_PROTOCOL_VERSION}, {MCP_PROTOCOL_VERSION_MODERN}.\n\ - \x20 May be repeated. Default: {MCP_PROTOCOL_VERSION} only.\n\ - \x20 --strict Serve exactly the revisions named with --protocol\n\ - \x20 (default {MCP_PROTOCOL_VERSION}) and reject any\n\ - \x20 non-conformant interaction — no fallback, no\n\ - \x20 `initialize` handshake unless {MCP_PROTOCOL_VERSION}\n\ - \x20 was explicitly enabled." - ) -} + #[tool(description = "Get current system time in the specified IANA timezone.")] + fn get_system_time( + &self, + Parameters(request): Parameters, + ) -> Result { + let timezone = request.timezone.as_deref().unwrap_or("UTC"); + + self.state.request_count.fetch_add(1, Ordering::Relaxed); + match parse_timezone(timezone) { + Ok(timezone) => Ok(CallToolResult::success(vec![ContentBlock::text( + timezone.format_utc(Utc::now()), + )])), + Err(err) => Ok(CallToolResult::error(vec![ContentBlock::text(format!( + "Invalid timezone '{timezone}': {err}" + ))])), + } + } -fn parse_args(args: &[String]) -> Result { - let mut modern = false; - let mut legacy_explicit = false; - let mut strict = false; - let mut args = args.iter(); - while let Some(arg) = args.next() { - match arg.as_str() { - "--strict" => strict = true, - "--protocol" => { - let version = args - .next() - .ok_or_else(|| "--protocol requires a version argument".to_string())?; - match version.as_str() { - MCP_PROTOCOL_VERSION => legacy_explicit = true, - MCP_PROTOCOL_VERSION_MODERN => modern = true, - other => { - return Err(format!( - "unsupported protocol version '{other}' \ - (supported: {MCP_PROTOCOL_VERSION}, {MCP_PROTOCOL_VERSION_MODERN})" - )); - } - } + #[tool( + description = "Convert a time value from a source IANA timezone to a target IANA timezone." + )] + fn convert_time( + &self, + Parameters(request): Parameters, + ) -> Result { + self.state.request_count.fetch_add(1, Ordering::Relaxed); + + let source_timezone = match parse_timezone(&request.source_timezone) { + Ok(timezone) => timezone, + Err(err) => { + return Ok(CallToolResult::error(vec![ContentBlock::text(format!( + "invalid source timezone: {err}" + ))])); + } + }; + let target_timezone = match parse_timezone(&request.target_timezone) { + Ok(timezone) => timezone, + Err(err) => { + return Ok(CallToolResult::error(vec![ContentBlock::text(format!( + "invalid target timezone: {err}" + ))])); } - other => return Err(format!("unknown argument '{other}'")), + }; + match parse_time_in_timezone(&request.time, &source_timezone) { + Ok(parsed) => Ok(CallToolResult::success(vec![ContentBlock::text( + target_timezone.format_utc(parsed), + )])), + Err(_) => Ok(CallToolResult::error(vec![ContentBlock::text(format!( + "invalid time format: {}", + request.time + ))])), } } - // Non-strict servers always keep the legacy revision available so older - // clients can fall back to it; strict servers serve exactly the revisions - // named with --protocol and reject everything else. - Ok(ServerConfig { - modern, - legacy: if strict { legacy_explicit } else { true }, - strict, - }) + + #[tool( + description = "Always returns isError=true.", + output_schema = rmcp::handler::server::tool::schema_for_output::() + )] + fn schema_error(&self) -> Result { + self.state.request_count.fetch_add(1, Ordering::Relaxed); + Ok(CallToolResult::error(vec![ContentBlock::text( + "You cannot send more than 200 points", + )])) + } + + #[tool(description = "Returns a JSON payload that conforms to the declared outputSchema.")] + fn schema_success(&self) -> Result, McpError> { + self.state.request_count.fetch_add(1, Ordering::Relaxed); + Ok(Json(RecognitionResult { + recognition_id: "rec-123".to_string(), + message: Some("ok".to_string()), + })) + } + + #[tool(description = "Get server statistics including request count and uptime.")] + fn get_stats(&self) -> Result { + let count = self.state.request_count.load(Ordering::Relaxed); + Ok(CallToolResult::success(vec![ContentBlock::text(format!( + "{{\n \"server\": \"{}\",\n \"version\": \"{}\",\n \"requests_handled\": {}\n}}", + APP_NAME, APP_VERSION, count + ))])) + } + + #[tool( + name = "verify-protocol", + description = "Report the MCP protocol version active for the current request." + )] + fn verify_protocol( + &self, + context: RequestContext, + ) -> Result, McpError> { + let negotiated = context + .peer + .peer_info() + .map(|info| info.protocol_version.clone()); + Ok(Json(protocol_report( + context.meta.protocol_version(), + negotiated, + ))) + } +} + +#[tool_handler(router = self.tool_router)] +impl ServerHandler for FastTimeServer { + fn get_info(&self) -> ServerInfo { + ServerInfo::new(ServerCapabilities::builder().enable_tools().build()) + .with_server_info(Implementation::new(APP_NAME, APP_VERSION)) + .with_protocol_version(ProtocolVersion::V_2025_11_25) + .with_instructions("Ultra-fast MCP test server.".to_string()) + } + + fn supported_protocol_versions(&self) -> Cow<'static, [ProtocolVersion]> { + Cow::Borrowed(SUPPORTED_PROTOCOL_VERSIONS) + } } // ============================================================================ // Main Entry Point // ============================================================================ +fn build_router() -> Router { + let server = Arc::new(FastTimeServer::new()); + let ct = tokio_util::sync::CancellationToken::new(); + let mcp_service = StreamableHttpService::new( + move || Ok(server.clone()), + LocalSessionManager::default().into(), + StreamableHttpServerConfig::default() + .with_json_response(true) + // The pre-SDK server performed no Host validation; keep it open so + // container and LAN benchmarks are not rejected as DNS rebinding. + .with_allowed_hosts(Vec::::new()) + .with_cancellation_token(ct.clone()), + ); + + Router::new() + // Health & version + .route("/health", axum::routing::get(health_handler)) + .route("/version", axum::routing::get(version_handler)) + // REST API for benchmarking (bypasses MCP session overhead) + .route("/api/echo", axum::routing::post(rest_echo_handler)) + .route("/api/time", axum::routing::get(rest_time_handler)) + // MCP protocol endpoint (POST + DELETE; GET opens an SSE stream) + .nest_service("/mcp", mcp_service) +} + #[tokio::main] async fn main() -> anyhow::Result<()> { // Initialize logging @@ -292,20 +494,6 @@ async fn main() -> anyhow::Result<()> { .with(tracing_subscriber::fmt::layer()) .init(); - // Parse CLI arguments - let raw_args: Vec = env::args().skip(1).collect(); - if raw_args.iter().any(|arg| arg == "-h" || arg == "--help") { - println!("{}", usage()); - return Ok(()); - } - let config = match parse_args(&raw_args) { - Ok(config) => Arc::new(config), - Err(err) => { - eprintln!("error: {err}\n\n{}", usage()); - std::process::exit(2); - } - }; - // Get bind address from environment or use default let bind_address = env::var("BIND_ADDRESS").unwrap_or_else(|_| DEFAULT_BIND_ADDRESS.to_string()); @@ -313,27 +501,11 @@ async fn main() -> anyhow::Result<()> { info!("{} v{} starting...", APP_NAME, APP_VERSION); info!("Binding to: {}", bind_address); info!( - "MCP protocol versions: {}", - config.supported_versions().join(", ") + "MCP protocol versions: {}, {}", + MCP_PROTOCOL_VERSION, MCP_PROTOCOL_VERSION_MODERN ); - if config.strict { - info!("Strict negotiation: unsupported versions rejected (no fallback)"); - } - // Build router with health check endpoint and REST API for benchmarking - let router = Router::new() - // Health & version - .route("/health", axum::routing::get(health_handler)) - .route("/version", axum::routing::get(version_handler)) - // REST API for benchmarking (bypasses MCP session overhead) - .route("/api/echo", axum::routing::post(rest_echo_handler)) - .route("/api/time", axum::routing::get(rest_time_handler)) - // MCP protocol endpoint - .route( - "/mcp", - axum::routing::post(mcp_handler).delete(mcp_delete_handler), - ) - .with_state(config); + let router = build_router(); // Bind and serve let tcp_listener = tokio::net::TcpListener::bind(&bind_address) @@ -379,582 +551,15 @@ async fn health_handler() -> axum::Json { } // Version handler -async fn version_handler( - axum::extract::State(config): axum::extract::State>, -) -> axum::Json { +async fn version_handler() -> axum::Json { axum::Json(json!({ "name": APP_NAME, "version": APP_VERSION, "mcp_version": MCP_PROTOCOL_VERSION, - "mcp_versions": config.supported_versions(), - "strict": config.strict + "mcp_versions": [MCP_PROTOCOL_VERSION, MCP_PROTOCOL_VERSION_MODERN] })) } -// ============================================================================ -// Fast Streamable HTTP MCP Handler -// ============================================================================ - -async fn mcp_delete_handler(headers: HeaderMap) -> StatusCode { - let Some(session_id) = mcp_session_id(&headers) else { - return StatusCode::BAD_REQUEST; - }; - if remove_session(session_id) { - StatusCode::OK - } else { - StatusCode::NOT_FOUND - } -} - -async fn mcp_handler( - axum::extract::State(config): axum::extract::State>, - headers: HeaderMap, - axum::Json(req): axum::Json, -) -> Response { - let method = req - .get("method") - .and_then(serde_json::Value::as_str) - .unwrap_or_default(); - let id = req.get("id"); - - if method != "initialize" { - // A request carrying a per-request protocol version in `_meta` speaks - // the modern era and is served statelessly, without a session. - if let Some(requested) = modern_protocol_version(&req) { - return mcp_modern_dispatch(&config, &headers, id, method, requested, &req).await; - } - let Err(status) = mcp_validate_active_session(&headers) else { - if id.is_none() { - return StatusCode::ACCEPTED.into_response(); - } - return match method { - "ping" => mcp_empty_result_response(id, false), - "tools/list" => mcp_tools_list_response(id, false), - "tools/call" => mcp_tools_call_response(id, &req, false).await, - _ => mcp_error_response(id, -32601, "Method not found", None), - }; - }; - if id.is_none() { - return status.into_response(); - } - return mcp_error_response_with_status(status, id, -32000, "Invalid session ID", None); - } - - if id.is_none() { - return StatusCode::ACCEPTED.into_response(); - } - - match method { - "initialize" => mcp_initialize_response(id, &req, &config), - _ => mcp_error_response(id, -32601, "Method not found", None), - } -} - -fn modern_protocol_version(req: &serde_json::Value) -> Option<&str> { - req.get("params")? - .get("_meta")? - .get(PROTOCOL_VERSION_META_KEY)? - .as_str() -} - -async fn mcp_modern_dispatch( - config: &ServerConfig, - headers: &HeaderMap, - id: Option<&serde_json::Value>, - method: &str, - requested: &str, - req: &serde_json::Value, -) -> Response { - // The mirrored MCP-Protocol-Version header must agree with the body; a - // missing header is tolerated (the body `_meta` is authoritative here). - if let Some(header_version) = headers - .get(PROTOCOL_VERSION_HEADER) - .and_then(|value| value.to_str().ok()) - && header_version != requested - { - return mcp_error_response_with_status( - StatusCode::BAD_REQUEST, - id, - ERR_HEADER_MISMATCH, - &format!( - "Header mismatch: {PROTOCOL_VERSION_HEADER} header value \ - '{header_version}' does not match body value '{requested}'" - ), - None, - ); - } - if !config.modern || requested != MCP_PROTOCOL_VERSION_MODERN { - return mcp_error_response_with_status( - StatusCode::BAD_REQUEST, - id, - ERR_UNSUPPORTED_PROTOCOL_VERSION, - "Unsupported protocol version", - Some(json!({ - "supported": config.supported_versions(), - "requested": requested, - })), - ); - } - let Some(id) = id else { - return StatusCode::ACCEPTED.into_response(); - }; - match method { - "ping" => mcp_empty_result_response(Some(id), true), - "server/discover" => mcp_discover_response(Some(id), config), - "tools/list" => mcp_tools_list_response(Some(id), true), - "tools/call" => mcp_tools_call_response(Some(id), req, true).await, - _ => mcp_error_response_with_status( - StatusCode::NOT_FOUND, - Some(id), - -32601, - "Method not found", - None, - ), - } -} - -fn mcp_discover_response(id: Option<&serde_json::Value>, config: &ServerConfig) -> Response { - mcp_json_response( - json!({ - "jsonrpc": "2.0", - "id": id.cloned().unwrap_or(serde_json::Value::Null), - "result": { - "resultType": "complete", - "supportedVersions": config.supported_versions(), - "capabilities": { "tools": {} }, - "serverInfo": { "name": APP_NAME, "version": APP_VERSION }, - "cacheScope": "private", - "ttlMs": 0, - "_meta": { - SERVER_INFO_META_KEY: { "name": APP_NAME, "version": APP_VERSION } - }, - "instructions": "Ultra-fast MCP test server." - } - }) - .to_string(), - ) -} - -fn mcp_json_response(body: String) -> Response { - ([(header::CONTENT_TYPE, "application/json")], body).into_response() -} - -fn mcp_id_json(id: Option<&serde_json::Value>) -> String { - id.and_then(|value| serde_json::to_string(value).ok()) - .unwrap_or_else(|| "null".to_string()) -} - -fn mcp_initialize_response( - id: Option<&serde_json::Value>, - req: &serde_json::Value, - config: &ServerConfig, -) -> Response { - let requested = req - .get("params") - .and_then(|params| params.get("protocolVersion")) - .and_then(serde_json::Value::as_str); - if config.strict && (!config.legacy || requested != Some(MCP_PROTOCOL_VERSION)) { - return mcp_error_response( - id, - ERR_UNSUPPORTED_PROTOCOL_VERSION, - "Unsupported protocol version", - Some(json!({ - "supported": config.supported_versions(), - "requested": requested, - })), - ); - } - let session_id = Uuid::new_v4().to_string(); - let session_header = HeaderValue::from_str(&session_id) - .unwrap_or_else(|_| HeaderValue::from_static("fast-time")); - if !remember_session(session_id) { - return mcp_error_response_with_status( - StatusCode::SERVICE_UNAVAILABLE, - id, - -32000, - "Maximum active sessions reached", - Some(json!({ "max_sessions": MAX_ACTIVE_SESSIONS })), - ); - } - let mut response = mcp_json_response(format!( - r#"{{"jsonrpc":"2.0","id":{},"result":{{"protocolVersion":"{}","capabilities":{{"tools":{{}}}},"serverInfo":{{"name":"{}","version":"{}"}},"instructions":"Ultra-fast MCP test server."}}}}"#, - mcp_id_json(id), - MCP_PROTOCOL_VERSION, - APP_NAME, - APP_VERSION - )); - response - .headers_mut() - .insert(SESSION_HEADER, session_header); - response -} - -fn remember_session(session_id: String) -> bool { - if let Ok(mut sessions) = ACTIVE_SESSIONS.write() { - remember_session_in(&mut sessions, session_id) - } else { - false - } -} - -fn remember_session_in(sessions: &mut HashSet, session_id: String) -> bool { - if sessions.len() >= MAX_ACTIVE_SESSIONS { - return false; - } - sessions.insert(session_id) -} - -fn remove_session(session_id: &str) -> bool { - ACTIVE_SESSIONS - .write() - .map(|mut sessions| sessions.remove(session_id)) - .unwrap_or(false) -} - -fn mcp_validate_active_session(headers: &HeaderMap) -> Result<(), StatusCode> { - let Some(session_id) = mcp_session_id(headers) else { - return Err(StatusCode::BAD_REQUEST); - }; - if ACTIVE_SESSIONS - .read() - .map(|sessions| sessions.contains(session_id)) - .unwrap_or(false) - { - Ok(()) - } else { - Err(StatusCode::NOT_FOUND) - } -} - -fn result_type_field(modern: bool) -> &'static str { - if modern { - r#""resultType":"complete","# - } else { - "" - } -} - -// CacheableResult responses (tools/list, server/discover) additionally -// require cacheScope and ttlMs on the 2026-07-28 wire. -fn cacheable_result_fields(modern: bool) -> &'static str { - if modern { - r#""resultType":"complete","cacheScope":"private","ttlMs":0,"# - } else { - "" - } -} - -fn mcp_tools_list_response(id: Option<&serde_json::Value>, modern: bool) -> Response { - mcp_json_response(format!( - r#"{{"jsonrpc":"2.0","id":{},"result":{{{}"tools":[{{"name":"echo","description":"Echo back the provided message.","inputSchema":{{"type":"object","properties":{{"message":{{"type":"string"}},"delay":{{"type":"integer","minimum":0,"maximum":60000}},"delay_stddev":{{"type":"number","minimum":0}}}},"required":["message"]}}}},{{"name":"flaky","description":"Return isError=true for the first fail_times calls per key, then succeed (retry testing).","inputSchema":{{"type":"object","properties":{{"key":{{"type":"string","description":"Unique key to track attempt count across retries"}},"fail_times":{{"type":"integer","minimum":0,"description":"Number of times to return isError=true before succeeding (default 0)"}}}},"required":["key"]}}}},{{"name":"get_system_time","description":"Get current system time in the specified IANA timezone.","inputSchema":{{"type":"object","properties":{{"timezone":{{"type":"string"}}}}}}}},{{"name":"convert_time","description":"Convert a time value from a source IANA timezone to a target IANA timezone.","inputSchema":{{"type":"object","properties":{{"time":{{"type":"string"}},"source_timezone":{{"type":"string"}},"target_timezone":{{"type":"string"}}}},"required":["time","source_timezone","target_timezone"]}}}},{{"name":"schema_error","description":"Always returns isError=true.","inputSchema":{{"type":"object","properties":{{}}}},"outputSchema":{{"type":"object","properties":{{"recognitionId":{{"type":"string"}},"message":{{"type":"string"}}}},"required":["recognitionId"]}}}},{{"name":"schema_success","description":"Returns a JSON payload that conforms to the declared outputSchema.","inputSchema":{{"type":"object","properties":{{}}}},"outputSchema":{{"type":"object","properties":{{"recognitionId":{{"type":"string"}},"message":{{"type":"string"}}}},"required":["recognitionId"]}}}},{{"name":"get_stats","description":"Get server statistics including request count and uptime.","inputSchema":{{"type":"object","properties":{{}}}}}}]}}}}"#, - mcp_id_json(id), - cacheable_result_fields(modern) - )) -} - -fn mcp_session_id(headers: &HeaderMap) -> Option<&str> { - headers.get(SESSION_HEADER)?.to_str().ok() -} - -async fn mcp_tools_call_response( - id: Option<&serde_json::Value>, - req: &serde_json::Value, - modern: bool, -) -> Response { - let params = req.get("params").unwrap_or(&serde_json::Value::Null); - let name = params - .get("name") - .and_then(serde_json::Value::as_str) - .unwrap_or_default(); - let arguments = params.get("arguments").unwrap_or(&serde_json::Value::Null); - - match name { - "echo" => { - let Some(arguments) = mcp_arguments_object(id, arguments) else { - return mcp_invalid_params_response(id, "arguments must be an object"); - }; - let Some(message) = mcp_required_string(id, arguments, "message") else { - return mcp_invalid_params_response(id, "message must be a string"); - }; - let Some(delay) = mcp_optional_u64(id, arguments, "delay") else { - return mcp_invalid_params_response(id, "delay must be an unsigned integer"); - }; - let Some(delay_stddev) = mcp_optional_f64(id, arguments, "delay_stddev") else { - return mcp_invalid_params_response(id, "delay_stddev must be a number"); - }; - let Ok(delay) = validate_delay(delay) else { - return mcp_invalid_params_response(id, "delay exceeds the 60000 ms limit"); - }; - - DIRECT_REQUEST_COUNT.fetch_add(1, Ordering::Relaxed); - if let Some(ms) = delay - && ms > 0 - { - let actual_ms = compute_delay(ms, delay_stddev); - tokio::time::sleep(std::time::Duration::from_millis(actual_ms)).await; - } - mcp_text_result_response(id, message, false, modern) - } - "flaky" => { - let Some(arguments) = mcp_arguments_object(id, arguments) else { - return mcp_invalid_params_response(id, "arguments must be an object"); - }; - let Some(key) = mcp_required_string(id, arguments, "key") else { - return mcp_invalid_params_response(id, "key must be a string"); - }; - let Some(fail_times) = mcp_optional_u64(id, arguments, "fail_times") else { - return mcp_invalid_params_response(id, "fail_times must be an unsigned integer"); - }; - let fail_times = fail_times.unwrap_or(0); - - DIRECT_REQUEST_COUNT.fetch_add(1, Ordering::Relaxed); - let mut state = FLAKY_STATE.lock().unwrap(); - let attempt = { - let counter = state.entry(key.to_string()).or_insert(0); - *counter += 1; - *counter - }; - if attempt <= fail_times { - mcp_text_result_response( - id, - &format!("flaky transient failure (attempt {attempt}/{fail_times})"), - true, - modern, - ) - } else { - state.remove(key); - mcp_text_result_response( - id, - &format!("flaky recovered after {attempt} attempt(s)"), - false, - modern, - ) - } - } - "get_system_time" => { - let timezone = if arguments.is_null() { - None - } else { - let Some(arguments) = mcp_arguments_object(id, arguments) else { - return mcp_invalid_params_response(id, "arguments must be an object"); - }; - let Some(timezone) = mcp_optional_string(id, arguments, "timezone") else { - return mcp_invalid_params_response(id, "timezone must be a string"); - }; - timezone - }; - let timezone = timezone.unwrap_or("UTC"); - - DIRECT_REQUEST_COUNT.fetch_add(1, Ordering::Relaxed); - match parse_timezone(timezone) { - Ok(timezone) => { - mcp_text_result_response(id, &timezone.format_utc(Utc::now()), false, modern) - } - Err(err) => mcp_text_result_response( - id, - &format!("Invalid timezone '{timezone}': {err}"), - true, - modern, - ), - } - } - "convert_time" => mcp_convert_time_response(id, arguments, modern), - "schema_error" => { - DIRECT_REQUEST_COUNT.fetch_add(1, Ordering::Relaxed); - mcp_text_result_response(id, "You cannot send more than 200 points", true, modern) - } - "schema_success" => mcp_json_response({ - DIRECT_REQUEST_COUNT.fetch_add(1, Ordering::Relaxed); - format!( - r#"{{"jsonrpc":"2.0","id":{},"result":{{{}"content":[{{"type":"text","text":"{{\"recognitionId\":\"rec-123\",\"message\":\"ok\"}}"}}],"structuredContent":{{"recognitionId":"rec-123","message":"ok"}},"isError":false}}}}"#, - mcp_id_json(id), - result_type_field(modern) - ) - }), - "get_stats" => { - let count = DIRECT_REQUEST_COUNT.load(Ordering::Relaxed); - mcp_text_result_response( - id, - &format!( - "{{\n \"server\": \"{}\",\n \"version\": \"{}\",\n \"requests_handled\": {}\n}}", - APP_NAME, APP_VERSION, count - ), - false, - modern, - ) - } - _ => mcp_error_response(id, -32602, "Unknown tool", Some(json!({ "tool": name }))), - } -} - -fn mcp_convert_time_response( - id: Option<&serde_json::Value>, - arguments: &serde_json::Value, - modern: bool, -) -> Response { - let Some(arguments) = mcp_arguments_object(id, arguments) else { - return mcp_invalid_params_response(id, "arguments must be an object"); - }; - let Some(time) = mcp_required_string(id, arguments, "time") else { - return mcp_invalid_params_response(id, "time must be a string"); - }; - let Some(source_timezone) = mcp_required_string(id, arguments, "source_timezone") else { - return mcp_invalid_params_response(id, "source_timezone must be a string"); - }; - let Some(target_timezone) = mcp_required_string(id, arguments, "target_timezone") else { - return mcp_invalid_params_response(id, "target_timezone must be a string"); - }; - - DIRECT_REQUEST_COUNT.fetch_add(1, Ordering::Relaxed); - - let source_timezone = match parse_timezone(source_timezone) { - Ok(timezone) => timezone, - Err(err) => { - return mcp_text_result_response( - id, - &format!("invalid source timezone: {err}"), - true, - modern, - ); - } - }; - let target_timezone = match parse_timezone(target_timezone) { - Ok(timezone) => timezone, - Err(err) => { - return mcp_text_result_response( - id, - &format!("invalid target timezone: {err}"), - true, - modern, - ); - } - }; - match parse_time_in_timezone(time, &source_timezone) { - Ok(parsed) => { - let converted = target_timezone.format_utc(parsed); - mcp_text_result_response(id, &converted, false, modern) - } - Err(_) => { - mcp_text_result_response(id, &format!("invalid time format: {time}"), true, modern) - } - } -} - -fn mcp_arguments_object<'a>( - _id: Option<&serde_json::Value>, - value: &'a serde_json::Value, -) -> Option<&'a serde_json::Map> { - value.as_object() -} - -fn mcp_required_string<'a>( - _id: Option<&serde_json::Value>, - arguments: &'a serde_json::Map, - field: &str, -) -> Option<&'a str> { - arguments.get(field)?.as_str() -} - -fn mcp_optional_string<'a>( - _id: Option<&serde_json::Value>, - arguments: &'a serde_json::Map, - field: &str, -) -> Option> { - match arguments.get(field) { - Some(value) if value.is_null() => Some(None), - Some(value) => value.as_str().map(Some), - None => Some(None), - } -} - -fn mcp_optional_u64( - _id: Option<&serde_json::Value>, - arguments: &serde_json::Map, - field: &str, -) -> Option> { - match arguments.get(field) { - Some(value) if value.is_null() => Some(None), - Some(value) => value.as_u64().map(Some), - None => Some(None), - } -} - -fn mcp_optional_f64( - _id: Option<&serde_json::Value>, - arguments: &serde_json::Map, - field: &str, -) -> Option> { - match arguments.get(field) { - Some(value) if value.is_null() => Some(None), - Some(value) => value.as_f64().map(Some), - None => Some(None), - } -} - -fn mcp_text_result_response( - id: Option<&serde_json::Value>, - text: &str, - is_error: bool, - modern: bool, -) -> Response { - let escaped = serde_json::to_string(text).unwrap_or_else(|_| "\"\"".to_string()); - mcp_json_response(format!( - r#"{{"jsonrpc":"2.0","id":{},"result":{{{}"content":[{{"type":"text","text":{}}}],"isError":{}}}}}"#, - mcp_id_json(id), - result_type_field(modern), - escaped, - is_error - )) -} - -fn mcp_empty_result_response(id: Option<&serde_json::Value>, modern: bool) -> Response { - let result = if modern { - r#"{"resultType":"complete"}"# - } else { - "{}" - }; - mcp_json_response(format!( - r#"{{"jsonrpc":"2.0","id":{},"result":{}}}"#, - mcp_id_json(id), - result - )) -} - -fn mcp_error_response( - id: Option<&serde_json::Value>, - code: i32, - message: &str, - data: Option, -) -> Response { - mcp_error_response_with_status(StatusCode::OK, id, code, message, data) -} - -fn mcp_error_response_with_status( - status: StatusCode, - id: Option<&serde_json::Value>, - code: i32, - message: &str, - data: Option, -) -> Response { - let escaped_message = serde_json::to_string(message).unwrap_or_else(|_| "\"\"".to_string()); - let data = data - .map(|value| format!(r#","data":{}"#, value)) - .unwrap_or_default(); - let mut response = mcp_json_response(format!( - r#"{{"jsonrpc":"2.0","id":{},"error":{{"code":{},"message":{}{}}}}}"#, - mcp_id_json(id), - code, - escaped_message, - data - )); - *response.status_mut() = status; - response -} - -fn mcp_invalid_params_response(id: Option<&serde_json::Value>, message: &str) -> Response { - mcp_error_response(id, -32602, message, None) -} - // ============================================================================ // REST API Handlers (for benchmarking - bypasses MCP session overhead) // ============================================================================ @@ -1018,10 +623,14 @@ async fn rest_time_handler( mod tests { use super::*; use axum::body; + use axum::http::{HeaderValue, Request, StatusCode}; + use tower::ServiceExt; - fn default_state() -> axum::extract::State> { - axum::extract::State(Arc::new(ServerConfig::default())) - } + const MCP_ACCEPT: &str = "application/json, text/event-stream"; + const SESSION_HEADER: &str = "mcp-session-id"; + const PROTOCOL_VERSION_HEADER: &str = "mcp-protocol-version"; + const PROTOCOL_VERSION_META_KEY: &str = "io.modelcontextprotocol/protocolVersion"; + const CLIENT_CAPABILITIES_META_KEY: &str = "io.modelcontextprotocol/clientCapabilities"; #[test] fn test_parse_utc() { @@ -1091,53 +700,53 @@ mod tests { } #[test] - fn test_server_advertises_latest_protocol() { - assert_eq!(MCP_PROTOCOL_VERSION, "2025-11-25"); + fn test_delay_validation_rejects_values_above_limit() { + assert_eq!(validate_delay(Some(MAX_DELAY_MS)), Ok(Some(MAX_DELAY_MS))); + assert!(validate_delay(Some(MAX_DELAY_MS + 1)).is_err()); } #[test] - fn test_active_session_validation() { - let session_id = "unit-test-session-validation"; - remove_session(session_id); - - let mut headers = HeaderMap::new(); - headers.insert(SESSION_HEADER, HeaderValue::from_static(session_id)); + fn test_supported_protocol_versions_advertises_exactly_two_eras() { + let server = FastTimeServer::new(); assert_eq!( - mcp_validate_active_session(&headers), - Err(StatusCode::NOT_FOUND) - ); - - assert!(remember_session(session_id.to_string())); - assert_eq!(mcp_validate_active_session(&headers), Ok(())); - - assert!(remove_session(session_id)); - assert_eq!( - mcp_validate_active_session(&headers), - Err(StatusCode::NOT_FOUND) + server.supported_protocol_versions().as_ref(), + [ProtocolVersion::V_2025_11_25, ProtocolVersion::V_2026_07_28] ); } #[test] - fn test_session_cap_rejects_new_session_when_full() { - let mut sessions = HashSet::with_capacity(MAX_ACTIVE_SESSIONS); - for idx in 0..MAX_ACTIVE_SESSIONS { - assert!(remember_session_in( - &mut sessions, - format!("test-session-{idx}") - )); - } + fn test_protocol_report_modern_meta_wins() { + let report = protocol_report(Some(ProtocolVersion::V_2026_07_28), None); + assert_eq!(report.protocol_version, "2026-07-28"); + assert_eq!(report.transport, "stateless"); + } - assert!(!remember_session_in( - &mut sessions, - "overflow-session".to_string() - )); - assert_eq!(sessions.len(), MAX_ACTIVE_SESSIONS); + #[test] + fn test_protocol_report_legacy_falls_back_to_negotiated() { + let report = protocol_report(None, Some(ProtocolVersion::V_2025_11_25)); + assert_eq!(report.protocol_version, "2025-11-25"); + assert_eq!(report.transport, "session"); } #[test] - fn test_delay_validation_rejects_values_above_limit() { - assert_eq!(validate_delay(Some(MAX_DELAY_MS)), Ok(Some(MAX_DELAY_MS))); - assert!(validate_delay(Some(MAX_DELAY_MS + 1)).is_err()); + fn test_protocol_report_without_any_version_is_unknown() { + let report = protocol_report(None, None); + assert_eq!(report.protocol_version, "unknown"); + assert_eq!(report.transport, "session"); + } + + // ======================================================================== + // HTTP integration helpers (tower oneshot against the real router) + // ======================================================================== + + fn mcp_post(body: serde_json::Value) -> Request { + Request::builder() + .method("POST") + .uri("http://localhost/mcp") + .header(header::CONTENT_TYPE, "application/json") + .header(header::ACCEPT, MCP_ACCEPT) + .body(axum::body::Body::from(body.to_string())) + .expect("request should build") } async fn response_text(response: Response) -> String { @@ -1147,263 +756,307 @@ mod tests { String::from_utf8(bytes.to_vec()).expect("response body should be utf-8") } - async fn response_json(response: Response) -> serde_json::Value { - serde_json::from_str(&response_text(response).await).expect("response body should be json") + /// Legacy session requests are answered as SSE streams; the JSON-RPC + /// message rides in the first non-empty `data:` line. + fn parse_sse_json(text: &str) -> serde_json::Value { + for line in text.lines() { + if let Some(data) = line.strip_prefix("data:") { + let data = data.trim(); + if !data.is_empty() { + return serde_json::from_str(data).expect("SSE data should be JSON"); + } + } + } + panic!("no SSE data line in response body: {text:?}"); } - async fn initialized_headers() -> HeaderMap { - let response = mcp_handler( - default_state(), - HeaderMap::new(), - axum::Json(json!({ - "jsonrpc": "2.0", - "method": "initialize", - "params": { - "protocolVersion": "2024-11-05", - "capabilities": {}, - "clientInfo": { - "name": "go-parity", - "version": "1.0" - } - }, - "id": 1 - })), + async fn oneshot(router: &Router, request: Request) -> Response { + tokio::time::timeout( + std::time::Duration::from_secs(10), + router.clone().oneshot(request), ) - .await; - let mut headers = HeaderMap::new(); - headers.insert( - SESSION_HEADER, - response - .headers() - .get(SESSION_HEADER) - .expect("initialize should issue session id") - .clone(), - ); - headers + .await + .expect("request timed out") + .expect("router should be infallible") } - #[tokio::test] - async fn test_version_endpoint_advertises_latest_protocol() { - let version = version_handler(default_state()).await; - assert_eq!(version.0["mcp_version"], MCP_PROTOCOL_VERSION); + fn initialize_request(protocol_version: &str) -> serde_json::Value { + json!({ + "jsonrpc": "2.0", + "method": "initialize", + "params": { + "protocolVersion": protocol_version, + "capabilities": {}, + "clientInfo": { "name": "test", "version": "1.0" } + }, + "id": 1 + }) } - #[tokio::test] - async fn test_initialize_accepts_older_protocol_and_advertises_latest() { - let response = mcp_handler( - default_state(), - HeaderMap::new(), - axum::Json(json!({ - "jsonrpc": "2.0", - "method": "initialize", - "params": { - "protocolVersion": "2024-11-05", - "capabilities": {}, - "clientInfo": { - "name": "go-compat-smoke", - "version": "1.0" - } - }, - "id": 1 - })), - ) - .await; - + /// Run the legacy handshake and return the issued session id. + async fn initialize_session(router: &Router) -> String { + let response = oneshot(router, mcp_post(initialize_request(MCP_PROTOCOL_VERSION))).await; assert_eq!(response.status(), StatusCode::OK); - assert!(response.headers().contains_key(SESSION_HEADER)); let session_id = response .headers() .get(SESSION_HEADER) - .expect("initialize should issue session id") + .expect("initialize should issue a session id") .to_str() .expect("session id should be ascii") .to_string(); - assert!(Uuid::parse_str(&session_id).is_ok()); - assert!(!session_id.starts_with("fast-time-")); - let body = response_text(response).await; - assert!(body.contains(r#""protocolVersion":"2025-11-25""#)); - assert!(remove_session(&session_id)); + assert!(!session_id.is_empty()); + let body = parse_sse_json(&response_text(response).await); + assert_eq!(body["result"]["protocolVersion"], MCP_PROTOCOL_VERSION); + + let mut initialized = mcp_post(json!({ + "jsonrpc": "2.0", + "method": "notifications/initialized" + })); + initialized + .headers_mut() + .insert(SESSION_HEADER, HeaderValue::from_str(&session_id).unwrap()); + let response = oneshot(router, initialized).await; + assert_eq!(response.status(), StatusCode::ACCEPTED); + + session_id + } + + async fn legacy_tool_call( + router: &Router, + session_id: &str, + name: &str, + arguments: serde_json::Value, + id: i64, + ) -> serde_json::Value { + let mut request = mcp_post(json!({ + "jsonrpc": "2.0", + "method": "tools/call", + "params": { "name": name, "arguments": arguments }, + "id": id + })); + request + .headers_mut() + .insert(SESSION_HEADER, HeaderValue::from_str(session_id).unwrap()); + let response = oneshot(router, request).await; + assert_eq!(response.status(), StatusCode::OK); + parse_sse_json(&response_text(response).await) } - #[tokio::test] - async fn test_flaky_fails_then_succeeds() { - let key = format!("test-flaky-{}", uuid::Uuid::new_v4()); - // First two calls should be errors - for attempt in 1..=2u64 { - let response = mcp_handler( - default_state(), - initialized_headers().await, - axum::Json(json!({ - "jsonrpc": "2.0", - "method": "tools/call", - "params": { - "name": "flaky", - "arguments": { - "key": key, - "fail_times": 2 - } - }, - "id": 100 + attempt - })), - ) - .await; - let body = response_json(response).await; - assert_eq!( - body["result"]["isError"], true, - "attempt {attempt} should be isError" - ); - } - // Third call should succeed - let response = mcp_handler( - default_state(), - initialized_headers().await, - axum::Json(json!({ - "jsonrpc": "2.0", - "method": "tools/call", - "params": { - "name": "flaky", - "arguments": { - "key": key, - "fail_times": 2 - } - }, - "id": 103 - })), - ) - .await; - let body = response_json(response).await; - assert_eq!( - body["result"]["isError"], false, - "third attempt should succeed" - ); - assert!( - body["result"]["content"][0]["text"] - .as_str() - .unwrap() - .contains("flaky recovered after 3 attempt(s)"), + fn modern_request(method: &str, version: &str, id: i64) -> serde_json::Value { + json!({ + "jsonrpc": "2.0", + "method": method, + "params": { + "_meta": { + PROTOCOL_VERSION_META_KEY: version, + CLIENT_CAPABILITIES_META_KEY: {} + } + }, + "id": id + }) + } + + /// The MCP-Protocol-Version header must mirror the version in `_meta`, + /// and 2026-07-28 requests must carry SEP-2243 headers: `Mcp-Method` + /// matching the body method, plus `Mcp-Name` for named methods. + async fn modern_call( + router: &Router, + body: serde_json::Value, + ) -> (StatusCode, serde_json::Value) { + let version = body["params"]["_meta"][PROTOCOL_VERSION_META_KEY] + .as_str() + .expect("modern request should carry a version") + .to_string(); + let method = body["method"].as_str().expect("request method").to_string(); + let name = body["params"]["name"].as_str().map(str::to_string); + let mut request = mcp_post(body); + let headers = request.headers_mut(); + headers.insert( + PROTOCOL_VERSION_HEADER, + HeaderValue::from_str(&version).unwrap(), ); + headers.insert("mcp-method", HeaderValue::from_str(&method).unwrap()); + if let Some(name) = name { + headers.insert("mcp-name", HeaderValue::from_str(&name).unwrap()); + } + let response = oneshot(router, request).await; + let status = response.status(); + let text = response_text(response).await; + let body = serde_json::from_str(&text).expect("modern responses should be JSON"); + (status, body) } + async fn modern_tool_call( + router: &Router, + name: &str, + arguments: serde_json::Value, + id: i64, + ) -> (StatusCode, serde_json::Value) { + let mut request = modern_request("tools/call", MCP_PROTOCOL_VERSION_MODERN, id); + request["params"]["name"] = json!(name); + request["params"]["arguments"] = arguments; + modern_call(router, request).await + } + + // ======================================================================== + // Legacy era (2025-11-25): initialize handshake + mcp-session-id sessions + // ======================================================================== + #[tokio::test] - async fn test_direct_mcp_session_lifecycle_matches_streamable_http() { - let initialize = mcp_handler( - default_state(), - HeaderMap::new(), - axum::Json(json!({ - "jsonrpc": "2.0", - "method": "initialize", - "params": { - "protocolVersion": "2024-11-05", - "capabilities": {}, - "clientInfo": { - "name": "go-compat-smoke", - "version": "1.0" - } - }, - "id": 1 - })), + async fn test_initialize_issues_session_and_echoes_legacy_version() { + let response = oneshot( + &build_router(), + mcp_post(initialize_request(MCP_PROTOCOL_VERSION)), ) .await; - let session_id = initialize - .headers() - .get(SESSION_HEADER) - .expect("initialize should issue session id") - .clone(); - - let mut valid_headers = HeaderMap::new(); - valid_headers.insert(SESSION_HEADER, session_id.clone()); - let valid = mcp_handler( - default_state(), - valid_headers.clone(), - axum::Json(json!({ - "jsonrpc": "2.0", - "method": "tools/list", - "id": 2 - })), - ) - .await; - assert_eq!(valid.status(), StatusCode::OK); + assert_eq!(response.status(), StatusCode::OK); + assert!(response.headers().contains_key(SESSION_HEADER)); + let body = parse_sse_json(&response_text(response).await); + let result = &body["result"]; + assert_eq!(result["protocolVersion"], MCP_PROTOCOL_VERSION); + assert_eq!(result["serverInfo"]["name"], APP_NAME); + assert!(result["capabilities"]["tools"].is_object()); + } - let ping = mcp_handler( - default_state(), - valid_headers.clone(), - axum::Json(json!({ - "jsonrpc": "2.0", - "method": "ping", - "id": 6 - })), - ) - .await; - assert_eq!(ping.status(), StatusCode::OK); - let ping_body = response_json(ping).await; - assert_eq!(ping_body["result"], json!({})); - - let missing = mcp_handler( - default_state(), - HeaderMap::new(), - axum::Json(json!({ + #[tokio::test] + async fn test_initialize_falls_back_to_legacy_for_unknown_version() { + let response = oneshot(&build_router(), mcp_post(initialize_request("1999-01-01"))).await; + assert_eq!(response.status(), StatusCode::OK); + let body = parse_sse_json(&response_text(response).await); + assert_eq!(body["result"]["protocolVersion"], MCP_PROTOCOL_VERSION); + } + + #[tokio::test] + async fn test_legacy_session_lifecycle() { + let router = build_router(); + let session_id = initialize_session(&router).await; + + let mut list = mcp_post(json!({ + "jsonrpc": "2.0", + "method": "tools/list", + "id": 2 + })); + list.headers_mut() + .insert(SESSION_HEADER, HeaderValue::from_str(&session_id).unwrap()); + let response = oneshot(&router, list).await; + assert_eq!(response.status(), StatusCode::OK); + let body = parse_sse_json(&response_text(response).await); + assert_eq!(body["result"]["tools"].as_array().map(Vec::len), Some(8)); + + let response = oneshot( + &router, + mcp_post(json!({ "jsonrpc": "2.0", "method": "tools/list", "id": 3 })), ) .await; - assert_eq!(missing.status(), StatusCode::BAD_REQUEST); - - let mut fake_headers = HeaderMap::new(); - fake_headers.insert(SESSION_HEADER, HeaderValue::from_static("fake-session")); - let fake = mcp_handler( - default_state(), - fake_headers, - axum::Json(json!({ - "jsonrpc": "2.0", - "method": "tools/list", - "id": 4 - })), - ) - .await; - assert_eq!(fake.status(), StatusCode::NOT_FOUND); + assert_eq!(response.status(), StatusCode::UNPROCESSABLE_ENTITY); + + let mut fake = mcp_post(json!({ + "jsonrpc": "2.0", + "method": "tools/list", + "id": 4 + })); + fake.headers_mut() + .insert(SESSION_HEADER, HeaderValue::from_static("fake-session")); + let response = oneshot(&router, fake).await; + assert_eq!(response.status(), StatusCode::NOT_FOUND); + let delete = Request::builder() + .method("DELETE") + .uri("http://localhost/mcp") + .header(SESSION_HEADER, HeaderValue::from_str(&session_id).unwrap()) + .body(axum::body::Body::empty()) + .expect("request should build"); + let response = oneshot(&router, delete).await; + assert_eq!(response.status(), StatusCode::ACCEPTED); + + let mut gone = mcp_post(json!({ + "jsonrpc": "2.0", + "method": "tools/list", + "id": 5 + })); + gone.headers_mut() + .insert(SESSION_HEADER, HeaderValue::from_str(&session_id).unwrap()); + let response = oneshot(&router, gone).await; + assert_eq!(response.status(), StatusCode::NOT_FOUND); + } + + #[tokio::test] + async fn test_legacy_verify_protocol_reports_session() { + let router = build_router(); + let session_id = initialize_session(&router).await; + let body = legacy_tool_call(&router, &session_id, "verify-protocol", json!({}), 10).await; + let result = &body["result"]; + assert_eq!(result["isError"], false); assert_eq!( - mcp_delete_handler(valid_headers.clone()).await, - StatusCode::OK + result["structuredContent"], + json!({ + "protocolVersion": MCP_PROTOCOL_VERSION, + "transport": "session" + }) ); - let deleted = mcp_handler( - default_state(), - valid_headers, - axum::Json(json!({ - "jsonrpc": "2.0", - "method": "tools/list", - "id": 5 - })), + let text: serde_json::Value = + serde_json::from_str(result["content"][0]["text"].as_str().expect("text content")) + .expect("text content should mirror the structured payload"); + assert_eq!(text["protocolVersion"], MCP_PROTOCOL_VERSION); + assert_eq!(text["transport"], "session"); + } + + #[tokio::test] + async fn test_flaky_fails_then_succeeds() { + let router = build_router(); + let session_id = initialize_session(&router).await; + let key = "test-flaky-sdk"; + for attempt in 1..=2i64 { + let body = legacy_tool_call( + &router, + &session_id, + "flaky", + json!({ "key": key, "fail_times": 2 }), + 100 + attempt, + ) + .await; + assert_eq!( + body["result"]["isError"], true, + "attempt {attempt} should be isError" + ); + } + let body = legacy_tool_call( + &router, + &session_id, + "flaky", + json!({ "key": key, "fail_times": 2 }), + 103, ) .await; - assert_eq!(deleted.status(), StatusCode::NOT_FOUND); + assert_eq!(body["result"]["isError"], false); + assert!( + body["result"]["content"][0]["text"] + .as_str() + .unwrap() + .contains("flaky recovered after 3 attempt(s)"), + ); } #[tokio::test] async fn test_convert_time_matches_go_fast_time_dst_behavior() { - let response = mcp_handler( - default_state(), - initialized_headers().await, - axum::Json(json!({ - "jsonrpc": "2.0", - "method": "tools/call", - "params": { - "name": "convert_time", - "arguments": { - "time": "2025-06-21T16:00:00Z", - "source_timezone": "UTC", - "target_timezone": "America/New_York" - } - }, - "id": 10 - })), + let router = build_router(); + let session_id = initialize_session(&router).await; + let body = legacy_tool_call( + &router, + &session_id, + "convert_time", + json!({ + "time": "2025-06-21T16:00:00Z", + "source_timezone": "UTC", + "target_timezone": "America/New_York" + }), + 11, ) .await; - - assert_eq!(response.status(), StatusCode::OK); - let body = response_json(response).await; assert_eq!( body["result"]["content"][0]["text"], "2025-06-21T12:00:00-04:00" @@ -1412,391 +1065,302 @@ mod tests { #[tokio::test] async fn test_convert_time_matches_go_fast_time_half_hour_zones() { - let response = mcp_handler( - default_state(), - initialized_headers().await, - axum::Json(json!({ - "jsonrpc": "2.0", - "method": "tools/call", - "params": { - "name": "convert_time", - "arguments": { - "time": "2025-01-10 10:00:00", - "source_timezone": "Asia/Kolkata", - "target_timezone": "UTC" - } - }, - "id": 11 - })), + let router = build_router(); + let session_id = initialize_session(&router).await; + let body = legacy_tool_call( + &router, + &session_id, + "convert_time", + json!({ + "time": "2025-01-10 10:00:00", + "source_timezone": "Asia/Kolkata", + "target_timezone": "UTC" + }), + 12, ) .await; - - assert_eq!(response.status(), StatusCode::OK); - let body = response_json(response).await; assert_eq!(body["result"]["content"][0]["text"], "2025-01-10T04:30:00Z"); } #[tokio::test] - async fn test_error_response_escapes_dynamic_message_text() { - let response = mcp_error_response_with_status( - StatusCode::BAD_REQUEST, - Some(&json!(99)), - -32602, - r#"bad "message" } ,"injected":true"#, - None, - ); - - assert_eq!(response.status(), StatusCode::BAD_REQUEST); - let body = response_json(response).await; - assert_eq!( - body["error"]["message"], - r#"bad "message" } ,"injected":true"# - ); - assert!(body["error"].get("injected").is_none()); - } - - #[tokio::test] - async fn test_mcp_echo_rejects_delay_above_limit() { - let response = mcp_handler( - default_state(), - initialized_headers().await, - axum::Json(json!({ - "jsonrpc": "2.0", - "method": "tools/call", - "params": { - "name": "echo", - "arguments": { - "message": "hello", - "delay": MAX_DELAY_MS + 1 - } - }, - "id": 12 - })), + async fn test_legacy_time_stats_and_schema_tools() { + let router = build_router(); + let session_id = initialize_session(&router).await; + + let body = legacy_tool_call(&router, &session_id, "get_system_time", json!({}), 20).await; + let text = body["result"]["content"][0]["text"].as_str().unwrap(); + assert!(text.ends_with('Z'), "UTC default should end with Z: {text}"); + + let body = legacy_tool_call( + &router, + &session_id, + "get_system_time", + json!({"timezone": "Mars/Olympus"}), + 21, ) .await; - - assert_eq!(response.status(), StatusCode::OK); - let body = response_json(response).await; - assert_eq!(body["error"]["code"], -32602); - assert_eq!(body["error"]["message"], "delay exceeds the 60000 ms limit"); - } - - #[test] - fn test_parse_args_defaults_to_legacy_only() { - let config = parse_args(&[]).unwrap(); - assert_eq!(config, ServerConfig::default()); - assert_eq!(config.supported_versions(), vec!["2025-11-25"]); - } - - #[test] - fn test_parse_args_enables_modern_and_strict() { - let args: Vec = ["--protocol", "2026-07-28", "--strict"] - .iter() - .map(ToString::to_string) - .collect(); - let config = parse_args(&args).unwrap(); - assert!(config.modern); - assert!(config.strict); - // Strict serves exactly the revisions named with --protocol. - assert!(!config.legacy); - assert_eq!(config.supported_versions(), vec!["2026-07-28"]); - } - - #[test] - fn test_parse_args_strict_with_explicit_versions_serves_both() { - let args: Vec = [ - "--protocol", - "2026-07-28", - "--protocol", - "2025-11-25", - "--strict", - ] - .iter() - .map(ToString::to_string) - .collect(); - let config = parse_args(&args).unwrap(); - assert_eq!( - config.supported_versions(), - vec!["2026-07-28", "2025-11-25"] + assert_eq!(body["result"]["isError"], true); + assert!( + body["result"]["content"][0]["text"] + .as_str() + .unwrap() + .starts_with("Invalid timezone 'Mars/Olympus'") ); - } - #[test] - fn test_parse_args_non_strict_keeps_legacy_fallback() { - let args: Vec = ["--protocol", "2026-07-28"] - .iter() - .map(ToString::to_string) - .collect(); - let config = parse_args(&args).unwrap(); - assert!(config.legacy); + let body = legacy_tool_call(&router, &session_id, "schema_success", json!({}), 22).await; + assert_eq!(body["result"]["isError"], false); + let expected = json!({ "recognitionId": "rec-123", "message": "ok" }); + assert_eq!(body["result"]["structuredContent"], expected); + let text: serde_json::Value = + serde_json::from_str(body["result"]["content"][0]["text"].as_str().unwrap()) + .expect("text content should mirror the structured payload"); + assert_eq!(text, expected); + + let body = legacy_tool_call(&router, &session_id, "schema_error", json!({}), 23).await; + assert_eq!(body["result"]["isError"], true); assert_eq!( - config.supported_versions(), - vec!["2026-07-28", "2025-11-25"] + body["result"]["content"][0]["text"], + "You cannot send more than 200 points" ); - } - #[test] - fn test_parse_args_rejects_unknown_version_and_missing_value() { - let bad: Vec = ["--protocol", "1999-01-01"] - .iter() - .map(ToString::to_string) - .collect(); - assert!(parse_args(&bad).is_err()); - let missing: Vec = ["--protocol"].iter().map(ToString::to_string).collect(); - assert!(parse_args(&missing).is_err()); - let unknown: Vec = ["--frobnicate"].iter().map(ToString::to_string).collect(); - assert!(parse_args(&unknown).is_err()); - } - - fn modern_state() -> axum::extract::State> { - axum::extract::State(Arc::new(ServerConfig { - modern: true, - legacy: true, - strict: false, - })) + let body = legacy_tool_call(&router, &session_id, "get_stats", json!({}), 24).await; + let text = body["result"]["content"][0]["text"].as_str().unwrap(); + assert!(text.contains(r#""server": "fast-time-server""#)); + assert!(text.contains(r#""requests_handled": "#)); } - fn strict_state() -> axum::extract::State> { - axum::extract::State(Arc::new(ServerConfig { - modern: true, - legacy: true, - strict: true, - })) - } + // ======================================================================== + // Modern era (2026-07-28): stateless, version in params._meta + header + // ======================================================================== - fn strict_modern_only_state() -> axum::extract::State> { - axum::extract::State(Arc::new(ServerConfig { - modern: true, - legacy: false, - strict: true, - })) + #[tokio::test] + async fn test_modern_verify_protocol_reports_stateless() { + let (status, body) = + modern_tool_call(&build_router(), "verify-protocol", json!({}), 1).await; + assert_eq!(status, StatusCode::OK); + let result = &body["result"]; + assert_eq!(result["resultType"], "complete"); + assert_eq!(result["isError"], false); + assert_eq!( + result["structuredContent"], + json!({ + "protocolVersion": MCP_PROTOCOL_VERSION_MODERN, + "transport": "stateless" + }) + ); + let text: serde_json::Value = + serde_json::from_str(result["content"][0]["text"].as_str().expect("text content")) + .expect("text content should mirror the structured payload"); + assert_eq!(text["protocolVersion"], MCP_PROTOCOL_VERSION_MODERN); + assert_eq!(text["transport"], "stateless"); } - fn modern_request(method: &str, version: &str, id: i64) -> serde_json::Value { - json!({ - "jsonrpc": "2.0", - "method": method, - "params": { - "_meta": { - "io.modelcontextprotocol/protocolVersion": version, - "io.modelcontextprotocol/clientCapabilities": {} - } - }, - "id": id - }) + #[tokio::test] + async fn test_modern_tools_call_needs_no_session() { + let (status, body) = + modern_tool_call(&build_router(), "echo", json!({ "message": "hi" }), 2).await; + assert_eq!(status, StatusCode::OK); + assert_eq!(body["result"]["content"][0]["text"], "hi"); + assert_eq!(body["result"]["isError"], false); } #[tokio::test] - async fn test_modern_request_rejected_when_modern_not_enabled() { - let response = mcp_handler( - default_state(), - HeaderMap::new(), - axum::Json(modern_request("tools/list", "2026-07-28", 1)), + async fn test_modern_tools_list_schemas() { + let (status, body) = modern_call( + &build_router(), + modern_request("tools/list", MCP_PROTOCOL_VERSION_MODERN, 3), ) .await; + assert_eq!(status, StatusCode::OK); + let tools = body["result"]["tools"].as_array().unwrap(); + assert_eq!(tools.len(), 8); - assert_eq!(response.status(), StatusCode::BAD_REQUEST); - let body = response_json(response).await; - assert_eq!(body["error"]["code"], -32022); - assert_eq!(body["error"]["data"]["supported"], json!(["2025-11-25"])); - assert_eq!(body["error"]["data"]["requested"], "2026-07-28"); + let echo = tools.iter().find(|tool| tool["name"] == "echo").unwrap(); + assert_eq!(echo["description"], "Echo back the provided message."); + assert_eq!(echo["inputSchema"]["type"], "object"); + assert_eq!( + echo["inputSchema"]["properties"]["message"]["type"], + "string" + ); + assert!( + echo["inputSchema"]["required"] + .as_array() + .unwrap() + .contains(&json!("message")) + ); + + for name in ["schema_error", "schema_success"] { + let tool = tools.iter().find(|tool| tool["name"] == name).unwrap(); + assert_eq!( + tool["outputSchema"]["properties"]["recognitionId"]["type"], "string", + "{name} should keep its outputSchema" + ); + assert!( + tool["outputSchema"]["required"] + .as_array() + .unwrap() + .contains(&json!("recognitionId")) + ); + } + + let verify = tools + .iter() + .find(|tool| tool["name"] == "verify-protocol") + .expect("verify-protocol should be listed"); + assert_eq!( + verify["outputSchema"]["properties"]["protocolVersion"]["type"], + "string" + ); + assert_eq!( + verify["outputSchema"]["properties"]["transport"]["type"], + "string" + ); } #[tokio::test] - async fn test_server_discover_lists_supported_versions() { - let response = mcp_handler( - modern_state(), - HeaderMap::new(), - axum::Json(modern_request("server/discover", "2026-07-28", 1)), + async fn test_server_discover_lists_both_eras() { + let (status, body) = modern_call( + &build_router(), + modern_request("server/discover", MCP_PROTOCOL_VERSION_MODERN, 4), ) .await; - - assert_eq!(response.status(), StatusCode::OK); - let body = response_json(response).await; + assert_eq!(status, StatusCode::OK); let result = &body["result"]; assert_eq!(result["resultType"], "complete"); assert_eq!( result["supportedVersions"], - json!(["2026-07-28", "2025-11-25"]) + json!([MCP_PROTOCOL_VERSION, MCP_PROTOCOL_VERSION_MODERN]) ); - // mcp_types 2.0.0b2 (v2026_07_28 wire models) requires serverInfo, - // cacheScope, and ttlMs as top-level DiscoverResult fields. - assert_eq!(result["serverInfo"]["name"], APP_NAME); + assert!(result["capabilities"]["tools"].is_object()); assert_eq!(result["cacheScope"], "private"); assert_eq!(result["ttlMs"], 0); assert_eq!( result["_meta"]["io.modelcontextprotocol/serverInfo"]["name"], APP_NAME ); - assert!(result["capabilities"]["tools"].is_object()); } #[tokio::test] - async fn test_modern_tools_list_includes_cacheable_result_fields() { - let response = mcp_handler( - modern_state(), - HeaderMap::new(), - axum::Json(modern_request("tools/list", "2026-07-28", 1)), + async fn test_modern_unsupported_version_rejected() { + let (status, body) = modern_call( + &build_router(), + modern_request("tools/list", "2025-06-18", 5), ) .await; - - assert_eq!(response.status(), StatusCode::OK); - let body = response_json(response).await; - let result = &body["result"]; - assert_eq!(result["resultType"], "complete"); - assert_eq!(result["cacheScope"], "private"); - assert_eq!(result["ttlMs"], 0); - assert!(result["tools"].as_array().unwrap().len() >= 7); - } - - #[tokio::test] - async fn test_modern_tools_call_needs_no_session() { - let mut request = modern_request("tools/call", "2026-07-28", 7); - request["params"]["name"] = json!("echo"); - request["params"]["arguments"] = json!({ "message": "hi" }); - let response = mcp_handler(modern_state(), HeaderMap::new(), axum::Json(request)).await; - - assert_eq!(response.status(), StatusCode::OK); - let body = response_json(response).await; - assert_eq!(body["result"]["resultType"], "complete"); - assert_eq!(body["result"]["content"][0]["text"], "hi"); - assert_eq!(body["result"]["isError"], false); - } - - #[tokio::test] - async fn test_modern_version_mismatch_returns_unsupported_error() { - let response = mcp_handler( - modern_state(), - HeaderMap::new(), - axum::Json(modern_request("tools/list", "2025-06-18", 1)), - ) - .await; - - assert_eq!(response.status(), StatusCode::BAD_REQUEST); - let body = response_json(response).await; + assert_eq!(status, StatusCode::BAD_REQUEST); assert_eq!(body["error"]["code"], -32022); + assert_eq!(body["error"]["message"], "Unsupported protocol version"); assert_eq!( body["error"]["data"]["supported"], - json!(["2026-07-28", "2025-11-25"]) + json!([MCP_PROTOCOL_VERSION, MCP_PROTOCOL_VERSION_MODERN]) ); + assert_eq!(body["error"]["data"]["requested"], "2025-06-18"); } #[tokio::test] async fn test_modern_header_mismatch_rejected() { - let mut headers = HeaderMap::new(); - headers.insert( + let mut request = mcp_post(modern_request("tools/list", MCP_PROTOCOL_VERSION_MODERN, 6)); + request.headers_mut().insert( PROTOCOL_VERSION_HEADER, HeaderValue::from_static("2025-06-18"), ); - let response = mcp_handler( - modern_state(), - headers, - axum::Json(modern_request("ping", "2026-07-28", 1)), - ) - .await; - + let response = oneshot(&build_router(), request).await; assert_eq!(response.status(), StatusCode::BAD_REQUEST); - let body = response_json(response).await; + let body: serde_json::Value = serde_json::from_str(&response_text(response).await).unwrap(); assert_eq!(body["error"]["code"], -32020); } #[tokio::test] - async fn test_modern_unknown_method_is_not_found() { - let response = mcp_handler( - modern_state(), - HeaderMap::new(), - axum::Json(modern_request("resources/list", "2026-07-28", 1)), + async fn test_modern_missing_client_capabilities_rejected() { + let mut request = modern_request("tools/list", MCP_PROTOCOL_VERSION_MODERN, 7); + request["params"]["_meta"] + .as_object_mut() + .unwrap() + .remove(CLIENT_CAPABILITIES_META_KEY); + let (status, body) = modern_call(&build_router(), request).await; + assert_eq!(status, StatusCode::BAD_REQUEST); + assert_eq!(body["error"]["code"], -32602); + } + + #[tokio::test] + async fn test_mcp_echo_rejects_delay_above_limit() { + let (status, body) = modern_tool_call( + &build_router(), + "echo", + json!({ "message": "hello", "delay": MAX_DELAY_MS + 1 }), + 8, ) .await; - - assert_eq!(response.status(), StatusCode::NOT_FOUND); - let body = response_json(response).await; - assert_eq!(body["error"]["code"], -32601); + assert_eq!(status, StatusCode::BAD_REQUEST); + assert_eq!(body["error"]["code"], -32602); + assert_eq!(body["error"]["message"], "delay exceeds the 60000 ms limit"); } - fn initialize_request(protocol_version: &str) -> serde_json::Value { - json!({ - "jsonrpc": "2.0", - "method": "initialize", - "params": { - "protocolVersion": protocol_version, - "capabilities": {}, - "clientInfo": { "name": "test", "version": "1.0" } - }, - "id": 1 - }) - } + // ======================================================================== + // REST endpoints survive alongside the SDK service + // ======================================================================== #[tokio::test] - async fn test_strict_initialize_rejects_unsupported_version() { - let response = mcp_handler( - strict_state(), - HeaderMap::new(), - axum::Json(initialize_request("2024-11-05")), + async fn test_rest_and_meta_endpoints() { + let router = build_router(); + let response = oneshot( + &router, + Request::builder() + .method("GET") + .uri("http://localhost/health") + .body(axum::body::Body::empty()) + .unwrap(), ) .await; - assert_eq!(response.status(), StatusCode::OK); - assert!(response.headers().get(SESSION_HEADER).is_none()); - let body = response_json(response).await; - assert_eq!(body["error"]["code"], -32022); - assert_eq!(body["error"]["message"], "Unsupported protocol version"); + let body: serde_json::Value = serde_json::from_str(&response_text(response).await).unwrap(); + assert_eq!(body["status"], "healthy"); + + let response = oneshot( + &router, + Request::builder() + .method("GET") + .uri("http://localhost/version") + .body(axum::body::Body::empty()) + .unwrap(), + ) + .await; + let body: serde_json::Value = serde_json::from_str(&response_text(response).await).unwrap(); assert_eq!( - body["error"]["data"]["supported"], - json!(["2026-07-28", "2025-11-25"]) + body["mcp_versions"], + json!([MCP_PROTOCOL_VERSION, MCP_PROTOCOL_VERSION_MODERN]) ); - assert_eq!(body["error"]["data"]["requested"], "2024-11-05"); - } - - #[tokio::test] - async fn test_strict_initialize_accepts_supported_version() { - let response = mcp_handler( - strict_state(), - HeaderMap::new(), - axum::Json(initialize_request("2025-11-25")), + assert!(body.get("strict").is_none()); + + let response = oneshot( + &router, + Request::builder() + .method("POST") + .uri("http://localhost/api/echo") + .header(header::CONTENT_TYPE, "application/json") + .body(axum::body::Body::from(r#"{"message":"hello"}"#)) + .unwrap(), ) .await; - assert_eq!(response.status(), StatusCode::OK); - assert!(response.headers().contains_key(SESSION_HEADER)); - let body = response_text(response).await; - assert!(body.contains(r#""protocolVersion":"2025-11-25""#)); - } - - #[tokio::test] - async fn test_strict_modern_only_rejects_all_initialize() { - for version in ["2025-11-25", "2024-11-05"] { - let response = mcp_handler( - strict_modern_only_state(), - HeaderMap::new(), - axum::Json(initialize_request(version)), - ) - .await; - - assert_eq!(response.status(), StatusCode::OK); - assert!(response.headers().get(SESSION_HEADER).is_none()); - let body = response_json(response).await; - assert_eq!(body["error"]["code"], -32022); - assert_eq!(body["error"]["message"], "Unsupported protocol version"); - assert_eq!( - body["error"]["data"]["supported"], - json!(["2026-07-28"]), - "initialize for {version} must be rejected naming only the strict set" - ); - } - } - - #[tokio::test] - async fn test_non_strict_initialize_falls_back_to_legacy_version() { - let response = mcp_handler( - modern_state(), - HeaderMap::new(), - axum::Json(initialize_request("2024-11-05")), + let body: serde_json::Value = serde_json::from_str(&response_text(response).await).unwrap(); + assert_eq!(body["message"], "hello"); + + let response = oneshot( + &router, + Request::builder() + .method("GET") + .uri("http://localhost/api/time?tz=America/New_York") + .body(axum::body::Body::empty()) + .unwrap(), ) .await; - assert_eq!(response.status(), StatusCode::OK); - let body = response_text(response).await; - assert!(body.contains(r#""protocolVersion":"2025-11-25""#)); + let body: serde_json::Value = serde_json::from_str(&response_text(response).await).unwrap(); + assert_eq!(body["timezone"], "America/New_York"); + assert!( + body["time"].as_str().unwrap().ends_with("-04:00") + || body["time"].as_str().unwrap().ends_with("-05:00") + ); } } From ca61da8369a2b52c2b8bd255207cb52ec9f32761 Mon Sep 17 00:00:00 2001 From: Jonathan Springer Date: Wed, 29 Jul 2026 15:53:07 +0100 Subject: [PATCH 2/3] Update fast-time-server docs and load test for SDK wire behavior README now documents the dual-era-only rmcp SDK server, the verify-protocol tool, and SSE streaming as it actually works on /mcp (replacing the stale /sse section and the --strict/--protocol flag docs). The locustfile SSE parser skips the SDK's empty priming data event on legacy responses. Co-authored-by: Sisyphus Signed-off-by: Jonathan Springer --- mcp-servers/rust/fast-time-server/README.md | 134 +++++++----------- .../rust/fast-time-server/locustfile.py | 6 +- 2 files changed, 57 insertions(+), 83 deletions(-) diff --git a/mcp-servers/rust/fast-time-server/README.md b/mcp-servers/rust/fast-time-server/README.md index c3ff5c9..e654933 100644 --- a/mcp-servers/rust/fast-time-server/README.md +++ b/mcp-servers/rust/fast-time-server/README.md @@ -2,13 +2,13 @@ > Author: Mihai Criveti -Ultra-fast MCP server written in Rust for performance testing and benchmarking. Hand-rolled on axum with no SDK in the hot path. +Ultra-fast MCP server written in Rust for performance testing and benchmarking. Built on the official [MCP Rust SDK](https://github.com/modelcontextprotocol/rust-sdk) (`rmcp`) with axum. ## Features - **Blazing fast**: Native Rust performance with zero-copy where possible -- **Streamable HTTP**: Modern HTTP transport with streaming support -- **Dual-era MCP**: Serves legacy `2025-11-25` (initialize handshake + sessions) by default, and optionally the modern `2026-07-28` revision (stateless, per-request `_meta`) via `--protocol` — see [Command-Line Flags](#command-line-flags) +- **Streamable HTTP**: MCP Streamable HTTP transport served by the SDK's `StreamableHttpService` +- **Dual-era MCP**: Legacy `2025-11-25` (initialize handshake + `mcp-session-id` sessions) and modern `2026-07-28` (stateless, per-request `_meta`) are served simultaneously on the same `/mcp` endpoint — no flags, no modes - **Minimal overhead**: No auth, no database, pure compute - **Tools**: - `echo` - Echoes back the provided message (with optional delay/jitter) @@ -17,21 +17,16 @@ Ultra-fast MCP server written in Rust for performance testing and benchmarking. - `convert_time` - Converts a time between IANA timezones - `schema_error` / `schema_success` - Output-schema validation fixtures - `get_stats` - Returns server statistics + - `verify-protocol` - Reports the MCP protocol version active for the current request ## Quick Start ```bash -# Build and run (legacy 2025-11-25 only) +# Build and run make run # Or release build for benchmarking make run-release - -# Also serve the modern 2026-07-28 revision -cargo run -- --protocol 2026-07-28 - -# Reject version fallback during initialize (strict negotiation) -cargo run -- --protocol 2026-07-28 --strict ``` Server starts at `http://localhost:9080/mcp` @@ -49,81 +44,84 @@ make test-echo make test-time ``` -Or with curl: +Or with curl, legacy era (`2025-11-25`): initialize a session, then send +requests with the `mcp-session-id` header. Session-mode POST responses are +`text/event-stream` (SSE); the JSON-RPC message rides in the `data:` line. ```bash -# Initialize session -SESSION_RESPONSE=$(curl -s -X POST http://localhost:9080/mcp \ +# Initialize session (response is SSE; session id comes back in a header) +curl -i -X POST http://localhost:9080/mcp \ -H 'Content-Type: application/json' \ - -d '{"jsonrpc":"2.0","method":"initialize","params":{"protocolVersion":"2024-11-05","capabilities":{},"clientInfo":{"name":"test","version":"1.0"}},"id":1}') + -H 'Accept: application/json, text/event-stream' \ + -d '{"jsonrpc":"2.0","method":"initialize","params":{"protocolVersion":"2025-11-25","capabilities":{},"clientInfo":{"name":"test","version":"1.0"}},"id":1}' -# Extract session ID from response header (if using httpie or similar) -# Or parse from mcp-session-id header - -# List tools +# Call echo tool (substitute the mcp-session-id from the initialize response) curl -X POST http://localhost:9080/mcp \ -H 'Content-Type: application/json' \ - -d '{"jsonrpc":"2.0","method":"tools/list","id":1}' + -H 'Accept: application/json, text/event-stream' \ + -H 'mcp-session-id: ' \ + -d '{"jsonrpc":"2.0","method":"tools/call","params":{"name":"echo","arguments":{"message":"Hello!"}},"id":2}' -# Call echo tool -curl -X POST http://localhost:9080/mcp \ - -H 'Content-Type: application/json' \ - -d '{"jsonrpc":"2.0","method":"tools/call","params":{"name":"echo","arguments":{"message":"Hello!"}},"id":1}' - -# Call get_system_time tool -curl -X POST http://localhost:9080/mcp \ - -H 'Content-Type: application/json' \ - -d '{"jsonrpc":"2.0","method":"tools/call","params":{"name":"get_system_time","arguments":{"timezone":"America/New_York"}},"id":1}' +# Terminate the session +curl -X DELETE http://localhost:9080/mcp -H 'mcp-session-id: ' ``` ### Modern Protocol (2026-07-28) -Start the server with `--protocol 2026-07-28` and skip the handshake entirely — -modern requests are stateless and carry their protocol version in -`params._meta` (plus the `MCP-Protocol-Version` header): +Modern requests are stateless: no handshake, no session. The protocol version +travels in `params._meta` plus the `MCP-Protocol-Version` header (the two must +agree), and every request must also mirror its method in the `Mcp-Method` +header — and its tool/prompt name in `Mcp-Name` for named methods — per the +2026-07-28 standard-headers rule (SEP-2243). Responses are plain +`application/json`. ```bash # Discover supported versions and capabilities (no session needed) curl -X POST http://localhost:9080/mcp \ -H 'Content-Type: application/json' \ + -H 'Accept: application/json, text/event-stream' \ -H 'MCP-Protocol-Version: 2026-07-28' \ + -H 'Mcp-Method: server/discover' \ -d '{"jsonrpc":"2.0","method":"server/discover","params":{"_meta":{"io.modelcontextprotocol/protocolVersion":"2026-07-28","io.modelcontextprotocol/clientCapabilities":{}}},"id":1}' # Call a tool directly - no initialize, no session curl -X POST http://localhost:9080/mcp \ -H 'Content-Type: application/json' \ + -H 'Accept: application/json, text/event-stream' \ -H 'MCP-Protocol-Version: 2026-07-28' \ + -H 'Mcp-Method: tools/call' \ + -H 'Mcp-Name: echo' \ -d '{"jsonrpc":"2.0","method":"tools/call","params":{"name":"echo","arguments":{"message":"Hello!"},"_meta":{"io.modelcontextprotocol/protocolVersion":"2026-07-28","io.modelcontextprotocol/clientCapabilities":{}}},"id":2}' ``` A request for an unsupported version is rejected with HTTP 400 and an -`UnsupportedProtocolVersionError` (`-32022`) whose `data.supported` lists the -versions the server speaks. - +`UnsupportedProtocolVersionError` (`-32022`) whose `data.supported` lists +exactly the two served eras (`2025-11-25`, `2026-07-28`). A mismatching +`MCP-Protocol-Version` header is rejected with `HeaderMismatch` (`-32020`). -### SSE Streaming Transport +### verify-protocol -The server supports Server-Sent Events (SSE) for streaming MCP protocol messages. Per the MCP SSE specification, clients connect to the SSE endpoint first to receive the POST endpoint URL, then initialize the session: +The `verify-protocol` tool reports which era served the current request. It +returns both text content and structured content: -```bash -# Step 1: Connect to SSE endpoint (no session required) -curl -N http://localhost:9080/sse +- Modern (stateless) requests: the version comes from the request's own + `_meta` → `{"protocolVersion": "2026-07-28", "transport": "stateless"}` +- Legacy (session) requests: the version is the one negotiated at `initialize` + → `{"protocolVersion": "2025-11-25", "transport": "session"}` -# Expected output: -# event: endpoint -# data: /mcp -# -# : (keep-alive comments every 15 seconds) +### SSE Streaming -# Step 2: Initialize session via the endpoint from SSE -curl -X POST http://localhost:9080/mcp \ - -H 'Content-Type: application/json' \ - -d '{"jsonrpc":"2.0","method":"initialize","params":{"protocolVersion":"2024-11-05","capabilities":{},"clientInfo":{"name":"test","version":"1.0"}},"id":1}' - -# Response includes mcp-session-id header for subsequent requests -``` +The `/mcp` endpoint itself speaks SSE — there is no separate `/sse` endpoint: -The SSE endpoint immediately sends an "endpoint" event with the POST endpoint URL (`/mcp`), then maintains the connection with periodic keep-alive comments. +- Legacy session POST responses (including `initialize`) are SSE streams + carrying the JSON-RPC response, so the server can interleave progress and + other notifications with the result. +- `GET /mcp` with `Accept: text/event-stream` and a valid `mcp-session-id` + opens a standalone stream for server-initiated messages; `Last-Event-ID` + resumes a broken stream. +- Modern stateless requests return plain `application/json` responses (the + server is configured with `json_response`), falling back to SSE only if a + handler emits intermediate messages. ## Benchmarking @@ -205,41 +203,15 @@ make docker-run | `/api/echo` | POST | Echo `{"message":"..."}` - pure performance test | | `/api/time` | GET | Get time, optional `?tz=America/New_York` | | `/health` | GET | Health check | -| `/version` | GET | Version info, supported MCP protocol versions, strict mode | +| `/version` | GET | Version info and supported MCP protocol versions | ### MCP Protocol | Endpoint | Method | Description | |----------|--------|-------------| -| `/mcp` | POST | MCP JSON-RPC. Legacy (`2025-11-25`): `initialize` handshake + `mcp-session-id` sessions. Modern (`2026-07-28`, if enabled): stateless requests with version in `params._meta`, including `server/discover` | +| `/mcp` | POST | MCP JSON-RPC. Legacy (`2025-11-25`): `initialize` handshake + `mcp-session-id` sessions, SSE responses. Modern (`2026-07-28`): stateless requests with version in `params._meta` + `MCP-Protocol-Version`/`Mcp-Method` headers, JSON responses, including `server/discover` | +| `/mcp` | GET | Open a standalone SSE stream for a legacy session (resume with `Last-Event-ID`) | | `/mcp` | DELETE | Terminate a legacy session | -| `/sse` | GET | Server-Sent Events streaming transport | - -## Command-Line Flags - -| Flag | Default | Description | -|------|---------|-------------| -| `--protocol ` | `2025-11-25` only | Also serve the given MCP protocol revision. Supported: `2025-11-25`, `2026-07-28`. May be repeated. | -| `--strict` | off | Serve exactly the revisions named with `--protocol` and reject any non-conformant interaction — no fallback, and no `initialize` handshake at all unless `2025-11-25` was explicitly enabled. | - -Without arguments the server speaks only the legacy `2025-11-25` revision -(`initialize` handshake + sessions). With `--protocol 2026-07-28` it becomes -dual-era: legacy `initialize` traffic is served as before, and requests that -carry `io.modelcontextprotocol/protocolVersion` in `params._meta` are served -statelessly per the modern revision, including the mandatory `server/discover` -method. Unsupported modern versions are rejected with HTTP 400 and an -`UnsupportedProtocolVersionError` (`-32022`) listing the supported versions; -a mismatching `MCP-Protocol-Version` header is rejected with `HeaderMismatch` -(`-32020`). - -`--strict` makes the served set exact. `--protocol 2026-07-28 --strict` runs a -pure `2026-07-28` server: every `initialize` call — including ones naming -`2025-11-25` — is rejected with an `UnsupportedProtocolVersionError` (`-32022`) -whose `data.supported` -lists only the configured revisions, and `server/discover` advertises only -`2026-07-28`. To run a strict server that still accepts the legacy handshake, -enable both revisions explicitly: -`--protocol 2025-11-25 --protocol 2026-07-28 --strict`. ## Environment Variables diff --git a/mcp-servers/rust/fast-time-server/locustfile.py b/mcp-servers/rust/fast-time-server/locustfile.py index 5f01504..b842935 100644 --- a/mcp-servers/rust/fast-time-server/locustfile.py +++ b/mcp-servers/rust/fast-time-server/locustfile.py @@ -198,11 +198,13 @@ def _mcp_request(self, method: str, params: dict = None, name: str = None): response.failure(f"Status {response.status_code}") return None - # Parse SSE response format (data: {...}) + # Parse SSE response format (data: {...}); the server may send a + # priming event with an empty payload first, so take the first + # non-empty data line. text = response.text.strip() for line in text.split("\n"): line = line.strip() - if line.startswith("data:"): + if line.startswith("data:") and line[5:].strip(): text = line[5:].strip() break From 16a5ea9009256fa93cf9adb4f5fbab14e14388bf Mon Sep 17 00:00:00 2001 From: Jonathan Springer Date: Wed, 29 Jul 2026 16:30:14 +0100 Subject: [PATCH 3/3] Update expected failures in conformance test Signed-off-by: Jonathan Springer --- .github/conformance/fast-time-server-xfail.yml | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/.github/conformance/fast-time-server-xfail.yml b/.github/conformance/fast-time-server-xfail.yml index e8769cd..74a55cf 100644 --- a/.github/conformance/fast-time-server-xfail.yml +++ b/.github/conformance/fast-time-server-xfail.yml @@ -2,7 +2,7 @@ # Remove an entry when the corresponding server capability is implemented. server: - logging-set-level - - completion-complete + # - completion-complete - tools-call-simple-text - tools-call-image - tools-call-audio @@ -15,13 +15,13 @@ server: - tools-call-elicitation - elicitation-sep1034-defaults - elicitation-sep1330-enums - - resources-list + # - resources-list - resources-read-text - resources-read-binary - resources-templates-read - resources-subscribe - resources-unsubscribe - - prompts-list + # - prompts-list - prompts-get-simple - prompts-get-with-args - prompts-get-embedded-resource