use std::sync::Arc;
use myko::{
command::CommandRegistration, operation_index::OperationSchema, query::QueryRegistration,
report::ReportRegistration, view::ViewRegistration,
};
use serde_json::{Value, json};
use super::{
exec::Executor,
filter::ClientFilters,
sandbox,
types::{McpError, McpRequest, McpResource, McpResponse, McpTool},
};
const CONNECTION_STATUS_TOOL: &str = "connection_status";
const SEARCH_TOOL: &str = "search";
const EXECUTE_TOOL: &str = "execute";
#[derive(Debug, Clone)]
pub struct ServerInfo {
pub name: String,
pub version: String,
pub instructions: Option<String>,
pub operation_index: Arc<Vec<OperationSchema>>,
}
impl Default for ServerInfo {
fn default() -> Self {
Self {
name: "myko-mcp".to_string(),
version: env!("CARGO_PKG_VERSION").to_string(),
instructions: None,
operation_index: Arc::new(myko::operation_index::build_operation_index()),
}
}
}
pub async fn handle_request(
request: McpRequest,
filter: &ClientFilters,
executor: &Executor,
info: &ServerInfo,
) -> Option<McpResponse> {
match request.method.as_str() {
"initialize" => Some(handle_initialize(request.id, info)),
"notifications/initialized" | "notifications/cancelled" => None,
"tools/list" => Some(handle_tools_list(request.id, filter)),
"tools/call" => {
Some(handle_tools_call(request.id, request.params, filter, executor, info).await)
}
"resources/list" => Some(handle_resources_list(request.id, filter)),
"resources/read" => Some(handle_resources_read(request.id, request.params, filter)),
_ => Some(McpResponse::error(
request.id,
McpError::method_not_found(&request.method),
)),
}
}
fn handle_initialize(id: Value, info: &ServerInfo) -> McpResponse {
let mut payload = json!({
"protocolVersion": "2024-11-05",
"capabilities": {
"tools": {},
"resources": {}
},
"serverInfo": {
"name": info.name,
"version": info.version,
}
});
if let Some(text) = &info.instructions {
payload
.as_object_mut()
.expect("payload is an object literal above")
.insert("instructions".to_string(), Value::String(text.clone()));
}
McpResponse::success(id, payload)
}
fn handle_tools_list(id: Value, filter: &ClientFilters) -> McpResponse {
let mut tools: Vec<McpTool> = Vec::new();
if filter.meta_tool_visible(CONNECTION_STATUS_TOOL) {
tools.push(McpTool {
name: CONNECTION_STATUS_TOOL.to_string(),
description: "Check the connection status to the Myko server".to_string(),
input_schema: json!({
"type": "object",
"properties": {},
"required": []
}),
});
}
if filter.meta_tool_visible(SEARCH_TOOL) {
tools.push(McpTool {
name: SEARCH_TOOL.to_string(),
description: "Search the index of available Myko queries/views/reports/commands. \
Returns compact {id, kind, args, outputType} entries — call this before \
`execute` to find operation ids and argument shapes."
.to_string(),
input_schema: json!({
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "Case-insensitive substring match against operation id/description."
},
"kind": {
"type": "string",
"enum": ["query", "view", "report", "command"],
"description": "Restrict results to one operation kind."
}
},
"required": []
}),
});
}
if filter.meta_tool_visible(EXECUTE_TOOL) {
tools.push(McpTool {
name: EXECUTE_TOOL.to_string(),
description: "Run JavaScript against the Myko API. The code runs as an async \
function body — use `await myko.query(id, args)`, `myko.view(id, args)`, \
`myko.report(id, args)`, or `myko.command(id, args)` (ids/args from `search`), \
and optionally `return` a JSON-serializable value. Chain multiple calls in one \
script instead of one `execute` call per operation. Each call resolves to a \
wrapper object, not the raw payload directly: query/view resolve to \
{query_id|view_id, item_type, count, items} (items is the payload); \
report resolves to {report_id, output_type, result} (result is the payload); \
command resolves to {command_id, success, result} (result is the payload). \
E.g. `(await myko.report('ServerStats', {})).result`, not `.serverStats`."
.to_string(),
input_schema: json!({
"type": "object",
"properties": {
"code": {
"type": "string",
"description": "JavaScript function body to run."
}
},
"required": ["code"]
}),
});
}
McpResponse::success(id, json!({ "tools": tools }))
}
async fn handle_tools_call(
id: Value,
params: Option<Value>,
filter: &ClientFilters,
executor: &Executor,
info: &ServerInfo,
) -> McpResponse {
let Some(params) = params else {
return McpResponse::error(id, McpError::invalid_params("Missing params"));
};
let Some(tool_name) = params
.get("name")
.and_then(|v| v.as_str())
.map(str::to_string)
else {
return McpResponse::error(id, McpError::invalid_params("Missing tool name"));
};
if !filter.meta_tool_visible(&tool_name) {
return McpResponse::error(
id,
McpError {
code: McpError::INVALID_PARAMS,
message: format!("Unknown tool: {}", tool_name),
data: None,
},
);
}
let arguments = params
.get("arguments")
.cloned()
.unwrap_or_else(|| json!({}));
if let Err(message) = filter.tool_callable(&tool_name, &arguments) {
return McpResponse::success(
id,
json!({
"content": [{
"type": "text",
"text": message,
}],
"isError": true,
}),
);
}
let result = execute_tool(executor, info, filter, &tool_name, arguments).await;
match result {
Ok(data) => McpResponse::success(
id,
json!({
"content": [{
"type": "text",
"text": serde_json::to_string_pretty(&data).unwrap_or_default()
}]
}),
),
Err(message) => McpResponse::success(
id,
json!({
"content": [{
"type": "text",
"text": format!("Error: {}", message)
}],
"isError": true,
}),
),
}
}
async fn execute_tool(
executor: &Executor,
info: &ServerInfo,
filter: &ClientFilters,
tool_name: &str,
args: Value,
) -> Result<Value, String> {
match tool_name {
CONNECTION_STATUS_TOOL => Ok(executor.connection_status(info)),
SEARCH_TOOL => Ok(handle_search(&args, filter, &info.operation_index)),
EXECUTE_TOOL => handle_execute(&args, executor, filter).await,
_ => Err(format!(
"Unknown tool: {tool_name}. This server uses search + execute — \
call `search` to discover operations, then `execute` to run them."
)),
}
}
fn handle_search(args: &Value, filter: &ClientFilters, index: &[OperationSchema]) -> Value {
let query = args
.get("query")
.and_then(|v| v.as_str())
.map(str::to_lowercase);
let kind = args.get("kind").and_then(|v| v.as_str());
let operations: Vec<&OperationSchema> = index
.iter()
.filter(|op| filter.tool_visible(&format!("{}_{}", op.kind, op.id)))
.filter(|op| kind.is_none_or(|k| op.kind == k))
.filter(|op| {
query.as_deref().is_none_or(|q| {
op.id.to_lowercase().contains(q)
|| op
.description
.as_deref()
.is_some_and(|d| d.to_lowercase().contains(q))
})
})
.collect();
json!({ "operations": operations })
}
async fn handle_execute(
args: &Value,
executor: &Executor,
filter: &ClientFilters,
) -> Result<Value, String> {
let Some(code) = args.get("code").and_then(|v| v.as_str()) else {
return Err("Missing required `code` argument".to_string());
};
sandbox::execute(code, Arc::new(executor.clone()), filter.clone()).await
}
fn handle_resources_list(id: Value, filter: &ClientFilters) -> McpResponse {
let mut resources: Vec<McpResource> = Vec::new();
for reg in inventory::iter::<QueryRegistration> {
let tool_name = format!("query_{}", reg.query_id);
if !filter.tool_visible(&tool_name) {
continue;
}
resources.push(McpResource {
uri: format!("myko://schema/query/{}", reg.query_id),
name: reg.query_id.to_string(),
description: Some(format!("Query returning {} entities", reg.query_item_type)),
mime_type: Some("application/json".to_string()),
});
}
for reg in inventory::iter::<ViewRegistration> {
let tool_name = format!("view_{}", reg.view_id);
if !filter.tool_visible(&tool_name) {
continue;
}
resources.push(McpResource {
uri: format!("myko://schema/view/{}", reg.view_id),
name: reg.view_id.to_string(),
description: Some(format!("View returning a list of {}", reg.view_item_type)),
mime_type: Some("application/json".to_string()),
});
}
for reg in inventory::iter::<ReportRegistration> {
let tool_name = format!("report_{}", reg.report_id);
if !filter.tool_visible(&tool_name) {
continue;
}
resources.push(McpResource {
uri: format!("myko://schema/report/{}", reg.report_id),
name: reg.report_id.to_string(),
description: Some(format!("Report returning {}", reg.output_type)),
mime_type: Some("application/json".to_string()),
});
}
for reg in inventory::iter::<CommandRegistration> {
let tool_name = format!("command_{}", reg.command_id);
if !filter.tool_visible(&tool_name) {
continue;
}
resources.push(McpResource {
uri: format!("myko://schema/command/{}", reg.command_id),
name: format!("{} (command)", reg.command_id),
description: Some(format!("Command returning {}", reg.result_type)),
mime_type: Some("application/json".to_string()),
});
}
McpResponse::success(id, json!({ "resources": resources }))
}
fn handle_resources_read(id: Value, params: Option<Value>, filter: &ClientFilters) -> McpResponse {
let Some(params) = params else {
return McpResponse::error(id, McpError::invalid_params("Missing params"));
};
let Some(uri) = params.get("uri").and_then(|v| v.as_str()) else {
return McpResponse::error(id, McpError::invalid_params("Missing uri"));
};
if let Some(path) = uri.strip_prefix("myko://schema/") {
let parts: Vec<&str> = path.splitn(2, '/').collect();
if parts.len() == 2 {
let (schema_type, schema_id) = (parts[0], parts[1]);
let tool_name = format!("{}:{}", schema_type, schema_id);
if !filter.tool_visible(&tool_name) {
return McpResponse::error(
id,
McpError {
code: McpError::INVALID_PARAMS,
message: format!("Resource not accessible: {}", uri),
data: None,
},
);
}
let content = match schema_type {
"query" => get_query_schema(schema_id),
"view" => get_view_schema(schema_id),
"report" => get_report_schema(schema_id),
"command" => get_command_schema(schema_id),
_ => None,
};
if let Some(content) = content {
return McpResponse::success(
id,
json!({
"contents": [{
"uri": uri,
"mimeType": "application/json",
"text": content,
}]
}),
);
}
}
}
McpResponse::error(
id,
McpError {
code: McpError::INVALID_PARAMS,
message: format!("Resource not found: {}", uri),
data: None,
},
)
}
fn get_query_schema(query_id: &str) -> Option<String> {
for reg in inventory::iter::<QueryRegistration> {
if reg.query_id == query_id {
let schema = json!({
"$schema": "http://json-schema.org/draft-07/schema#",
"title": reg.query_id,
"description": format!("Query returning {} entities", reg.query_item_type),
"type": "object",
"additionalProperties": true,
});
return Some(serde_json::to_string_pretty(&schema).unwrap_or_default());
}
}
None
}
fn get_view_schema(view_id: &str) -> Option<String> {
for reg in inventory::iter::<ViewRegistration> {
if reg.view_id == view_id {
let schema = json!({
"$schema": "http://json-schema.org/draft-07/schema#",
"title": reg.view_id,
"description": format!("View returning a list of {}", reg.view_item_type),
"type": "object",
"additionalProperties": true,
});
return Some(serde_json::to_string_pretty(&schema).unwrap_or_default());
}
}
None
}
fn get_report_schema(report_id: &str) -> Option<String> {
for reg in inventory::iter::<ReportRegistration> {
if reg.report_id == report_id {
let schema = json!({
"$schema": "http://json-schema.org/draft-07/schema#",
"title": reg.report_id,
"description": format!("Report returning {}", reg.output_type),
"type": "object",
"additionalProperties": true,
});
return Some(serde_json::to_string_pretty(&schema).unwrap_or_default());
}
}
None
}
fn get_command_schema(command_id: &str) -> Option<String> {
for reg in inventory::iter::<CommandRegistration> {
if reg.command_id == command_id {
let schema = json!({
"$schema": "http://json-schema.org/draft-07/schema#",
"title": reg.command_id,
"description": format!("Command returning {}", reg.result_type),
"type": "object",
"additionalProperties": true,
});
return Some(serde_json::to_string_pretty(&schema).unwrap_or_default());
}
}
None
}
#[cfg(test)]
mod tests {
use super::*;
use serde_json::Value;
fn make_request(method: &str) -> McpRequest {
McpRequest {
jsonrpc: "2.0".to_string(),
id: Value::Number(1.into()),
method: method.to_string(),
params: None,
}
}
#[test]
fn server_info_default_omits_instructions() {
let info = ServerInfo::default();
assert_eq!(info.instructions, None);
}
#[test]
fn server_info_can_carry_instructions() {
let info = ServerInfo {
name: "test".into(),
version: "0.0.0".into(),
instructions: Some("test instructions text".into()),
..Default::default()
};
assert_eq!(info.instructions.as_deref(), Some("test instructions text"));
}
#[tokio::test]
async fn initialize_returns_server_info() {
let filter = ClientFilters::allow_all();
let info = ServerInfo {
name: "test".into(),
version: "0.0.0".into(),
instructions: None,
..Default::default()
};
let client = std::sync::Arc::new(myko::client::MykoClient::new());
let executor = Executor::Client(client);
let response = handle_request(make_request("initialize"), &filter, &executor, &info)
.await
.expect("initialize must produce a response");
let result = response.result.expect("initialize must have a result");
assert_eq!(result["serverInfo"]["name"], "test");
assert_eq!(result["serverInfo"]["version"], "0.0.0");
}
#[tokio::test]
async fn initialize_includes_instructions_when_set() {
let filter = ClientFilters::allow_all();
let info = ServerInfo {
name: "pulse-mcp".into(),
version: "0.2.0".into(),
instructions: Some("teach me".into()),
..Default::default()
};
let client = std::sync::Arc::new(myko::client::MykoClient::new());
let executor = Executor::Client(client);
let resp = handle_request(make_request("initialize"), &filter, &executor, &info)
.await
.expect("initialize must return a response");
let result = resp.result.expect("initialize must succeed");
assert_eq!(result["serverInfo"]["name"], json!("pulse-mcp"));
assert_eq!(result["serverInfo"]["version"], json!("0.2.0"));
assert_eq!(result["instructions"], json!("teach me"));
}
#[tokio::test]
async fn initialize_omits_instructions_when_unset() {
let filter = ClientFilters::allow_all();
let info = ServerInfo::default();
let client = std::sync::Arc::new(myko::client::MykoClient::new());
let executor = Executor::Client(client);
let resp = handle_request(make_request("initialize"), &filter, &executor, &info)
.await
.expect("response");
let result = resp.result.expect("ok");
assert!(
result.get("instructions").is_none(),
"instructions must be omitted when ServerInfo.instructions is None"
);
}
#[tokio::test]
async fn notifications_produce_no_response() {
let filter = ClientFilters::allow_all();
let info = ServerInfo::default();
let client = std::sync::Arc::new(myko::client::MykoClient::new());
let executor = Executor::Client(client);
assert!(
handle_request(
make_request("notifications/initialized"),
&filter,
&executor,
&info,
)
.await
.is_none()
);
}
#[tokio::test]
async fn unknown_method_returns_error() {
let filter = ClientFilters::allow_all();
let info = ServerInfo::default();
let client = std::sync::Arc::new(myko::client::MykoClient::new());
let executor = Executor::Client(client);
let response = handle_request(make_request("unknown/method"), &filter, &executor, &info)
.await
.expect("must produce a response");
assert!(response.error.is_some());
}
fn make_tool_call(name: &str, arguments: Value) -> McpRequest {
McpRequest {
jsonrpc: "2.0".to_string(),
id: Value::Number(1.into()),
method: "tools/call".to_string(),
params: Some(json!({ "name": name, "arguments": arguments })),
}
}
fn dummy_executor() -> Executor {
Executor::Client(std::sync::Arc::new(myko::client::MykoClient::new()))
}
fn info_with_index() -> ServerInfo {
ServerInfo {
operation_index: Arc::new(vec![
OperationSchema {
id: "GetAllServers".to_string(),
kind: "query".to_string(),
description: Some("All servers".to_string()),
args: vec![],
output_type: "Server[]".to_string(),
},
OperationSchema {
id: "DeleteServer".to_string(),
kind: "command".to_string(),
description: Some("Delete a server".to_string()),
args: vec![],
output_type: "DeleteServerResult".to_string(),
},
]),
..Default::default()
}
}
#[tokio::test]
async fn tools_list_only_exposes_search_execute_and_connection_status() {
let filter = ClientFilters::allow_all();
let info = ServerInfo::default();
let executor = dummy_executor();
let resp = handle_request(make_request("tools/list"), &filter, &executor, &info)
.await
.expect("response");
let tools = resp.result.expect("ok")["tools"]
.as_array()
.expect("array")
.iter()
.map(|t| t["name"].as_str().unwrap().to_string())
.collect::<std::collections::HashSet<_>>();
assert_eq!(
tools,
std::collections::HashSet::from([
CONNECTION_STATUS_TOOL.to_string(),
SEARCH_TOOL.to_string(),
EXECUTE_TOOL.to_string(),
])
);
}
#[tokio::test]
async fn search_filters_by_kind_and_query_text() {
let filter = ClientFilters::allow_all();
let info = info_with_index();
let executor = dummy_executor();
let resp = handle_request(
make_tool_call("search", json!({ "kind": "command" })),
&filter,
&executor,
&info,
)
.await
.expect("response");
let text = resp.result.expect("ok")["content"][0]["text"]
.as_str()
.unwrap()
.to_string();
let parsed: Value = serde_json::from_str(&text).expect("valid JSON content");
let ops = parsed["operations"].as_array().expect("array");
assert_eq!(ops.len(), 1);
assert_eq!(ops[0]["id"], "DeleteServer");
}
#[tokio::test]
async fn search_respects_visibility_filter() {
let filter = ClientFilters::from_strings(None, Some("command_*"), None, None);
let info = info_with_index();
let executor = dummy_executor();
let resp = handle_request(
make_tool_call("search", json!({})),
&filter,
&executor,
&info,
)
.await
.expect("response");
let text = resp.result.expect("ok")["content"][0]["text"]
.as_str()
.unwrap()
.to_string();
let parsed: Value = serde_json::from_str(&text).unwrap();
let ids: Vec<&str> = parsed["operations"]
.as_array()
.unwrap()
.iter()
.map(|o| o["id"].as_str().unwrap())
.collect();
assert_eq!(ids, vec!["GetAllServers"]);
}
#[tokio::test]
async fn execute_runs_a_script_and_returns_its_value() {
let filter = ClientFilters::allow_all();
let info = ServerInfo::default();
let executor = dummy_executor();
let resp = handle_request(
make_tool_call("execute", json!({ "code": "return 21 * 2;" })),
&filter,
&executor,
&info,
)
.await
.expect("response");
let result = resp.result.expect("ok");
assert_ne!(result["isError"], json!(true));
let text = result["content"][0]["text"].as_str().unwrap();
assert_eq!(text.trim(), "42");
}
#[tokio::test]
async fn connection_status_identifies_the_server_instance() {
let filter = ClientFilters::allow_all();
let info = ServerInfo {
name: "pulse-ctx".into(),
version: "1.2.3".into(),
..Default::default()
};
let executor = dummy_executor();
let resp = handle_request(
make_tool_call("connection_status", json!({})),
&filter,
&executor,
&info,
)
.await
.expect("response");
let text = resp.result.expect("ok")["content"][0]["text"]
.as_str()
.unwrap()
.to_string();
let parsed: Value = serde_json::from_str(&text).unwrap();
assert_eq!(parsed["name"], "pulse-ctx");
assert_eq!(parsed["version"], "1.2.3");
}
#[tokio::test]
async fn execute_without_code_argument_is_a_tool_execution_error() {
let filter = ClientFilters::allow_all();
let info = ServerInfo::default();
let executor = dummy_executor();
let resp = handle_request(
make_tool_call("execute", json!({})),
&filter,
&executor,
&info,
)
.await
.expect("response");
let result = resp.result.expect("ok");
assert_eq!(result["isError"], json!(true));
}
#[tokio::test]
async fn hidden_execute_tool_is_a_protocol_error() {
let filter = ClientFilters::from_strings(None, Some("execute"), None, None);
let info = ServerInfo::default();
let executor = dummy_executor();
let resp = handle_request(
make_tool_call("execute", json!({ "code": "return 1;" })),
&filter,
&executor,
&info,
)
.await
.expect("response");
assert!(
resp.error.is_some(),
"denied tool must be a protocol error, not a tool result"
);
}
#[tokio::test]
async fn op_level_allow_list_scopes_operations_without_hiding_search_and_execute() {
let filter = ClientFilters::from_strings(Some("query_GetAllServers"), None, None, None);
let info = info_with_index();
let executor = dummy_executor();
let list_resp = handle_request(make_request("tools/list"), &filter, &executor, &info)
.await
.expect("response");
let tools: Vec<String> = list_resp.result.expect("ok")["tools"]
.as_array()
.expect("array")
.iter()
.map(|t| t["name"].as_str().unwrap().to_string())
.collect();
assert!(
tools.contains(&SEARCH_TOOL.to_string()) && tools.contains(&EXECUTE_TOOL.to_string()),
"op-level allow list must not hide search/execute, got {tools:?}"
);
let search_resp = handle_request(
make_tool_call("search", json!({})),
&filter,
&executor,
&info,
)
.await
.expect("response");
let text = search_resp.result.expect("ok")["content"][0]["text"]
.as_str()
.unwrap()
.to_string();
let parsed: Value = serde_json::from_str(&text).unwrap();
let ids: Vec<&str> = parsed["operations"]
.as_array()
.unwrap()
.iter()
.map(|o| o["id"].as_str().unwrap())
.collect();
assert_eq!(
ids,
vec!["GetAllServers"],
"search must only surface the allow-listed query, not the un-listed DeleteServer command"
);
let execute_resp = handle_request(
make_tool_call(
"execute",
json!({ "code": "try { await myko.command('DeleteServer', {id: 'x'}); return 'no-throw'; } catch (e) { return e.message; }" }),
),
&filter,
&executor,
&info,
)
.await
.expect("response");
let execute_result = execute_resp.result.expect("ok");
assert_ne!(execute_result["isError"], json!(true));
let message = execute_result["content"][0]["text"].as_str().unwrap();
assert_eq!(
message.trim(),
"\"Unknown operation: command_DeleteServer\""
);
}
}