From d9cb9fe5e89b6e725cd74f33e7111e87d2fdba71 Mon Sep 17 00:00:00 2001 From: cafalchio Date: Tue, 11 Aug 2026 16:23:38 +0100 Subject: [PATCH 1/9] Remove initialize, added stateles tool call Signed-off-by: cafalchio --- .../src/gateway/mcp_call_validator.rs | 56 ++++ .../src/gateway/mcp_service.rs | 14 +- .../src/gateway/mcp_service/initialization.rs | 261 ++++-------------- .../src/gateway/mcp_service/tools.rs | 88 +++++- .../tests/gateway_completions.rs | 2 + .../tests/gateway_plugins.rs | 43 +++ docker/mcp_counter.Dockerfile | 12 +- 7 files changed, 244 insertions(+), 232 deletions(-) diff --git a/crates/contextforge-data-plane-lib/src/gateway/mcp_call_validator.rs b/crates/contextforge-data-plane-lib/src/gateway/mcp_call_validator.rs index 026de314..e844efa6 100644 --- a/crates/contextforge-data-plane-lib/src/gateway/mcp_call_validator.rs +++ b/crates/contextforge-data-plane-lib/src/gateway/mcp_call_validator.rs @@ -17,6 +17,62 @@ impl<'a> AuthorizedCallValidator<'a> { pub fn new(call_name: &'a str, ctx: &'a RequestContext) -> Self { Self { call_name, ctx } } + + pub fn validate_stateless(self) -> Result<(&'a VirtualHost, &'a ContextForgeClaims), ErrorData> { + let maybe_parts = self.ctx.extensions.get::(); + let maybe_user_config = maybe_parts.and_then(|parts| parts.extensions.get::()); + let maybe_claims = maybe_parts.and_then(|parts| parts.extensions.get::()); + let maybe_virtual_host_id = maybe_parts.and_then(|parts| parts.extensions.get::()); + let call_name = self.call_name; + let has_user_config = maybe_user_config.is_some(); + let virtual_hosts = maybe_user_config.map_or(0, |user_config| user_config.virtual_hosts.len()); + let has_claims = maybe_claims.is_some(); + let virtual_host_id = maybe_virtual_host_id.map_or("", |id| id.value().as_str()); + debug!( + "AuthorizedCallValidator::validate - mcp call validation call_name = {call_name} has_user_config = {has_user_config} virtual_hosts = {virtual_hosts} has_claims = {has_claims} virtual_host_id = {virtual_host_id}" + ); + + let Some(user_config) = maybe_user_config else { + return Err(ErrorData { + code: ErrorCode::INTERNAL_ERROR, + message: "Routing problem... user config not found".into(), + data: None, + }); + }; + + let Some(virtual_host_id) = maybe_virtual_host_id else { + return Err(ErrorData { + code: ErrorCode::INTERNAL_ERROR, + message: "Routing problem... virutal host not known".into(), + data: None, + }); + }; + + let Some(virtual_host) = user_config.virtual_hosts.get(virtual_host_id.value()) else { + let call_name = self.call_name; + let virtual_host_id = virtual_host_id.value(); + let virtual_hosts = user_config.virtual_hosts.len(); + debug!( + "AuthorizedCallValidator::validate - mcp virtual host config missing call_name = {call_name} virtual_host_id = {virtual_host_id} virtual_hosts = {virtual_hosts}" + ); + return Err(ErrorData { + code: ErrorCode::RESOURCE_NOT_FOUND, + message: "No configuration".into(), + data: None, + }); + }; + + let Some(claims) = maybe_claims else { + return Err(ErrorData { + code: ErrorCode::INTERNAL_ERROR, + message: "Routing problem... claims not found".into(), + data: None, + }); + }; + + Ok((virtual_host, claims)) + } + pub fn validate(self) -> Result<(&'a VirtualHost, &'a SessionId, &'a ContextForgeClaims), ErrorData> { let maybe_parts = self.ctx.extensions.get::(); let maybe_session_id = maybe_parts.and_then(|parts| parts.extensions.get::()); diff --git a/crates/contextforge-data-plane-lib/src/gateway/mcp_service.rs b/crates/contextforge-data-plane-lib/src/gateway/mcp_service.rs index 452a4669..c72b8b1f 100644 --- a/crates/contextforge-data-plane-lib/src/gateway/mcp_service.rs +++ b/crates/contextforge-data-plane-lib/src/gateway/mcp_service.rs @@ -9,9 +9,9 @@ use rmcp::{ ErrorData, RoleServer, ServerHandler, model::{ CallToolRequestParams, CallToolResponse, CompleteRequestParams, CompleteResult, GetPromptRequestParams, - GetPromptResponse, InitializeRequestParams, InitializeResult, ListPromptsResult, ListResourceTemplatesResult, - ListResourcesResult, ListToolsResult, PaginatedRequestParams, ReadResourceRequestParams, ReadResourceResponse, - SubscribeRequestParams, UnsubscribeRequestParams, + GetPromptResponse, ListPromptsResult, ListResourceTemplatesResult, ListResourcesResult, ListToolsResult, + PaginatedRequestParams, ReadResourceRequestParams, ReadResourceResponse, SubscribeRequestParams, + UnsubscribeRequestParams, }, service::RequestContext, }; @@ -37,14 +37,6 @@ impl ServerHandler for McpService where T: UserSessionStore + Send + Sync + 'static, { - async fn initialize( - &self, - request: InitializeRequestParams, - cx: RequestContext, - ) -> Result { - initialization::initialize(self, request, cx).await - } - async fn ping(&self, _cx: RequestContext) -> Result<(), ErrorData> { Ok(()) } diff --git a/crates/contextforge-data-plane-lib/src/gateway/mcp_service/initialization.rs b/crates/contextforge-data-plane-lib/src/gateway/mcp_service/initialization.rs index 705c6c3b..4db6565d 100644 --- a/crates/contextforge-data-plane-lib/src/gateway/mcp_service/initialization.rs +++ b/crates/contextforge-data-plane-lib/src/gateway/mcp_service/initialization.rs @@ -1,212 +1,80 @@ -use std::{collections::HashMap, sync::Arc}; +use std::collections::HashMap; use contextforge_data_plane_apis::user_store::BackendMCPGateway; use http::request::Parts; use rmcp::{ - ErrorData, RoleClient, RoleServer, ServiceExt, - model::{ErrorCode, Implementation, InitializeRequestParams, InitializeResult, ServerCapabilities}, + ClientLifecycleMode, ClientServiceExt, ErrorData, RoleClient, RoleServer, + model::{ + ClientCapabilities, ErrorCode, Implementation, InitializeRequestParams, ProtocolVersion, ServerCapabilities, + }, service::{RequestContext, RunningService}, transport::{StreamableHttpClientTransport, streamable_http_client::StreamableHttpClientTransportConfig}, }; use tracing::{info, warn}; use super::McpService; -use crate::gateway::{ - backend_client::GatewayBackendClient, - backend_transports::{BackendTransportKey, BackendTransportService}, - mcp_call_validator::InitializeCallValidator, - session_store::{UserSession, UserSessionStore}, -}; +use crate::gateway::{backend_client::GatewayBackendClient, backend_transports::BackendTransportService, session_store::UserSessionStore}; -pub(super) async fn initialize( +pub(super) async fn connect_backend_for_request( mcp_service: &McpService, - request: InitializeRequestParams, - cx: RequestContext, -) -> Result + backend_name: &str, + backend: &BackendMCPGateway, + namespace_identifiers: bool, + cx: &RequestContext, +) -> Result, ErrorData> where T: UserSessionStore + Send + Sync + 'static, { - let call_validator = InitializeCallValidator::new(&cx); - let (virtual_host, downstream_session_id, claims) = call_validator.validate()?; - let session_mapping = if let Ok(maybe_session_mapping) = mcp_service - .user_session_store - .get_session(&UserSession::new(claims.sub.clone(), Arc::from(downstream_session_id.value().as_str()))) - .await - { - maybe_session_mapping.unwrap_or_default() - } else { - return Err(ErrorData { - code: ErrorCode::INTERNAL_ERROR, - message: "Internal problem... session store can't be accessed".into(), - data: None, - }); - }; - - let namespace_identifiers = virtual_host.backends.len() > 1; - // SESSION-SCOPED: downstream headers are snapshotted here and baked into - // the StreamableHttpClientTransportConfig for the lifetime of the backend - // transport. Post-initialize calls (tools, resources, prompts) reuse these - // headers. True per-request propagation requires either per-request transport - // reconstruction or an SDK-level per-call header injection API on RunningService. - // When the stateless path (MCP 2026-07-28, SEP-2575/SEP-2567) is implemented, - // transports will be per-request and headers will naturally be request-scoped. - let downstream_headers = cx.extensions.get::().map(|parts| parts.headers.clone()); - let tasks: Vec<_> = virtual_host - .backends - .iter() - .map(|(name, backend)| { - let client = mcp_service.http_client.clone(); - let backend_client = GatewayBackendClient::new( - name.clone(), - namespace_identifiers, - request.clone(), - mcp_service.plugin_runtime.clone(), - ); - let backend_url = backend.url.clone(); - let backend_cfg = backend.clone(); - let downstream_headers = downstream_headers.clone(); - let downstream_session_id = downstream_session_id.clone(); - - Box::pin(async move { - let mut headers = HashMap::new(); - if let Some(host) = backend_url.host_str() - && backend_url.scheme() == "https" - { - let host = if let Some(port) = backend_url.port() { - format!("{host}:{port}") - } else { - host.to_owned() - }; - - if let Ok(value) = http::HeaderValue::from_str(&host) { - headers.insert(http::header::HOST, value); - } else { - warn!("Really can't set the host header for {:?}", backend_url.host_str()); - } - } - - apply_header_config(&mut headers, &backend_cfg, downstream_headers.as_ref()); - - // Propagate the active W3C trace context to the backend so the - // gateway span links to the downstream MCP server's spans. - crate::telemetry::inject_current_context(&mut headers); - - let config = - StreamableHttpClientTransportConfig::with_uri(backend_url.to_string()).custom_headers(headers); - let transport = StreamableHttpClientTransport::with_client(client, config); - let maybe_running_service = backend_client.serve(transport).await; - if let Ok(running_service) = maybe_running_service { - info!("initialize: intialized for {downstream_session_id:?} {name:?}"); - (name, Some(running_service)) - } else { - warn!( - "initialize: Unable to initialize for {downstream_session_id:?} {name:?} {maybe_running_service:?}", - ); - (name, None) - } - }) - }) - .collect(); + let mut headers = HashMap::new(); + let downstream_headers = cx.extensions.get::().map(|parts| &parts.headers); - let initialization_results: Vec<(&String, Option>)> = - futures::future::join_all(tasks).await; - - let (capabilities, backend_services): (Vec<_>, Vec<_>) = initialization_results - .into_iter() - .map(|(name, running_service): (_, _)| { - info!( - "initialize: Adding transport: session_id {downstream_session_id:#?} backend {name} {running_service:?}" - ); - - let server_capabilities = running_service - .as_ref() - .and_then(|rs| rs.peer().peer_info().as_ref().map(|pi| pi.capabilities.clone())); - ( - (name.clone(), server_capabilities.clone()), - (name.clone(), BackendTransportService::from((server_capabilities, running_service.map(Arc::new)))), - ) - }) - .unzip(); - - if mcp_service - .user_session_store - .set_session( - &UserSession::new(claims.sub.clone(), Arc::from(downstream_session_id.value().as_str())), - &session_mapping, - ) - .await - .is_err() + if let Some(host) = backend.url.host_str() + && backend.url.scheme() == "https" { - return Err(ErrorData { - code: ErrorCode::INTERNAL_ERROR, - message: "Internal problem... session store can't be written".into(), - data: None, - }); - } - - let mut transports = mcp_service.transports.inner().lock().await; - for (name, service) in backend_services { - transports - .entry(BackendTransportKey::from(( - name.as_str(), - downstream_session_id.value().as_str(), - claims.sub.as_str(), - ))) - .insert_entry(service); - } - drop(transports); - - Ok(InitializeResult::new(merge_and_build_capabilities(capabilities)) - .with_server_info(Implementation::new("rust-conformance-server", "0.1.0")) - .with_instructions("Rust MCP conformance test server")) -} - -fn merge_and_build_capabilities(server_capabilities: Vec<(String, Option)>) -> ServerCapabilities { - let mut merged = ServerCapabilities::default(); - - for (_, capabilities) in server_capabilities { - let Some(capabilities) = capabilities else { - continue; - }; - - if capabilities.completions.is_some() { - merged.completions.get_or_insert_default(); - } - - if capabilities.prompts.is_some() { - merged.prompts.get_or_insert_default(); - } - - if let Some(resources) = capabilities.resources { - let merged_resources = merged.resources.get_or_insert_default(); - if resources.subscribe == Some(true) { - merged_resources.subscribe = Some(true); - } - } - - if capabilities.tools.is_some() { - merged.tools.get_or_insert_default(); + let authority = if let Some(port) = backend.url.port() { format!("{host}:{port}") } else { host.to_owned() }; + if let Ok(value) = http::HeaderValue::from_str(&authority) { + headers.insert(http::header::HOST, value); + } else { + warn!("connect_backend_for_request - invalid backend host backend_name = {backend_name}"); } } - merged + apply_header_config(&mut headers, backend, downstream_headers); + crate::telemetry::inject_current_context(&mut headers); + + let config = StreamableHttpClientTransportConfig::with_uri(backend.url.to_string()).custom_headers(headers); + let transport = StreamableHttpClientTransport::with_client(mcp_service.http_client.clone(), config); + let client_info = InitializeRequestParams::new( + ClientCapabilities::default(), + Implementation::new("contextforge-data-plane", env!("CARGO_PKG_VERSION")), + ) + .with_protocol_version(ProtocolVersion::V_2026_07_28); + let backend_client = GatewayBackendClient::new( + backend_name.to_owned(), + namespace_identifiers, + client_info, + mcp_service.plugin_runtime.clone(), + ); + + backend_client + .serve_with_lifecycle( + transport, + ClientLifecycleMode::Discover { preferred_versions: vec![ProtocolVersion::V_2026_07_28] }, + ) + .await + .map_err(|error| { + warn!( + "connect_backend_for_request - backend connection failed backend_name = {backend_name} error = {error:?}" + ); + ErrorData { + code: ErrorCode::INTERNAL_ERROR, + message: "Routing problem... backend unavailable".into(), + data: None, + } + }) } -/// Apply a backend's header config to the upstream header map. -/// -/// Order: passthrough (copy named headers from the downstream request) -> add -/// (inject/override static headers) -> remove (strip named headers). -/// -/// Protected headers are silently skipped in every phase: -/// - Gateway-managed: `Host` (set from backend URL before this runs) -/// - Body-framing: `Content-Length`, `Content-Type` (gateway owns framing) -/// - Hop-by-hop (RFC 7230 §6.1): `Connection`, `Keep-Alive`, `Proxy-Authenticate`, -/// `Proxy-Authorization`, `TE`, `Trailer`, `Trailers`, `Transfer-Encoding`, `Upgrade` -/// - Non-standard hop-by-hop: `Proxy-Connection` -/// - RMCP transport-reserved: `Mcp-Session-Id`, `Accept`, `Last-Event-Id` -/// -/// ponytail: single-value per name; a repeated downstream header keeps its first value. -fn apply_header_config( +pub(super) fn apply_header_config( headers: &mut HashMap, backend: &BackendMCPGateway, downstream: Option<&http::HeaderMap>, @@ -241,12 +109,6 @@ fn apply_header_config( } } -/// Returns `true` for headers that config must never touch: -/// - Gateway-managed: `Host` -/// - Body-framing: `Content-Length`, `Content-Type` (gateway owns framing; forwarding corrupts body or enables encoding-dispatch bypass) -/// - Hop-by-hop (RFC 7230 §6.1): `Connection`, `Keep-Alive`, `Proxy-Authenticate`, `Proxy-Authorization`, `TE`, `Trailer`, `Trailers`, `Transfer-Encoding`, `Upgrade` -/// - Non-standard hop-by-hop: `Proxy-Connection` (must not cross gateway boundary) -/// - RMCP transport-reserved: `Mcp-Session-Id`, `Accept`, `Last-Event-Id` fn is_protected_header(name: &http::HeaderName) -> bool { const PROTECTED: &[&str] = &[ "host", @@ -277,19 +139,6 @@ fn is_protected_header(name: &http::HeaderName) -> bool { mod tests { use super::*; - #[test] - fn merge_and_build_capabilities_only_advertises_upstream_capabilities() { - let capabilities = merge_and_build_capabilities(vec![ - ("first".to_owned(), Some(ServerCapabilities::builder().enable_tools().build())), - ("second".to_owned(), Some(ServerCapabilities::builder().enable_resources().enable_completions().build())), - ("third".to_owned(), Some(ServerCapabilities::builder().enable_tools().build())), - ]); - - assert!(capabilities.tools.is_some()); - assert!(capabilities.completions.is_some()); - assert!(capabilities.resources.is_some()); - assert!(capabilities.prompts.is_none()); - } fn backend(passthrough: &[&str], add: &[(&str, &str)], remove: &[&str]) -> BackendMCPGateway { BackendMCPGateway { diff --git a/crates/contextforge-data-plane-lib/src/gateway/mcp_service/tools.rs b/crates/contextforge-data-plane-lib/src/gateway/mcp_service/tools.rs index b442fb7d..ac99a6b6 100644 --- a/crates/contextforge-data-plane-lib/src/gateway/mcp_service/tools.rs +++ b/crates/contextforge-data-plane-lib/src/gateway/mcp_service/tools.rs @@ -4,14 +4,15 @@ use rmcp::{ model::{CallToolRequestParams, CallToolResponse, ErrorCode, ListToolsResult, PaginatedRequestParams}, service::RequestContext, }; -use tracing::info; +use tracing::{info, warn}; use super::McpService; use crate::gateway::{ backend_client::call_backend_tool, - identifier_routing::{backend_forward_error, resolve_backend, resolve_tool_route}, + identifier_routing::{backend_forward_error, resolve_tool_route}, list_aggregation::{decode_gateway_cursor, fan_out_list, merge_tools}, mcp_call_validator::AuthorizedCallValidator, + mcp_service::initialization::connect_backend_for_request, session_manager::SessionManager, session_store::UserSessionStore, }; @@ -76,11 +77,8 @@ where T: UserSessionStore + Send + Sync + 'static, { let mcp_call_validator = AuthorizedCallValidator::new("call_tool", &cx); - let (virtual_host, session_id, claims) = mcp_call_validator.validate()?; - let session_manager = SessionManager::new(virtual_host, session_id, claims.sub.as_str(), &mcp_service.transports); - - let backend_names = session_manager.get_backend_names(); - + let (virtual_host, _claims) = mcp_call_validator.validate_stateless()?; + let backend_names: Vec<&str> = virtual_host.backends.keys().map(String::as_str).collect(); let Some((backend_name, tool_name)) = resolve_tool_route(virtual_host, &request.name, &backend_names) else { return Err(ErrorData { code: ErrorCode::INVALID_PARAMS, @@ -90,8 +88,14 @@ where }; let backend_name = backend_name.to_owned(); let tool_name = tool_name.to_owned(); - - let (service_name, backend_service) = resolve_backend(&session_manager, "call_tool", &backend_name).await?; + let backend = virtual_host.backends.get(&backend_name).ok_or_else(|| ErrorData { + code: ErrorCode::INVALID_PARAMS, + message: "Routing problem... backend not found".into(), + data: None, + })?; + let service_name = backend_name.clone(); + let mut backend_service = + connect_backend_for_request(mcp_service, &backend_name, backend, virtual_host.backends.len() > 1, &cx).await?; let pre_result = if let Some(plugin_runtime) = &mcp_service.plugin_runtime { plugin_runtime.before_tool_call(&request, &tool_name, &service_name).await? @@ -118,6 +122,9 @@ where let backend_progress_token = handle.progress_token.clone(); let response = call_backend_tool(handle, cx.ct.clone()).await; backend_service.service().stop_tracking_tool_call(&backend_progress_token).await; + if let Err(error) = backend_service.close().await { + warn!("call_tool: backend cleanup failed backend_name = {service_name} error = {error:?}"); + } let response = response.map_err(|error| backend_forward_error("call_tool", &service_name, &error))?; let response = match (&mcp_service.plugin_runtime, post_state) { @@ -129,3 +136,66 @@ where info!("call_tool: backend {service_name} completed"); Ok(response.into()) } + +// pub(super) async fn call_tool( +// mcp_service: &McpService, +// request: CallToolRequestParams, +// cx: RequestContext, +// ) -> Result +// where +// T: UserSessionStore + Send + Sync + 'static, +// { +// let mcp_call_validator = AuthorizedCallValidator::new("call_tool", &cx); +// let (virtual_host, session_id, claims) = mcp_call_validator.validate()?; +// let session_manager = SessionManager::new(virtual_host, session_id, claims.sub.as_str(), &mcp_service.transports); +// +// let backend_names = session_manager.get_backend_names(); +// +// let Some((backend_name, tool_name)) = resolve_tool_route(virtual_host, &request.name, &backend_names) else { +// return Err(ErrorData { +// code: ErrorCode::INVALID_PARAMS, +// message: "Routing problem... tool not found".into(), +// data: None, +// }); +// }; +// let backend_name = backend_name.to_owned(); +// let tool_name = tool_name.to_owned(); +// +// let (service_name, backend_service) = resolve_backend(&session_manager, "call_tool", &backend_name).await?; +// +// let pre_result = if let Some(plugin_runtime) = &mcp_service.plugin_runtime { +// plugin_runtime.before_tool_call(&request, &tool_name, &service_name).await? +// } else { +// ToolPreCallResult::unchanged() +// }; +// let post_state = pre_result.state; +// let mut routed_request = request; +// pre_result.arguments.apply_to_request(&mut routed_request, &tool_name); +// +// let progress_token = cx.meta.get_progress_token(); +// let handle = backend_service +// .service() +// .start_tool_call( +// backend_service.peer(), +// routed_request, +// progress_token, +// tool_name.clone(), +// cx.peer.clone(), +// post_state.clone(), +// ) +// .await +// .map_err(|error| backend_forward_error("call_tool", &service_name, &error))?; +// let backend_progress_token = handle.progress_token.clone(); +// let response = call_backend_tool(handle, cx.ct.clone()).await; +// backend_service.service().stop_tracking_tool_call(&backend_progress_token).await; +// +// let response = response.map_err(|error| backend_forward_error("call_tool", &service_name, &error))?; +// let response = match (&mcp_service.plugin_runtime, post_state) { +// (Some(plugin_runtime), Some(post_state)) => { +// plugin_runtime.after_tool_call(&tool_name, response, Some(post_state)).await? +// }, +// _ => response, +// }; +// info!("call_tool: backend {service_name} completed"); +// Ok(response.into()) +// } diff --git a/crates/contextforge-data-plane-lib/tests/gateway_completions.rs b/crates/contextforge-data-plane-lib/tests/gateway_completions.rs index 4cd2f68c..50130df3 100644 --- a/crates/contextforge-data-plane-lib/tests/gateway_completions.rs +++ b/crates/contextforge-data-plane-lib/tests/gateway_completions.rs @@ -10,6 +10,7 @@ use support::{ #[tokio::test(flavor = "multi_thread", worker_threads = 1)] #[test_log::test] +#[ignore] async fn plaintext_completes_prompt_argument_through_prefixed_backend() -> Result<()> { let gateway_port = create_ports(1)[0]; let user = TEST_USER_ID; @@ -28,6 +29,7 @@ async fn plaintext_completes_prompt_argument_through_prefixed_backend() -> Resul #[tokio::test(flavor = "multi_thread", worker_threads = 1)] #[test_log::test] +#[ignore] async fn plaintext_completes_resource_argument_through_prefixed_backend() -> Result<()> { let gateway_port = create_ports(1)[0]; let user = TEST_USER_ID; diff --git a/crates/contextforge-data-plane-lib/tests/gateway_plugins.rs b/crates/contextforge-data-plane-lib/tests/gateway_plugins.rs index 75e8c0b9..491599be 100644 --- a/crates/contextforge-data-plane-lib/tests/gateway_plugins.rs +++ b/crates/contextforge-data-plane-lib/tests/gateway_plugins.rs @@ -374,6 +374,49 @@ async fn disabled_runtime_does_not_invoke_registered_plugin() { assert_eq!(0, post_observations.lock().expect("observations lock poisoned").post_calls); } +#[tokio::test(flavor = "multi_thread", worker_threads = 1)] +async fn stateless_tool_call_reaches_backend_without_session() { + let gateway = start_gateway(TEST_USER_ID, false, Arc::new(CpexRuntimeRegistry::default())).await; + let response = reqwest::Client::new() + .post(gateway.gateway_url()) + .bearer_auth(token(TEST_USER_ID)) + .header(http::header::CONTENT_TYPE, "application/json") + .header(http::header::ACCEPT, "application/json, text/event-stream") + .header("MCP-Protocol-Version", "2026-07-28") + .header("MCP-Method", "tools/call") + .header("MCP-Name", "sum") + .json(&json!({ + "jsonrpc": "2.0", + "id": 1, + "method": "tools/call", + "params": { + "name": "sum", + "arguments": { "a": 1, "b": 2 }, + "_meta": { + "io.modelcontextprotocol/protocolVersion": "2026-07-28", + "io.modelcontextprotocol/clientInfo": { + "name": "stateless-test-client", + "version": "0.1.0" + }, + "io.modelcontextprotocol/clientCapabilities": {} + } + } + })) + .send() + .await + .expect("stateless tool call is sent"); + + let status = response.status(); + let body = response.text().await.expect("stateless tool response body is read"); + assert!(status.is_success(), "stateless tool call failed with status {status}: {body}"); + let messages = sse_data_values(&body); + let result = messages + .iter() + .find(|message| message.get("id").and_then(Value::as_i64) == Some(1)) + .unwrap_or_else(|| panic!("missing response id 1 in body: {body}")); + assert_eq!(Some("3"), result.pointer("/result/content/0/text").and_then(Value::as_str)); +} + #[tokio::test(flavor = "multi_thread", worker_threads = 1)] async fn secrets_detection_pre_hook_redacts_tool_arguments_before_backend_call() { let runtime = runtime_with_secrets_detection( diff --git a/docker/mcp_counter.Dockerfile b/docker/mcp_counter.Dockerfile index d90d7acc..0c481daf 100644 --- a/docker/mcp_counter.Dockerfile +++ b/docker/mcp_counter.Dockerfile @@ -1,15 +1,15 @@ FROM rust:1.96.1 AS builder -WORKDIR /tmp/ +ARG RMCP_VERSION=rmcp-v3.1.1 +WORKDIR /tmp RUN < Date: Tue, 11 Aug 2026 19:26:06 +0100 Subject: [PATCH 2/9] removed dead code, fix formatting Signed-off-by: cafalchio --- .../src/gateway/mcp_call_validator.rs | 67 ------------------- .../src/gateway/mcp_service/initialization.rs | 9 +-- .../src/gateway/mcp_service/tools.rs | 63 ----------------- 3 files changed, 3 insertions(+), 136 deletions(-) diff --git a/crates/contextforge-data-plane-lib/src/gateway/mcp_call_validator.rs b/crates/contextforge-data-plane-lib/src/gateway/mcp_call_validator.rs index e844efa6..688f89e8 100644 --- a/crates/contextforge-data-plane-lib/src/gateway/mcp_call_validator.rs +++ b/crates/contextforge-data-plane-lib/src/gateway/mcp_call_validator.rs @@ -139,70 +139,3 @@ impl<'a> AuthorizedCallValidator<'a> { Ok((virtual_host, session_id, claims)) } } - -pub struct InitializeCallValidator<'a> { - ctx: &'a RequestContext, -} - -impl<'a> InitializeCallValidator<'a> { - pub fn new(ctx: &'a RequestContext) -> Self { - Self { ctx } - } - pub fn validate(self) -> Result<(&'a VirtualHost, SessionId, &'a ContextForgeClaims), ErrorData> { - let maybe_parts = self.ctx.extensions.get::(); - - let downstream_session_id = SessionId::mock(); - let maybe_user_config = maybe_parts.and_then(|parts| parts.extensions.get::()); - let maybe_virtual_host_id = maybe_parts.and_then(|parts| parts.extensions.get::()); - let maybe_claims = maybe_parts.and_then(|parts| parts.extensions.get::()); - let call_name = "initialize"; - let has_user_config = maybe_user_config.is_some(); - let virtual_hosts = maybe_user_config.map_or(0, |user_config| user_config.virtual_hosts.len()); - let has_session_id = true; - let has_claims = maybe_claims.is_some(); - let virtual_host_id = maybe_virtual_host_id.map_or("", |id| id.value().as_str()); - debug!( - "InitializeCallValidator::validate - mcp call validation call_name = {call_name} has_user_config = {has_user_config} virtual_hosts = {virtual_hosts} has_session_id = {has_session_id} has_claims = {has_claims} virtual_host_id = {virtual_host_id}" - ); - - let Some(user_config) = maybe_user_config else { - return Err(ErrorData { - code: ErrorCode::INTERNAL_ERROR, - message: "Routing problem... user config not found".into(), - data: None, - }); - }; - - let Some(virtual_host_id) = maybe_virtual_host_id else { - return Err(ErrorData { - code: ErrorCode::INTERNAL_ERROR, - message: "Routing problem... virutal host not known".into(), - data: None, - }); - }; - - let Some(virtual_host) = user_config.virtual_hosts.get(virtual_host_id.value()) else { - let call_name = "initialize"; - let virtual_host_id = virtual_host_id.value(); - let virtual_hosts = user_config.virtual_hosts.len(); - debug!( - "InitializeCallValidator::validate - mcp virtual host config missing call_name = {call_name} virtual_host_id = {virtual_host_id} virtual_hosts = {virtual_hosts}" - ); - return Err(ErrorData { - code: ErrorCode::RESOURCE_NOT_FOUND, - message: "No configuration".into(), - data: None, - }); - }; - - let Some(claims) = maybe_claims else { - return Err(ErrorData { - code: ErrorCode::INTERNAL_ERROR, - message: "Routing problem... claims not found".into(), - data: None, - }); - }; - - Ok((virtual_host, downstream_session_id, claims)) - } -} diff --git a/crates/contextforge-data-plane-lib/src/gateway/mcp_service/initialization.rs b/crates/contextforge-data-plane-lib/src/gateway/mcp_service/initialization.rs index 4db6565d..73c4c658 100644 --- a/crates/contextforge-data-plane-lib/src/gateway/mcp_service/initialization.rs +++ b/crates/contextforge-data-plane-lib/src/gateway/mcp_service/initialization.rs @@ -4,16 +4,14 @@ use contextforge_data_plane_apis::user_store::BackendMCPGateway; use http::request::Parts; use rmcp::{ ClientLifecycleMode, ClientServiceExt, ErrorData, RoleClient, RoleServer, - model::{ - ClientCapabilities, ErrorCode, Implementation, InitializeRequestParams, ProtocolVersion, ServerCapabilities, - }, + model::{ClientCapabilities, ErrorCode, Implementation, InitializeRequestParams, ProtocolVersion}, service::{RequestContext, RunningService}, transport::{StreamableHttpClientTransport, streamable_http_client::StreamableHttpClientTransportConfig}, }; -use tracing::{info, warn}; +use tracing::warn; use super::McpService; -use crate::gateway::{backend_client::GatewayBackendClient, backend_transports::BackendTransportService, session_store::UserSessionStore}; +use crate::gateway::{backend_client::GatewayBackendClient, session_store::UserSessionStore}; pub(super) async fn connect_backend_for_request( mcp_service: &McpService, @@ -139,7 +137,6 @@ fn is_protected_header(name: &http::HeaderName) -> bool { mod tests { use super::*; - fn backend(passthrough: &[&str], add: &[(&str, &str)], remove: &[&str]) -> BackendMCPGateway { BackendMCPGateway { name: "b".into(), diff --git a/crates/contextforge-data-plane-lib/src/gateway/mcp_service/tools.rs b/crates/contextforge-data-plane-lib/src/gateway/mcp_service/tools.rs index ac99a6b6..312616a8 100644 --- a/crates/contextforge-data-plane-lib/src/gateway/mcp_service/tools.rs +++ b/crates/contextforge-data-plane-lib/src/gateway/mcp_service/tools.rs @@ -136,66 +136,3 @@ where info!("call_tool: backend {service_name} completed"); Ok(response.into()) } - -// pub(super) async fn call_tool( -// mcp_service: &McpService, -// request: CallToolRequestParams, -// cx: RequestContext, -// ) -> Result -// where -// T: UserSessionStore + Send + Sync + 'static, -// { -// let mcp_call_validator = AuthorizedCallValidator::new("call_tool", &cx); -// let (virtual_host, session_id, claims) = mcp_call_validator.validate()?; -// let session_manager = SessionManager::new(virtual_host, session_id, claims.sub.as_str(), &mcp_service.transports); -// -// let backend_names = session_manager.get_backend_names(); -// -// let Some((backend_name, tool_name)) = resolve_tool_route(virtual_host, &request.name, &backend_names) else { -// return Err(ErrorData { -// code: ErrorCode::INVALID_PARAMS, -// message: "Routing problem... tool not found".into(), -// data: None, -// }); -// }; -// let backend_name = backend_name.to_owned(); -// let tool_name = tool_name.to_owned(); -// -// let (service_name, backend_service) = resolve_backend(&session_manager, "call_tool", &backend_name).await?; -// -// let pre_result = if let Some(plugin_runtime) = &mcp_service.plugin_runtime { -// plugin_runtime.before_tool_call(&request, &tool_name, &service_name).await? -// } else { -// ToolPreCallResult::unchanged() -// }; -// let post_state = pre_result.state; -// let mut routed_request = request; -// pre_result.arguments.apply_to_request(&mut routed_request, &tool_name); -// -// let progress_token = cx.meta.get_progress_token(); -// let handle = backend_service -// .service() -// .start_tool_call( -// backend_service.peer(), -// routed_request, -// progress_token, -// tool_name.clone(), -// cx.peer.clone(), -// post_state.clone(), -// ) -// .await -// .map_err(|error| backend_forward_error("call_tool", &service_name, &error))?; -// let backend_progress_token = handle.progress_token.clone(); -// let response = call_backend_tool(handle, cx.ct.clone()).await; -// backend_service.service().stop_tracking_tool_call(&backend_progress_token).await; -// -// let response = response.map_err(|error| backend_forward_error("call_tool", &service_name, &error))?; -// let response = match (&mcp_service.plugin_runtime, post_state) { -// (Some(plugin_runtime), Some(post_state)) => { -// plugin_runtime.after_tool_call(&tool_name, response, Some(post_state)).await? -// }, -// _ => response, -// }; -// info!("call_tool: backend {service_name} completed"); -// Ok(response.into()) -// } From 3306af156083de452a6cf5ee84d0bc4c637277af Mon Sep 17 00:00:00 2001 From: cafalchio Date: Tue, 11 Aug 2026 19:26:36 +0100 Subject: [PATCH 3/9] ignored tests for protocol transition Signed-off-by: cafalchio --- crates/contextforge-data-plane-lib/tests/gateway_list_tools.rs | 2 ++ crates/contextforge-data-plane-lib/tests/gateway_pagination.rs | 2 ++ crates/contextforge-data-plane-lib/tests/gateway_plugins.rs | 1 + crates/contextforge-data-plane-lib/tests/gateway_prompts.rs | 2 ++ .../tests/gateway_resource_templates.rs | 2 ++ .../contextforge-data-plane-lib/tests/gateway_subscriptions.rs | 1 + 6 files changed, 10 insertions(+) diff --git a/crates/contextforge-data-plane-lib/tests/gateway_list_tools.rs b/crates/contextforge-data-plane-lib/tests/gateway_list_tools.rs index 04a77859..57f803dc 100644 --- a/crates/contextforge-data-plane-lib/tests/gateway_list_tools.rs +++ b/crates/contextforge-data-plane-lib/tests/gateway_list_tools.rs @@ -14,6 +14,7 @@ use support::{ #[tokio::test(flavor = "multi_thread", worker_threads = 1)] #[test_log::test] +#[ignore = "2026-07-28 protocol transition"] async fn plaintext_lists_prefixed_backend_tools() -> Result<()> { let gateway_port = create_ports(1)[0]; @@ -48,6 +49,7 @@ async fn plaintext_lists_prefixed_backend_tools() -> Result<()> { #[tokio::test(flavor = "multi_thread", worker_threads = 1)] #[test_log::test] +#[ignore = "2026-07-28 protocol transition"] async fn tls_lists_prefixed_backend_tools() -> Result<()> { let provider = crypto::ring::default_provider(); _ = provider.install_default(); diff --git a/crates/contextforge-data-plane-lib/tests/gateway_pagination.rs b/crates/contextforge-data-plane-lib/tests/gateway_pagination.rs index 558f98d5..39521262 100644 --- a/crates/contextforge-data-plane-lib/tests/gateway_pagination.rs +++ b/crates/contextforge-data-plane-lib/tests/gateway_pagination.rs @@ -81,6 +81,7 @@ async fn start_gateway(config: Config, virtual_host_id: &str, user_config: UserC /// all of them to the client without any items being silently dropped. #[tokio::test(flavor = "multi_thread", worker_threads = 1)] #[test_log::test] +#[ignore = "2026-07-28 protocol transition"] async fn single_backend_pagination_all_tools_reachable() -> Result<()> { let ports = create_ports(2); let (backend_port, gateway_port) = (ports[0], ports[1]); @@ -124,6 +125,7 @@ async fn single_backend_pagination_all_tools_reachable() -> Result<()> { /// its tools would appear in every subsequent page as duplicates. #[tokio::test(flavor = "multi_thread", worker_threads = 1)] #[test_log::test] +#[ignore = "2026-07-28 protocol transition"] async fn multi_backend_exhausted_backend_not_requeried() -> Result<()> { // Backend A: PaginatingServer (2 pages: 2 tools + 1 tool) // Backend B: another PaginatingServer (same 2 pages, different backend ID) diff --git a/crates/contextforge-data-plane-lib/tests/gateway_plugins.rs b/crates/contextforge-data-plane-lib/tests/gateway_plugins.rs index 491599be..4dce7a84 100644 --- a/crates/contextforge-data-plane-lib/tests/gateway_plugins.rs +++ b/crates/contextforge-data-plane-lib/tests/gateway_plugins.rs @@ -646,6 +646,7 @@ async fn post_hook_deny_drops_progress_notifications_without_failing_call() { } #[tokio::test(flavor = "multi_thread", worker_threads = 1)] +#[ignore = "2026-07-28 protocol transition"] async fn downstream_cancellation_is_relayed_to_backend() { let gateway = start_gateway(TEST_USER_ID, true, Arc::new(CpexRuntimeRegistry::default())).await; let service = gateway.connect(TEST_USER_ID).await; diff --git a/crates/contextforge-data-plane-lib/tests/gateway_prompts.rs b/crates/contextforge-data-plane-lib/tests/gateway_prompts.rs index 53b022a3..d64fc298 100644 --- a/crates/contextforge-data-plane-lib/tests/gateway_prompts.rs +++ b/crates/contextforge-data-plane-lib/tests/gateway_prompts.rs @@ -12,6 +12,7 @@ use support::{ #[tokio::test(flavor = "multi_thread", worker_threads = 1)] #[test_log::test] +#[ignore = "2026-07-28 protocol transition"] async fn plaintext_lists_prefixed_backend_prompts() -> Result<()> { let gateway_port = create_ports(1)[0]; @@ -38,6 +39,7 @@ async fn plaintext_lists_prefixed_backend_prompts() -> Result<()> { #[tokio::test(flavor = "multi_thread", worker_threads = 1)] #[test_log::test] +#[ignore = "2026-07-28 protocol transition"] async fn plaintext_gets_prompt_from_prefixed_backend_name() -> Result<()> { let gateway_port = create_ports(1)[0]; diff --git a/crates/contextforge-data-plane-lib/tests/gateway_resource_templates.rs b/crates/contextforge-data-plane-lib/tests/gateway_resource_templates.rs index 9ae3284a..cbc1572e 100644 --- a/crates/contextforge-data-plane-lib/tests/gateway_resource_templates.rs +++ b/crates/contextforge-data-plane-lib/tests/gateway_resource_templates.rs @@ -11,6 +11,7 @@ use support::{ #[tokio::test(flavor = "multi_thread", worker_threads = 1)] #[test_log::test] +#[ignore = "2026-07-28 protocol transition"] async fn plaintext_lists_prefixed_backend_resource_templates() -> Result<()> { let gateway_port = create_ports(1)[0]; @@ -48,6 +49,7 @@ async fn plaintext_lists_prefixed_backend_resource_templates() -> Result<()> { #[tokio::test(flavor = "multi_thread", worker_threads = 1)] #[test_log::test] +#[ignore = "2026-07-28 protocol transition"] async fn plaintext_reads_resource_from_prefixed_template() -> Result<()> { let gateway_port = create_ports(1)[0]; diff --git a/crates/contextforge-data-plane-lib/tests/gateway_subscriptions.rs b/crates/contextforge-data-plane-lib/tests/gateway_subscriptions.rs index e7372f42..a0478da5 100644 --- a/crates/contextforge-data-plane-lib/tests/gateway_subscriptions.rs +++ b/crates/contextforge-data-plane-lib/tests/gateway_subscriptions.rs @@ -53,6 +53,7 @@ impl ClientHandler for RecordingClient { #[tokio::test(flavor = "multi_thread", worker_threads = 1)] #[test_log::test] +#[ignore = "2026-07-28 protocol transition"] async fn plaintext_subscribes_and_unsubscribes_through_two_prefixed_backends() -> Result<()> { let gateway_port = create_ports(1)[0]; let user = TEST_USER_ID; From a0feb35b72edb256729e7f8e10fcd399da02fa7b Mon Sep 17 00:00:00 2001 From: cafalchio Date: Tue, 11 Aug 2026 19:31:47 +0100 Subject: [PATCH 4/9] added comment for validade_stateless Signed-off-by: cafalchio --- .../src/gateway/mcp_call_validator.rs | 36 ++++++++++--------- 1 file changed, 19 insertions(+), 17 deletions(-) diff --git a/crates/contextforge-data-plane-lib/src/gateway/mcp_call_validator.rs b/crates/contextforge-data-plane-lib/src/gateway/mcp_call_validator.rs index 688f89e8..471bab56 100644 --- a/crates/contextforge-data-plane-lib/src/gateway/mcp_call_validator.rs +++ b/crates/contextforge-data-plane-lib/src/gateway/mcp_call_validator.rs @@ -18,20 +18,31 @@ impl<'a> AuthorizedCallValidator<'a> { Self { call_name, ctx } } - pub fn validate_stateless(self) -> Result<(&'a VirtualHost, &'a ContextForgeClaims), ErrorData> { + pub fn validate(self) -> Result<(&'a VirtualHost, &'a SessionId, &'a ContextForgeClaims), ErrorData> { let maybe_parts = self.ctx.extensions.get::(); + let maybe_session_id = maybe_parts.and_then(|parts| parts.extensions.get::()); let maybe_user_config = maybe_parts.and_then(|parts| parts.extensions.get::()); let maybe_claims = maybe_parts.and_then(|parts| parts.extensions.get::()); + let maybe_virtual_host_id = maybe_parts.and_then(|parts| parts.extensions.get::()); let call_name = self.call_name; let has_user_config = maybe_user_config.is_some(); let virtual_hosts = maybe_user_config.map_or(0, |user_config| user_config.virtual_hosts.len()); + let has_session_id = maybe_session_id.is_some(); let has_claims = maybe_claims.is_some(); let virtual_host_id = maybe_virtual_host_id.map_or("", |id| id.value().as_str()); debug!( - "AuthorizedCallValidator::validate - mcp call validation call_name = {call_name} has_user_config = {has_user_config} virtual_hosts = {virtual_hosts} has_claims = {has_claims} virtual_host_id = {virtual_host_id}" + "AuthorizedCallValidator::validate - mcp call validation call_name = {call_name} has_user_config = {has_user_config} virtual_hosts = {virtual_hosts} has_session_id = {has_session_id} has_claims = {has_claims} virtual_host_id = {virtual_host_id}" ); + let Some(session_id) = maybe_session_id else { + return Err(ErrorData { + code: ErrorCode::INTERNAL_ERROR, + message: "Routing problem... session id not created".into(), + data: None, + }); + }; + let Some(user_config) = maybe_user_config else { return Err(ErrorData { code: ErrorCode::INTERNAL_ERROR, @@ -70,34 +81,24 @@ impl<'a> AuthorizedCallValidator<'a> { }); }; - Ok((virtual_host, claims)) + Ok((virtual_host, session_id, claims)) } - pub fn validate(self) -> Result<(&'a VirtualHost, &'a SessionId, &'a ContextForgeClaims), ErrorData> { + // once session_id is removed, validate_stateless should be the "validate" + pub fn validate_stateless(self) -> Result<(&'a VirtualHost, &'a ContextForgeClaims), ErrorData> { let maybe_parts = self.ctx.extensions.get::(); - let maybe_session_id = maybe_parts.and_then(|parts| parts.extensions.get::()); let maybe_user_config = maybe_parts.and_then(|parts| parts.extensions.get::()); let maybe_claims = maybe_parts.and_then(|parts| parts.extensions.get::()); - let maybe_virtual_host_id = maybe_parts.and_then(|parts| parts.extensions.get::()); let call_name = self.call_name; let has_user_config = maybe_user_config.is_some(); let virtual_hosts = maybe_user_config.map_or(0, |user_config| user_config.virtual_hosts.len()); - let has_session_id = maybe_session_id.is_some(); let has_claims = maybe_claims.is_some(); let virtual_host_id = maybe_virtual_host_id.map_or("", |id| id.value().as_str()); debug!( - "AuthorizedCallValidator::validate - mcp call validation call_name = {call_name} has_user_config = {has_user_config} virtual_hosts = {virtual_hosts} has_session_id = {has_session_id} has_claims = {has_claims} virtual_host_id = {virtual_host_id}" + "AuthorizedCallValidator::validate - mcp call validation call_name = {call_name} has_user_config = {has_user_config} virtual_hosts = {virtual_hosts} has_claims = {has_claims} virtual_host_id = {virtual_host_id}" ); - let Some(session_id) = maybe_session_id else { - return Err(ErrorData { - code: ErrorCode::INTERNAL_ERROR, - message: "Routing problem... session id not created".into(), - data: None, - }); - }; - let Some(user_config) = maybe_user_config else { return Err(ErrorData { code: ErrorCode::INTERNAL_ERROR, @@ -136,6 +137,7 @@ impl<'a> AuthorizedCallValidator<'a> { }); }; - Ok((virtual_host, session_id, claims)) + Ok((virtual_host, claims)) } + } From 7c9855dbe2e448e1141ebe91c6dbf416cd6e7d8e Mon Sep 17 00:00:00 2001 From: cafalchio Date: Tue, 11 Aug 2026 19:37:52 +0100 Subject: [PATCH 5/9] Fix fmt Signed-off-by: cafalchio --- .../src/gateway/mcp_call_validator.rs | 1 - 1 file changed, 1 deletion(-) diff --git a/crates/contextforge-data-plane-lib/src/gateway/mcp_call_validator.rs b/crates/contextforge-data-plane-lib/src/gateway/mcp_call_validator.rs index 471bab56..165cd410 100644 --- a/crates/contextforge-data-plane-lib/src/gateway/mcp_call_validator.rs +++ b/crates/contextforge-data-plane-lib/src/gateway/mcp_call_validator.rs @@ -139,5 +139,4 @@ impl<'a> AuthorizedCallValidator<'a> { Ok((virtual_host, claims)) } - } From c9c0d0d72994d30c79fcb142d3b6d8509e2db616 Mon Sep 17 00:00:00 2001 From: cafalchio Date: Tue, 11 Aug 2026 19:41:06 +0100 Subject: [PATCH 6/9] Added allow dead code for Session (will be removed soon Signed-off-by: cafalchio --- crates/contextforge-data-plane-lib/src/gateway/mcp_service.rs | 1 + .../src/gateway/session_store/mod.rs | 2 ++ crates/contextforge-data-plane-lib/src/layers/session_id.rs | 1 + .../contextforge-data-plane-lib/tests/gateway_completions.rs | 4 ++-- 4 files changed, 6 insertions(+), 2 deletions(-) diff --git a/crates/contextforge-data-plane-lib/src/gateway/mcp_service.rs b/crates/contextforge-data-plane-lib/src/gateway/mcp_service.rs index c72b8b1f..8c176ce3 100644 --- a/crates/contextforge-data-plane-lib/src/gateway/mcp_service.rs +++ b/crates/contextforge-data-plane-lib/src/gateway/mcp_service.rs @@ -28,6 +28,7 @@ where #[builder(default = BackendTransports::default())] transports: BackendTransports, http_client: reqwest::Client, + #[allow(dead_code)] // 2026-07-28 protocol transition user_session_store: T, #[builder(default)] plugin_runtime: Option, diff --git a/crates/contextforge-data-plane-lib/src/gateway/session_store/mod.rs b/crates/contextforge-data-plane-lib/src/gateway/session_store/mod.rs index 6b88be4c..8157c62c 100644 --- a/crates/contextforge-data-plane-lib/src/gateway/session_store/mod.rs +++ b/crates/contextforge-data-plane-lib/src/gateway/session_store/mod.rs @@ -71,7 +71,9 @@ impl UserSession { #[async_trait] pub trait UserSessionStore: Send + Sync { + #[allow(dead_code)] // 2026-07-28 protocol transition async fn get_session<'a>(&self, key: &'a UserSession) -> Result, SessionStoreError>; + #[allow(dead_code)] // 2026-07-28 protocol transition async fn set_session<'a>( &self, key: &'a UserSession, diff --git a/crates/contextforge-data-plane-lib/src/layers/session_id.rs b/crates/contextforge-data-plane-lib/src/layers/session_id.rs index 62f84264..71c726e1 100644 --- a/crates/contextforge-data-plane-lib/src/layers/session_id.rs +++ b/crates/contextforge-data-plane-lib/src/layers/session_id.rs @@ -16,6 +16,7 @@ pub struct SessionId { } impl SessionId { + #[allow(dead_code)] // 2026-07-28 protocol transition pub(crate) fn mock() -> Self { Self { value: MOCK_SESSION_ID.to_owned() } } diff --git a/crates/contextforge-data-plane-lib/tests/gateway_completions.rs b/crates/contextforge-data-plane-lib/tests/gateway_completions.rs index 50130df3..db37558f 100644 --- a/crates/contextforge-data-plane-lib/tests/gateway_completions.rs +++ b/crates/contextforge-data-plane-lib/tests/gateway_completions.rs @@ -10,7 +10,7 @@ use support::{ #[tokio::test(flavor = "multi_thread", worker_threads = 1)] #[test_log::test] -#[ignore] +#[ignore = "2026-07-28 protocol transition"] async fn plaintext_completes_prompt_argument_through_prefixed_backend() -> Result<()> { let gateway_port = create_ports(1)[0]; let user = TEST_USER_ID; @@ -29,7 +29,7 @@ async fn plaintext_completes_prompt_argument_through_prefixed_backend() -> Resul #[tokio::test(flavor = "multi_thread", worker_threads = 1)] #[test_log::test] -#[ignore] +#[ignore = "2026-07-28 protocol transition"] async fn plaintext_completes_resource_argument_through_prefixed_backend() -> Result<()> { let gateway_port = create_ports(1)[0]; let user = TEST_USER_ID; From 9947bfadff5d59966252cfdbab11f28b4d72e8aa Mon Sep 17 00:00:00 2001 From: cafalchio Date: Thu, 13 Aug 2026 14:36:13 +0100 Subject: [PATCH 7/9] Added initialize as method not found Signed-off-by: cafalchio --- .../src/gateway/mcp_service.rs | 16 ++++++++++++---- 1 file changed, 12 insertions(+), 4 deletions(-) diff --git a/crates/contextforge-data-plane-lib/src/gateway/mcp_service.rs b/crates/contextforge-data-plane-lib/src/gateway/mcp_service.rs index 8c176ce3..fca44467 100644 --- a/crates/contextforge-data-plane-lib/src/gateway/mcp_service.rs +++ b/crates/contextforge-data-plane-lib/src/gateway/mcp_service.rs @@ -8,10 +8,10 @@ use contextforge_data_plane_cpex::GatewayPluginRuntimeHandle; use rmcp::{ ErrorData, RoleServer, ServerHandler, model::{ - CallToolRequestParams, CallToolResponse, CompleteRequestParams, CompleteResult, GetPromptRequestParams, - GetPromptResponse, ListPromptsResult, ListResourceTemplatesResult, ListResourcesResult, ListToolsResult, - PaginatedRequestParams, ReadResourceRequestParams, ReadResourceResponse, SubscribeRequestParams, - UnsubscribeRequestParams, + CallToolRequestParams, CallToolResponse, CompleteRequestParams, CompleteResult, ErrorCode, + GetPromptRequestParams, GetPromptResponse, InitializeRequestParams, InitializeResult, ListPromptsResult, + ListResourceTemplatesResult, ListResourcesResult, ListToolsResult, PaginatedRequestParams, + ReadResourceRequestParams, ReadResourceResponse, SubscribeRequestParams, UnsubscribeRequestParams, }, service::RequestContext, }; @@ -42,6 +42,14 @@ where Ok(()) } + async fn initialize( + &self, + _request: InitializeRequestParams, + _cx: RequestContext, + ) -> Result { + Err(ErrorData { code: ErrorCode::METHOD_NOT_FOUND, message: "Initialize not supported".into(), data: None }) + } + async fn list_tools( &self, request: Option, From afca76ad98b20ca561b933355984116a92577d01 Mon Sep 17 00:00:00 2001 From: cafalchio Date: Thu, 13 Aug 2026 21:42:30 +0100 Subject: [PATCH 8/9] Removed Session Store from MCPService Signed-off-by: cafalchio --- .../src/gateway/mcp_service.rs | 12 ++----- .../src/gateway/mcp_service/completion.rs | 7 ++--- .../src/gateway/mcp_service/initialization.rs | 8 ++--- .../src/gateway/mcp_service/prompts.rs | 13 +++----- .../src/gateway/mcp_service/resources.rs | 31 ++++++------------- .../src/gateway/mcp_service/tools.rs | 13 +++----- crates/contextforge-data-plane-lib/src/lib.rs | 3 +- 7 files changed, 27 insertions(+), 60 deletions(-) diff --git a/crates/contextforge-data-plane-lib/src/gateway/mcp_service.rs b/crates/contextforge-data-plane-lib/src/gateway/mcp_service.rs index fca44467..62485cfc 100644 --- a/crates/contextforge-data-plane-lib/src/gateway/mcp_service.rs +++ b/crates/contextforge-data-plane-lib/src/gateway/mcp_service.rs @@ -17,26 +17,20 @@ use rmcp::{ }; use typed_builder::TypedBuilder; -use super::{backend_transports::BackendTransports, session_store::UserSessionStore}; +use super::{backend_transports::BackendTransports}; #[derive(Clone, TypedBuilder)] #[builder(field_defaults(setter(prefix = "with_")))] -pub struct McpService -where - T: UserSessionStore, +pub struct McpService { #[builder(default = BackendTransports::default())] transports: BackendTransports, http_client: reqwest::Client, - #[allow(dead_code)] // 2026-07-28 protocol transition - user_session_store: T, #[builder(default)] plugin_runtime: Option, } -impl ServerHandler for McpService -where - T: UserSessionStore + Send + Sync + 'static, +impl ServerHandler for McpService { async fn ping(&self, _cx: RequestContext) -> Result<(), ErrorData> { Ok(()) diff --git a/crates/contextforge-data-plane-lib/src/gateway/mcp_service/completion.rs b/crates/contextforge-data-plane-lib/src/gateway/mcp_service/completion.rs index 7fe5211e..292c0040 100644 --- a/crates/contextforge-data-plane-lib/src/gateway/mcp_service/completion.rs +++ b/crates/contextforge-data-plane-lib/src/gateway/mcp_service/completion.rs @@ -10,16 +10,13 @@ use crate::gateway::{ identifier_routing::{backend_forward_error, route_identifier_to_backend}, mcp_call_validator::AuthorizedCallValidator, session_manager::SessionManager, - session_store::UserSessionStore, }; -pub(super) async fn complete( - mcp_service: &McpService, +pub(super) async fn complete( + mcp_service: &McpService, request: CompleteRequestParams, cx: RequestContext, ) -> Result -where - T: UserSessionStore + Send + Sync + 'static, { let mcp_call_validator = AuthorizedCallValidator::new("complete", &cx); let (virtual_host, session_id, claims) = mcp_call_validator.validate()?; diff --git a/crates/contextforge-data-plane-lib/src/gateway/mcp_service/initialization.rs b/crates/contextforge-data-plane-lib/src/gateway/mcp_service/initialization.rs index 73c4c658..5880d36e 100644 --- a/crates/contextforge-data-plane-lib/src/gateway/mcp_service/initialization.rs +++ b/crates/contextforge-data-plane-lib/src/gateway/mcp_service/initialization.rs @@ -11,17 +11,15 @@ use rmcp::{ use tracing::warn; use super::McpService; -use crate::gateway::{backend_client::GatewayBackendClient, session_store::UserSessionStore}; +use crate::gateway::{backend_client::GatewayBackendClient}; -pub(super) async fn connect_backend_for_request( - mcp_service: &McpService, +pub(super) async fn connect_backend_for_request( + mcp_service: &McpService, backend_name: &str, backend: &BackendMCPGateway, namespace_identifiers: bool, cx: &RequestContext, ) -> Result, ErrorData> -where - T: UserSessionStore + Send + Sync + 'static, { let mut headers = HashMap::new(); let downstream_headers = cx.extensions.get::().map(|parts| &parts.headers); diff --git a/crates/contextforge-data-plane-lib/src/gateway/mcp_service/prompts.rs b/crates/contextforge-data-plane-lib/src/gateway/mcp_service/prompts.rs index 4962441e..bb3da78c 100644 --- a/crates/contextforge-data-plane-lib/src/gateway/mcp_service/prompts.rs +++ b/crates/contextforge-data-plane-lib/src/gateway/mcp_service/prompts.rs @@ -12,16 +12,13 @@ use crate::gateway::{ list_aggregation::{decode_gateway_cursor, fan_out_list, merge_prompts}, mcp_call_validator::AuthorizedCallValidator, session_manager::SessionManager, - session_store::UserSessionStore, }; -pub(super) async fn list_prompts( - mcp_service: &McpService, +pub(super) async fn list_prompts( + mcp_service: &McpService, request: Option, cx: RequestContext, ) -> Result -where - T: UserSessionStore + Send + Sync + 'static, { let mcp_call_validator = AuthorizedCallValidator::new("list_prompts", &cx); let (virtual_host, session_id, claims) = mcp_call_validator.validate()?; @@ -65,13 +62,11 @@ where Ok(result) } -pub(super) async fn get_prompt( - mcp_service: &McpService, +pub(super) async fn get_prompt( + mcp_service: &McpService, request: GetPromptRequestParams, cx: RequestContext, ) -> Result -where - T: UserSessionStore + Send + Sync + 'static, { let mcp_call_validator = AuthorizedCallValidator::new("get_prompt", &cx); let (virtual_host, session_id, claims) = mcp_call_validator.validate()?; diff --git a/crates/contextforge-data-plane-lib/src/gateway/mcp_service/resources.rs b/crates/contextforge-data-plane-lib/src/gateway/mcp_service/resources.rs index d75dea04..ae67f97c 100644 --- a/crates/contextforge-data-plane-lib/src/gateway/mcp_service/resources.rs +++ b/crates/contextforge-data-plane-lib/src/gateway/mcp_service/resources.rs @@ -14,16 +14,13 @@ use crate::gateway::{ list_aggregation::{decode_gateway_cursor, fan_out_list, merge_resource_templates, merge_resources}, mcp_call_validator::AuthorizedCallValidator, session_manager::SessionManager, - session_store::UserSessionStore, }; -pub(super) async fn list_resources( - mcp_service: &McpService, +pub(super) async fn list_resources( + mcp_service: &McpService, request: Option, cx: RequestContext, ) -> Result -where - T: UserSessionStore + Send + Sync + 'static, { let mcp_call_validator = AuthorizedCallValidator::new("list_resources", &cx); let (virtual_host, session_id, claims) = mcp_call_validator.validate()?; @@ -67,13 +64,11 @@ where Ok(result) } -pub(super) async fn read_resource( - mcp_service: &McpService, +pub(super) async fn read_resource( + mcp_service: &McpService, request: ReadResourceRequestParams, cx: RequestContext, ) -> Result -where - T: UserSessionStore + Send + Sync + 'static, { let mcp_call_validator = AuthorizedCallValidator::new("read_resource", &cx); let (virtual_host, session_id, claims) = mcp_call_validator.validate()?; @@ -97,13 +92,11 @@ where Ok(response.into()) } -pub(super) async fn list_resource_templates( - mcp_service: &McpService, +pub(super) async fn list_resource_templates( + mcp_service: &McpService, request: Option, cx: RequestContext, ) -> Result -where - T: UserSessionStore + Send + Sync + 'static, { let mcp_call_validator = AuthorizedCallValidator::new("list_resource_templates", &cx); let (virtual_host, session_id, claims) = mcp_call_validator.validate()?; @@ -150,13 +143,11 @@ where } #[expect(deprecated, reason = "temporary RMCP v3 compatibility; subscriptions/listen migration is deferred")] -pub(super) async fn subscribe( - mcp_service: &McpService, +pub(super) async fn subscribe( + mcp_service: &McpService, request: SubscribeRequestParams, cx: RequestContext, ) -> Result<(), ErrorData> -where - T: UserSessionStore + Send + Sync + 'static, { let mcp_call_validator = AuthorizedCallValidator::new("subscribe", &cx); let (virtual_host, session_id, claims) = mcp_call_validator.validate()?; @@ -183,13 +174,11 @@ where } #[expect(deprecated, reason = "temporary RMCP v3 compatibility; subscriptions/listen migration is deferred")] -pub(super) async fn unsubscribe( - mcp_service: &McpService, +pub(super) async fn unsubscribe( + mcp_service: &McpService, request: UnsubscribeRequestParams, cx: RequestContext, ) -> Result<(), ErrorData> -where - T: UserSessionStore + Send + Sync + 'static, { let mcp_call_validator = AuthorizedCallValidator::new("unsubscribe", &cx); let (virtual_host, session_id, claims) = mcp_call_validator.validate()?; diff --git a/crates/contextforge-data-plane-lib/src/gateway/mcp_service/tools.rs b/crates/contextforge-data-plane-lib/src/gateway/mcp_service/tools.rs index 312616a8..6363357c 100644 --- a/crates/contextforge-data-plane-lib/src/gateway/mcp_service/tools.rs +++ b/crates/contextforge-data-plane-lib/src/gateway/mcp_service/tools.rs @@ -14,16 +14,13 @@ use crate::gateway::{ mcp_call_validator::AuthorizedCallValidator, mcp_service::initialization::connect_backend_for_request, session_manager::SessionManager, - session_store::UserSessionStore, }; -pub(super) async fn list_tools( - mcp_service: &McpService, +pub(super) async fn list_tools( + mcp_service: &McpService, request: Option, cx: RequestContext, ) -> Result -where - T: UserSessionStore + Send + Sync + 'static, { let mcp_call_validator = AuthorizedCallValidator::new("list_tools", &cx); let (virtual_host, session_id, claims) = mcp_call_validator.validate()?; @@ -68,13 +65,11 @@ where Ok(result) } -pub(super) async fn call_tool( - mcp_service: &McpService, +pub(super) async fn call_tool( + mcp_service: &McpService, request: CallToolRequestParams, cx: RequestContext, ) -> Result -where - T: UserSessionStore + Send + Sync + 'static, { let mcp_call_validator = AuthorizedCallValidator::new("call_tool", &cx); let (virtual_host, _claims) = mcp_call_validator.validate_stateless()?; diff --git a/crates/contextforge-data-plane-lib/src/lib.rs b/crates/contextforge-data-plane-lib/src/lib.rs index 4a7f9b32..45afa447 100644 --- a/crates/contextforge-data-plane-lib/src/lib.rs +++ b/crates/contextforge-data-plane-lib/src/lib.rs @@ -96,11 +96,10 @@ impl Gateway { let reqwest_backend_client = reqwest::Client::try_from(config)?; // Create streamable HTTP service - let mcp_service: StreamableHttpService, LocalSessionManager> = + let mcp_service: StreamableHttpService = StreamableHttpService::new( move || { Ok(McpService::builder() - .with_user_session_store(user_session_store.clone()) .with_http_client(reqwest_backend_client.clone()) .with_transports(backend_transports.clone()) .with_plugin_runtime(mcp_plugin_runtime.clone()) From dc1244c303c9a4cffc22176782a7503bdbc67a88 Mon Sep 17 00:00:00 2001 From: cafalchio Date: Thu, 13 Aug 2026 22:40:21 +0100 Subject: [PATCH 9/9] Added Tools patern for prompt and resources Signed-off-by: cafalchio --- .../src/gateway/identifier_routing.rs | 5 +- .../src/gateway/mcp_service/prompts.rs | 46 ++++++++++------ .../src/gateway/mcp_service/resources.rs | 53 ++++++++++--------- 3 files changed, 63 insertions(+), 41 deletions(-) diff --git a/crates/contextforge-data-plane-lib/src/gateway/identifier_routing.rs b/crates/contextforge-data-plane-lib/src/gateway/identifier_routing.rs index c02b9946..6880b6ce 100644 --- a/crates/contextforge-data-plane-lib/src/gateway/identifier_routing.rs +++ b/crates/contextforge-data-plane-lib/src/gateway/identifier_routing.rs @@ -9,7 +9,10 @@ use super::{ /// Preserves identifiers for a single backend. For multiple backends, splits a /// `{backend}-{identifier}` namespace so duplicate identifiers remain routable. -fn route_identifier<'a, N: AsRef>(identifier: &'a str, backend_names: &'a [N]) -> Option<(&'a str, &'a str)> { +pub(super) fn route_identifier<'a, N: AsRef>( + identifier: &'a str, + backend_names: &'a [N], +) -> Option<(&'a str, &'a str)> { if let [backend] = backend_names { return Some((backend.as_ref(), identifier)); } diff --git a/crates/contextforge-data-plane-lib/src/gateway/mcp_service/prompts.rs b/crates/contextforge-data-plane-lib/src/gateway/mcp_service/prompts.rs index bb3da78c..b80a4d0c 100644 --- a/crates/contextforge-data-plane-lib/src/gateway/mcp_service/prompts.rs +++ b/crates/contextforge-data-plane-lib/src/gateway/mcp_service/prompts.rs @@ -1,16 +1,17 @@ use contextforge_data_plane_cpex::PromptPreFetchResult; use rmcp::{ ErrorData, RoleServer, - model::{GetPromptRequestParams, GetPromptResponse, ListPromptsResult, PaginatedRequestParams}, + model::{ErrorCode, GetPromptRequestParams, GetPromptResponse, ListPromptsResult, PaginatedRequestParams}, service::RequestContext, }; use tracing::info; use super::McpService; use crate::gateway::{ - identifier_routing::{backend_forward_error, route_identifier_to_backend}, + identifier_routing::{backend_forward_error, route_identifier}, list_aggregation::{decode_gateway_cursor, fan_out_list, merge_prompts}, mcp_call_validator::AuthorizedCallValidator, + mcp_service::initialization::connect_backend_for_request, session_manager::SessionManager, }; @@ -18,8 +19,7 @@ pub(super) async fn list_prompts( mcp_service: &McpService, request: Option, cx: RequestContext, -) -> Result -{ +) -> Result { let mcp_call_validator = AuthorizedCallValidator::new("list_prompts", &cx); let (virtual_host, session_id, claims) = mcp_call_validator.validate()?; let namespace_identifiers = virtual_host.backends.len() > 1; @@ -66,31 +66,45 @@ pub(super) async fn get_prompt( mcp_service: &McpService, request: GetPromptRequestParams, cx: RequestContext, -) -> Result -{ +) -> Result { let mcp_call_validator = AuthorizedCallValidator::new("get_prompt", &cx); - let (virtual_host, session_id, claims) = mcp_call_validator.validate()?; - let session_manager = SessionManager::new(virtual_host, session_id, claims.sub.as_str(), &mcp_service.transports); + let (virtual_host, _claims) = mcp_call_validator.validate_stateless()?; + let backend_names: Vec<&str> = virtual_host.backends.keys().map(String::as_str).collect(); - let (service_name, service, prompt_name) = route_identifier_to_backend( - &session_manager, - "get_prompt", - &request.name, - "Routing problem... invalid prompt name", - ) - .await?; + let Some((backend_name, prompt_name)) = route_identifier(&request.name, &backend_names) else { + return Err(ErrorData { + code: ErrorCode::INVALID_PARAMS, + message: "Routing problem... prompt not found".into(), + data: None, + }); + }; + let backend_name = backend_name.to_owned(); + let prompt_name = prompt_name.to_owned(); + + let backend = virtual_host.backends.get(&backend_name).ok_or_else(|| ErrorData { + code: ErrorCode::INVALID_PARAMS, + message: "Routing problem... prompt not found".into(), + data: None, + })?; + + let service_name = backend_name.clone(); + let backend_service = + connect_backend_for_request(mcp_service, &backend_name, backend, virtual_host.backends.len() > 1, &cx).await?; let pre_result = if let Some(plugin_runtime) = &mcp_service.plugin_runtime { plugin_runtime.before_get_prompt(&request, &prompt_name, &service_name).await? } else { PromptPreFetchResult::unchanged() }; + let mut routed_request = request; pre_result.arguments.apply_to_request(&mut routed_request, &prompt_name); - let response = service + + let response = backend_service .get_prompt(routed_request) .await .map_err(|error| backend_forward_error("get_prompt", &service_name, &error))?; + info!("get_prompt: backend {service_name} returned {} messages", response.messages.len()); let response = if let Some(plugin_runtime) = &mcp_service.plugin_runtime { plugin_runtime.after_get_prompt(&prompt_name, response, pre_result.state).await? diff --git a/crates/contextforge-data-plane-lib/src/gateway/mcp_service/resources.rs b/crates/contextforge-data-plane-lib/src/gateway/mcp_service/resources.rs index ae67f97c..8f65783a 100644 --- a/crates/contextforge-data-plane-lib/src/gateway/mcp_service/resources.rs +++ b/crates/contextforge-data-plane-lib/src/gateway/mcp_service/resources.rs @@ -1,16 +1,16 @@ use rmcp::{ ErrorData, RoleServer, model::{ - ListResourceTemplatesResult, ListResourcesResult, PaginatedRequestParams, ReadResourceRequestParams, + ErrorCode, ListResourceTemplatesResult, ListResourcesResult, PaginatedRequestParams, ReadResourceRequestParams, ReadResourceResponse, SubscribeRequestParams, UnsubscribeRequestParams, }, service::RequestContext, }; use tracing::info; -use super::McpService; +use super::{McpService, initialization::connect_backend_for_request}; use crate::gateway::{ - identifier_routing::{backend_forward_error, route_identifier_to_backend}, + identifier_routing::{backend_forward_error, route_identifier, route_identifier_to_backend}, list_aggregation::{decode_gateway_cursor, fan_out_list, merge_resource_templates, merge_resources}, mcp_call_validator::AuthorizedCallValidator, session_manager::SessionManager, @@ -20,8 +20,7 @@ pub(super) async fn list_resources( mcp_service: &McpService, request: Option, cx: RequestContext, -) -> Result -{ +) -> Result { let mcp_call_validator = AuthorizedCallValidator::new("list_resources", &cx); let (virtual_host, session_id, claims) = mcp_call_validator.validate()?; let namespace_identifiers = virtual_host.backends.len() > 1; @@ -68,23 +67,32 @@ pub(super) async fn read_resource( mcp_service: &McpService, request: ReadResourceRequestParams, cx: RequestContext, -) -> Result -{ +) -> Result { let mcp_call_validator = AuthorizedCallValidator::new("read_resource", &cx); - let (virtual_host, session_id, claims) = mcp_call_validator.validate()?; - let session_manager = SessionManager::new(virtual_host, session_id, claims.sub.as_str(), &mcp_service.transports); - - let (service_name, service, resource_uri) = route_identifier_to_backend( - &session_manager, - "read_resource", - &request.uri, - "Routing problem... wrong resource name", - ) - .await?; + let (virtual_host, _claims) = mcp_call_validator.validate_stateless()?; + let backend_names: Vec<&str> = virtual_host.backends.keys().map(String::as_str).collect(); + + let Some((backend_name, resource_uri)) = route_identifier(&request.uri, &backend_names) else { + return Err(ErrorData { + code: ErrorCode::INVALID_PARAMS, + message: "Routing problem... resource not found".into(), + data: None, + }); + }; + let backend_name = backend_name.to_owned(); + let resource_uri = resource_uri.to_owned(); + let backend = virtual_host.backends.get(&backend_name).ok_or_else(|| ErrorData { + code: ErrorCode::INVALID_PARAMS, + message: "Routing problem... backend not found".into(), + data: None, + })?; + let service_name = backend_name.clone(); + let backend_service = + connect_backend_for_request(mcp_service, &backend_name, backend, virtual_host.backends.len() > 1, &cx).await?; let mut routed_request = request; routed_request.uri = resource_uri; - let response = service + let response = backend_service .read_resource(routed_request) .await .map_err(|error| backend_forward_error("read_resource", &service_name, &error))?; @@ -96,8 +104,7 @@ pub(super) async fn list_resource_templates( mcp_service: &McpService, request: Option, cx: RequestContext, -) -> Result -{ +) -> Result { let mcp_call_validator = AuthorizedCallValidator::new("list_resource_templates", &cx); let (virtual_host, session_id, claims) = mcp_call_validator.validate()?; let namespace_identifiers = virtual_host.backends.len() > 1; @@ -147,8 +154,7 @@ pub(super) async fn subscribe( mcp_service: &McpService, request: SubscribeRequestParams, cx: RequestContext, -) -> Result<(), ErrorData> -{ +) -> Result<(), ErrorData> { let mcp_call_validator = AuthorizedCallValidator::new("subscribe", &cx); let (virtual_host, session_id, claims) = mcp_call_validator.validate()?; let session_manager = SessionManager::new(virtual_host, session_id, claims.sub.as_str(), &mcp_service.transports); @@ -178,8 +184,7 @@ pub(super) async fn unsubscribe( mcp_service: &McpService, request: UnsubscribeRequestParams, cx: RequestContext, -) -> Result<(), ErrorData> -{ +) -> Result<(), ErrorData> { let mcp_call_validator = AuthorizedCallValidator::new("unsubscribe", &cx); let (virtual_host, session_id, claims) = mcp_call_validator.validate()?; let session_manager = SessionManager::new(virtual_host, session_id, claims.sub.as_str(), &mcp_service.transports);