From f5c04520058b939e4c31ef19519eed6799912a65 Mon Sep 17 00:00:00 2001 From: cafalchio Date: Thu, 27 Aug 2026 17:13:16 +0100 Subject: [PATCH 1/3] Added complete/completion Signed-off-by: cafalchio --- .../src/gateway/mcp_service/completion.rs | 60 ++++++++++++++++--- 1 file changed, 52 insertions(+), 8 deletions(-) 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 c8d83e14..ef0c0846 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 @@ -1,20 +1,64 @@ use rmcp::{ ErrorData, RoleServer, - model::{CompleteRequestParams, CompleteResult, ErrorCode}, + model::{CompleteRequestParams, CompleteResult, ErrorCode, Reference}, service::RequestContext, }; +use tracing::info; + +use crate::gateway::{ + mcp_call_validator::AuthorizedCallValidator, + mcp_service::initialization::connect_backend_for_request, + routing_error::backend_forward_error, +}; use super::McpService; #[allow(clippy::unused_async)] pub(super) async fn complete( - _: &McpService, - _: CompleteRequestParams, - _: RequestContext, + mcp_service: &McpService, + request: CompleteRequestParams, + cx: RequestContext, ) -> Result { - Err(ErrorData { - code: ErrorCode::INVALID_REQUEST, - message: "Fan out not supported at the moment. Go to control plane".into(), + info!("\n\n\n---------- request ----- {:?}", &request); + let mcp_call_validator = AuthorizedCallValidator::new("complete", &cx); + let (virtual_host, _claims) = mcp_call_validator.validate_stateless()?; + + let route = match &request.r#ref { + Reference::Prompt(prompt) => virtual_host.prompts.get(&prompt.name), + Reference::Resource(resource) => virtual_host.resource_templates.get(&resource.uri), + }; + let Some(route) = route else { + return Err(ErrorData { + code: ErrorCode::INVALID_PARAMS, + message: "Routing problem... completion not found".into(), + data: None, + }); + }; + let backend_name = route.backend_name.clone(); + let upstream_name = route.upstream_name.clone(); + + 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, &cx).await?; + + let mut routed_request = request; + match &mut routed_request.r#ref { + Reference::Prompt(prompt) => prompt.name = upstream_name, + Reference::Resource(resource) => resource.uri = upstream_name, + } + + let response = backend_service.complete(routed_request).await; + if let Err(error) = backend_service.close().await { + tracing::warn!("complete: backend cleanup failed backend_name = {service_name} error = {error:?}"); + } + let response = response.map_err(|error| backend_forward_error("complete", &service_name, &error))?; + + info!("read_resource: backend {service_name} returned {} contents", response.completion.values.len()); + + Ok(response.into()) } From 0779647ecacf3e697bc5358be32ce22f92ed54f4 Mon Sep 17 00:00:00 2001 From: cafalchio Date: Fri, 28 Aug 2026 00:16:36 +0100 Subject: [PATCH 2/3] fmt clippy and small cleanup Signed-off-by: cafalchio --- .../src/gateway/mcp_service/completion.rs | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) 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 ef0c0846..79e98c71 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 @@ -19,7 +19,6 @@ pub(super) async fn complete( request: CompleteRequestParams, cx: RequestContext, ) -> Result { - info!("\n\n\n---------- request ----- {:?}", &request); let mcp_call_validator = AuthorizedCallValidator::new("complete", &cx); let (virtual_host, _claims) = mcp_call_validator.validate_stateless()?; @@ -58,7 +57,7 @@ pub(super) async fn complete( } let response = response.map_err(|error| backend_forward_error("complete", &service_name, &error))?; - info!("read_resource: backend {service_name} returned {} contents", response.completion.values.len()); + info!("complete: backend {service_name} returned {} contents", response.completion.values.len()); - Ok(response.into()) + Ok(response) } From a34675f3b13dc97bfee5fc9243dd09633e90c0c9 Mon Sep 17 00:00:00 2001 From: Matheus Date: Thu, 3 Sep 2026 22:21:44 +0100 Subject: [PATCH 3/3] Rebased completion/complete Signed-off-by: Matheus --- .../src/gateway/mcp_service/completion.rs | 31 ++++++++++++------- 1 file changed, 19 insertions(+), 12 deletions(-) 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 79e98c71..d3becb46 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 @@ -6,8 +6,7 @@ use rmcp::{ use tracing::info; use crate::gateway::{ - mcp_call_validator::AuthorizedCallValidator, - mcp_service::initialization::connect_backend_for_request, + mcp_call_validator::AuthorizedCallValidator, mcp_service::initialization::connect_backend_for_request, routing_error::backend_forward_error, }; @@ -22,19 +21,28 @@ pub(super) async fn complete( let mcp_call_validator = AuthorizedCallValidator::new("complete", &cx); let (virtual_host, _claims) = mcp_call_validator.validate_stateless()?; - let route = match &request.r#ref { - Reference::Prompt(prompt) => virtual_host.prompts.get(&prompt.name), - Reference::Resource(resource) => virtual_host.resource_templates.get(&resource.uri), - }; - let Some(route) = route else { + let Some(downstream_name) = (match &request.r#ref { + Reference::Prompt(_) => request.r#ref.as_prompt_name(), + Reference::Resource(_) => request.r#ref.as_resource_uri(), + _ => None, + }) else { return Err(ErrorData { code: ErrorCode::INVALID_PARAMS, message: "Routing problem... completion not found".into(), data: None, }); }; + + let Some(route) = virtual_host.tools.get(downstream_name) else { + return Err(ErrorData { + code: ErrorCode::INVALID_PARAMS, + message: "Routing problem... tool not found".into(), + data: None, + }); + }; + let backend_name = route.backend_name.clone(); - let upstream_name = route.upstream_name.clone(); + let completion_name = route.upstream_name.clone(); let backend = virtual_host.backends.get(&backend_name).ok_or_else(|| ErrorData { code: ErrorCode::INVALID_PARAMS, @@ -46,15 +54,14 @@ pub(super) async fn complete( let mut backend_service = connect_backend_for_request(mcp_service, &backend_name, backend, &cx).await?; let mut routed_request = request; - match &mut routed_request.r#ref { - Reference::Prompt(prompt) => prompt.name = upstream_name, - Reference::Resource(resource) => resource.uri = upstream_name, - } + routed_request.argument.name = completion_name; let response = backend_service.complete(routed_request).await; + if let Err(error) = backend_service.close().await { tracing::warn!("complete: backend cleanup failed backend_name = {service_name} error = {error:?}"); } + let response = response.map_err(|error| backend_forward_error("complete", &service_name, &error))?; info!("complete: backend {service_name} returned {} contents", response.completion.values.len());