feat(mcp): upgrade rmcp 1.8.0 → 3.1.2 — serve MCP 2026-07-28, pin MRTR/cache/version defaults

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

crates/aisix-mcp/examples/conformance_server.rs

@@ -0,0 +1,203 @@
+//! 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
+//! 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<PaginatedRequestParams>,
+        _context: RequestContext<RoleServer>,
+    ) -> Result<ListToolsResult, ErrorData> {
+        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<RoleServer>,
+    ) -> Result<CallToolResponse, ErrorData> {
+        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<ContentBlock> = 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));
+    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");
+}

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;
@@ -368,6 +369,20 @@ 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. This bridge is shared across
+        // every AISIX caller that reaches the same upstream, so any cached
+        // upstream response would be served across principals — a
+        // cross-tenant leak the SDK's own migration guide warns about
+        // (`private_partition`). 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 +469,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 {