alkdev/alkhttp
6
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(adapters): MCP tool-gateway fidelity (PRJ-07..10, PRJ-13)

Browse Source 
- search honors the optional query substring filter (PRJ-07)
- search excludes both Sub andPub ops per ADR-041/ADR-068; dead match
  arms removed (PRJ-08)
- batch returns {"results": [...]} with {isError, output|error} items;
  tool description states the shape (PRJ-09)
- structuredContent is always an object: non-object outputs wrapped as
  {"result": <output>} (PRJ-10)
- argument errors are CallError values (retryable always present);
  non-string operation no longer reports 'missing required field'
  (PRJ-13)

Verified: cargo test --all-features, cargo test, clippy -D warnings,
fmt --check
This commit is contained in:
glm-5.3-flash
2026-08-29 12:53:03 +00:00
parent a9d17405a7
commit 6fa2d4c036
2 changed files with 329 additions and 64 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/adapters/to_mcp.rs
+286 -57
View File
@@ -4,9 +4,14 @@
//!
//! This is the tool-gateway pattern (ADR-041): the LLM gets a fixed set of
//! meta-tools (`search`, `schema`, `call`, `batch`) and discovers operations
//! on demand — not one MCP tool per registry operation. `Sub` ops are
//! excluded from `search` and cannot be invoked via `call` (MCP tool calls
//! are request/response — ADR-041 §2).
//! on demand — not one MCP tool per registry operation. `Sub` and `Pub`
//! ops are excluded from `search` and cannot be invoked via `call` (MCP
//! tool calls are request/response — ADR-041 §2, ADR-068).
//!
//! `structuredContent` is always a JSON object (strict MCP clients reject
//! non-object structured content): successful `call` results pass object
//! outputs through verbatim and wrap non-object outputs as
//! `{"result": <output>}`; `batch` returns `{"results": [...]}`.
//!
//! `to_mcp` is a pure projection (ADR-017 §5): it consumes the registry and
//! does not produce entries for it. It is not an `OperationAdapter`. The
@@ -139,12 +144,16 @@ impl ToMcpGateway {
.and_then(Option::clone)
}
async fn handle_search(&self, identity: Option<Identity>) -> CallToolResult {
async fn handle_search(
&self,
query: Option<String>,
identity: Option<Identity>,
) -> CallToolResult {
let response = self
.dispatch
.invoke(identity.clone(), OP_SERVICES_LIST, Value::Null)
.await;
map_search_response(response)
map_search_response(response, query.as_deref())
}
async fn handle_schema(
@@ -158,10 +167,9 @@ impl ToMcpGateway {
{
Some(n) => n,
None => {
return CallToolResult::structured_error(serde_json::json!({
"code": "INVALID_INPUT",
"message": "missing required field: name"
}));
return call_error_to_structured_error(CallError::invalid_input(
"invalid arguments: `name` is required and must be a string",
));
}
};
if let Some(error) =
@@ -187,7 +195,7 @@ impl ToMcpGateway {
) -> CallToolResult {
let (operation, input) = match parse_call_arguments(arguments) {
Ok(pair) => pair,
Err(err) => return err,
Err(err) => return call_error_to_structured_error(err),
};
let response = self.dispatch.invoke(identity, &operation, input).await;
envelope_to_call_tool_result(response)
@@ -195,6 +203,7 @@ impl ToMcpGateway {
async fn handle_batch(
&self,
search_filter: Option<String>,
arguments: Option<JsonObject>,
identity: Option<Identity>,
) -> CallToolResult {
@@ -204,10 +213,9 @@ impl ToMcpGateway {
{
Some(arr) => arr,
None => {
return CallToolResult::structured_error(serde_json::json!({
"code": "INVALID_INPUT",
"message": "missing required field: calls"
}));
return call_error_to_structured_error(CallError::invalid_input(
"invalid arguments: `calls` is required and must be an array",
));
}
};
@@ -216,7 +224,10 @@ impl ToMcpGateway {
let (operation, input) = match parse_call_arguments(call.as_object().cloned()) {
Ok(pair) => pair,
Err(err) => {
results.push(batch_error_value(err));
results.push(serde_json::json!({
"isError": true,
"error": serde_json::to_value(&err).unwrap_or(Value::Null),
}));
continue;
}
};
@@ -226,42 +237,47 @@ impl ToMcpGateway {
.await;
results.push(envelope_to_value(response));
}
CallToolResult::structured(Value::Array(results))
let _ = search_filter;
CallToolResult::structured(serde_json::json!({ "results": results }))
}
}
fn parse_call_arguments(arguments: Option<JsonObject>) -> Result<(String, Value), CallToolResult> {
fn parse_call_arguments(arguments: Option<JsonObject>) -> Result<(String, Value), CallError> {
let mut map = match arguments {
Some(m) => m,
None => {
return Err(CallToolResult::structured_error(serde_json::json!({
"code": "INVALID_INPUT",
"message": "missing required field: operation"
})));
return Err(CallError::invalid_input(
"invalid arguments: `operation` is required and must be a string",
));
}
};
let operation = match map
.remove("operation")
.and_then(|v| v.as_str().map(str::to_string))
{
Some(s) => s,
let operation = match map.remove("operation") {
Some(Value::String(s)) => s,
Some(other) => {
return Err(CallError::invalid_input(format!(
"invalid arguments: `operation` must be a string, got {}",
json_type_name(&other)
)));
}
None => {
return Err(CallToolResult::structured_error(serde_json::json!({
"code": "INVALID_INPUT",
"message": "missing required field: operation"
})));
return Err(CallError::invalid_input(
"invalid arguments: missing required field `operation`",
));
}
};
let input = map.remove("input").unwrap_or(Value::Object(Map::new()));
Ok((operation, input))
}
fn batch_error_value(result: CallToolResult) -> Value {
serde_json::json!({
"isError": result.is_error.unwrap_or(false),
"structuredContent": result.structured_content,
"content": result.content,
})
fn json_type_name(value: &Value) -> &'static str {
match value {
Value::Null => "null",
Value::Bool(_) => "a boolean",
Value::Number(_) => "a number",
Value::String(_) => "a string",
Value::Array(_) => "an array",
Value::Object(_) => "an object",
}
}
/// Symmetric with HTTP `GET /schema` (PRJ-06): internal ops are
@@ -286,7 +302,7 @@ fn schema_visibility_and_access_denial(
None
}
fn map_search_response(response: ResponseEnvelope) -> CallToolResult {
fn map_search_response(response: ResponseEnvelope, query: Option<&str>) -> CallToolResult {
match response.result {
Ok(value) => {
let operations = value
@@ -298,7 +314,14 @@ fn map_search_response(response: ResponseEnvelope) -> CallToolResult {
.into_iter()
.filter(|op| {
let op_type = op.get("op_type").and_then(Value::as_str).unwrap_or("");
!matches!(op_type, "sub" | "subscription" | "Sub")
!matches!(op_type, "sub" | "pub")
})
.filter(|op| match query {
Some(q) => op
.get("name")
.and_then(Value::as_str)
.is_some_and(|name| name.contains(q)),
None => true,
})
.map(|op| op_to_search_listing(&op))
.collect();
@@ -321,7 +344,7 @@ fn op_to_search_listing(op: &Value) -> Value {
fn envelope_to_call_tool_result(response: ResponseEnvelope) -> CallToolResult {
match response.result {
Ok(value) => CallToolResult::structured(value),
Ok(output) => CallToolResult::structured(object_result(output)),
Err(err) => call_error_to_structured_error(err),
}
}
@@ -335,7 +358,7 @@ fn envelope_to_value(response: ResponseEnvelope) -> Value {
match response.result {
Ok(output) => serde_json::json!({
"isError": false,
"output": output,
"output": object_result(output),
}),
Err(err) => {
let details = serde_json::to_value(&err).unwrap_or(Value::Null);
@@ -347,12 +370,23 @@ fn envelope_to_value(response: ResponseEnvelope) -> Value {
}
}
/// `structuredContent` must be a JSON object (strict MCP clients reject
/// non-object values). Object outputs pass through verbatim; any other
/// output is wrapped as `{"result": <output>}`.
fn object_result(output: Value) -> Value {
if output.is_object() {
output
} else {
serde_json::json!({ "result": output })
}
}
pub(crate) fn gateway_tools() -> Vec<Tool> {
vec![
Tool::new(
Cow::Borrowed(TOOL_SEARCH),
Cow::Borrowed(
"List available operations (filtered by the caller's AccessControl). Returns names + descriptions, not full schemas. Subscription operations are excluded.",
"List available operations (filtered by the caller's AccessControl). Returns names + descriptions, not full schemas. Subscription and publish operations are excluded.",
),
value_to_object(search_input_schema()),
),
@@ -366,20 +400,27 @@ pub(crate) fn gateway_tools() -> Vec<Tool> {
Tool::new(
Cow::Borrowed(TOOL_CALL),
Cow::Borrowed(
"Invoke an operation by name with a JSON input. Returns the output as structuredContent, or isError with typed error details for a CallError.",
"Invoke an operation by name with a JSON input. Returns the output as structuredContent (object outputs verbatim; non-object outputs wrapped as {\"result\": <output>}), or isError with typed error details for a CallError.",
),
value_to_object(call_input_schema()),
),
Tool::new(
Cow::Borrowed(TOOL_BATCH),
Cow::Borrowed(
"Invoke multiple operations in one tool call. Returns an array of results, each shaped like a `call` result.",
"Invoke multiple operations in one tool call. Returns {\"results\": [...]} where each entry is {\"isError\": false, \"output\": ...} on success or {\"isError\": true, \"error\": {code, message, retryable}} on failure.",
),
value_to_object(batch_input_schema()),
),
]
}
fn parse_search_query(arguments: Option<&JsonObject>) -> Option<String> {
arguments
.and_then(|a| a.get("query"))
.and_then(Value::as_str)
.map(str::to_string)
}
fn value_to_object(value: Value) -> Arc<JsonObject> {
match value {
Value::Object(map) => Arc::new(map),
@@ -405,13 +446,14 @@ impl rmcp::handler::server::ServerHandler for ToMcpGateway {
let identity = Self::extract_identity(&context);
let name = request.name.to_string();
let arguments = request.arguments;
let search_filter = parse_search_query(arguments.as_ref());
let this = self;
async move {
let result = match name.as_str() {
TOOL_SEARCH => this.handle_search(identity).await,
TOOL_SEARCH => this.handle_search(search_filter, identity).await,
TOOL_SCHEMA => this.handle_schema(arguments, identity).await,
TOOL_CALL => this.handle_call(arguments, identity).await,
TOOL_BATCH => this.handle_batch(arguments, identity).await,
TOOL_BATCH => this.handle_batch(search_filter, arguments, identity).await,
unknown => {
let err = CallError::new(
"NOT_FOUND",
@@ -454,15 +496,16 @@ mod tests {
use super::*;
use alkcall::core::auth::{AuthToken, IdentityProvider};
use alkcall::core::types::Capabilities;
use alkcall::registry::context::ScopedPeerEnv;
use alkcall::registry::context::{OperationContext, ScopedPeerEnv};
use alkcall::registry::discovery::{
services_list_handler, services_list_spec, services_schema_handler, services_schema_spec,
};
use alkcall::registry::registration::{
make_handler, make_streaming_handler, HandlerKind, HandlerRegistration,
make_handler, make_sink_handler, make_streaming_handler, HandlerKind, HandlerRegistration,
OperationProvenance, OperationRegistry,
};
use alkcall::registry::spec::{AccessControl, OperationSpec, OperationType, Visibility};
use futures::StreamExt;
use rmcp::model::Extensions;
use std::collections::HashMap;
use std::sync::Mutex as StdMutex;
@@ -553,12 +596,23 @@ mod tests {
OperationType::Query | OperationType::Mutation => {
HandlerKind::Once(make_echo_handler())
}
OperationType::Pub => {
unreachable!("to_mcp tests never register Pub ops")
}
OperationType::Pub => HandlerKind::Sink(make_echo_sink_handler()),
}
}
fn make_echo_sink_handler() -> alkcall::registry::registration::SinkHandler {
make_sink_handler(|_input, context: OperationContext, mut stream| async move {
let mut last = Value::Null;
while let Some(item) = stream.next().await {
match item {
Ok(chunk) => last = chunk,
Err(err) => return ResponseEnvelope::error(context.request_id, err),
}
}
ResponseEnvelope::ok(context.request_id, last)
})
}
fn full_registry_with_ops(
specs: Vec<(String, OperationType, AccessControl)>,
) -> Arc<OperationRegistry> {
@@ -643,11 +697,16 @@ mod tests {
arguments: Option<JsonObject>,
identity: Option<Identity>,
) -> CallToolResult {
let search_filter = parse_search_query(arguments.as_ref());
match name {
TOOL_SEARCH => gateway.handle_search(identity).await,
TOOL_SEARCH => gateway.handle_search(search_filter, identity).await,
TOOL_SCHEMA => gateway.handle_schema(arguments, identity).await,
TOOL_CALL => gateway.handle_call(arguments, identity).await,
TOOL_BATCH => gateway.handle_batch(arguments, identity).await,
TOOL_BATCH => {
gateway
.handle_batch(search_filter, arguments, identity)
.await
}
unknown => {
let err = CallError::new(
"NOT_FOUND",
@@ -752,6 +811,79 @@ mod tests {
}
}
#[tokio::test]
async fn search_excludes_pub_ops() {
let registry = full_registry_with_ops(vec![
(
"public/echo".to_string(),
OperationType::Query,
AccessControl::default(),
),
(
"metrics/ingest".to_string(),
OperationType::Pub,
AccessControl::default(),
),
]);
let gateway = ToMcpGateway::new(dispatch(registry, provider()));
let result = invoke_tool(&gateway, "search", None, None).await;
assert_eq!(result.is_error, Some(false));
let structured = result.structured_content.expect("structured present");
let names: Vec<String> = structured
.get("operations")
.and_then(Value::as_array)
.expect("operations array")
.iter()
.filter_map(|o| o.get("name").and_then(Value::as_str).map(str::to_string))
.collect();
assert!(
!names.contains(&"metrics/ingest".to_string()),
"publish op must be excluded from search: {names:?}"
);
assert_eq!(names, vec!["public/echo".to_string()]);
}
#[tokio::test]
async fn search_honors_query_substring_filter() {
let registry = full_registry_with_ops(vec![
(
"fs/readFile".to_string(),
OperationType::Query,
AccessControl::default(),
),
(
"fs/writeFile".to_string(),
OperationType::Mutation,
AccessControl::default(),
),
(
"mail/send".to_string(),
OperationType::Query,
AccessControl::default(),
),
]);
let gateway = ToMcpGateway::new(dispatch(registry, provider()));
let mut args = Map::new();
args.insert("query".to_string(), Value::String("fs".to_string()));
let result = invoke_tool(&gateway, "search", Some(args), None).await;
assert_eq!(result.is_error, Some(false));
let structured = result.structured_content.expect("structured present");
let names: Vec<String> = structured
.get("operations")
.and_then(Value::as_array)
.expect("operations array")
.iter()
.filter_map(|o| o.get("name").and_then(Value::as_str).map(str::to_string))
.collect();
assert_eq!(
names,
vec!["fs/readFile".to_string(), "fs/writeFile".to_string()],
"query filter must keep only matching operations"
);
}
#[tokio::test]
async fn schema_returns_full_operation_spec() {
let registry = full_registry_with_ops(vec![(
@@ -941,6 +1073,75 @@ mod tests {
);
}
#[tokio::test]
async fn call_wraps_non_object_output_into_object_structured_content() {
for output in [
Value::String("just a string".to_string()),
serde_json::json!([1, 2, 3]),
Value::Null,
Value::Bool(true),
] {
let registry = full_registry_with_ops(vec![(
"echo/run".to_string(),
OperationType::Query,
AccessControl::default(),
)]);
let gateway = ToMcpGateway::new(dispatch(registry, provider()));
let mut args = Map::new();
args.insert(
"operation".to_string(),
Value::String("echo/run".to_string()),
);
args.insert("input".to_string(), output.clone());
let result = invoke_tool(&gateway, "call", Some(args), None).await;
assert_eq!(result.is_error, Some(false));
assert_eq!(
result.structured_content,
Some(serde_json::json!({ "result": output })),
"non-object output must be wrapped into an object"
);
}
}
#[tokio::test]
async fn call_argument_errors_carry_retryable_and_truthy_messages() {
let registry = full_registry_with_ops(vec![]);
let gateway = ToMcpGateway::new(dispatch(registry, provider()));
let missing = invoke_tool(&gateway, "call", None, None).await;
assert_eq!(missing.is_error, Some(true));
let structured = missing
.structured_content
.expect("structured error present");
assert_eq!(
structured.get("code"),
Some(&Value::String("INVALID_INPUT".to_string()))
);
assert_eq!(structured.get("retryable"), Some(&Value::Bool(false)));
let mut args = Map::new();
args.insert("operation".to_string(), Value::Number(42.into()));
let non_string = invoke_tool(&gateway, "call", Some(args), None).await;
assert_eq!(non_string.is_error, Some(true));
let structured = non_string
.structured_content
.expect("structured error present");
assert_eq!(
structured.get("code"),
Some(&Value::String("INVALID_INPUT".to_string()))
);
assert_eq!(structured.get("retryable"), Some(&Value::Bool(false)));
let message = structured
.get("message")
.and_then(Value::as_str)
.expect("message");
assert!(
message.contains("must be a string"),
"non-string operation must not report a missing field: {message}"
);
}
#[tokio::test]
async fn call_returns_structured_error_for_call_error() {
let registry = full_registry_with_ops(vec![]);
@@ -962,7 +1163,7 @@ mod tests {
}
#[tokio::test]
async fn batch_returns_array_of_results() {
async fn batch_returns_object_with_result_entries() {
let registry = full_registry_with_ops(vec![(
"echo/run".to_string(),
OperationType::Query,
@@ -976,15 +1177,43 @@ mod tests {
serde_json::json!([
{ "operation": "echo/run", "input": { "n": 1 } },
{ "operation": "no/such", "input": {} },
{ "operation": 7, "input": {} },
]),
);
let result = invoke_tool(&gateway, "batch", Some(args), None).await;
assert_eq!(result.is_error, Some(false));
let structured = result.structured_content.expect("structured present");
let arr = structured.as_array().expect("batch returns array");
assert_eq!(arr.len(), 2);
assert_eq!(arr[0].get("isError"), Some(&Value::Bool(false)));
assert_eq!(arr[1].get("isError"), Some(&Value::Bool(true)));
assert!(
structured.is_object(),
"batch structuredContent must be an object, not an array"
);
let results = structured
.get("results")
.and_then(Value::as_array)
.expect("results array");
assert_eq!(results.len(), 3);
assert_eq!(results[0].get("isError"), Some(&Value::Bool(false)));
assert_eq!(
results[0].get("output"),
Some(&serde_json::json!({ "n": 1 }))
);
assert!(results[0].get("error").is_none());
assert_eq!(results[1].get("isError"), Some(&Value::Bool(true)));
let not_found = results[1].get("error").expect("error object");
assert_eq!(
not_found.get("code"),
Some(&Value::String("NOT_FOUND".to_string()))
);
assert_eq!(
not_found.get("retryable"),
Some(&Value::Bool(false)),
"batch error items carry the full CallError shape"
);
assert_eq!(results[2].get("isError"), Some(&Value::Bool(true)));
assert_eq!(
results[2].get("error").and_then(|e| e.get("code")),
Some(&Value::String("INVALID_INPUT".to_string()))
);
}
#[tokio::test]