use std::borrow::Cow;
use std::sync::Arc;
use alkcall::core::auth::Identity;
use alkcall::protocol::wire::{CallError, ResponseEnvelope};
use alkcall::registry::registration::OperationRegistry;
use rmcp::model::{
CallToolRequestParams, CallToolResult, Implementation, JsonObject, ListToolsResult,
PaginatedRequestParams, ServerCapabilities, ServerInfo, Tool,
};
use rmcp::service::{RequestContext, RoleServer};
use rmcp::transport::{
streamable_http_server::{session::local::LocalSessionManager, tower::StreamableHttpService},
StreamableHttpServerConfig,
};
use serde_json::{Map, Value};
use crate::gateway::MAX_BATCH_OPERATIONS;
use alkcall::gateway::GatewayDispatch;
const TOOL_SEARCH: &str = "search";
const TOOL_SCHEMA: &str = "schema";
const TOOL_CALL: &str = "call";
const TOOL_BATCH: &str = "batch";
const OP_SERVICES_LIST: &str = "services/list";
const OP_SERVICES_SCHEMA: &str = "services/schema";
fn search_input_schema() -> Value {
serde_json::json!({
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "Optional substring filter on operation name."
}
}
})
}
fn schema_input_schema() -> Value {
serde_json::json!({
"type": "object",
"properties": {
"name": {
"type": "string",
"description": "The fully-qualified operation name (e.g. `fs/readFile`)."
}
},
"required": ["name"]
})
}
fn call_input_schema() -> Value {
serde_json::json!({
"type": "object",
"properties": {
"operation": {
"type": "string",
"description": "The fully-qualified operation name to invoke."
},
"input": {
"type": "object",
"description": "The JSON input object to pass to the operation."
}
},
"required": ["operation"]
})
}
fn batch_input_schema() -> Value {
serde_json::json!({
"type": "object",
"properties": {
"calls": {
"type": "array",
"items": {
"type": "object",
"properties": {
"operation": { "type": "string" },
"input": { "type": "object" }
},
"required": ["operation"]
},
"maxItems": MAX_BATCH_OPERATIONS,
"description": "The operations to invoke in this batch. At most 100 operations per batch.",
}
},
"required": ["calls"]
})
}
pub struct ToMcpGateway {
dispatch: Arc<GatewayDispatch>,
}
impl ToMcpGateway {
pub fn new(dispatch: Arc<GatewayDispatch>) -> Self {
Self { dispatch }
}
pub fn dispatch(&self) -> &Arc<GatewayDispatch> {
&self.dispatch
}
fn extract_identity(context: &RequestContext<RoleServer>) -> Option<Identity> {
Self::extract_identity_from_extensions(&context.extensions)
}
fn extract_identity_from_extensions(extensions: &rmcp::model::Extensions) -> Option<Identity> {
let parts = extensions.get::<http::request::Parts>()?;
parts
.extensions
.get::<Option<Identity>>()
.and_then(Option::clone)
}
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, query.as_deref())
}
async fn handle_schema(
&self,
arguments: Option<JsonObject>,
identity: Option<Identity>,
) -> CallToolResult {
let name = match arguments
.and_then(|mut a| a.remove("name"))
.and_then(|v| v.as_str().map(str::to_string))
{
Some(n) => n,
None => {
return call_error_to_structured_error(CallError::invalid_input(
"invalid arguments: `name` is required and must be a string",
));
}
};
if let Some(error) =
schema_visibility_and_access_denial(self.dispatch.registry(), &name, identity.as_ref())
{
return call_error_to_structured_error(error);
}
let response = self
.dispatch
.invoke(
identity,
OP_SERVICES_SCHEMA,
serde_json::json!({ "name": name }),
)
.await;
envelope_to_call_tool_result(response)
}
async fn handle_call(
&self,
arguments: Option<JsonObject>,
identity: Option<Identity>,
) -> CallToolResult {
let (operation, input) = match parse_call_arguments(arguments) {
Ok(pair) => pair,
Err(err) => return call_error_to_structured_error(err),
};
let response = self.dispatch.invoke(identity, &operation, input).await;
envelope_to_call_tool_result(response)
}
async fn handle_batch(
&self,
arguments: Option<JsonObject>,
identity: Option<Identity>,
) -> CallToolResult {
let calls = match arguments
.and_then(|mut a| a.remove("calls"))
.and_then(|v| v.as_array().cloned())
{
Some(arr) => arr,
None => {
return call_error_to_structured_error(CallError::invalid_input(
"invalid arguments: `calls` is required and must be an array",
));
}
};
if calls.len() > MAX_BATCH_OPERATIONS {
return call_error_to_structured_error(CallError::invalid_input(format!(
"batch exceeds the maximum of {MAX_BATCH_OPERATIONS} operations"
)));
}
let mut results: Vec<Value> = Vec::with_capacity(calls.len());
for call in calls {
let (operation, input) = match parse_call_arguments(call.as_object().cloned()) {
Ok(pair) => pair,
Err(err) => {
results.push(serde_json::json!({
"isError": true,
"error": serde_json::to_value(&err).unwrap_or(Value::Null),
}));
continue;
}
};
let response = self
.dispatch
.invoke(identity.clone(), &operation, input)
.await;
results.push(envelope_to_value(response));
}
CallToolResult::structured(serde_json::json!({ "results": results }))
}
}
fn parse_call_arguments(arguments: Option<JsonObject>) -> Result<(String, Value), CallError> {
let mut map = match arguments {
Some(m) => m,
None => {
return Err(CallError::invalid_input(
"invalid arguments: `operation` is required and must be a string",
));
}
};
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(CallError::invalid_input(
"invalid arguments: missing required field `operation`",
));
}
};
let input = map.remove("input").unwrap_or(Value::Object(Map::new()));
Ok((operation, input))
}
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",
}
}
fn schema_visibility_and_access_denial(
registry: &OperationRegistry,
operation: &str,
identity: Option<&Identity>,
) -> Option<CallError> {
alkcall::gateway::schema_disclosure_denial(registry, operation, identity)
}
fn map_search_response(response: ResponseEnvelope, query: Option<&str>) -> CallToolResult {
match response.result {
Ok(value) => {
let operations = value
.get("operations")
.and_then(Value::as_array)
.cloned()
.unwrap_or_default();
let filtered: Vec<Value> = operations
.into_iter()
.filter(|op| {
let op_type = op.get("op_type").and_then(Value::as_str).unwrap_or("");
!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();
CallToolResult::structured(serde_json::json!({ "operations": filtered }))
}
Err(err) => call_error_to_structured_error(err),
}
}
fn op_to_search_listing(op: &Value) -> Value {
let name = op.get("name").and_then(Value::as_str).unwrap_or("");
let op_type = op.get("op_type").and_then(Value::as_str).unwrap_or("query");
let namespace = op.get("namespace").and_then(Value::as_str).unwrap_or("");
let description = format!("{op_type} operation `{name}` in namespace `{namespace}`");
serde_json::json!({
"name": name,
"description": description,
})
}
fn envelope_to_call_tool_result(response: ResponseEnvelope) -> CallToolResult {
match response.result {
Ok(output) => CallToolResult::structured(object_result(output)),
Err(err) => call_error_to_structured_error(err),
}
}
fn call_error_to_structured_error(err: CallError) -> CallToolResult {
let details = serde_json::to_value(&err).unwrap_or(Value::Null);
CallToolResult::structured_error(details)
}
fn envelope_to_value(response: ResponseEnvelope) -> Value {
match response.result {
Ok(output) => serde_json::json!({
"isError": false,
"output": object_result(output),
}),
Err(err) => {
let details = serde_json::to_value(&err).unwrap_or(Value::Null);
serde_json::json!({
"isError": true,
"error": details,
})
}
}
}
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 and publish operations are excluded.",
),
value_to_object(search_input_schema()),
),
Tool::new(
Cow::Borrowed(TOOL_SCHEMA),
Cow::Borrowed(
"Get the full OperationSpec for an operation (input/output JSON Schemas, error schemas).",
),
value_to_object(schema_input_schema()),
),
Tool::new(
Cow::Borrowed(TOOL_CALL),
Cow::Borrowed(
"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, executed serially in order (each bounded by the 30 s deadline). Returns {\"results\": [...]} where each entry is {\"isError\": false, \"output\": ...} on success or {\"isError\": true, \"error\": {code, message, retryable}} on failure. At most 100 operations per batch; larger batches are rejected with INVALID_INPUT without dispatching.",
),
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),
_ => Arc::new(Map::new()),
}
}
impl rmcp::handler::server::ServerHandler for ToMcpGateway {
fn list_tools(
&self,
_request: Option<PaginatedRequestParams>,
_context: RequestContext<RoleServer>,
) -> impl futures::Future<Output = Result<ListToolsResult, rmcp::ErrorData>> + Send + '_ {
let tools = gateway_tools();
std::future::ready(Ok(ListToolsResult::with_all_items(tools)))
}
fn call_tool(
&self,
request: CallToolRequestParams,
context: RequestContext<RoleServer>,
) -> impl futures::Future<Output = Result<CallToolResult, rmcp::ErrorData>> + Send + '_ {
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(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,
unknown => {
let err = CallError::new(
"NOT_FOUND",
format!("unknown gateway tool: {unknown}"),
false,
);
call_error_to_structured_error(err)
}
};
Ok(result)
}
}
fn get_info(&self) -> ServerInfo {
let capabilities = ServerCapabilities::builder().enable_tools().build();
ServerInfo::new(capabilities)
.with_server_info(Implementation::new(
"alkhttp-to-mcp",
env!("CARGO_PKG_VERSION"),
))
.with_instructions(
"alk MCP gateway. Call `search` to discover operations, `schema` for an operation's full spec, `call` to invoke, `batch` to invoke many.",
)
}
}
pub type ToMcpService = StreamableHttpService<ToMcpGateway, LocalSessionManager>;
pub fn to_mcp_service(dispatch: Arc<GatewayDispatch>) -> ToMcpService {
let gateway = ToMcpGateway::new(dispatch);
StreamableHttpService::new(
move || Ok(ToMcpGateway::new(Arc::clone(gateway.dispatch()))),
LocalSessionManager::default().into(),
StreamableHttpServerConfig::default(),
)
}
#[cfg(test)]
mod tests {
use super::*;
use alkcall::core::types::Capabilities;
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_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;
fn identity_with_scopes(id: &str, scopes: &[&str]) -> Identity {
Identity {
id: id.to_string(),
scopes: scopes.iter().map(|s| s.to_string()).collect(),
resources: HashMap::new(),
}
}
fn external_spec(name: &str, op_type: OperationType, acl: AccessControl) -> OperationSpec {
OperationSpec::new(
name,
op_type,
Visibility::External,
serde_json::json!({}),
serde_json::json!({}),
vec![],
acl,
None,
)
}
fn internal_spec(name: &str) -> OperationSpec {
OperationSpec::new(
name,
OperationType::Query,
Visibility::Internal,
serde_json::json!({}),
serde_json::json!({}),
vec![],
AccessControl::default(),
None,
)
}
fn make_echo_handler() -> alkcall::registry::registration::Handler {
make_handler(
|input, context| async move { ResponseEnvelope::ok(context.request_id, input) },
)
}
fn make_echo_streaming_handler() -> alkcall::registry::registration::StreamingHandler {
make_streaming_handler(|input, context| {
futures::stream::iter(vec![ResponseEnvelope::ok(context.request_id, input)])
})
}
fn handler_kind_for(op_type: OperationType) -> HandlerKind {
match op_type {
OperationType::Sub => HandlerKind::Stream(make_echo_streaming_handler()),
OperationType::Query | OperationType::Mutation => {
HandlerKind::Once(make_echo_handler())
}
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> {
let inner = OperationRegistry::new();
for (name, op_type, acl) in specs {
inner
.register(HandlerRegistration::new(
external_spec(&name, op_type, acl),
handler_kind_for(op_type),
OperationProvenance::Local,
None,
None,
Capabilities::new(),
))
.unwrap();
}
let inner = Arc::new(inner);
let dispatch_registry = OperationRegistry::new();
for op in inner.list_operations() {
dispatch_registry
.register(HandlerRegistration::new(
external_spec(&op.name, op.op_type, op.access_control.clone()),
handler_kind_for(op.op_type),
OperationProvenance::Local,
None,
None,
Capabilities::new(),
))
.unwrap();
}
dispatch_registry
.register(HandlerRegistration::new(
services_list_spec(),
HandlerKind::Once(services_list_handler(Arc::clone(&inner))),
OperationProvenance::Local,
None,
Some(ScopedPeerEnv::empty()),
Capabilities::new(),
))
.unwrap();
dispatch_registry
.register(HandlerRegistration::new(
services_schema_spec(),
HandlerKind::Once(services_schema_handler(Arc::clone(&inner))),
OperationProvenance::Local,
None,
Some(ScopedPeerEnv::empty()),
Capabilities::new(),
))
.unwrap();
Arc::new(dispatch_registry)
}
fn dispatch(registry: Arc<OperationRegistry>) -> Arc<GatewayDispatch> {
Arc::new(GatewayDispatch::new(registry))
}
fn extensions_with_identity(identity: Option<Identity>) -> Extensions {
let request = http::Request::builder()
.method(http::Method::POST)
.uri("/mcp")
.body(())
.expect("valid request");
let (mut parts, _) = request.into_parts();
parts.extensions.insert(identity);
let mut extensions = Extensions::new();
extensions.insert(parts);
extensions
}
async fn invoke_tool(
gateway: &ToMcpGateway,
name: &str,
arguments: Option<JsonObject>,
identity: Option<Identity>,
) -> CallToolResult {
let search_filter = parse_search_query(arguments.as_ref());
match name {
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,
unknown => {
let err = CallError::new(
"NOT_FOUND",
format!("unknown gateway tool: {unknown}"),
false,
);
call_error_to_structured_error(err)
}
}
}
#[tokio::test]
async fn list_tools_returns_exactly_four_gateway_tools() {
let _gateway = ToMcpGateway::new(dispatch(full_registry_with_ops(vec![])));
let tools = gateway_tools();
let names: Vec<String> = tools.iter().map(|t| t.name.to_string()).collect();
assert_eq!(names.len(), 4);
assert!(names.contains(&"search".to_string()));
assert!(names.contains(&"schema".to_string()));
assert!(names.contains(&"call".to_string()));
assert!(names.contains(&"batch".to_string()));
}
#[tokio::test]
async fn list_tools_does_not_leak_registry_operations() {
let registry = full_registry_with_ops(vec![(
"fs/readFile".to_string(),
OperationType::Query,
AccessControl::default(),
)]);
let _gateway = ToMcpGateway::new(dispatch(registry));
let tools = gateway_tools();
for tool in &tools {
assert_ne!(tool.name, "fs/readFile");
assert_ne!(tool.name, "services/list");
assert_ne!(tool.name, "services/schema");
}
assert_eq!(tools.len(), 4);
}
#[tokio::test]
async fn search_returns_access_control_filtered_ops_excluding_subscriptions() {
let registry = full_registry_with_ops(vec![
(
"public/echo".to_string(),
OperationType::Query,
AccessControl::default(),
),
(
"admin/secret".to_string(),
OperationType::Query,
AccessControl {
required_scopes: vec!["admin".to_string()],
..Default::default()
},
),
(
"events/stream".to_string(),
OperationType::Sub,
AccessControl::default(),
),
]);
let gateway = ToMcpGateway::new(dispatch(registry));
let result = invoke_tool(
&gateway,
"search",
None,
Some(identity_with_scopes("user", &["user"])),
)
.await;
assert_eq!(result.is_error, Some(false));
let structured = result.structured_content.expect("structured present");
let ops = structured
.get("operations")
.and_then(Value::as_array)
.expect("operations array");
let names: Vec<&str> = ops
.iter()
.filter_map(|o| o.get("name").and_then(Value::as_str))
.collect();
assert!(names.contains(&"public/echo"));
assert!(
!names.contains(&"admin/secret"),
"ACL-filtered op must not appear"
);
assert!(
!names.contains(&"events/stream"),
"Subscription op must be excluded"
);
for op in ops {
assert!(
op.get("description").is_some(),
"each entry has a description"
);
assert!(
op.get("input_schema").is_none(),
"search must not return full schemas"
);
}
}
#[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));
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));
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 mut 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();
names.sort();
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![(
"fs/readFile".to_string(),
OperationType::Query,
AccessControl::default(),
)]);
let gateway = ToMcpGateway::new(dispatch(registry));
let mut args = Map::new();
args.insert("name".to_string(), Value::String("fs/readFile".to_string()));
let result = invoke_tool(&gateway, "schema", Some(args), None).await;
assert_eq!(result.is_error, Some(false));
let structured = result.structured_content.expect("structured present");
assert_eq!(
structured.get("name"),
Some(&Value::String("fs/readFile".to_string()))
);
assert!(structured.get("input_schema").is_some());
assert!(structured.get("output_schema").is_some());
assert!(structured.get("error_schemas").is_some());
}
#[tokio::test]
async fn schema_returns_full_spec_for_authorized_identity() {
let registry = full_registry_with_ops(vec![(
"fs/readFile".to_string(),
OperationType::Query,
AccessControl {
required_scopes: vec!["fs:read".to_string()],
..Default::default()
},
)]);
let gateway = ToMcpGateway::new(dispatch(registry));
let mut args = Map::new();
args.insert("name".to_string(), Value::String("fs/readFile".to_string()));
let result = invoke_tool(
&gateway,
"schema",
Some(args),
Some(identity_with_scopes("reader", &["fs:read"])),
)
.await;
assert_eq!(result.is_error, Some(false));
let structured = result.structured_content.expect("structured present");
assert_eq!(
structured.get("name"),
Some(&Value::String("fs/readFile".to_string()))
);
assert!(structured.get("access_control").is_some());
}
#[tokio::test]
async fn schema_denies_acl_forbidden_op_symmetrically_with_http() {
let registry = full_registry_with_ops(vec![(
"admin/secret".to_string(),
OperationType::Query,
AccessControl {
required_scopes: vec!["admin".to_string()],
..Default::default()
},
)]);
let gateway = ToMcpGateway::new(dispatch(registry));
let mut args = Map::new();
args.insert(
"name".to_string(),
Value::String("admin/secret".to_string()),
);
let result = invoke_tool(
&gateway,
"schema",
Some(args),
Some(identity_with_scopes("user", &["user"])),
)
.await;
assert_eq!(result.is_error, Some(true));
let structured = result.structured_content.expect("structured error present");
assert_eq!(
structured.get("code"),
Some(&Value::String("FORBIDDEN".to_string()))
);
assert!(
structured.get("input_schema").is_none(),
"a denied schema lookup must not leak the op's schemas"
);
}
#[tokio::test]
async fn schema_denies_internal_op_with_not_found_symmetrically_with_http() {
let inner = OperationRegistry::new();
inner
.register(HandlerRegistration::new(
internal_spec("secret/op"),
HandlerKind::Once(make_echo_handler()),
OperationProvenance::Local,
None,
None,
Capabilities::new(),
))
.unwrap();
let inner = Arc::new(inner);
let dispatch_registry = OperationRegistry::new();
dispatch_registry
.register(HandlerRegistration::new(
internal_spec("secret/op"),
HandlerKind::Once(make_echo_handler()),
OperationProvenance::Local,
None,
None,
Capabilities::new(),
))
.unwrap();
dispatch_registry
.register(HandlerRegistration::new(
services_list_spec(),
HandlerKind::Once(services_list_handler(Arc::clone(&inner))),
OperationProvenance::Local,
None,
Some(ScopedPeerEnv::empty()),
Capabilities::new(),
))
.unwrap();
dispatch_registry
.register(HandlerRegistration::new(
services_schema_spec(),
HandlerKind::Once(services_schema_handler(Arc::clone(&inner))),
OperationProvenance::Local,
None,
Some(ScopedPeerEnv::empty()),
Capabilities::new(),
))
.unwrap();
let gateway = ToMcpGateway::new(dispatch(Arc::new(dispatch_registry)));
let mut args = Map::new();
args.insert("name".to_string(), Value::String("secret/op".to_string()));
let unauthenticated = invoke_tool(&gateway, "schema", Some(args.clone()), None).await;
assert_eq!(unauthenticated.is_error, Some(true));
let structured = unauthenticated
.structured_content
.expect("structured error present");
assert_eq!(
structured.get("code"),
Some(&Value::String("NOT_FOUND".to_string()))
);
assert!(structured.get("input_schema").is_none());
let unauthorized = invoke_tool(
&gateway,
"schema",
Some(args),
Some(identity_with_scopes("user", &["user"])),
)
.await;
assert_eq!(unauthorized.is_error, Some(true));
let structured = unauthorized
.structured_content
.expect("structured error present");
assert_eq!(
structured.get("code"),
Some(&Value::String("NOT_FOUND".to_string()))
);
}
#[tokio::test]
async fn call_returns_structured_for_success() {
let registry = full_registry_with_ops(vec![(
"echo/run".to_string(),
OperationType::Query,
AccessControl::default(),
)]);
let gateway = ToMcpGateway::new(dispatch(registry));
let mut args = Map::new();
args.insert(
"operation".to_string(),
Value::String("echo/run".to_string()),
);
args.insert("input".to_string(), serde_json::json!({ "msg": "hi" }));
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!({ "msg": "hi" }))
);
}
#[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));
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));
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 schema_tool_without_arguments_is_invalid_input() {
let registry = full_registry_with_ops(vec![(
"fs/readFile".to_string(),
OperationType::Query,
AccessControl::default(),
)]);
let gateway = ToMcpGateway::new(dispatch(registry));
let result = invoke_tool(&gateway, "schema", None, None).await;
assert_eq!(result.is_error, Some(true));
let structured = result.structured_content.expect("structured error present");
assert_eq!(
structured.get("code"),
Some(&Value::String("INVALID_INPUT".to_string()))
);
let message = structured
.get("message")
.and_then(Value::as_str)
.expect("message");
assert!(
message.contains("`name` is required"),
"missing tool arguments must name the required field: {message}"
);
}
#[tokio::test]
async fn batch_tool_without_calls_argument_is_invalid_input_without_dispatching() {
let registry = full_registry_with_ops(vec![(
"echo/run".to_string(),
OperationType::Query,
AccessControl::default(),
)]);
let gateway = ToMcpGateway::new(dispatch(registry));
let none_args = invoke_tool(&gateway, "batch", None, None).await;
assert_eq!(none_args.is_error, Some(true));
let structured = none_args
.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("`calls` is required"),
"missing batch arguments must name the required field: {message}"
);
let mut args = Map::new();
args.insert("calls".to_string(), Value::String("nope".to_string()));
let non_array = invoke_tool(&gateway, "batch", Some(args), None).await;
assert_eq!(non_array.is_error, Some(true));
let structured = non_array
.structured_content
.expect("structured error present");
assert_eq!(
structured.get("code"),
Some(&Value::String("INVALID_INPUT".to_string()))
);
}
#[tokio::test]
async fn call_returns_structured_error_for_call_error() {
let registry = full_registry_with_ops(vec![]);
let gateway = ToMcpGateway::new(dispatch(registry));
let mut args = Map::new();
args.insert(
"operation".to_string(),
Value::String("no/such".to_string()),
);
args.insert("input".to_string(), Value::Object(Map::new()));
let result = invoke_tool(&gateway, "call", Some(args), None).await;
assert_eq!(result.is_error, Some(true));
let structured = result.structured_content.expect("structured error present");
assert_eq!(
structured.get("code"),
Some(&Value::String("NOT_FOUND".to_string()))
);
}
#[tokio::test]
async fn call_tool_via_services_schema_with_internal_name_returns_not_found() {
let inner = OperationRegistry::new();
inner
.register(HandlerRegistration::new(
internal_spec("secret/op"),
HandlerKind::Once(make_echo_handler()),
OperationProvenance::Local,
None,
None,
Capabilities::new(),
))
.unwrap();
let inner = Arc::new(inner);
let dispatch_registry = OperationRegistry::new();
dispatch_registry
.register(HandlerRegistration::new(
internal_spec("secret/op"),
HandlerKind::Once(make_echo_handler()),
OperationProvenance::Local,
None,
None,
Capabilities::new(),
))
.unwrap();
dispatch_registry
.register(HandlerRegistration::new(
services_list_spec(),
HandlerKind::Once(services_list_handler(Arc::clone(&inner))),
OperationProvenance::Local,
None,
Some(ScopedPeerEnv::empty()),
Capabilities::new(),
))
.unwrap();
dispatch_registry
.register(HandlerRegistration::new(
services_schema_spec(),
HandlerKind::Once(services_schema_handler(Arc::clone(&inner))),
OperationProvenance::Local,
None,
Some(ScopedPeerEnv::empty()),
Capabilities::new(),
))
.unwrap();
let gateway = ToMcpGateway::new(dispatch(Arc::new(dispatch_registry)));
let mut args = Map::new();
args.insert(
"operation".to_string(),
Value::String("services/schema".to_string()),
);
args.insert(
"input".to_string(),
serde_json::json!({ "name": "secret/op" }),
);
let result = invoke_tool(&gateway, "call", Some(args), None).await;
assert_eq!(
result.is_error,
Some(true),
"the MCP call tool must deny a spec the schema tool denies (PRJ-16)"
);
let structured = result.structured_content.expect("structured error present");
assert_eq!(
structured.get("code"),
Some(&Value::String("NOT_FOUND".to_string()))
);
assert!(
structured.get("input_schema").is_none(),
"the internal op's spec must not leak: {structured}"
);
}
#[tokio::test]
async fn batch_tool_via_services_schema_with_internal_name_yields_not_found_entry() {
let inner = OperationRegistry::new();
inner
.register(HandlerRegistration::new(
internal_spec("secret/op"),
HandlerKind::Once(make_echo_handler()),
OperationProvenance::Local,
None,
None,
Capabilities::new(),
))
.unwrap();
inner
.register(HandlerRegistration::new(
external_spec("echo/run", OperationType::Query, AccessControl::default()),
HandlerKind::Once(make_echo_handler()),
OperationProvenance::Local,
None,
None,
Capabilities::new(),
))
.unwrap();
let inner = Arc::new(inner);
let dispatch_registry = OperationRegistry::new();
dispatch_registry
.register(HandlerRegistration::new(
internal_spec("secret/op"),
HandlerKind::Once(make_echo_handler()),
OperationProvenance::Local,
None,
None,
Capabilities::new(),
))
.unwrap();
dispatch_registry
.register(HandlerRegistration::new(
services_list_spec(),
HandlerKind::Once(services_list_handler(Arc::clone(&inner))),
OperationProvenance::Local,
None,
Some(ScopedPeerEnv::empty()),
Capabilities::new(),
))
.unwrap();
dispatch_registry
.register(HandlerRegistration::new(
services_schema_spec(),
HandlerKind::Once(services_schema_handler(Arc::clone(&inner))),
OperationProvenance::Local,
None,
Some(ScopedPeerEnv::empty()),
Capabilities::new(),
))
.unwrap();
dispatch_registry
.register(HandlerRegistration::new(
external_spec("echo/run", OperationType::Query, AccessControl::default()),
HandlerKind::Once(make_echo_handler()),
OperationProvenance::Local,
None,
None,
Capabilities::new(),
))
.unwrap();
let gateway = ToMcpGateway::new(dispatch(Arc::new(dispatch_registry)));
let mut args = Map::new();
args.insert(
"calls".to_string(),
serde_json::json!([
{ "operation": "services/schema", "input": { "name": "secret/op" } },
{ "operation": "services/schema", "input": { "name": "echo/run" } },
]),
);
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 results = structured
.get("results")
.and_then(Value::as_array)
.expect("results array");
assert_eq!(results.len(), 2);
assert_eq!(results[0].get("isError"), Some(&Value::Bool(true)));
assert_eq!(
results[0]
.get("error")
.and_then(|e| e.get("code"))
.cloned()
.unwrap_or(Value::Null),
Value::String("NOT_FOUND".to_string())
);
assert_eq!(results[1].get("isError"), Some(&Value::Bool(false)));
}
#[tokio::test]
async fn batch_over_cap_returns_invalid_input_without_dispatching() {
let registry = full_registry_with_ops(vec![(
"echo/run".to_string(),
OperationType::Query,
AccessControl::default(),
)]);
let calls: Vec<Value> = (0..MAX_BATCH_OPERATIONS + 1)
.map(|i| serde_json::json!({ "operation": "echo/run", "input": { "n": i } }))
.collect();
let mut args = Map::new();
args.insert("calls".to_string(), Value::Array(calls));
let dispatch_spine = dispatch(registry);
let gateway = ToMcpGateway::new(Arc::clone(&dispatch_spine));
let result = invoke_tool(&gateway, "batch", Some(args), None).await;
assert_eq!(result.is_error, Some(true));
let structured = result.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("maximum of 100"),
"the cap message must state the limit: {message}"
);
assert_eq!(
dispatch_spine.invoke_count(),
0,
"no operation may be dispatched when the batch is over cap"
);
}
#[tokio::test]
async fn batch_at_cap_is_accepted() {
let registry = full_registry_with_ops(vec![(
"echo/run".to_string(),
OperationType::Query,
AccessControl::default(),
)]);
let calls: Vec<Value> = (0..MAX_BATCH_OPERATIONS)
.map(|i| serde_json::json!({ "operation": "echo/run", "input": { "n": i } }))
.collect();
let mut args = Map::new();
args.insert("calls".to_string(), Value::Array(calls));
let dispatch_spine = dispatch(registry);
let gateway = ToMcpGateway::new(Arc::clone(&dispatch_spine));
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 results = structured
.get("results")
.and_then(Value::as_array)
.expect("results array");
assert_eq!(results.len(), MAX_BATCH_OPERATIONS);
for entry in results {
assert_eq!(entry.get("isError"), Some(&Value::Bool(false)));
}
assert_eq!(
dispatch_spine.invoke_count(),
MAX_BATCH_OPERATIONS,
"every entry in an at-cap batch is dispatched"
);
}
#[tokio::test]
async fn batch_returns_object_with_result_entries() {
let registry = full_registry_with_ops(vec![(
"echo/run".to_string(),
OperationType::Query,
AccessControl::default(),
)]);
let gateway = ToMcpGateway::new(dispatch(registry));
let mut args = Map::new();
args.insert(
"calls".to_string(),
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");
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]
async fn call_with_restricted_op_and_unauthorized_identity_returns_forbidden_error() {
let registry = full_registry_with_ops(vec![(
"admin/run".to_string(),
OperationType::Query,
AccessControl {
required_scopes: vec!["admin".to_string()],
..Default::default()
},
)]);
let gateway = ToMcpGateway::new(dispatch(registry));
let mut args = Map::new();
args.insert(
"operation".to_string(),
Value::String("admin/run".to_string()),
);
args.insert("input".to_string(), Value::Object(Map::new()));
let result = invoke_tool(&gateway, "call", Some(args), None).await;
assert_eq!(result.is_error, Some(true));
let structured = result.structured_content.expect("structured error present");
assert_eq!(
structured.get("code"),
Some(&Value::String("FORBIDDEN".to_string()))
);
}
#[tokio::test]
async fn unknown_tool_name_returns_not_found_structured_error() {
let gateway = ToMcpGateway::new(dispatch(Arc::new(OperationRegistry::new())));
let result = invoke_tool(&gateway, "bogus", None, None).await;
assert_eq!(result.is_error, Some(true));
let structured = result.structured_content.expect("structured error present");
assert_eq!(
structured.get("code"),
Some(&Value::String("NOT_FOUND".to_string()))
);
}
#[tokio::test]
async fn identity_survives_rmcp_framing_into_call_tool() {
let registry = full_registry_with_ops(vec![(
"admin/run".to_string(),
OperationType::Query,
AccessControl {
required_scopes: vec!["admin".to_string()],
..Default::default()
},
)]);
let gateway = ToMcpGateway::new(dispatch(registry));
let admin_identity = identity_with_scopes("admin-peer", &["admin"]);
let extensions = extensions_with_identity(Some(admin_identity.clone()));
let extracted = ToMcpGateway::extract_identity_from_extensions(&extensions);
assert_eq!(
extracted.as_ref().map(|i| &i.id),
Some(&"admin-peer".to_string())
);
let mut args = Map::new();
args.insert(
"operation".to_string(),
Value::String("admin/run".to_string()),
);
args.insert("input".to_string(), serde_json::json!({ "ok": 1 }));
let result = gateway.handle_call(Some(args), extracted).await;
assert_eq!(result.is_error, Some(false));
assert_eq!(
result.structured_content,
Some(serde_json::json!({ "ok": 1 }))
);
}
#[test]
fn extract_identity_returns_none_when_no_parts_in_extensions() {
let extensions = Extensions::new();
assert!(ToMcpGateway::extract_identity_from_extensions(&extensions).is_none());
}
#[test]
fn extract_identity_returns_none_when_parts_have_no_identity() {
let extensions = extensions_with_identity(None);
assert!(ToMcpGateway::extract_identity_from_extensions(&extensions).is_none());
}
#[test]
fn extract_identity_reads_stashed_option_identity_from_parts() {
let id = identity_with_scopes("caller", &["read"]);
let extensions = extensions_with_identity(Some(id.clone()));
let extracted = ToMcpGateway::extract_identity_from_extensions(&extensions);
assert_eq!(
extracted.as_ref().map(|i| i.id.clone()),
Some("caller".to_string())
);
assert_eq!(
extracted.as_ref().map(|i| i.scopes.clone()),
Some(vec!["read".to_string()])
);
}
#[test]
fn to_mcp_is_not_an_operation_adapter() {
fn assert_not_adapter<T>() {}
assert_not_adapter::<ToMcpGateway>();
}
#[test]
fn gateway_tools_definition_is_stable() {
let tools = gateway_tools();
assert_eq!(tools.len(), 4);
assert_eq!(tools[0].name, "search");
assert_eq!(tools[1].name, "schema");
assert_eq!(tools[2].name, "call");
assert_eq!(tools[3].name, "batch");
}
#[tokio::test]
async fn search_schema_call_round_trip() {
let registry = full_registry_with_ops(vec![(
"fs/readFile".to_string(),
OperationType::Query,
AccessControl::default(),
)]);
let gateway = ToMcpGateway::new(dispatch(registry));
let search_result = invoke_tool(&gateway, "search", None, None).await;
let ops = search_result
.structured_content
.as_ref()
.and_then(|v| v.get("operations"))
.and_then(Value::as_array)
.expect("search ops");
let first_name = ops[0].get("name").and_then(Value::as_str).expect("name");
assert_eq!(first_name, "fs/readFile");
let mut schema_args = Map::new();
schema_args.insert("name".to_string(), Value::String(first_name.to_string()));
let schema_result = invoke_tool(&gateway, "schema", Some(schema_args), None).await;
assert_eq!(
schema_result
.structured_content
.as_ref()
.and_then(|v| v.get("name"))
.and_then(Value::as_str),
Some("fs/readFile")
);
let mut call_args = Map::new();
call_args.insert(
"operation".to_string(),
Value::String(first_name.to_string()),
);
call_args.insert(
"input".to_string(),
serde_json::json!({ "path": "/etc/hosts" }),
);
let call_result = invoke_tool(&gateway, "call", Some(call_args), None).await;
assert_eq!(
call_result.structured_content,
Some(serde_json::json!({ "path": "/etc/hosts" }))
);
}
}