diff --git a/Cargo.lock b/Cargo.lock index 78c36da31..70caa4bb5 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4058,12 +4058,12 @@ dependencies = [ [[package]] name = "rmcp" -version = "1.8.0" +version = "3.1.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1d1f571c72940a19d9532fe52dbea8bc9912bf1d766c2970bb824056b86f3f59" +checksum = "c8dddc5b1924b9a59fba420166160ca2c4663a4e01803e52eda33070f56d63c8" dependencies = [ "async-trait", - "base64 0.22.1", + "base64 0.23.1", "bytes", "chrono", "futures", @@ -4662,9 +4662,9 @@ checksum = "6980e8d7511241f8acf4aebddbb1ff938df5eebe98691418c4468d0b72a96a67" [[package]] name = "sse-stream" -version = "0.2.3" +version = "0.2.5" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f3962b63f038885f15bce2c6e02c0e7925c072f1ac86bb60fd44c5c6b762fb72" +checksum = "c123f296ade4ec4b8b0f6162116e6629f5146922ca5ab40ca9d3c2e73ab4761e" dependencies = [ "bytes", "futures-util", diff --git a/crates/aisix-mcp/Cargo.toml b/crates/aisix-mcp/Cargo.toml index 5153d2a42..39e9ecd6c 100644 --- a/crates/aisix-mcp/Cargo.toml +++ b/crates/aisix-mcp/Cargo.toml @@ -43,21 +43,24 @@ hex.workspace = true # Path-parameter encoding for OpenAPI-backed tool calls (`crate::openapi`). percent-encoding.workspace = true -# Official MCP Rust SDK. Pinned exactly: rmcp is <16 months old and still +# Official MCP Rust SDK. Pinned exactly: rmcp is <2 years old and still # ships breaking changes on a roughly-monthly cadence, so we hold a fixed # version and keep all rmcp types behind this crate's own surface (the # `McpBridge` trait and the gateway DTOs) to absorb API drift. # +# The 3.x line implements the MCP 2026-07-28 revision (stateless serving, +# `resultType`, cache hints, `Mcp-Method`/`Mcp-Name` header validation) and +# passes the official conformance suite's dated 2026-07-28 core suites. +# # The gateway is dual-role — an MCP client to upstream servers AND an MCP # server to downstream agents — so both the `client` and `server` halves are # production dependencies. [dependencies.rmcp] -version = "=1.8.0" +version = "=3.1.2" default-features = false features = [ "client", "server", - "base64", "transport-streamable-http-client", "transport-streamable-http-client-reqwest", # `transport-streamable-http-client-reqwest` only wires `dep:reqwest` @@ -74,3 +77,6 @@ aisix-core = { path = "../aisix-core" } tokio = { workspace = true, features = ["macros", "rt-multi-thread"] } axum.workspace = true async-trait.workspace = true +# Raw-HTTP wire-shape assertions in tests/protocol_generations.rs: real +# clients hide the headers and envelope details those tests pin. +reqwest.workspace = true diff --git a/crates/aisix-mcp/examples/conformance_server.rs b/crates/aisix-mcp/examples/conformance_server.rs new file mode 100644 index 000000000..fea297509 --- /dev/null +++ b/crates/aisix-mcp/examples/conformance_server.rs @@ -0,0 +1,205 @@ +//! Serve the MCP gateway on a local port so the OFFICIAL MCP conformance +//! suite can be run against the exact `/mcp/{server}` protocol surface the +//! data plane ships (AISIX-Cloud#1144 acceptance): +//! +//! ```text +//! cargo run -p aisix-mcp --example conformance_server +//! # optional: bind a different address (default 127.0.0.1:3111) +//! cargo run -p aisix-mcp --example conformance_server -- 127.0.0.1:4000 +//! npx -y @modelcontextprotocol/conformance server --url http://127.0.0.1:3111/mcp +//! ``` +//! +//! The chain is the production one end to end: the conformance client talks +//! to a SCOPED [`McpGateway`] (original tool names, as `/mcp/{server}` +//! serves them), which reaches an in-process upstream MCP server through the +//! real `EphemeralBridge` — connect, list, call, disconnect per operation. +//! The upstream implements the tool set the suite's `tools-call-*` scenarios +//! prescribe (`test_simple_text`, `test_image_content`, …), each returning +//! the exact content the scenario checks. +//! +//! Expected NON-passes, all deliberate surface decisions rather than bugs: +//! prompts/resources/completions scenarios (tools-only gateway), +//! sampling / elicitation / progress / logging scenarios (server-initiated +//! frames a cross-upstream aggregator does not relay; `logging/setLevel` is +//! also deleted by the 2026-07-28 revision), and the `Host`-allowlist half +//! of `dns-rebinding-protection` (deliberately disabled — the endpoint is +//! server-to-server and API-key-gated in production; see +//! `streamable_http_service`). +//! +//! Auth, quota, and guardrails live in the proxy layer above this service +//! and are deliberately absent here — the suite probes the protocol, not +//! AISIX governance. + +use std::sync::Arc; + +use aisix_mcp::{streamable_http_service, McpGateway}; +use rmcp::model::{ + CallToolRequestParams, CallToolResponse, CallToolResult, ContentBlock, ErrorData, + ListToolsResult, PaginatedRequestParams, ServerCapabilities, ServerInfo, Tool, +}; +use rmcp::service::RequestContext; +use rmcp::transport::streamable_http_server::session::local::LocalSessionManager; +use rmcp::transport::streamable_http_server::{StreamableHttpServerConfig, StreamableHttpService}; +use rmcp::{RoleServer, ServerHandler}; + +/// 1x1 red-pixel PNG. +const TEST_PNG_BASE64: &str = "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mP8z8BQDwAEhQGAhKmMIQAAAABJRU5ErkJggg=="; +/// Minimal (headers-only) WAV. +const TEST_WAV_BASE64: &str = "UklGRiQAAABXQVZFZm10IBAAAAABAAEAQB8AAEAfAAABAAgAZGF0YQAAAAA="; + +/// The upstream: implements the conformance suite's prescribed test tools. +#[derive(Clone, Default)] +struct ConformanceUpstream; + +fn tool(name: &str, description: &str) -> Tool { + let schema = serde_json::json!({ "type": "object", "properties": {} }); + let schema_obj = schema.as_object().expect("schema is an object").clone(); + Tool::new(name.to_string(), description.to_string(), schema_obj) +} + +/// The `(content, is_error)` each conformance tool must answer with. +fn conformance_result(name: &str) -> Option<(serde_json::Value, bool)> { + match name { + "test_simple_text" => Some(( + serde_json::json!([ + { "type": "text", "text": "This is a simple text response for testing." } + ]), + false, + )), + "test_image_content" => Some(( + serde_json::json!([ + { "type": "image", "data": TEST_PNG_BASE64, "mimeType": "image/png" } + ]), + false, + )), + "test_audio_content" => Some(( + serde_json::json!([ + { "type": "audio", "data": TEST_WAV_BASE64, "mimeType": "audio/wav" } + ]), + false, + )), + "test_embedded_resource" => Some(( + serde_json::json!([{ + "type": "resource", + "resource": { + "uri": "test://embedded-resource", + "mimeType": "text/plain", + "text": "This is an embedded resource content.", + } + }]), + false, + )), + "test_multiple_content_types" => Some(( + serde_json::json!([ + { "type": "text", "text": "Multiple content types test:" }, + { "type": "image", "data": TEST_PNG_BASE64, "mimeType": "image/png" }, + { + "type": "resource", + "resource": { + "uri": "test://mixed-content-resource", + "mimeType": "application/json", + "text": "{\"test\":\"data\",\"value\":123}", + } + } + ]), + false, + )), + "test_error_handling" => Some(( + serde_json::json!([ + { "type": "text", "text": "This tool intentionally returns an error for testing" } + ]), + true, + )), + _ => None, + } +} + +impl ServerHandler for ConformanceUpstream { + async fn list_tools( + &self, + _request: Option, + _context: RequestContext, + ) -> Result { + Ok(ListToolsResult::with_all_items(vec![ + tool("test_simple_text", "Returns simple text content"), + tool("test_image_content", "Returns image content"), + tool("test_audio_content", "Returns audio content"), + tool("test_embedded_resource", "Returns an embedded resource"), + tool("test_multiple_content_types", "Returns mixed content"), + tool("test_error_handling", "Always returns a tool error"), + ])) + } + + async fn call_tool( + &self, + request: CallToolRequestParams, + _context: RequestContext, + ) -> Result { + let (content, is_error) = conformance_result(request.name.as_ref()).ok_or_else(|| { + ErrorData::invalid_params(format!("unknown tool: {}", request.name), None) + })?; + let blocks: Vec = serde_json::from_value(content) + .map_err(|e| ErrorData::internal_error(format!("bad fixture: {e}"), None))?; + let result = if is_error { + CallToolResult::error(blocks) + } else { + CallToolResult::success(blocks) + }; + Ok(result.into()) + } + + fn get_info(&self) -> ServerInfo { + let mut info = ServerInfo::default(); + info.capabilities = ServerCapabilities::builder().enable_tools().build(); + info + } +} + +#[tokio::main] +async fn main() { + // The in-process upstream the gateway bridges to, on an ephemeral port. + let upstream_service = StreamableHttpService::new( + move || Ok(ConformanceUpstream), + Arc::new(LocalSessionManager::default()), + StreamableHttpServerConfig::default(), + ); + let upstream_listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind upstream port"); + let upstream_addr = upstream_listener.local_addr().expect("upstream addr"); + tokio::spawn(async move { + axum::serve( + upstream_listener, + axum::Router::new().nest_service("/mcp", upstream_service), + ) + .await + .expect("serve upstream"); + }); + + // Register it as a snapshot `mcp_servers` row and serve the SCOPED + // gateway — the same construction `/mcp/{server}` uses in production, + // so the suite exercises the shipped bridge + gateway chain. + let server: aisix_core::McpServer = serde_json::from_value(serde_json::json!({ + "display_name": "conformance", + "url": format!("http://{upstream_addr}/mcp"), + "auth_type": "none", + })) + .expect("valid mcp_servers row"); + let snapshot = aisix_core::AisixSnapshot::new(); + snapshot + .mcp_servers + .insert(aisix_core::ResourceEntry::new("mcp-conf", server, 1)); + let gateway = McpGateway::from_snapshot_scoped(&snapshot, "conformance") + .expect("scoped gateway over the registered upstream"); + + let addr = std::env::args() + .nth(1) + .unwrap_or_else(|| "127.0.0.1:3111".to_string()); + let app = axum::Router::new().nest_service("/mcp", streamable_http_service(gateway, 0)); + let listener = tokio::net::TcpListener::bind(&addr) + .await + .unwrap_or_else(|e| panic!("bind {addr}: {e}")); + println!("upstream: http://{upstream_addr}/mcp"); + println!("conformance target: http://{addr}/mcp"); + axum::serve(listener, app).await.expect("serve"); +} diff --git a/crates/aisix-mcp/src/bridge.rs b/crates/aisix-mcp/src/bridge.rs index 795cf8aec..7fd4879ce 100644 --- a/crates/aisix-mcp/src/bridge.rs +++ b/crates/aisix-mcp/src/bridge.rs @@ -17,12 +17,13 @@ use std::time::Duration; use aisix_core::{McpAuthType, McpServer}; use async_trait::async_trait; use http::{HeaderName, HeaderValue}; -use rmcp::model::CallToolRequestParams; +use rmcp::model::{CallToolRequestParams, CallToolResponse}; use rmcp::service::{ClientInitializeError, RoleClient, RunningService}; use rmcp::transport::streamable_http_client::{ StreamableHttpClientTransportConfig, StreamableHttpError, }; use rmcp::transport::StreamableHttpClientTransport; +use rmcp::ClientCacheConfig; use rmcp::ServiceExt; use crate::error::McpError; @@ -299,6 +300,29 @@ pub struct RmcpBridge { timeout: Duration, } +/// Base transport configuration for one upstream connection. Two rmcp +/// defaults are deliberately overridden, both to keep "one inbound call = +/// one upstream execution" true and the 1.8-era transport behavior stable: +/// +/// - `reinit_on_expired_session = false`. On a session-expired 404 the SDK +/// default transparently re-runs `initialize` and RE-SENDS the in-flight +/// request — for `tools/call` that is a silent second execution of a +/// possibly side-effectful tool, outside quota and budget accounting +/// (the session-layer sibling of the MRTR auto-retry disabled in +/// [`McpBridge::call_tool`]). The gateway surfaces the failure instead; +/// [`EphemeralBridge`] opens a fresh session on the next operation anyway. +/// - `max_sse_event_size = usize::MAX`. 3.x introduced a 16 MiB per-event +/// cap on upstream SSE frames that 1.8 never had — a large image/audio +/// tool result delivered over SSE would fail while the identical JSON +/// response succeeds. Unbounded preserves the shipped behavior; the +/// per-operation deadline still bounds the read. +fn transport_config(url: &str) -> StreamableHttpClientTransportConfig { + let mut config = StreamableHttpClientTransportConfig::with_uri(url.to_string()); + config.reinit_on_expired_session = false; + config.max_sse_event_size = usize::MAX; + config +} + impl RmcpBridge { /// Open a session to `upstream`: build the Streamable HTTP transport /// (injecting gateway-held auth — for `oauth2` this mints or reuses a @@ -310,16 +334,16 @@ impl RmcpBridge { // Every arm goes through `with_client(shared_http_client(), ..)` // so the transport inherits the deployment's `upstream.*` // connection settings — `from_uri`/`from_config` would build - // rmcp's own default client with none of them. + // rmcp's own default client with none of them — and through + // `transport_config` for the pinned transport defaults. let transport = match &upstream.auth { McpAuth::None => StreamableHttpClientTransport::with_client( shared_http_client(), - StreamableHttpClientTransportConfig::with_uri(upstream.url.clone()), + transport_config(&upstream.url), ), McpAuth::Bearer(token) => StreamableHttpClientTransport::with_client( shared_http_client(), - StreamableHttpClientTransportConfig::with_uri(upstream.url.clone()) - .auth_header(token.clone()), + transport_config(&upstream.url).auth_header(token.clone()), ), McpAuth::ApiKey(key) => { // A key with non-header-safe bytes is a clean config error, @@ -335,16 +359,14 @@ impl RmcpBridge { let headers = HashMap::from([(HeaderName::from_static(API_KEY_HEADER), value)]); StreamableHttpClientTransport::with_client( shared_http_client(), - StreamableHttpClientTransportConfig::with_uri(upstream.url.clone()) - .custom_headers(headers), + transport_config(&upstream.url).custom_headers(headers), ) } McpAuth::OAuth2(cfg) => { let token = crate::oauth::get_or_fetch(cfg).await?; StreamableHttpClientTransport::with_client( shared_http_client(), - StreamableHttpClientTransportConfig::with_uri(upstream.url.clone()) - .auth_header(token), + transport_config(&upstream.url).auth_header(token), ) } }; @@ -368,6 +390,23 @@ impl RmcpBridge { let running = tokio::time::timeout(upstream.timeout, establish) .await .map_err(|_| McpError::Connect("upstream MCP connect timed out".to_string()))??; + // rmcp 3.x ships a client response cache that is ENABLED by default + // (`ClientCacheConfig::default()`), keyed per peer and honoring the + // server's `ttlMs`/`cacheScope` hints. A pooled or persistent bridge + // is shared across every AISIX caller that reaches the same + // upstream, so a cached upstream response would be served across + // principals — a cross-tenant leak the SDK's own migration guide + // warns about (`private_partition`). Today's production path + // (`EphemeralBridge`) reconnects per operation, so this pins the + // posture BEFORE the "connection pooling is a later optimization" + // note above ever lands. The gateway's caching story is a wire-level + // hint only (no cache engine), so the cache is disabled outright + // rather than partitioned. Do not re-enable without keying the + // partition to the calling principal. + running + .peer() + .set_response_cache_config(ClientCacheConfig::disabled()) + .await; Ok(Self { running, timeout: upstream.timeout, @@ -454,10 +493,43 @@ impl McpBridge for RmcpBridge { )) } }; - let result = tokio::time::timeout(self.timeout, self.running.call_tool(params)) + // `call_tool_once`, NOT `call_tool`: the 3.x high-level helper drives + // MRTR (SEP-2322) automatically — on an `input_required` result it + // re-sends the request with client-fulfilled inputs, up to + // `DEFAULT_MRTR_MAX_ROUNDS` (10) upstream round trips for ONE inbound + // call. That would silently multiply upstream cost and defeat every + // per-call quota, budget, and timeout assumption in the gateway. The + // `_once` variant sends exactly one request; a non-final response is + // surfaced as a clean tool failure instead of a hidden retry loop. + let response = tokio::time::timeout(self.timeout, self.running.call_tool_once(params)) .await .map_err(|_| McpError::Request("upstream tools/call timed out".to_string()))? .map_err(|e| McpError::Request(e.to_string()))?; + let result = match response { + CallToolResponse::Complete(result) => result, + CallToolResponse::InputRequired(_) => { + return Err(McpError::Request( + "upstream tool requires additional interactive input (MRTR), which the \ + gateway does not relay" + .to_string(), + )) + } + CallToolResponse::Task(_) => { + return Err(McpError::Request( + "upstream tool deferred to an asynchronous task, which the gateway does \ + not support" + .to_string(), + )) + } + // `CallToolResponse` is #[non_exhaustive]; treat future variants + // as unsupported rather than mis-mapping them to a result. Kept + // payload-free: the message lands in gateway logs verbatim. + _ => { + return Err(McpError::Request( + "upstream returned an unsupported tools/call response kind".to_string(), + )) + } + }; let content = serde_json::to_value(&result.content) .map_err(|e| McpError::Request(format!("failed to encode tool result: {e}")))?; Ok(McpToolResult { diff --git a/crates/aisix-mcp/src/gateway.rs b/crates/aisix-mcp/src/gateway.rs index 5c687af87..b9950ae10 100644 --- a/crates/aisix-mcp/src/gateway.rs +++ b/crates/aisix-mcp/src/gateway.rs @@ -32,8 +32,8 @@ use std::borrow::Cow; use std::sync::Arc; use rmcp::model::{ - CallToolRequestParams, CallToolResult, Content, ErrorData, ListToolsResult, - PaginatedRequestParams, ServerCapabilities, ServerInfo, Tool, + CacheScope, CallToolRequestParams, CallToolResponse, CallToolResult, ContentBlock, ErrorData, + ListToolsResult, PaginatedRequestParams, ProtocolVersion, ServerCapabilities, ServerInfo, Tool, }; use rmcp::service::RequestContext; use rmcp::transport::streamable_http_server::session::local::LocalSessionManager; @@ -53,6 +53,24 @@ use crate::openapi::OpenApiBridge; /// not contain it; tool names may (we split on the first occurrence). pub const TOOL_NAMESPACE_SEPARATOR: &str = "__"; +/// Every protocol version `/mcp` serves, oldest first: the Streamable HTTP +/// generations (`2025-03-26` onward) plus the stateless `2026-07-28` +/// revision. The single source of truth for `initialize` negotiation, +/// `server/discover`, and the proxy layer's `MCP-Protocol-Version` header +/// gate — one list so the three can never drift. +pub const SUPPORTED_PROTOCOL_VERSIONS: &[ProtocolVersion] = &[ + ProtocolVersion::V_2025_03_26, + ProtocolVersion::V_2025_06_18, + ProtocolVersion::V_2025_11_25, + ProtocolVersion::V_2026_07_28, +]; + +/// [`SUPPORTED_PROTOCOL_VERSIONS`] as plain strings, for callers that gate on +/// the `MCP-Protocol-Version` HTTP header without depending on rmcp types +/// (the proxy layer). Kept in lockstep by `supported_version_lists_agree`. +pub const SUPPORTED_PROTOCOL_VERSION_NAMES: &[&str] = + &["2025-03-26", "2025-06-18", "2025-11-25", "2026-07-28"]; + /// One registered upstream: its gateway-facing name and the live bridge to it. struct NamedUpstream { name: String, @@ -387,7 +405,7 @@ impl ServerHandler for McpGateway { async fn list_tools( &self, _request: Option, - _context: RequestContext, + context: RequestContext, ) -> Result { // Fan out concurrently; each upstream call is already deadline-bounded // by its bridge, so a slow upstream cannot stall the aggregate. @@ -439,14 +457,35 @@ impl ServerHandler for McpGateway { } } } - Ok(ListToolsResult::with_all_items(tools)) + // Cache hints (SEP-2549, required on 2026-07-28 list results) are a + // wire-level statement only — the gateway runs no cache engine. The + // honest values here are "do not cache": the list is filtered by the + // CALLER's per-key ACL (so it is `private` to that authorization + // context), and upstream tool sets can change between requests (so + // its freshness window is zero). + // + // Version-gated because rmcp only strips `resultType` for legacy + // peers, not these fields — setting them unconditionally would add + // two fields legacy clients have never seen. Modern requests carry + // the version in `_meta`; legacy sessions fall back to the + // handshake's negotiated version. (ISO `YYYY-MM-DD` versions compare + // lexically the same as chronologically.) + let result = ListToolsResult::with_all_items(tools); + let modern = context + .protocol_version() + .is_some_and(|v| v.as_str() >= ProtocolVersion::V_2026_07_28.as_str()); + Ok(if modern { + result.with_ttl_ms(0).with_cache_scope(CacheScope::Private) + } else { + result + }) } async fn call_tool( &self, request: CallToolRequestParams, _context: RequestContext, - ) -> Result { + ) -> Result { // Resolve `(server, tool, namespaced)` from the caller's tool name. // Scoped: the name is the upstream's original one, but a caller that // sends the namespaced form anyway (an aggregated-endpoint client @@ -537,11 +576,34 @@ impl ServerHandler for McpGateway { ) })?; - into_call_tool_result(result) + // Always a final (`Complete`) response: the gateway never asks the + // downstream agent for more input (MRTR stays un-relayed — see + // `RmcpBridge::call_tool`), so `InputRequired` is never produced here. + into_call_tool_result(result).map(CallToolResponse::from) + } + + /// The protocol versions this endpoint implements, replacing the SDK + /// default (`ProtocolVersion::KNOWN_VERSIONS`, which would advertise + /// every version rmcp has ever heard of — including ones this endpoint + /// does not serve). Consulted by rmcp for `initialize` negotiation (an + /// exact match is echoed; anything else falls back to + /// `get_info().protocol_version`) and advertised verbatim in + /// `server/discover`. + /// + /// `2024-11-05` is deliberately absent: that generation's transport is + /// HTTP+SSE, which this Streamable-HTTP-only endpoint has never served. + fn supported_protocol_versions(&self) -> Cow<'static, [ProtocolVersion]> { + Cow::Borrowed(SUPPORTED_PROTOCOL_VERSIONS) } fn get_info(&self) -> ServerInfo { let mut info = ServerInfo::default(); + // Pinned, not `ProtocolVersion::default()`: this is the version a + // legacy `initialize` falls back to when the client requests one we + // don't list in `supported_protocol_versions` — the same answer the + // endpoint has always given. Riding the SDK's `LATEST` alias would + // silently move this fallback whenever rmcp bumps it. + info.protocol_version = ProtocolVersion::V_2025_11_25; info.capabilities = ServerCapabilities::builder().enable_tools().build(); match &self.scoped { // The scoped endpoint presents as the upstream server itself, so @@ -572,11 +634,29 @@ impl ServerHandler for McpGateway { /// Configured stateless (no sticky session, JSON responses): the aggregator /// keeps no per-session state, so the endpoint can sit behind a plain load /// balancer — matching the MCP 2026-07-28 transport direction. +/// +/// `request_body_limit_bytes` is the deployment's request-body cap +/// (`0` = unlimited, the same convention as everywhere else in the data +/// plane). It must be threaded in because rmcp 3.x added its OWN inbound +/// cap (4 MiB default) inside this service — beneath the gateway's limit +/// middleware — which would silently override any configured limit above +/// 4 MiB (and the documented unlimited mode) with a plain-text 413. pub fn streamable_http_service( gateway: McpGateway, + request_body_limit_bytes: usize, ) -> StreamableHttpService { let mut config = StreamableHttpServerConfig::default(); - config.stateful_mode = false; + config.max_request_body_bytes = if request_body_limit_bytes == 0 { + usize::MAX + } else { + request_body_limit_bytes + }; + // rmcp 3.x rename of `stateful_mode` (same semantics for legacy + // protocol versions; 2026-07-28 requests are stateless regardless, per + // SEP-2567). Kept `false`: the aggregator holds no session state, so + // legacy clients get sessionless serving too and the endpoint stays + // safe behind a plain load balancer. + config.legacy_session_mode = false; config.json_response = true; // Disable rmcp's `Host`-header allowlist. Its default // (`localhost`/`127.0.0.1`/`::1`) is a DNS-rebinding guard for @@ -614,7 +694,7 @@ fn prefixed_tool(server: &str, tool: crate::McpTool) -> Tool { /// preserving the upstream's tool-level error flag (a tool-level error is /// propagated as `Ok(error_result)`, not turned into a protocol error). fn into_call_tool_result(result: crate::McpToolResult) -> Result { - let content: Vec = serde_json::from_value(result.content).map_err(|e| { + let content: Vec = serde_json::from_value(result.content).map_err(|e| { ErrorData::internal_error(format!("malformed tool content from upstream: {e}"), None) })?; let mut call_result = if result.is_error { @@ -625,3 +705,44 @@ fn into_call_tool_result(result: crate::McpToolResult) -> Result = SUPPORTED_PROTOCOL_VERSIONS + .iter() + .map(|v| v.as_str()) + .collect(); + assert_eq!(typed, SUPPORTED_PROTOCOL_VERSION_NAMES); + } + + /// `2024-11-05` (the HTTP+SSE generation) must stay excluded: this + /// endpoint serves Streamable HTTP only. + #[test] + fn http_sse_generation_stays_unsupported() { + assert!(!SUPPORTED_PROTOCOL_VERSION_NAMES.contains(&"2024-11-05")); + assert!(!SUPPORTED_PROTOCOL_VERSIONS.contains(&ProtocolVersion::V_2024_11_05)); + } + + /// The exact served set, as literals: growing or shrinking it is a + /// deliberate protocol-surface decision that must show up as a failing + /// test, not ride along inside a refactor that edits the constants. + /// (The lockstep test above only proves the two constants AGREE — it + /// would pass if a fifth version were added to both.) + #[test] + fn supported_version_set_is_pinned() { + assert_eq!( + SUPPORTED_PROTOCOL_VERSION_NAMES, + &["2025-03-26", "2025-06-18", "2025-11-25", "2026-07-28"], + "update this pin together with the negotiation/discover/header-gate \ + docs and the tracking issue when the served set changes" + ); + } +} diff --git a/crates/aisix-mcp/src/lib.rs b/crates/aisix-mcp/src/lib.rs index 736a561dc..a00110ffc 100644 --- a/crates/aisix-mcp/src/lib.rs +++ b/crates/aisix-mcp/src/lib.rs @@ -25,6 +25,7 @@ pub use bridge::{ }; pub use error::McpError; pub use gateway::{ - streamable_http_service, strip_server_prefix, McpGateway, ToolAcl, TOOL_NAMESPACE_SEPARATOR, + streamable_http_service, strip_server_prefix, McpGateway, ToolAcl, + SUPPORTED_PROTOCOL_VERSION_NAMES, TOOL_NAMESPACE_SEPARATOR, }; pub use openapi::{validate_spec, OpenApiBridge}; diff --git a/crates/aisix-mcp/tests/gateway_aggregation.rs b/crates/aisix-mcp/tests/gateway_aggregation.rs index 876f4408e..281152537 100644 --- a/crates/aisix-mcp/tests/gateway_aggregation.rs +++ b/crates/aisix-mcp/tests/gateway_aggregation.rs @@ -19,8 +19,8 @@ use aisix_mcp::{ McpTool, McpToolResult, McpUpstream, RmcpBridge, ToolAcl, }; use rmcp::model::{ - CallToolRequestParams, CallToolResult, Content, ErrorData, ListToolsResult, - PaginatedRequestParams, ServerCapabilities, ServerInfo, Tool, + CallToolRequestParams, CallToolResponse, CallToolResult, ContentBlock, ErrorData, + ListToolsResult, PaginatedRequestParams, ServerCapabilities, ServerInfo, Tool, }; use rmcp::service::RequestContext; use rmcp::transport::streamable_http_server::session::local::LocalSessionManager; @@ -58,7 +58,7 @@ impl ServerHandler for LabeledEcho { &self, request: CallToolRequestParams, _context: RequestContext, - ) -> Result { + ) -> Result { if request.name != "echo" { return Err(ErrorData::invalid_params( format!("unknown tool: {}", request.name), @@ -74,12 +74,12 @@ impl ServerHandler for LabeledEcho { // `fail` drives the tool-level-error path: a valid call whose tool // reports failure, returned as `Ok(CallToolResult::error(..))`. if text == "fail" { - return Ok(CallToolResult::error(vec![Content::text("boom")])); + return Ok(CallToolResult::error(vec![ContentBlock::text("boom")]).into()); } - Ok(CallToolResult::success(vec![Content::text(format!( - "{}:{text}", - self.label - ))])) + Ok( + CallToolResult::success(vec![ContentBlock::text(format!("{}:{text}", self.label))]) + .into(), + ) } fn get_info(&self) -> ServerInfo { @@ -101,7 +101,7 @@ async fn spawn_upstream(label: &'static str) -> SocketAddr { /// Serve the gateway itself; return its bound address. async fn spawn_gateway(gateway: McpGateway) -> SocketAddr { - serve(axum::Router::new().nest_service("/mcp", streamable_http_service(gateway))).await + serve(axum::Router::new().nest_service("/mcp", streamable_http_service(gateway, 0))).await } async fn serve(app: axum::Router) -> SocketAddr { diff --git a/crates/aisix-mcp/tests/gateway_scoped.rs b/crates/aisix-mcp/tests/gateway_scoped.rs index 1e4328786..30c2918e9 100644 --- a/crates/aisix-mcp/tests/gateway_scoped.rs +++ b/crates/aisix-mcp/tests/gateway_scoped.rs @@ -19,8 +19,8 @@ use std::sync::Arc; use aisix_core::{AisixSnapshot, McpServer, ResourceEntry}; use aisix_mcp::{streamable_http_service, McpGateway, ToolAcl}; use rmcp::model::{ - CallToolRequestParams, CallToolResult, Content, ErrorData, ListToolsResult, - PaginatedRequestParams, ServerCapabilities, ServerInfo, Tool, + CallToolRequestParams, CallToolResponse, CallToolResult, ContentBlock, ErrorData, + ListToolsResult, PaginatedRequestParams, ServerCapabilities, ServerInfo, Tool, }; use rmcp::service::RequestContext; use rmcp::transport::streamable_http_server::session::local::LocalSessionManager; @@ -59,7 +59,7 @@ impl ServerHandler for LabeledEcho { &self, request: CallToolRequestParams, _context: RequestContext, - ) -> Result { + ) -> Result { if request.name != self.tool_name { return Err(ErrorData::invalid_params( format!("unknown tool: {}", request.name), @@ -72,10 +72,10 @@ impl ServerHandler for LabeledEcho { .and_then(|m| m.get("text")) .and_then(|v| v.as_str()) .unwrap_or_default(); - Ok(CallToolResult::success(vec![Content::text(format!( - "{}:{text}", - self.label - ))])) + Ok( + CallToolResult::success(vec![ContentBlock::text(format!("{}:{text}", self.label))]) + .into(), + ) } fn get_info(&self) -> ServerInfo { @@ -97,7 +97,7 @@ async fn spawn_upstream(label: &'static str, tool_name: &'static str) -> SocketA /// Serve the gateway itself; return its bound address. async fn spawn_gateway(gateway: McpGateway) -> SocketAddr { - serve(axum::Router::new().nest_service("/mcp", streamable_http_service(gateway))).await + serve(axum::Router::new().nest_service("/mcp", streamable_http_service(gateway, 0))).await } async fn serve(app: axum::Router) -> SocketAddr { @@ -165,7 +165,11 @@ async fn scoped_serves_original_names_and_accepts_both_call_forms() { // `initialize` presents the scoped server, not the aggregate. let info = client.peer_info().expect("initialize completed"); - assert_eq!(info.server_info.name, "alpha"); + let server_info = info + .server_info + .as_ref() + .expect("legacy initialize always reports server identity"); + assert_eq!(server_info.name, "alpha"); // tools/list carries the upstream's original names — no `alpha__` prefix. let tools = client.list_all_tools().await.expect("list tools"); diff --git a/crates/aisix-mcp/tests/openapi_tool_roundtrip.rs b/crates/aisix-mcp/tests/openapi_tool_roundtrip.rs index 7afb63701..9f7d23aee 100644 --- a/crates/aisix-mcp/tests/openapi_tool_roundtrip.rs +++ b/crates/aisix-mcp/tests/openapi_tool_roundtrip.rs @@ -128,7 +128,7 @@ fn openapi_entry(id: &str, config: Value) -> ResourceEntry { } async fn spawn_gateway(gateway: McpGateway) -> SocketAddr { - serve(axum::Router::new().nest_service("/mcp", streamable_http_service(gateway))).await + serve(axum::Router::new().nest_service("/mcp", streamable_http_service(gateway, 0))).await } fn first_text(result: &CallToolResult) -> String { diff --git a/crates/aisix-mcp/tests/protocol_generations.rs b/crates/aisix-mcp/tests/protocol_generations.rs new file mode 100644 index 000000000..2f14d6c27 --- /dev/null +++ b/crates/aisix-mcp/tests/protocol_generations.rs @@ -0,0 +1,538 @@ +//! Two-generation contract tests for the downstream `/mcp` surface after the +//! rmcp 3.x upgrade: the stateless MCP `2026-07-28` revision and the legacy +//! Streamable HTTP generations must BOTH work against the same endpoint, and +//! the version-negotiation behavior pinned by AISIX-Cloud#1144/#1148 must not +//! drift: +//! +//! - `initialize` echoes a supported requested version exactly, and falls +//! back to `2025-11-25` (the endpoint's historical answer) for anything +//! else — never to whatever the SDK's `LATEST` alias happens to be. +//! - `server/discover` advertises exactly `SUPPORTED_PROTOCOL_VERSION_NAMES`. +//! - Modern requests are served without any `Mcp-Session-Id`, carry +//! `resultType: "complete"`, and get the honest do-not-cache hints +//! (`ttlMs: 0`, `cacheScope: "private"`); legacy responses keep their +//! historical shape (no `resultType`). +//! - The deliberate `allowed_hosts` opt-out survives the upgrade: a +//! non-loopback `Host` is served, not 403'd (the AISIX API key is the real +//! access control on this endpoint). +//! +//! Raw HTTP is used for the wire-shape pins (a real client hides headers and +//! envelope details); rmcp's own modern client covers the end-to-end modern +//! lifecycle. + +use std::net::SocketAddr; +use std::sync::Arc; + +use aisix_mcp::{streamable_http_service, McpBridge, McpError, McpGateway, McpTool, McpToolResult}; +use rmcp::model::{CallToolRequestParams, ClientInfo, ProtocolVersion}; +use rmcp::transport::StreamableHttpClientTransport; +use rmcp::{ClientLifecycleMode, ClientServiceExt, ServiceExt}; + +/// A self-contained upstream: one `echo` tool, no network. The generation +/// tests exercise the DOWNSTREAM protocol surface; a live upstream session +/// would only add noise. +struct StaticEcho; + +#[async_trait::async_trait] +impl McpBridge for StaticEcho { + async fn list_tools(&self) -> Result, McpError> { + Ok(vec![McpTool { + name: "echo".to_string(), + description: Some("echo back the text argument".to_string()), + input_schema: serde_json::json!({ + "type": "object", + "properties": { "text": { "type": "string" } }, + }), + }]) + } + + async fn call_tool( + &self, + name: &str, + arguments: serde_json::Value, + ) -> Result { + assert_eq!(name, "echo"); + let text = arguments + .get("text") + .and_then(|v| v.as_str()) + .unwrap_or_default(); + Ok(McpToolResult { + content: serde_json::json!([{ "type": "text", "text": text }]), + structured_content: None, + is_error: false, + }) + } +} + +async fn spawn_gateway() -> SocketAddr { + let gateway = McpGateway::new([( + "alpha".to_string(), + Arc::new(StaticEcho) as Arc, + )]); + let app = axum::Router::new().nest_service("/mcp", streamable_http_service(gateway, 0)); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind ephemeral port"); + let addr = listener.local_addr().expect("local addr"); + tokio::spawn(async move { + axum::serve(listener, app).await.expect("serve"); + }); + addr +} + +/// POST one JSON-RPC message; return `(status, headers, parsed body)`. +async fn post_raw( + addr: SocketAddr, + extra_headers: &[(&str, &str)], + body: serde_json::Value, +) -> ( + reqwest::StatusCode, + reqwest::header::HeaderMap, + serde_json::Value, +) { + let client = reqwest::Client::new(); + let mut request = client + .post(format!("http://{addr}/mcp")) + .header("content-type", "application/json") + .header("accept", "application/json, text/event-stream"); + for (name, value) in extra_headers { + request = request.header(*name, *value); + } + let response = request.body(body.to_string()).send().await.expect("send"); + let status = response.status(); + let headers = response.headers().clone(); + let text = response.text().await.expect("read body"); + let parsed = serde_json::from_str(&text).unwrap_or(serde_json::Value::String(text)); + (status, headers, parsed) +} + +fn initialize_body(protocol_version: &str) -> serde_json::Value { + serde_json::json!({ + "jsonrpc": "2.0", + "id": 1, + "method": "initialize", + "params": { + "protocolVersion": protocol_version, + "capabilities": {}, + "clientInfo": { "name": "generation-test", "version": "0.0.0" }, + } + }) +} + +/// `initialize` echoes a requested version exactly when it is one we +/// support, for BOTH generations' Streamable HTTP clients. +#[tokio::test] +async fn initialize_echoes_supported_versions_exactly() { + let addr = spawn_gateway().await; + for version in aisix_mcp::SUPPORTED_PROTOCOL_VERSION_NAMES { + let (status, headers, body) = post_raw(addr, &[], initialize_body(version)).await; + assert_eq!(status, 200, "initialize {version}: {body}"); + assert_eq!( + body["result"]["protocolVersion"], *version, + "supported version must be echoed exactly: {body}" + ); + // Stateless serving on every generation: no session is minted. + assert!( + headers.get("mcp-session-id").is_none(), + "no Mcp-Session-Id may be issued ({version})" + ); + } +} + +/// A version we do NOT list falls back to the endpoint's historical answer, +/// `2025-11-25` — pinned so an SDK `LATEST` bump can never silently move it. +#[tokio::test] +async fn initialize_falls_back_to_2025_11_25() { + let addr = spawn_gateway().await; + for unsupported in ["2024-11-05", "9999-12-31"] { + let (status, _headers, body) = post_raw(addr, &[], initialize_body(unsupported)).await; + assert_eq!(status, 200, "initialize {unsupported}: {body}"); + assert_eq!( + body["result"]["protocolVersion"], "2025-11-25", + "unsupported request must fall back to the pinned version: {body}" + ); + } +} + +/// The deliberate DNS-rebinding-allowlist opt-out (`allowed_hosts` cleared) +/// survives the upgrade: a request whose `Host` is the deployment's real DNS +/// name is served, not 403'd. rmcp's DEFAULT config would reject this. +#[tokio::test] +async fn non_loopback_host_is_served() { + let addr = spawn_gateway().await; + let (status, _headers, body) = post_raw( + addr, + &[("host", "aisix.example.com")], + initialize_body("2025-11-25"), + ) + .await; + assert_eq!( + status, 200, + "non-loopback Host must be accepted (API-key auth is the real gate): {body}" + ); +} + +/// `server/discover` advertises exactly the supported list — the same list +/// `initialize` negotiates against and the proxy header gate enforces. +#[tokio::test] +async fn discover_advertises_the_supported_list() { + let addr = spawn_gateway().await; + let (status, headers, body) = post_raw( + addr, + &[ + ("mcp-protocol-version", "2026-07-28"), + ("mcp-method", "server/discover"), + ], + serde_json::json!({ + "jsonrpc": "2.0", + "id": 1, + "method": "server/discover", + "params": { + "_meta": { + "io.modelcontextprotocol/protocolVersion": "2026-07-28", + "io.modelcontextprotocol/clientInfo": { + "name": "generation-test", "version": "0.0.0" + }, + "io.modelcontextprotocol/clientCapabilities": {}, + } + } + }), + ) + .await; + assert_eq!(status, 200, "server/discover: {body}"); + let advertised: Vec<&str> = body["result"]["supportedVersions"] + .as_array() + .unwrap_or_else(|| panic!("supportedVersions missing: {body}")) + .iter() + .filter_map(|v| v.as_str()) + .collect(); + assert_eq!( + advertised, + aisix_mcp::SUPPORTED_PROTOCOL_VERSION_NAMES.to_vec(), + "discover must advertise the pinned list" + ); + assert!( + headers.get("mcp-session-id").is_none(), + "discover is stateless" + ); +} + +/// The modern stateless flow, raw: no handshake, per-request metadata, the +/// SEP-2243 mirrored headers. Pins the wire shape a `2026-07-28` client +/// observes: `resultType: "complete"`, the do-not-cache hints on +/// `tools/list`, and no session header anywhere. +#[tokio::test] +async fn modern_stateless_list_and_call_without_handshake() { + let addr = spawn_gateway().await; + let meta = serde_json::json!({ + "io.modelcontextprotocol/protocolVersion": "2026-07-28", + "io.modelcontextprotocol/clientInfo": { + "name": "generation-test", "version": "0.0.0" + }, + "io.modelcontextprotocol/clientCapabilities": {}, + }); + + let (status, headers, body) = post_raw( + addr, + &[ + ("mcp-protocol-version", "2026-07-28"), + ("mcp-method", "tools/list"), + ], + serde_json::json!({ + "jsonrpc": "2.0", + "id": 2, + "method": "tools/list", + "params": { "_meta": meta } + }), + ) + .await; + assert_eq!(status, 200, "modern tools/list: {body}"); + assert!( + headers.get("mcp-session-id").is_none(), + "modern tools/list is stateless" + ); + let result = &body["result"]; + assert_eq!(result["resultType"], "complete", "modern result: {body}"); + assert_eq!( + result["ttlMs"], 0, + "the ACL-filtered list must not advertise cacheability: {body}" + ); + assert_eq!(result["cacheScope"], "private", "caller-specific: {body}"); + let names: Vec<&str> = result["tools"] + .as_array() + .unwrap_or_else(|| panic!("tools missing: {body}")) + .iter() + .filter_map(|t| t["name"].as_str()) + .collect(); + assert_eq!(names, ["alpha__echo"], "namespaced tool list"); + + let (status, _headers, body) = post_raw( + addr, + &[ + ("mcp-protocol-version", "2026-07-28"), + ("mcp-method", "tools/call"), + ("mcp-name", "alpha__echo"), + ], + serde_json::json!({ + "jsonrpc": "2.0", + "id": 3, + "method": "tools/call", + "params": { + "name": "alpha__echo", + "arguments": { "text": "hello-modern" }, + "_meta": meta, + } + }), + ) + .await; + assert_eq!(status, 200, "modern tools/call: {body}"); + assert_eq!(body["result"]["resultType"], "complete", "{body}"); + assert_eq!( + body["result"]["content"][0]["text"], "hello-modern", + "tool routed and executed: {body}" + ); +} + +/// SEP-2243: a mirrored header that contradicts the body is rejected before +/// dispatch — the mirror is only trustworthy because mismatches are fatal. +#[tokio::test] +async fn modern_header_body_mismatch_is_rejected() { + let addr = spawn_gateway().await; + let (status, _headers, body) = post_raw( + addr, + &[ + ("mcp-protocol-version", "2026-07-28"), + ("mcp-method", "tools/call"), + ("mcp-name", "alpha__echo"), + ], + serde_json::json!({ + "jsonrpc": "2.0", + "id": 4, + "method": "tools/list", + "params": { + "_meta": { + "io.modelcontextprotocol/protocolVersion": "2026-07-28", + "io.modelcontextprotocol/clientInfo": { + "name": "generation-test", "version": "0.0.0" + }, + "io.modelcontextprotocol/clientCapabilities": {}, + } + } + }), + ) + .await; + assert_eq!( + status, 400, + "Mcp-Method header contradicting the body must be rejected: {body}" + ); +} + +/// Legacy responses keep their historical wire shape across EVERY legacy +/// generation this endpoint serves, on both `tools/list` and `tools/call`: +/// no `resultType` discriminator, and no `2026-07-28` cache-hint fields +/// (rmcp strips only `resultType` for legacy peers, so the gateway +/// version-gates the hints itself). +#[tokio::test] +async fn legacy_results_keep_their_wire_shape() { + let addr = spawn_gateway().await; + let client = () + .serve(StreamableHttpClientTransport::from_uri(format!( + "http://{addr}/mcp" + ))) + .await + .expect("legacy client connects"); + let tools = client.list_all_tools().await.expect("list tools"); + assert_eq!(tools.len(), 1); + assert_eq!(tools[0].name.as_ref(), "alpha__echo"); + + for version in ["2025-03-26", "2025-06-18", "2025-11-25"] { + let (status, _headers, body) = post_raw( + addr, + &[("mcp-protocol-version", version)], + serde_json::json!({ + "jsonrpc": "2.0", + "id": 5, + "method": "tools/list", + "params": {} + }), + ) + .await; + assert_eq!(status, 200, "legacy {version} tools/list: {body}"); + for field in ["resultType", "ttlMs", "cacheScope"] { + assert!( + body["result"].get(field).is_none(), + "no {field} on a legacy {version} tools/list response: {body}" + ); + } + + let (status, _headers, body) = post_raw( + addr, + &[("mcp-protocol-version", version)], + serde_json::json!({ + "jsonrpc": "2.0", + "id": 6, + "method": "tools/call", + "params": { "name": "alpha__echo", "arguments": { "text": "legacy" } } + }), + ) + .await; + assert_eq!(status, 200, "legacy {version} tools/call: {body}"); + assert_eq!( + body["result"]["content"][0]["text"], "legacy", + "legacy {version} call executes: {body}" + ); + assert!( + body["result"].get("resultType").is_none(), + "no resultType on a legacy {version} tools/call response: {body}" + ); + } +} + +/// The end-to-end modern lifecycle through rmcp's own client: `Discover` +/// startup (no initialize handshake), then list + call. +#[tokio::test] +async fn modern_rmcp_client_lifecycle_works_end_to_end() { + let addr = spawn_gateway().await; + let client = ClientInfo::default() + .serve_with_lifecycle( + StreamableHttpClientTransport::from_uri(format!("http://{addr}/mcp")), + ClientLifecycleMode::Discover { + preferred_versions: vec![ProtocolVersion::V_2026_07_28], + }, + ) + .await + .expect("modern client startup via server/discover"); + + let tools = client.list_all_tools().await.expect("list tools"); + assert_eq!(tools.len(), 1); + assert_eq!(tools[0].name.as_ref(), "alpha__echo"); + + let mut params = CallToolRequestParams::new("alpha__echo".to_string()); + params = params.with_arguments( + serde_json::json!({ "text": "modern-lifecycle" }) + .as_object() + .expect("object") + .clone(), + ); + let result = client.call_tool(params).await.expect("call tool"); + let content = serde_json::to_value(&result.content).expect("encode"); + assert_eq!(content[0]["text"], "modern-lifecycle"); +} + +/// A real MCP server for the full-chain test below: one `echo` tool served +/// over actual Streamable HTTP. +#[derive(Clone, Default)] +struct UpstreamEcho; + +impl rmcp::ServerHandler for UpstreamEcho { + async fn list_tools( + &self, + _request: Option, + _context: rmcp::service::RequestContext, + ) -> Result { + let schema = serde_json::json!({ + "type": "object", + "properties": { "text": { "type": "string" } }, + }); + let schema_obj = schema.as_object().expect("schema is an object").clone(); + let tool = rmcp::model::Tool::new("echo", "Echo the text argument", schema_obj); + Ok(rmcp::model::ListToolsResult::with_all_items(vec![tool])) + } + + async fn call_tool( + &self, + request: CallToolRequestParams, + _context: rmcp::service::RequestContext, + ) -> Result { + let text = request + .arguments + .as_ref() + .and_then(|m| m.get("text")) + .and_then(|v| v.as_str()) + .unwrap_or_default(); + Ok( + rmcp::model::CallToolResult::success(vec![rmcp::model::ContentBlock::text(text)]) + .into(), + ) + } + + fn get_info(&self) -> rmcp::model::ServerInfo { + let mut info = rmcp::model::ServerInfo::default(); + info.capabilities = rmcp::model::ServerCapabilities::builder() + .enable_tools() + .build(); + info + } +} + +/// The full shipped chain under a modern client: `Discover`-lifecycle +/// client → gateway → `EphemeralBridge` (real connect/list/call per +/// operation, legacy handshake) → real upstream MCP server over HTTP. Pins +/// that the modern downstream surface composes with the production bridge — +/// not just with an in-process stub — including the cross-generation seam +/// (modern downstream, legacy upstream session). +#[tokio::test] +async fn modern_client_through_real_bridge_and_upstream() { + use rmcp::transport::streamable_http_server::session::local::LocalSessionManager; + use rmcp::transport::streamable_http_server::{ + StreamableHttpServerConfig, StreamableHttpService, + }; + + // Real upstream server on an ephemeral port. + let upstream_service = StreamableHttpService::new( + move || Ok(UpstreamEcho), + Arc::new(LocalSessionManager::default()), + StreamableHttpServerConfig::default(), + ); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind upstream port"); + let upstream_addr = listener.local_addr().expect("upstream addr"); + tokio::spawn(async move { + axum::serve( + listener, + axum::Router::new().nest_service("/mcp", upstream_service), + ) + .await + .expect("serve upstream"); + }); + + // Gateway fronting it through the production bridge type. + let bridge = aisix_mcp::EphemeralBridge::new(aisix_mcp::McpUpstream::new(format!( + "http://{upstream_addr}/mcp" + ))); + let gateway = McpGateway::new([("alpha".to_string(), Arc::new(bridge) as Arc)]); + let app = axum::Router::new().nest_service("/mcp", streamable_http_service(gateway, 0)); + let gw_listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind gateway port"); + let gw_addr = gw_listener.local_addr().expect("gateway addr"); + tokio::spawn(async move { + axum::serve(gw_listener, app).await.expect("serve gateway"); + }); + + let client = ClientInfo::default() + .serve_with_lifecycle( + StreamableHttpClientTransport::from_uri(format!("http://{gw_addr}/mcp")), + ClientLifecycleMode::Discover { + preferred_versions: vec![ProtocolVersion::V_2026_07_28], + }, + ) + .await + .expect("modern client startup against the real chain"); + + let tools = client.list_all_tools().await.expect("list tools"); + let names: Vec<&str> = tools.iter().map(|t| t.name.as_ref()).collect(); + assert_eq!(names, ["alpha__echo"], "real upstream tool aggregated"); + + let mut params = CallToolRequestParams::new("alpha__echo".to_string()); + params = params.with_arguments( + serde_json::json!({ "text": "through-the-real-chain" }) + .as_object() + .expect("object") + .clone(), + ); + let result = client.call_tool(params).await.expect("call tool"); + let content = serde_json::to_value(&result.content).expect("encode"); + assert_eq!(content[0]["text"], "through-the-real-chain"); +} diff --git a/crates/aisix-mcp/tests/upstream_defaults.rs b/crates/aisix-mcp/tests/upstream_defaults.rs new file mode 100644 index 000000000..a86ab4552 --- /dev/null +++ b/crates/aisix-mcp/tests/upstream_defaults.rs @@ -0,0 +1,270 @@ +//! Regression tests for the two rmcp 3.x client-side defaults that would +//! silently change the gateway's cost and isolation posture if left at their +//! SDK values (AISIX-Cloud#1144): +//! +//! 1. **MRTR auto-retry.** `RunningService::call_tool` drives SEP-2322 +//! multi-round-trip requests automatically — up to 10 upstream round +//! trips for ONE inbound call. The bridge must send exactly one request +//! per inbound `tools/call` and surface a non-final (`input_required`) +//! answer as a clean error, never a hidden retry loop. +//! 2. **Client response cache.** rmcp enables a per-peer response cache by +//! default, honoring the server's `ttlMs` hint. The bridge is shared +//! across every AISIX caller reaching the same upstream, so a cached +//! response would cross tenant boundaries. The bridge must disable it: +//! two identical requests on ONE session must both reach the upstream, +//! even when the upstream advertises a large `ttlMs`. +//! +//! The upstream here is a raw axum JSON-RPC stub, not an rmcp server: the +//! tests need full control of the wire (an `input_required` result, a large +//! `ttlMs`) and an exact request counter. + +use std::net::SocketAddr; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::Arc; + +use aisix_mcp::{McpBridge, McpUpstream, RmcpBridge}; +use axum::extract::State; +use axum::response::IntoResponse; + +/// Counts upstream `tools/list` and `tools/call` requests, so the tests can +/// assert exactly how many times the bridge actually hit the wire. +#[derive(Default)] +struct StubCounters { + list: AtomicUsize, + call: AtomicUsize, +} + +/// Behavior switch for the stub's `tools/call` answer. +#[derive(Clone, Copy)] +enum CallBehavior { + /// Answer with a SEP-2322 `input_required` result (non-final). + InputRequired, + /// Answer with a SEP-2663 `task` result (asynchronous execution). + Task, + /// Answer `404 Not Found` — the Streamable HTTP signal for an expired + /// session (the stub also mints a session id on `initialize` so the + /// bridge's requests are session-bearing). + ExpiredSession, +} + +#[derive(Clone)] +struct Stub { + counters: Arc, + call_behavior: Option, +} + +/// Minimal legacy Streamable HTTP endpoint: answers `initialize`, accepts +/// `notifications/initialized`, serves `tools/list` / `tools/call`. +async fn stub_mcp(State(stub): State, body: axum::body::Bytes) -> axum::response::Response { + let message: serde_json::Value = match serde_json::from_slice(&body) { + Ok(v) => v, + Err(_) => return (axum::http::StatusCode::BAD_REQUEST, "not json").into_response(), + }; + let method = message["method"].as_str().unwrap_or_default(); + let id = message["id"].clone(); + let respond = |result: serde_json::Value| { + ( + [(axum::http::header::CONTENT_TYPE, "application/json")], + serde_json::json!({ "jsonrpc": "2.0", "id": id, "result": result }).to_string(), + ) + .into_response() + }; + match method { + "initialize" => { + let result = serde_json::json!({ + "protocolVersion": "2025-11-25", + "capabilities": { "tools": {} }, + "serverInfo": { "name": "stub", "version": "0.0.0" }, + }); + // The expired-session scenario needs session-BEARING requests, + // so its handshake mints a session id like a stateful server. + if matches!(stub.call_behavior, Some(CallBehavior::ExpiredSession)) { + return ( + [ + (axum::http::header::CONTENT_TYPE, "application/json"), + ( + axum::http::HeaderName::from_static("mcp-session-id"), + "sess-1", + ), + ], + serde_json::json!({ "jsonrpc": "2.0", "id": id, "result": result }).to_string(), + ) + .into_response(); + } + respond(result) + } + "notifications/initialized" => axum::http::StatusCode::ACCEPTED.into_response(), + "tools/list" => { + // Distinct payload per hit + a huge freshness hint: if any layer + // between the bridge and this stub cached the first answer, the + // second request would never arrive (the counter stays at 1) and + // the second result would repeat hit-1's description. + let hit = stub.counters.list.fetch_add(1, Ordering::SeqCst) + 1; + respond(serde_json::json!({ + "tools": [{ + "name": "echo", + "description": format!("hit-{hit}"), + "inputSchema": { "type": "object" }, + }], + "ttlMs": 3_600_000u64, + "cacheScope": "public", + })) + } + "tools/call" => { + stub.counters.call.fetch_add(1, Ordering::SeqCst); + match stub.call_behavior { + // `resultType: "input_required"` alone is a valid + // `InputRequiredResult` (both detail fields are optional) — + // and it is the exact discriminator rmcp's untagged result + // parsing keys on, so the shape cannot mis-parse as a + // completed tool result. + Some(CallBehavior::InputRequired) => respond(serde_json::json!({ + "resultType": "input_required", + "requestState": "opaque-state", + })), + // A minimal SEP-2663 `CreateTaskResult`: `resultType: "task"` + // is its parse discriminator, with the task fields flattened. + Some(CallBehavior::Task) => respond(serde_json::json!({ + "resultType": "task", + "taskId": "task-1", + "status": "working", + "createdAt": "2026-08-17T00:00:00Z", + "lastUpdatedAt": "2026-08-17T00:00:00Z", + "ttlMs": null, + })), + Some(CallBehavior::ExpiredSession) => { + axum::http::StatusCode::NOT_FOUND.into_response() + } + None => respond(serde_json::json!({ + "content": [{ "type": "text", "text": "ok" }], + "isError": false, + })), + } + } + other => ( + axum::http::StatusCode::BAD_REQUEST, + format!("unexpected method {other}"), + ) + .into_response(), + } +} + +async fn spawn_stub(call_behavior: Option) -> (SocketAddr, Arc) { + let counters = Arc::new(StubCounters::default()); + let stub = Stub { + counters: Arc::clone(&counters), + call_behavior, + }; + let app = axum::Router::new() + .route("/mcp", axum::routing::post(stub_mcp)) + .with_state(stub); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind ephemeral port"); + let addr = listener.local_addr().expect("local addr"); + tokio::spawn(async move { + axum::serve(listener, app).await.expect("serve"); + }); + (addr, counters) +} + +/// One inbound `tools/call` = exactly ONE upstream request, even when the +/// upstream answers `input_required`. With rmcp's default `call_tool` the +/// SDK would re-send the request itself (up to 10 rounds) — every extra +/// round an unmetered upstream call outside AISIX's quota and budget +/// accounting. +#[tokio::test] +async fn input_required_is_one_round_trip_and_a_clean_error() { + let (addr, counters) = spawn_stub(Some(CallBehavior::InputRequired)).await; + let bridge = RmcpBridge::connect(&McpUpstream::new(format!("http://{addr}/mcp"))) + .await + .expect("connect"); + + let error = bridge + .call_tool("echo", serde_json::json!({ "text": "x" })) + .await + .expect_err("a non-final result must surface as an error"); + assert!( + error.to_string().contains("interactive input"), + "error names the MRTR condition: {error}" + ); + assert_eq!( + counters.call.load(Ordering::SeqCst), + 1, + "the bridge must not silently retry an input_required response" + ); +} + +/// A tool call the upstream defers to a SEP-2663 asynchronous task is a +/// clean error (the gateway does not poll tasks) — and still exactly one +/// upstream request. +#[tokio::test] +async fn task_response_is_one_round_trip_and_a_clean_error() { + let (addr, counters) = spawn_stub(Some(CallBehavior::Task)).await; + let bridge = RmcpBridge::connect(&McpUpstream::new(format!("http://{addr}/mcp"))) + .await + .expect("connect"); + + let error = bridge + .call_tool("echo", serde_json::json!({ "text": "x" })) + .await + .expect_err("a task handle must surface as an error"); + assert!( + error.to_string().contains("asynchronous task"), + "error names the task condition: {error}" + ); + assert_eq!(counters.call.load(Ordering::SeqCst), 1); +} + +/// A session-expired `404` on `tools/call` must NOT be answered by silently +/// re-initializing and REPLAYING the call — for a side-effectful tool that +/// would be a second execution outside quota/budget accounting. rmcp's +/// transport default (`reinit_on_expired_session: true`) does exactly that; +/// the bridge pins it off and surfaces the failure instead. +#[tokio::test] +async fn expired_session_does_not_replay_the_tool_call() { + let (addr, counters) = spawn_stub(Some(CallBehavior::ExpiredSession)).await; + let bridge = RmcpBridge::connect(&McpUpstream::new(format!("http://{addr}/mcp"))) + .await + .expect("connect"); + + let result = bridge + .call_tool("echo", serde_json::json!({ "text": "x" })) + .await; + assert!( + result.is_err(), + "an expired upstream session must surface as an error, got {result:?}" + ); + assert_eq!( + counters.call.load(Ordering::SeqCst), + 1, + "the SDK must not re-initialize and replay the tools/call" + ); +} + +/// Two identical `tools/list` calls on ONE upstream session both reach the +/// upstream, despite a one-hour `ttlMs` hint: the SDK's default-enabled +/// response cache is disabled in the bridge. A cached second answer would +/// mean one AISIX caller could be served another caller's upstream view. +#[tokio::test] +async fn upstream_cache_hints_do_not_short_circuit_the_bridge() { + let (addr, counters) = spawn_stub(None).await; + let bridge = RmcpBridge::connect(&McpUpstream::new(format!("http://{addr}/mcp"))) + .await + .expect("connect"); + + let first = bridge.list_tools().await.expect("first list"); + assert_eq!(first[0].description.as_deref(), Some("hit-1")); + + let second = bridge.list_tools().await.expect("second list"); + assert_eq!( + second[0].description.as_deref(), + Some("hit-2"), + "the second list must be the upstream's SECOND answer, not a replay" + ); + assert_eq!( + counters.list.load(Ordering::SeqCst), + 2, + "both list calls must reach the upstream" + ); +} diff --git a/crates/aisix-mcp/tests/upstream_roundtrip.rs b/crates/aisix-mcp/tests/upstream_roundtrip.rs index d77bce159..b0459a844 100644 --- a/crates/aisix-mcp/tests/upstream_roundtrip.rs +++ b/crates/aisix-mcp/tests/upstream_roundtrip.rs @@ -14,8 +14,8 @@ use std::sync::Arc; use aisix_mcp::{EphemeralBridge, McpBridge, McpUpstream, OAuthClientConfig, RmcpBridge}; use rmcp::model::{ - CallToolRequestParams, CallToolResult, Content, ListToolsResult, PaginatedRequestParams, - ServerCapabilities, ServerInfo, Tool, + CallToolRequestParams, CallToolResponse, CallToolResult, ContentBlock, ListToolsResult, + PaginatedRequestParams, ServerCapabilities, ServerInfo, Tool, }; use rmcp::service::RequestContext; use rmcp::transport::streamable_http_server::session::local::LocalSessionManager; @@ -47,7 +47,7 @@ impl ServerHandler for EchoServer { &self, request: CallToolRequestParams, _context: RequestContext, - ) -> Result { + ) -> Result { if request.name != "echo" { return Err(ErrorData::invalid_params( format!("unknown tool: {}", request.name), @@ -64,7 +64,7 @@ impl ServerHandler for EchoServer { if text == "sleep" { tokio::time::sleep(std::time::Duration::from_secs(2)).await; } - Ok(CallToolResult::success(vec![Content::text(text)])) + Ok(CallToolResult::success(vec![ContentBlock::text(text)]).into()) } fn get_info(&self) -> ServerInfo { diff --git a/crates/aisix-proxy/src/mcp.rs b/crates/aisix-proxy/src/mcp.rs index 14886c8c1..db4b5abe5 100644 --- a/crates/aisix-proxy/src/mcp.rs +++ b/crates/aisix-proxy/src/mcp.rs @@ -208,6 +208,21 @@ async fn dispatch( }; let peek = serde_json::from_slice::(&bytes).ok(); + + // Converge the accepted `MCP-Protocol-Version` set before any quota, + // guardrail, or upstream work (AISIX-Cloud#1148). rmcp's own transport + // check admits its whole hardcoded KNOWN_VERSIONS list — including + // `2024-11-05`, whose HTTP+SSE transport this endpoint has never served — + // and renders violations as a bare text/plain 400 outside the JSON-RPC + // envelope. This gate rejects against the SAME list that `initialize` + // negotiation and `server/discover` advertise, in the same envelope every + // other synthesized `/mcp` error uses. An absent header passes: the + // spec's backwards-compatibility rule (assume `2025-03-26`) applies + // downstream. + if let Some(response) = reject_unsupported_protocol_version(&parts.headers, peek.as_ref()) { + return response; + } + let is_tool_call = peek.as_ref().and_then(|p| p.method.as_deref()) == Some("tools/call"); // Resolve the called (server, tool) up front, owned, so it survives the // body being consumed when the request is rebuilt. Aggregated: split the @@ -353,7 +368,10 @@ async fn dispatch( None => aisix_mcp::McpGateway::from_snapshot(&snapshot), } .with_tool_acl(acl); - let service = aisix_mcp::streamable_http_service(gateway); + // The deployment's body cap replaces rmcp's own 4 MiB default inside + // the service; the proxy-level read above already enforced the same + // limit, so the two layers can never disagree. + let service = aisix_mcp::streamable_http_service(gateway, state.request_body_limit_bytes); let request = Request::from_parts(parts, Body::from(bytes)); // `StreamableHttpService` is a tower service that dispatches on method and // never fails (`Error = Infallible`); map its boxed body back to axum's. @@ -563,6 +581,61 @@ fn emit_tool_call_usage( .fan_out(&event, None, exporters.iter().map(|e| &e.value)); } +/// Reject a request whose `MCP-Protocol-Version` header names a version +/// `/mcp` does not serve: HTTP 400 with a JSON-RPC error envelope (echoing +/// the request id when one was parseable), listing the supported versions. +/// Returns `None` when the header is absent (spec backwards-compatibility: +/// treated as `2025-03-26` downstream) or names a supported version. +/// +/// The header value is caller-controlled; the echo in the message is +/// control-stripped and length-bounded so a hostile value cannot inject log +/// lines or bloat the response. +fn reject_unsupported_protocol_version( + headers: &axum::http::HeaderMap, + peek: Option<&JsonRpcPeek>, +) -> Option { + const MAX_ECHO: usize = 64; + let value = headers.get("mcp-protocol-version")?; + let message = match value.to_str() { + Ok(v) if aisix_mcp::SUPPORTED_PROTOCOL_VERSION_NAMES.contains(&v) => return None, + Ok(v) => { + let echoed: String = v + .chars() + .filter(|c| !c.is_control()) + .take(MAX_ECHO) + .collect(); + format!("unsupported MCP-Protocol-Version: {echoed}") + } + Err(_) => "invalid MCP-Protocol-Version header".to_string(), + }; + // Echo only a VALID JSON-RPC id (string or number). A malformed request + // carrying an object/array id would otherwise be reflected into an + // id-invalid response envelope. + let id = peek + .and_then(|p| p.id.clone()) + .filter(|id| id.is_string() || id.is_number()) + .unwrap_or(serde_json::Value::Null); + let body = serde_json::json!({ + "jsonrpc": "2.0", + "id": id, + "error": { + // JSON-RPC 2.0 "Invalid Request": defined identically across + // every MCP generation, unlike the 2026-07-28-only -32022. + "code": -32600, + "message": message, + "data": { "supported": aisix_mcp::SUPPORTED_PROTOCOL_VERSION_NAMES }, + } + }); + Some( + ( + StatusCode::BAD_REQUEST, + [(axum::http::header::CONTENT_TYPE, "application/json")], + body.to_string(), + ) + .into_response(), + ) +} + /// Build the MCP-native response for a guardrail block: a `tools/call` result /// flagged `isError`, echoing the request id, served as HTTP 200 with a JSON /// body (the MCP Streamable HTTP shape). Both the input and output hooks funnel @@ -1844,4 +1917,224 @@ mod tests { .expect("router responds"); assert_eq!(scoped.status(), StatusCode::TOO_MANY_REQUESTS); } + + /// A `/mcp` request for `method` carrying an explicit + /// `MCP-Protocol-Version` header (and matching request `_meta` for the + /// stateless `2026-07-28` shape when `modern` is set). + fn versioned_request( + version: &str, + method: &str, + params: serde_json::Value, + ) -> HttpRequest { + let body = serde_json::json!({ + "jsonrpc": "2.0", + "id": 7, + "method": method, + "params": params + }); + HttpRequest::post("/mcp") + .header("host", "mcp.aisix.example.com") + .header("content-type", "application/json") + .header("accept", "application/json, text/event-stream") + .header("authorization", format!("Bearer {TOKEN}")) + .header("mcp-protocol-version", version) + .header("mcp-method", method) + .body(Body::from(body.to_string())) + .unwrap() + } + + /// A version outside the served set — including `2024-11-05`, which + /// rmcp's own transport check would ADMIT (it is in KNOWN_VERSIONS) — + /// is rejected by the proxy gate: HTTP 400, JSON-RPC envelope, the + /// request id echoed, the supported list attached + /// (AISIX-Cloud#1148 / #1144). + #[tokio::test] + async fn unsupported_protocol_version_header_is_rejected_in_the_envelope() { + let router = router_with(snapshot_with_key()); + for version in ["2024-11-05", "not-a-version"] { + let response = router + .clone() + .oneshot(versioned_request( + version, + "tools/list", + serde_json::json!({}), + )) + .await + .expect("router responds"); + assert_eq!( + response.status(), + StatusCode::BAD_REQUEST, + "{version} must be rejected" + ); + let content_type = response + .headers() + .get("content-type") + .and_then(|v| v.to_str().ok()) + .unwrap_or_default() + .to_string(); + assert!( + content_type.starts_with("application/json"), + "the rejection must use the JSON envelope, not text/plain \ + (got {content_type})" + ); + let bytes = to_bytes(response.into_body(), 1_048_576) + .await + .expect("read body"); + let body: serde_json::Value = serde_json::from_slice(&bytes).expect("json body"); + assert_eq!(body["jsonrpc"], "2.0", "{body}"); + assert_eq!(body["id"], 7, "request id must be echoed: {body}"); + assert_eq!(body["error"]["code"], -32600, "{body}"); + let supported: Vec<&str> = body["error"]["data"]["supported"] + .as_array() + .unwrap_or_else(|| panic!("supported list missing: {body}")) + .iter() + .filter_map(|v| v.as_str()) + .collect(); + assert_eq!( + supported, + aisix_mcp::SUPPORTED_PROTOCOL_VERSION_NAMES.to_vec() + ); + } + } + + /// Hostile header values cannot abuse the rejection envelope: an + /// overlong value is truncated to 64 echoed characters, a non-ASCII + /// (obs-text) value gets the static message (never echoed), and an + /// object request id — invalid JSON-RPC — is sanitized to `null` + /// rather than reflected. + #[tokio::test] + async fn version_gate_rejection_envelope_is_hardened() { + let router = router_with(snapshot_with_key()); + + // Overlong: 300 chars in, at most 64 echoed back out. + let long_version = "v".repeat(300); + let response = router + .clone() + .oneshot(versioned_request( + &long_version, + "tools/list", + serde_json::json!({}), + )) + .await + .expect("router responds"); + assert_eq!(response.status(), StatusCode::BAD_REQUEST); + let bytes = to_bytes(response.into_body(), 1_048_576) + .await + .expect("read body"); + let body: serde_json::Value = serde_json::from_slice(&bytes).expect("json body"); + let message = body["error"]["message"].as_str().unwrap_or_default(); + let echoed = message + .strip_prefix("unsupported MCP-Protocol-Version: ") + .unwrap_or_else(|| panic!("unexpected message shape: {message}")); + assert_eq!(echoed.chars().count(), 64, "echo must be truncated"); + + // Non-ASCII (obs-text) header bytes: `to_str()` fails, the static + // message is used, nothing of the value is reflected. + let request = HttpRequest::post("/mcp") + .header("host", "mcp.aisix.example.com") + .header("content-type", "application/json") + .header("accept", "application/json, text/event-stream") + .header("authorization", format!("Bearer {TOKEN}")) + .header( + "mcp-protocol-version", + axum::http::HeaderValue::from_bytes(&[0x32, 0x30, 0x32, 0x35, 0x80, 0xff]) + .expect("obs-text bytes are valid header bytes"), + ) + .body(Body::from( + serde_json::json!({ + "jsonrpc": "2.0", "id": 9, "method": "tools/list", "params": {} + }) + .to_string(), + )) + .unwrap(); + let response = router + .clone() + .oneshot(request) + .await + .expect("router responds"); + assert_eq!(response.status(), StatusCode::BAD_REQUEST); + let bytes = to_bytes(response.into_body(), 1_048_576) + .await + .expect("read body"); + let body: serde_json::Value = serde_json::from_slice(&bytes).expect("json body"); + assert_eq!( + body["error"]["message"], "invalid MCP-Protocol-Version header", + "non-UTF8 values must never be echoed: {body}" + ); + + // Object id: invalid per JSON-RPC — sanitized to null, not echoed. + let body = serde_json::json!({ + "jsonrpc": "2.0", + "id": { "not": "a valid id" }, + "method": "tools/list", + "params": {} + }); + let request = HttpRequest::post("/mcp") + .header("host", "mcp.aisix.example.com") + .header("content-type", "application/json") + .header("accept", "application/json, text/event-stream") + .header("authorization", format!("Bearer {TOKEN}")) + .header("mcp-protocol-version", "2024-11-05") + .body(Body::from(body.to_string())) + .unwrap(); + let response = router_with(snapshot_with_key()) + .oneshot(request) + .await + .expect("router responds"); + assert_eq!(response.status(), StatusCode::BAD_REQUEST); + let bytes = to_bytes(response.into_body(), 1_048_576) + .await + .expect("read body"); + let body: serde_json::Value = serde_json::from_slice(&bytes).expect("json body"); + assert_eq!( + body["id"], + serde_json::Value::Null, + "an object id is not a valid JSON-RPC id and must not be reflected: {body}" + ); + } + + /// Every version the endpoint serves passes the gate end-to-end: the + /// legacy generations as plain stateless requests, `2026-07-28` with its + /// required per-request metadata. + #[tokio::test] + async fn supported_protocol_version_headers_pass_the_gate() { + let router = router_with(snapshot_with_key()); + for version in ["2025-03-26", "2025-06-18", "2025-11-25"] { + let response = router + .clone() + .oneshot(versioned_request( + version, + "tools/list", + serde_json::json!({}), + )) + .await + .expect("router responds"); + assert_eq!( + response.status(), + StatusCode::OK, + "legacy {version} tools/list must be served" + ); + } + let modern = router + .oneshot(versioned_request( + "2026-07-28", + "tools/list", + serde_json::json!({ + "_meta": { + "io.modelcontextprotocol/protocolVersion": "2026-07-28", + "io.modelcontextprotocol/clientInfo": { + "name": "gate-test", "version": "0.0.0" + }, + "io.modelcontextprotocol/clientCapabilities": {}, + } + }), + )) + .await + .expect("router responds"); + assert_eq!( + modern.status(), + StatusCode::OK, + "a stateless 2026-07-28 tools/list must be served" + ); + } }