diff --git a/Cargo.lock b/Cargo.lock index e0ac8c005..6418b05aa 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4714,7 +4714,7 @@ dependencies = [ [[package]] name = "parseable" -version = "3.2.4" +version = "3.2.5" dependencies = [ "actix-cors", "actix-web", diff --git a/Cargo.toml b/Cargo.toml index 8bcc5bf6b..8674a5b0e 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "parseable" -version = "3.2.4" +version = "3.2.5" authors = ["Parseable Team "] edition = "2024" rust-version = "1.94.0" @@ -234,8 +234,8 @@ arrow = "59.2.0" temp-dir = "0.1.14" [package.metadata.parseable_ui] -assets-url = "https://parseableprismbuild.blob.core.windows.net/parseable-prism-build/v3.2.4/build.zip" -assets-sha1 = "5e3ec81fbb35b2b8aa6486a2bd38257d46b0264b" +assets-url = "https://parseableprismbuild.blob.core.windows.net/parseable-prism-build/v3.2.5/build.zip" +assets-sha1 = "cbaf98f0431844677d30f519b3da941599eb2e0a" [features] debug = [] diff --git a/docs/tool-api.md b/docs/tool-api.md new file mode 100644 index 000000000..720aaa30b --- /dev/null +++ b/docs/tool-api.md @@ -0,0 +1,91 @@ +# Parseable tool REST API + +Parseable owns the canonical tool catalog and execution semantics. External MCP +servers adapt these REST endpoints to MCP `tools/list` and `tools/call`. + +Both endpoints require normal Parseable authentication. Tool execution also +applies the RBAC action required by the selected tool and any referenced +dataset or resource. + +## List tools + +```http +GET /api/prism/v1/llm/tools/list +``` + +Response: + +```json +{ + "tools": [ + { + "name": "get_dataset_schema", + "title": "Get dataset schema", + "description": "Get columns and data types for a dataset.", + "inputSchema": { + "type": "object", + "properties": { "dataset": { "type": "string", "minLength": 1 } }, + "required": ["dataset"], + "additionalProperties": false + } + } + ] +} +``` + +The response includes an `ETag` and +`Cache-Control: private, max-age=300, must-revalidate`. A request with a +matching `If-None-Match` receives `304 Not Modified`. + +The catalog is edition-specific: + +- OSS returns externally callable OSS read tools. +- Enterprise returns OSS read tools plus Enterprise read tools. +- Parseable Cloud uses the Enterprise server catalog and exposes the tools + enabled by that deployment. + +## Call a tool + +```http +POST /api/prism/v1/llm/tools/call +Content-Type: application/json +``` + +Request: + +```json +{ + "name": "get_dataset_schema", + "arguments": { "dataset": "application-logs" } +} +``` + +`arguments` must satisfy the tool's `inputSchema`. The external registry +currently exposes only read-only and expensive-read tools. Mutation and +external-side-effect tools remain unavailable until Parseable supports a +server-verifiable confirmation flow bound to the session, tool, and arguments. + +Successful and failed executions use the MCP `CallToolResult` shape: + +```json +{ + "content": [ + { "type": "text", "text": "{\"fields\":[...]}" } + ], + "structuredContent": { + "result": { "fields": [] } + }, + "isError": false +} +``` + +Tool failures set `isError` to `true`, include a text error content item, and +put the message under `structuredContent.error.message`. A completed tool call +returns HTTP 200 even when the tool reports `isError: true`, matching MCP +`tools/call` semantics. REST status codes distinguish failures that prevent a +call from starting, such as malformed input, authentication, or an unknown +tool. + +The REST API is not itself an MCP transport. An MCP adapter maps the registry +response to `tools/list`, forwards calls to this endpoint, and returns the +`CallToolResult` from Parseable through MCP `tools/call`. diff --git a/src/handlers/http/modal/server.rs b/src/handlers/http/modal/server.rs index 12e67c670..1ef3fbd04 100644 --- a/src/handlers/http/modal/server.rs +++ b/src/handlers/http/modal/server.rs @@ -202,13 +202,21 @@ impl ParseableServer for Server { impl Server { pub fn get_llm_webscope() -> Scope { - web::scope("/llm").service( - web::resource("/tools").route( - web::get() - .to(crate::tool_catalog::list_tool_registry) - .authorize(Action::Query), - ), - ) + web::scope("/llm") + .service( + web::resource("/tools/list").route( + web::get() + .to(crate::tool_catalog::list_tool_registry) + .authorize(Action::Query), + ), + ) + .service( + web::resource("/tools/call").route( + web::post() + .to(crate::tool_executor::call_tool_registry) + .authorize(Action::Query), + ), + ) } pub fn get_prism_home() -> Scope { diff --git a/src/lib.rs b/src/lib.rs index d48dbb4a3..e570de9f3 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -59,6 +59,7 @@ pub mod sync; pub mod telemetry; pub mod tenants; pub mod tool_catalog; +pub mod tool_executor; pub mod users; pub mod utils; pub mod validator; diff --git a/src/sync.rs b/src/sync.rs index 39681ec42..8766ba84d 100644 --- a/src/sync.rs +++ b/src/sync.rs @@ -451,20 +451,24 @@ pub async fn alert_runtime(mut rx: mpsc::Receiver) -> Result<(), anyh let id = *alert.get_id(); let handle = tokio::spawn(async move { let mut retry_counter = 0; - let mut sleep_duration = alert.get_eval_frequency(); loop { - match alerts_utils::evaluate_alert(&*alert).await { + let sleep_duration = match alerts_utils::evaluate_alert(&*alert).await { Ok(_) => { retry_counter = 0; + alert.get_eval_frequency() } Err(err) => { + let retry_delay = if alert.get_query_type() + == crate::alerts::AlertQueryType::Promql + { + alert.get_eval_frequency() + } else { + 1 + }; warn!( - "Error while evaluation- {}\nRetrying after sleeping for 1 minute", - err + "Error while evaluation- {}\nRetrying after sleeping for {} minute(s)", + err, retry_delay ); - if alert.get_query_type() != crate::alerts::AlertQueryType::Promql { - sleep_duration = 1; - } retry_counter += 1; if retry_counter > 3 @@ -477,8 +481,9 @@ pub async fn alert_runtime(mut rx: mpsc::Receiver) -> Result<(), anyh ); break; } + retry_delay } - } + }; tokio::time::sleep(Duration::from_secs(sleep_duration * 60)).await; } }); diff --git a/src/tool_catalog.rs b/src/tool_catalog.rs index cc294a15c..3e2d898fd 100644 --- a/src/tool_catalog.rs +++ b/src/tool_catalog.rs @@ -1,6 +1,68 @@ -use actix_web::HttpResponse; +use actix_web::{ + HttpRequest, HttpResponse, + http::header::{CACHE_CONTROL, ETAG, IF_NONE_MATCH}, +}; +use serde::{Deserialize, Serialize}; use serde_json::{Value, json}; +use crate::utils::get_hash; + +#[derive(Clone, Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct ToolCallRequest { + pub name: String, + #[serde(default = "empty_arguments")] + pub arguments: Value, +} + +fn empty_arguments() -> Value { + json!({}) +} + +#[derive(Clone, Debug, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct ToolCallResult { + pub content: Vec, + pub structured_content: Value, + pub is_error: bool, +} + +#[derive(Clone, Debug, Serialize)] +pub struct ToolContent { + #[serde(rename = "type")] + pub content_type: &'static str, + pub text: String, +} + +impl ToolCallResult { + pub fn success(result: Value) -> Self { + let text = match &result { + Value::String(text) => text.clone(), + result => serde_json::to_string(result).unwrap_or_else(|_| "null".to_owned()), + }; + Self { + content: vec![ToolContent { + content_type: "text", + text, + }], + structured_content: json!({ "result": result }), + is_error: false, + } + } + + pub fn error(message: impl Into) -> Self { + let message = message.into(); + Self { + content: vec![ToolContent { + content_type: "text", + text: message.clone(), + }], + structured_content: json!({ "error": { "message": message } }), + is_error: true, + } + } +} + #[derive(Clone, Copy, Debug, Eq, PartialEq)] pub enum ToolEffect { ReadOnly, @@ -38,7 +100,6 @@ impl ToolSpec { | "sample_events" | "get_log_context" | "query_sql" - | "query_promql" | "get_cluster_metrics" | "explain_query" | "discover_datasets" @@ -60,6 +121,147 @@ impl ToolSpec { _ => ToolEffect::ReadOnly, } } + + pub fn validate_arguments(&self, arguments: &Value) -> Result<(), String> { + validate_tool_arguments(&self.input_schema, arguments) + } +} + +pub fn validate_tool_arguments(schema: &Value, arguments: &Value) -> Result<(), String> { + validate_schema_value(schema, arguments, "arguments") +} + +fn validate_schema_value(schema: &Value, value: &Value, path: &str) -> Result<(), String> { + if let Some(result) = validate_any_of(schema, value, path) { + return result; + } + + validate_enum(schema, value, path)?; + validate_type(schema, value, path)?; + validate_object(schema, value, path)?; + validate_array(schema, value, path)?; + validate_string(schema, value, path)?; + validate_number(schema, value, path) +} + +fn validate_any_of(schema: &Value, value: &Value, path: &str) -> Option> { + schema + .get("anyOf") + .and_then(Value::as_array) + .map(|variants| { + variants + .iter() + .any(|variant| validate_schema_value(variant, value, path).is_ok()) + .then_some(()) + .ok_or_else(|| format!("{path} does not match any allowed schema")) + }) +} + +fn validate_enum(schema: &Value, value: &Value, path: &str) -> Result<(), String> { + if let Some(allowed) = schema.get("enum").and_then(Value::as_array) + && !allowed.contains(value) + { + return Err(format!("{path} must be one of {allowed:?}")); + } + Ok(()) +} + +fn validate_type(schema: &Value, value: &Value, path: &str) -> Result<(), String> { + let Some(expected_type) = schema.get("type").and_then(Value::as_str) else { + return Ok(()); + }; + let type_matches = match expected_type { + "object" => value.is_object(), + "array" => value.is_array(), + "string" => value.is_string(), + "integer" => value.as_i64().is_some() || value.as_u64().is_some(), + "number" => value.is_number(), + "boolean" => value.is_boolean(), + "null" => value.is_null(), + _ => true, + }; + if type_matches { + Ok(()) + } else { + Err(format!("{path} must be a {expected_type}")) + } +} + +fn validate_object(schema: &Value, value: &Value, path: &str) -> Result<(), String> { + let Some(object) = value.as_object() else { + return Ok(()); + }; + let properties = schema.get("properties").and_then(Value::as_object); + if let Some(required) = schema.get("required").and_then(Value::as_array) { + for name in required.iter().filter_map(Value::as_str) { + if !object.contains_key(name) { + return Err(format!("{path}.{name} is required")); + } + } + } + if schema.get("additionalProperties").and_then(Value::as_bool) == Some(false) + && let Some(properties) = properties + && let Some(name) = object.keys().find(|name| !properties.contains_key(*name)) + { + return Err(format!("{path}.{name} is not allowed")); + } + if let Some(properties) = properties { + for (name, value) in object { + if let Some(property_schema) = properties.get(name) { + validate_schema_value(property_schema, value, &format!("{path}.{name}"))?; + } + } + } + Ok(()) +} + +fn validate_array(schema: &Value, value: &Value, path: &str) -> Result<(), String> { + let Some(array) = value.as_array() else { + return Ok(()); + }; + if let Some(minimum) = schema.get("minItems").and_then(Value::as_u64) + && array.len() < minimum as usize + { + return Err(format!("{path} must contain at least {minimum} items")); + } + if let Some(maximum) = schema.get("maxItems").and_then(Value::as_u64) + && array.len() > maximum as usize + { + return Err(format!("{path} must contain at most {maximum} items")); + } + if let Some(item_schema) = schema.get("items") { + for (index, item) in array.iter().enumerate() { + validate_schema_value(item_schema, item, &format!("{path}[{index}]"))?; + } + } + Ok(()) +} + +fn validate_string(schema: &Value, value: &Value, path: &str) -> Result<(), String> { + if let Some(string) = value.as_str() + && let Some(minimum) = schema.get("minLength").and_then(Value::as_u64) + && string.chars().count() < minimum as usize + { + return Err(format!("{path} must contain at least {minimum} characters")); + } + Ok(()) +} + +fn validate_number(schema: &Value, value: &Value, path: &str) -> Result<(), String> { + let Some(number) = value.as_f64() else { + return Ok(()); + }; + if let Some(minimum) = schema.get("minimum").and_then(Value::as_f64) + && number < minimum + { + return Err(format!("{path} must be at least {minimum}")); + } + if let Some(maximum) = schema.get("maximum").and_then(Value::as_f64) + && number > maximum + { + return Err(format!("{path} must be at most {maximum}")); + } + Ok(()) } fn object_schema(properties: Value, required: &[&str]) -> Value { @@ -100,11 +302,9 @@ fn target_schema() -> Value { ) } -/// Tools implemented by both Parseable OSS and Enterprise. -/// -/// Enterprise may extend this catalog, but an OSS server only advertises this -/// list so MCP clients never discover enterprise-only capabilities. -pub fn oss_tool_specs() -> Vec { +/// Read-only tools implemented by Parseable OSS and safe for generic external +/// invocation without a confirmation flow. +pub fn oss_read_tool_specs() -> Vec { vec![ ToolSpec::new( "list_datasets", @@ -215,25 +415,6 @@ pub fn oss_tool_specs() -> Vec { &["query", "startTime", "endTime"], ), ), - ToolSpec::new( - "query_promql", - "Run PromQL query", - "Execute a PromQL instant or range query.", - object_schema( - json!({ - "query": string_property("PromQL expression."), - "stream": string_property("Metrics dataset name."), - "start": { "type": "string" }, - "end": { "type": "string" }, - "step": { "type": "string" }, - "time": { "type": "string" }, - "timeout": { "type": "number", "minimum": 0 }, - "limit": { "type": "integer", "minimum": 1, "maximum": 500, "default": 500 }, - "timestamp_format": { "type": "string", "enum": ["rfc3339", "unix"] }, - }), - &["query", "stream"], - ), - ), ToolSpec::new( "list_alerts", "List alerts", @@ -252,33 +433,6 @@ pub fn oss_tool_specs() -> Vec { "List tags used across alerts.", empty_schema(), ), - ToolSpec::new( - "enable_alert", - "Enable alert", - "Enable a disabled alert.", - alert_schema(), - ), - ToolSpec::new( - "disable_alert", - "Disable alert", - "Disable an active alert.", - alert_schema(), - ), - ToolSpec::new( - "evaluate_alert", - "Evaluate alert now", - "Evaluate an alert immediately. This may trigger notifications.", - alert_schema(), - ), - ToolSpec::new( - "create_alert", - "Create alert", - "Create an alert from a complete Parseable alert specification.", - object_schema( - json!({ "spec": { "type": "object", "additionalProperties": true } }), - &["spec"], - ), - ), ToolSpec::new( "list_alert_targets", "List alert targets", @@ -291,15 +445,6 @@ pub fn oss_tool_specs() -> Vec { "Get the configuration for one alert target.", target_schema(), ), - ToolSpec::new( - "create_alert_target", - "Create alert target", - "Create an alert notification target.", - object_schema( - json!({ "spec": { "type": "object", "additionalProperties": true } }), - &["spec"], - ), - ), ToolSpec::new( "get_traces", "Get traces", @@ -456,6 +601,50 @@ pub fn oss_tool_specs() -> Vec { &[], ), ), + ] +} + +/// Effectful OSS tools retained in the canonical registry for internal agents. +/// Generic external invocation must not use this catalog until confirmation is +/// server-verifiable. +pub fn oss_action_tool_specs() -> Vec { + vec![ + ToolSpec::new( + "enable_alert", + "Enable alert", + "Enable a disabled alert.", + alert_schema(), + ), + ToolSpec::new( + "disable_alert", + "Disable alert", + "Disable an active alert.", + alert_schema(), + ), + ToolSpec::new( + "evaluate_alert", + "Evaluate alert now", + "Evaluate an alert immediately. This may trigger notifications.", + alert_schema(), + ), + ToolSpec::new( + "create_alert", + "Create alert", + "Create an alert from a complete Parseable alert specification.", + object_schema( + json!({ "spec": { "type": "object", "additionalProperties": true } }), + &["spec"], + ), + ), + ToolSpec::new( + "create_alert_target", + "Create alert target", + "Create an alert notification target.", + object_schema( + json!({ "spec": { "type": "object", "additionalProperties": true } }), + &["spec"], + ), + ), ToolSpec::new( "update_alert", "Update alert", @@ -562,13 +751,48 @@ pub fn oss_tool_specs() -> Vec { ] } +/// Canonical OSS catalog used by internal agents and imported by Enterprise. +pub fn oss_tool_specs() -> Vec { + let mut tools = oss_read_tool_specs(); + tools.extend(oss_action_tool_specs()); + tools +} + pub fn is_oss_tool(name: &str) -> bool { oss_tool_specs().iter().any(|tool| tool.name == name) } -/// `GET /api/prism/v1/llm/tools` -pub async fn list_tool_registry() -> HttpResponse { - let tools = oss_tool_specs() +pub fn is_oss_read_tool(name: &str) -> bool { + oss_read_tool_specs().iter().any(|tool| tool.name == name) +} + +pub fn registry_http_response(request: &HttpRequest, tools: Vec) -> HttpResponse { + let payload = json!({ "tools": tools }); + let body = serde_json::to_string(&payload).expect("tool registry is JSON serializable"); + let etag = format!("\"{}\"", get_hash(&body)); + let cache_control = "private, max-age=300, must-revalidate"; + if request + .headers() + .get(IF_NONE_MATCH) + .and_then(|value| value.to_str().ok()) + == Some(etag.as_str()) + { + return HttpResponse::NotModified() + .insert_header((ETAG, etag)) + .insert_header((CACHE_CONTROL, cache_control)) + .finish(); + } + + HttpResponse::Ok() + .insert_header((ETAG, etag)) + .insert_header((CACHE_CONTROL, cache_control)) + .content_type("application/json") + .body(body) +} + +/// `GET /api/prism/v1/llm/tools/list` +pub async fn list_tool_registry(request: HttpRequest) -> HttpResponse { + let tools = oss_read_tool_specs() .into_iter() .map(|tool| { json!({ @@ -580,53 +804,168 @@ pub async fn list_tool_registry() -> HttpResponse { }) .collect::>(); - HttpResponse::Ok().json(json!({ "tools": tools })) + registry_http_response(&request, tools) } #[cfg(test)] mod tests { use std::collections::HashSet; - use actix_web::{App, test, web}; + use actix_web::{App, http::header, test, web}; use serde_json::Value; - use super::{list_tool_registry, oss_tool_specs}; + use super::{ + ToolCallRequest, ToolCallResult, ToolEffect, is_oss_read_tool, list_tool_registry, + oss_action_tool_specs, oss_read_tool_specs, oss_tool_specs, + }; #[actix_web::test] async fn oss_catalog_has_unique_names() { let tools = oss_tool_specs(); let names = tools.iter().map(|tool| tool.name).collect::>(); - assert_eq!(tools.len(), 48); + assert_eq!(tools.len(), 47); assert_eq!(tools.len(), names.len()); } #[actix_web::test] async fn registry_api_returns_only_oss_tools() { let app = - test::init_service(App::new().route("/tools", web::get().to(list_tool_registry))).await; + test::init_service(App::new().route("/tools/list", web::get().to(list_tool_registry))) + .await; let response: Value = test::call_and_read_body_json( &app, - test::TestRequest::get().uri("/tools").to_request(), + test::TestRequest::get().uri("/tools/list").to_request(), ) .await; - assert_eq!(response["tools"].as_array().unwrap().len(), 48); + assert_eq!( + response["tools"].as_array().unwrap().len(), + oss_read_tool_specs().len() + ); assert_eq!(response["tools"][0]["name"], "list_datasets"); assert_eq!(response["tools"][0]["inputSchema"]["type"], "object"); + assert!(oss_read_tool_specs().iter().all(|tool| matches!( + tool.effect(), + ToolEffect::ReadOnly | ToolEffect::ExpensiveRead + ))); + assert!( + response["tools"] + .as_array() + .unwrap() + .iter() + .all(|tool| tool["name"] != "create_alert") + ); } #[actix_web::test] - async fn promql_execution_limits_use_numeric_schema_types() { - let tools = oss_tool_specs(); - let promql = tools - .iter() - .find(|tool| tool.name == "query_promql") + async fn read_and_action_catalogs_are_structurally_separated() { + let read_tools = oss_read_tool_specs(); + let action_tools = oss_action_tool_specs(); + + assert_eq!(read_tools.len(), 34); + assert_eq!(action_tools.len(), 13); + assert!(read_tools.iter().all(|tool| matches!( + tool.effect(), + ToolEffect::ReadOnly | ToolEffect::ExpensiveRead + ))); + assert!(action_tools.iter().all(|tool| matches!( + tool.effect(), + ToolEffect::Mutation | ToolEffect::ExternalSideEffect + ))); + assert!(read_tools.iter().all(|tool| is_oss_read_tool(tool.name))); + assert!(action_tools.iter().all(|tool| !is_oss_read_tool(tool.name))); + } + + #[actix_web::test] + async fn registry_api_supports_etag_revalidation() { + let app = + test::init_service(App::new().route("/tools/list", web::get().to(list_tool_registry))) + .await; + let first = test::call_service( + &app, + test::TestRequest::get().uri("/tools/list").to_request(), + ) + .await; + let etag = first.headers().get(header::ETAG).unwrap().clone(); + assert_eq!( + first.headers().get(header::CACHE_CONTROL).unwrap(), + "private, max-age=300, must-revalidate" + ); + + let second = test::call_service( + &app, + test::TestRequest::get() + .uri("/tools/list") + .insert_header((header::IF_NONE_MATCH, etag)) + .to_request(), + ) + .await; + assert_eq!(second.status(), actix_web::http::StatusCode::NOT_MODIFIED); + } + + #[actix_web::test] + async fn oss_catalog_excludes_enterprise_promql_tool() { + assert!( + oss_tool_specs() + .iter() + .all(|tool| tool.name != "query_promql") + ); + } + + #[actix_web::test] + async fn tool_call_defaults_to_empty_arguments() { + let call: ToolCallRequest = serde_json::from_value(serde_json::json!({ + "name": "list_datasets" + })) + .unwrap(); + + assert_eq!(call.arguments, serde_json::json!({})); + } + + #[actix_web::test] + async fn tool_schema_validation_rejects_missing_and_extra_arguments() { + let schema_tool = oss_tool_specs() + .into_iter() + .find(|tool| tool.name == "get_dataset_schema") .unwrap(); - let properties = &promql.input_schema["properties"]; - assert_eq!(properties["timeout"]["type"], "number"); - assert_eq!(properties["limit"]["type"], "integer"); - assert_eq!(properties["limit"]["maximum"], 500); + assert!( + schema_tool + .validate_arguments(&serde_json::json!({})) + .is_err() + ); + assert!( + schema_tool + .validate_arguments(&serde_json::json!({ + "dataset": "logs", + "unexpected": true + })) + .is_err() + ); + assert!( + schema_tool + .validate_arguments(&serde_json::json!({ "dataset": "logs" })) + .is_ok() + ); + } + + #[actix_web::test] + async fn tool_results_use_mcp_call_tool_result_shape() { + let result = serde_json::to_value(ToolCallResult::success(serde_json::json!({ + "rows": [] + }))) + .unwrap(); + + assert_eq!(result["content"][0]["type"], "text"); + assert_eq!( + result["structuredContent"]["result"]["rows"], + serde_json::json!([]) + ); + assert!( + !result["isError"] + .as_bool() + .expect("isError must be a boolean") + ); } } diff --git a/src/tool_executor.rs b/src/tool_executor.rs new file mode 100644 index 000000000..458f249bc --- /dev/null +++ b/src/tool_executor.rs @@ -0,0 +1,984 @@ +use std::collections::HashMap; + +use actix_web::{HttpRequest, HttpResponse, body::to_bytes_limited, http::header::HeaderMap, web}; +use chrono::{DateTime, Duration, Utc}; +use serde_json::{Value, json}; +use ulid::Ulid; + +use crate::{ + handlers::{ + TelemetryType, + http::{ + alerts, cluster, health_check, logstream, + modal::query::querier_logstream, + query, rbac, role, targets, + users::{dashboards, filters}, + }, + }, + parseable::PARSEABLE, + prism::logstream::get_prism_logstream_info, + rbac::{Response, Users, map::SessionKey, role::Action}, + tool_catalog::{ToolCallRequest, ToolCallResult, oss_read_tool_specs}, + utils::{actix::extract_session_key_from_req, get_tenant_id_from_request, time::TimeRange}, +}; + +fn required_string<'a>(arguments: &'a Value, name: &str) -> Result<&'a str, String> { + arguments + .get(name) + .and_then(Value::as_str) + .filter(|value| !value.is_empty()) + .ok_or_else(|| format!("missing required string argument `{name}`")) +} + +fn bounded_integer(arguments: &Value, name: &str, default: u64, min: u64, max: u64) -> u64 { + arguments + .get(name) + .and_then(Value::as_u64) + .unwrap_or(default) + .clamp(min, max) +} + +fn quote_identifier(identifier: &str) -> String { + format!("\"{}\"", identifier.replace('"', "\"\"")) +} + +fn quote_literal(value: &str) -> String { + format!("'{}'", value.replace('\'', "''")) +} + +fn bounded_select(query: &str) -> Result { + let query = query.trim(); + let query = query.strip_suffix(';').unwrap_or(query); + if query.contains(';') { + return Err("exactly one SQL statement is required".to_owned()); + } + let normalized = query.to_ascii_lowercase(); + if !(normalized.starts_with("select ") || normalized.starts_with("with ")) { + return Err("only SQL SELECT queries are allowed".to_owned()); + } + Ok(format!( + "SELECT * FROM ({query}) AS __parseable_tool_query LIMIT 500" + )) +} + +fn visible_datasets(tenant_id: &Option, session_key: &SessionKey) -> Vec { + let mut datasets = PARSEABLE + .streams + .list(tenant_id) + .into_iter() + .filter(|dataset| { + Users.authorize(session_key.clone(), Action::ListStream, Some(dataset), None) + == Response::Authorized + }) + .collect::>(); + datasets.sort_unstable(); + datasets +} + +fn resolve_time_range(arguments: &Value) -> Result { + let expression = required_string(arguments, "expression")?; + let timezone = arguments + .get("timezone") + .and_then(Value::as_str) + .unwrap_or("UTC"); + if !timezone.eq_ignore_ascii_case("utc") { + return Err("OSS time resolution currently requires timezone `UTC`".to_owned()); + } + let now = Utc::now(); + let normalized = expression.trim().to_ascii_lowercase(); + let (start, end) = match normalized.as_str() { + "today" => ( + now.date_naive() + .and_hms_opt(0, 0, 0) + .expect("midnight is valid") + .and_utc(), + now, + ), + "yesterday" => { + let today = now + .date_naive() + .and_hms_opt(0, 0, 0) + .expect("midnight is valid") + .and_utc(); + (today - Duration::days(1), today) + } + _ => { + let duration = normalized + .strip_prefix("last ") + .or_else(|| normalized.strip_prefix("past ")) + .unwrap_or(&normalized); + let duration = if duration.chars().any(|character| character.is_ascii_digit()) { + duration.to_owned() + } else { + format!("1 {duration}") + }; + let duration = humantime::parse_duration(&duration) + .map_err(|error| format!("invalid time expression `{expression}`: {error}"))?; + let duration = Duration::from_std(duration).map_err(|error| error.to_string())?; + (now - duration, now) + } + }; + Ok(json!({ + "expression": expression, + "timezone": "UTC", + "startTime": start.to_rfc3339(), + "endTime": end.to_rfc3339(), + })) +} + +fn nested_string<'a>(value: &'a Value, name: &str) -> Option<&'a str> { + match value { + Value::Object(object) => object + .get(name) + .and_then(Value::as_str) + .or_else(|| object.values().find_map(|value| nested_string(value, name))), + Value::Array(array) => array.iter().find_map(|value| nested_string(value, name)), + _ => None, + } +} + +async fn response_value(response: HttpResponse) -> Result { + let status = response.status(); + let bytes = to_bytes_limited(response.into_body(), 1024 * 1024) + .await + .map_err(|error| format!("tool response body error: {error:?}"))? + .map_err(|error| format!("tool response exceeded 1 MiB: {error}"))?; + let value = serde_json::from_slice(&bytes) + .unwrap_or_else(|_| Value::String(String::from_utf8_lossy(&bytes).into_owned())); + if status.is_success() { + Ok(value) + } else { + Err(format!("handler returned HTTP {status}: {value}")) + } +} + +fn authorize( + session_key: &SessionKey, + action: Action, + resource: Option<&str>, + user: Option<&str>, +) -> Result<(), String> { + if !Users.session_exists(session_key) { + return Err( + "Your session has expired or is no longer valid. Please re-authenticate to access this resource." + .to_owned(), + ); + } + match Users.authorize(session_key.clone(), action, resource, user) { + Response::Authorized => Ok(()), + Response::Suspended(message) => Err(message), + Response::UnAuthorized | Response::ReloadRequired => { + Err(format!("Caller is not authorized to perform {action:?}")) + } + } +} + +#[derive(Clone, Copy)] +struct OssToolContext<'a> { + session_key: &'a SessionKey, + tenant_id: &'a Option, + request_headers: &'a HeaderMap, +} + +fn execute_dataset_resolution_tool( + name: &str, + arguments: &Value, + context: OssToolContext<'_>, +) -> Result { + match name { + "list_datasets" => Ok(visible_datasets(context.tenant_id, context.session_key) + .into_iter() + .map(|name| json!({ "name": name })) + .collect()), + "resolve_dataset" => { + let datasets = visible_datasets(context.tenant_id, context.session_key); + let selected = arguments + .get("selectedDataset") + .or_else(|| arguments.get("contextDataset")) + .and_then(Value::as_str) + .or_else(|| arguments.get("query").and_then(Value::as_str)); + if let Some(selected) = selected + && let Some(dataset) = datasets.iter().find(|dataset| dataset.as_str() == selected) + { + return Ok(json!({ + "status": "resolved", + "datasets": [{ "name": dataset }], + "selectedDatasets": [dataset], + "confidence": 1.0, + "reason": "exact caller-visible dataset name", + })); + } + Ok(json!({ + "status": if datasets.is_empty() { "not_found" } else { "ambiguous" }, + "datasets": [], + "candidates": datasets.into_iter().map(|name| json!({ "name": name })).collect::>(), + "message": "Select one exact caller-visible dataset.", + })) + } + "resolve_time_range" => resolve_time_range(arguments), + _ => unreachable!("dataset resolution dispatcher only receives known tools"), + } +} + +async fn discover_datasets( + arguments: &Value, + context: OssToolContext<'_>, +) -> Result { + let telemetry_type = required_string(arguments, "telemetryType")?; + if !matches!(telemetry_type, "logs" | "metrics" | "traces") { + return Err("telemetryType must be logs, metrics, or traces".to_owned()); + } + let start = DateTime::parse_from_rfc3339(required_string(arguments, "startTime")?) + .map_err(|error| error.to_string())?; + let end = DateTime::parse_from_rfc3339(required_string(arguments, "endTime")?) + .map_err(|error| error.to_string())?; + let limit = bounded_integer(arguments, "limit", 50, 1, 500) as usize; + let mut datasets = Vec::new(); + let mut errors = Vec::new(); + for dataset in visible_datasets(context.tenant_id, context.session_key) { + match get_prism_logstream_info(&dataset, context.tenant_id).await { + Ok(info) => { + let info = serde_json::to_value(info).map_err(|error| error.to_string())?; + let matches_type = nested_string(&info, "telemetryType") + .is_some_and(|value| value.eq_ignore_ascii_case(telemetry_type)); + let active = nested_string(&info, "latestEventAt") + .and_then(|value| DateTime::parse_from_rfc3339(value).ok()) + .is_some_and(|latest| latest >= start && latest <= end); + if matches_type && active { + datasets.push(json!({ "name": dataset, "info": info })); + if datasets.len() == limit { + break; + } + } + } + Err(error) => errors.push(json!({ + "dataset": dataset, + "message": error.to_string(), + })), + } + } + Ok(json!({ + "telemetryType": telemetry_type, + "startTime": start.to_rfc3339(), + "endTime": end.to_rfc3339(), + "datasets": datasets, + "errors": errors, + })) +} + +async fn execute_dataset_metadata_tool( + name: &str, + arguments: &Value, + context: OssToolContext<'_>, +) -> Result { + let dataset = required_string(arguments, "dataset")?; + match name { + "get_dataset_schema" => { + authorize(context.session_key, Action::GetSchema, Some(dataset), None)?; + serde_json::to_value( + logstream::get_schema_internal(dataset, context.tenant_id) + .await + .map_err(|error| error.to_string())?, + ) + .map_err(|error| error.to_string()) + } + "get_dataset_info" => { + authorize( + context.session_key, + Action::GetStreamInfo, + Some(dataset), + None, + )?; + serde_json::to_value( + get_prism_logstream_info(dataset, context.tenant_id) + .await + .map_err(|error| error.to_string())?, + ) + .map_err(|error| error.to_string()) + } + "get_dataset_stats" => { + authorize(context.session_key, Action::GetStats, Some(dataset), None)?; + querier_logstream::get_stats_internal(dataset, context.tenant_id, None) + .await + .map_err(|error| error.to_string()) + } + _ => unreachable!("dataset metadata dispatcher only receives known tools"), + } +} + +async fn sample_events(arguments: &Value, context: OssToolContext<'_>) -> Result { + let dataset = required_string(arguments, "dataset")?; + authorize(context.session_key, Action::Query, Some(dataset), None)?; + let limit = bounded_integer(arguments, "limit", 10, 1, 100); + let minutes = bounded_integer(arguments, "minutes", 60, 1, 1440); + let schema = logstream::get_schema_internal(dataset, context.tenant_id) + .await + .map_err(|error| error.to_string())?; + let fields = schema + .fields() + .iter() + .take(10) + .map(|field| quote_identifier(field.name())) + .collect::>(); + if fields.is_empty() { + return Err("dataset schema has no fields".to_owned()); + } + let end = Utc::now(); + let response = query::query_internal( + query::Query { + query: format!( + "SELECT {} FROM {} LIMIT {limit}", + fields.join(", "), + quote_identifier(dataset) + ), + start_time: (end - Duration::minutes(minutes as i64)).to_rfc3339(), + end_time: end.to_rfc3339(), + send_null: true, + fields: false, + streaming: false, + filter_tags: None, + }, + context.session_key, + context.tenant_id, + ) + .await + .map_err(|error| error.to_string())?; + response_value(response).await +} + +fn normalized_log_context_bounds( + arguments: &Value, + anchor: DateTime, +) -> Result<(String, String), String> { + let start_time = arguments + .get("contextStartTime") + .and_then(Value::as_str) + .map(str::to_owned) + .unwrap_or_else(|| (anchor - Duration::minutes(5)).to_rfc3339()); + let end_time = arguments + .get("contextEndTime") + .and_then(Value::as_str) + .map(str::to_owned) + .unwrap_or_else(|| (anchor + Duration::minutes(5)).to_rfc3339()); + let time_range = TimeRange::parse_human_time(&start_time, &end_time) + .map_err(|error| format!("invalid log context time range: {error}"))?; + Ok((time_range.start.to_rfc3339(), time_range.end.to_rfc3339())) +} + +async fn get_log_context(arguments: &Value, context: OssToolContext<'_>) -> Result { + let dataset = required_string(arguments, "dataset")?; + authorize(context.session_key, Action::Query, Some(dataset), None)?; + let anchor = DateTime::parse_from_rfc3339(required_string(arguments, "pTimestamp")?) + .map_err(|error| error.to_string())? + .with_timezone(&Utc); + let (start_time, end_time) = normalized_log_context_bounds(arguments, anchor)?; + let limit = bounded_integer(arguments, "pageSize", 100, 1, 500); + let schema = logstream::get_schema_internal(dataset, context.tenant_id) + .await + .map_err(|error| error.to_string())?; + let fields = schema + .fields() + .iter() + .take(10) + .map(|field| quote_identifier(field.name())) + .collect::>(); + if fields.is_empty() { + return Err("dataset schema has no fields".to_owned()); + } + let response = query::query_internal( + query::Query { + query: format!( + "SELECT {} FROM {} WHERE {} >= {} AND {} <= {} ORDER BY {} LIMIT {limit}", + fields.join(", "), + quote_identifier(dataset), + quote_identifier("p_timestamp"), + quote_literal(&start_time), + quote_identifier("p_timestamp"), + quote_literal(&end_time), + quote_identifier("p_timestamp"), + ), + start_time, + end_time, + send_null: true, + fields: false, + streaming: false, + filter_tags: None, + }, + context.session_key, + context.tenant_id, + ) + .await + .map_err(|error| error.to_string())?; + response_value(response).await +} + +async fn execute_sql_tool( + name: &str, + arguments: &Value, + context: OssToolContext<'_>, +) -> Result { + authorize(context.session_key, Action::Query, None, None)?; + let query_text = required_string(arguments, "query")?; + let query_text = if name == "explain_query" { + format!("EXPLAIN {}", bounded_select(query_text)?) + } else { + bounded_select(query_text)? + }; + let response = query::query_internal( + query::Query { + query: query_text, + start_time: required_string(arguments, "startTime")?.to_owned(), + end_time: required_string(arguments, "endTime")?.to_owned(), + send_null: true, + fields: false, + streaming: false, + filter_tags: None, + }, + context.session_key, + context.tenant_id, + ) + .await + .map_err(|error| error.to_string())?; + response_value(response).await +} + +async fn execute_alert_tool( + name: &str, + arguments: &Value, + context: OssToolContext<'_>, +) -> Result { + match name { + "list_alerts" => { + authorize(context.session_key, Action::GetAlert, None, None)?; + serde_json::to_value( + alerts::list_internal(context.session_key.clone(), &HashMap::new()) + .await + .map_err(|error| error.to_string())?, + ) + .map_err(|error| error.to_string()) + } + "get_alert" => { + authorize(context.session_key, Action::GetAlert, None, None)?; + let id = Ulid::from_string(required_string(arguments, "alert_id")?) + .map_err(|error| error.to_string())?; + serde_json::to_value( + alerts::get_internal(context.session_key, id, context.tenant_id) + .await + .map_err(|error| error.to_string())?, + ) + .map_err(|error| error.to_string()) + } + "list_alert_tags" => { + authorize(context.session_key, Action::GetAlert, None, None)?; + serde_json::to_value( + alerts::list_tags_internal(context.tenant_id) + .await + .map_err(|error| error.to_string())?, + ) + .map_err(|error| error.to_string()) + } + "enable_alert" | "disable_alert" | "evaluate_alert" => { + authorize(context.session_key, Action::PutAlert, None, None)?; + let id = Ulid::from_string(required_string(arguments, "alert_id")?) + .map_err(|error| error.to_string())?; + let response = match name { + "enable_alert" => { + alerts::enable_alert_internal(context.session_key, id, context.tenant_id).await + } + "disable_alert" => { + alerts::disable_alert_internal(context.session_key, id, context.tenant_id).await + } + _ => { + alerts::evaluate_alert_internal(context.session_key, id, context.tenant_id) + .await + } + } + .map_err(|error| error.to_string())?; + serde_json::to_value(response).map_err(|error| error.to_string()) + } + "create_alert" => create_alert(arguments, context).await, + _ => unreachable!("alert dispatcher only receives known tools"), + } +} + +async fn create_alert(arguments: &Value, context: OssToolContext<'_>) -> Result { + authorize(context.session_key, Action::PutAlert, None, None)?; + let request = serde_json::from_value( + arguments + .get("spec") + .cloned() + .ok_or_else(|| "missing required object argument `spec`".to_owned())?, + ) + .map_err(|error| format!("invalid alert spec: {error}"))?; + serde_json::to_value( + alerts::post_internal(request, context.session_key, context.tenant_id) + .await + .map_err(|error| error.to_string())?, + ) + .map_err(|error| error.to_string()) +} + +async fn execute_alert_target_tool( + name: &str, + arguments: &Value, + context: OssToolContext<'_>, +) -> Result { + match name { + "list_alert_targets" => { + authorize(context.session_key, Action::GetAlert, None, None)?; + serde_json::to_value( + targets::list_internal(context.tenant_id) + .await + .map_err(|error| error.to_string())?, + ) + .map_err(|error| error.to_string()) + } + "get_alert_target" => { + authorize(context.session_key, Action::GetAlert, None, None)?; + let id = Ulid::from_string(required_string(arguments, "target_id")?) + .map_err(|error| error.to_string())?; + targets::get_internal(id, context.tenant_id) + .await + .map_err(|error| error.to_string()) + } + "create_alert_target" => { + authorize(context.session_key, Action::PutAlert, None, None)?; + let target = serde_json::from_value( + arguments + .get("spec") + .cloned() + .ok_or_else(|| "missing required object argument `spec`".to_owned())?, + ) + .map_err(|error| format!("invalid alert target spec: {error}"))?; + targets::post_internal(target, context.tenant_id) + .await + .map_err(|error| error.to_string()) + } + _ => unreachable!("alert target dispatcher only receives known tools"), + } +} + +fn trace_datasets(arguments: &Value, context: OssToolContext<'_>) -> Vec { + arguments + .get("datasets") + .and_then(Value::as_array) + .map(|datasets| { + datasets + .iter() + .filter_map(Value::as_str) + .take(10) + .map(str::to_owned) + .collect::>() + }) + .filter(|datasets| !datasets.is_empty()) + .unwrap_or_else(|| { + visible_datasets(context.tenant_id, context.session_key) + .into_iter() + .filter(|dataset| { + PARSEABLE + .get_stream(dataset, context.tenant_id) + .is_ok_and(|stream| { + stream.metadata.read().is_ok_and(|metadata| { + metadata.telemetry_type == TelemetryType::Traces + }) + }) + }) + .take(10) + .collect() + }) +} + +async fn execute_trace_tool( + name: &str, + arguments: &Value, + context: OssToolContext<'_>, +) -> Result { + let datasets = trace_datasets(arguments, context); + let start_time = required_string(arguments, "startTime")?.to_owned(); + let end_time = required_string(arguments, "endTime")?.to_owned(); + let limit = bounded_integer(arguments, "limit", 100, 1, 500) as usize; + let mut results = Vec::new(); + let mut errors = Vec::new(); + for dataset in datasets { + if let Err(message) = authorize(context.session_key, Action::Query, Some(&dataset), None) { + errors.push(json!({ "dataset": dataset, "message": message })); + continue; + } + let response = if name == "get_trace" { + crate::handlers::http::traces::get_trace_detail_internal( + crate::handlers::http::traces::TraceDetailRequest { + dataset: dataset.clone(), + trace_id: required_string(arguments, "traceId")?.to_owned(), + start_time: start_time.clone(), + end_time: end_time.clone(), + }, + context.tenant_id, + context.session_key, + Some(context.request_headers.clone()), + ) + .await + } else { + crate::handlers::http::traces::list_traces_internal( + crate::handlers::http::traces::TraceListRequest { + dataset: dataset.clone(), + service_name: arguments + .get("serviceName") + .and_then(Value::as_str) + .map(str::to_owned), + start_time: start_time.clone(), + end_time: end_time.clone(), + sort_by: None, + conditions: None, + options: None, + limit: Some(limit), + offset: Some(0), + }, + context.tenant_id, + context.session_key, + Some(context.request_headers.clone()), + ) + .await + }; + match response { + Ok(response) => results.push(json!({ "dataset": dataset, "response": response })), + Err(error) => errors.push(json!({ + "dataset": dataset, + "message": error.to_string(), + })), + } + } + if results.is_empty() { + return Err(format!("all selected datasets failed: {errors:?}")); + } + Ok(json!({ "results": results, "errors": errors })) +} + +async fn execute_identity_tool( + name: &str, + arguments: &Value, + context: OssToolContext<'_>, +) -> Result { + match name { + "list_users" => { + authorize(context.session_key, Action::ListUser, None, None)?; + Ok(rbac::list_users_internal(context.tenant_id)) + } + "get_user" => { + let user = required_string(arguments, "userid")?; + authorize(context.session_key, Action::GetUserRoles, None, Some(user))?; + serde_json::to_value( + rbac::get_user_internal(user, context.tenant_id) + .map_err(|error| error.to_string())?, + ) + .map_err(|error| error.to_string()) + } + "get_user_roles" => { + let user = required_string(arguments, "userid")?; + authorize(context.session_key, Action::GetUserRoles, None, Some(user))?; + rbac::get_role_internal(user, context.tenant_id).map_err(|error| error.to_string()) + } + "list_roles" => { + authorize(context.session_key, Action::ListRole, None, None)?; + serde_json::to_value( + role::list_internal(context.tenant_id) + .await + .map_err(|error| error.to_string())?, + ) + .map_err(|error| error.to_string()) + } + "get_role" => { + authorize(context.session_key, Action::GetRole, None, None)?; + serde_json::to_value( + role::get_internal(required_string(arguments, "name")?, context.tenant_id) + .await + .map_err(|error| error.to_string())?, + ) + .map_err(|error| error.to_string()) + } + "get_default_role" => { + authorize(context.session_key, Action::GetRole, None, None)?; + Ok(role::get_default_internal(context.tenant_id)) + } + _ => unreachable!("identity dispatcher only receives known tools"), + } +} + +async fn execute_system_tool( + name: &str, + arguments: &Value, + context: OssToolContext<'_>, +) -> Result { + match name { + "ping" => { + authorize(context.session_key, Action::GetAbout, None, None)?; + Ok(json!({ + "about": crate::handlers::http::about::about().await.into_inner(), + "liveness": { "status": health_check::liveness_internal().as_u16() }, + "readiness": { + "status": health_check::readiness_internal(context.tenant_id).await.as_u16() + }, + })) + } + "get_cluster_status" => { + authorize(context.session_key, Action::ListCluster, None, None)?; + serde_json::to_value( + cluster::get_cluster_info_internal(context.tenant_id) + .await + .map_err(|error| error.to_string())?, + ) + .map_err(|error| error.to_string()) + } + "get_cluster_metrics" => { + authorize(context.session_key, Action::ListClusterMetrics, None, None)?; + serde_json::to_value( + cluster::get_cluster_metrics_internal(context.tenant_id) + .await + .map_err(|error| error.to_string())?, + ) + .map_err(|error| error.to_string()) + } + "get_retention" => { + let dataset = required_string(arguments, "dataset")?; + authorize( + context.session_key, + Action::GetRetention, + Some(dataset), + None, + )?; + serde_json::to_value( + logstream::get_retention_internal(dataset, context.tenant_id) + .await + .map_err(|error| error.to_string())?, + ) + .map_err(|error| error.to_string()) + } + "get_hot_tier_config" => get_hot_tier_config(arguments, context).await, + _ => unreachable!("system dispatcher only receives known tools"), + } +} + +async fn get_hot_tier_config( + arguments: &Value, + context: OssToolContext<'_>, +) -> Result { + let dataset = required_string(arguments, "dataset")?; + authorize( + context.session_key, + Action::GetHotTierEnabled, + Some(dataset), + None, + )?; + serde_json::to_value( + logstream::get_stream_hot_tier_internal(dataset, context.tenant_id) + .await + .map_err(|error| error.to_string())?, + ) + .map_err(|error| error.to_string()) +} + +async fn execute_saved_object_tool( + name: &str, + arguments: &Value, + context: OssToolContext<'_>, +) -> Result { + match name { + "list_filters" => { + authorize(context.session_key, Action::ListFilter, None, None)?; + serde_json::to_value(filters::list_internal(context.session_key).await) + .map_err(|error| error.to_string()) + } + "get_filter" => { + authorize(context.session_key, Action::GetFilter, None, None)?; + serde_json::to_value( + filters::get_internal( + required_string(arguments, "id")?, + context.session_key, + context.tenant_id, + ) + .await + .map_err(|error| error.to_string())?, + ) + .map_err(|error| error.to_string()) + } + "list_dashboard_tags" => { + authorize(context.session_key, Action::ListDashboard, None, None)?; + serde_json::to_value(dashboards::list_tags_internal(context.tenant_id).await) + .map_err(|error| error.to_string()) + } + "get_dashboard" => { + authorize(context.session_key, Action::GetDashboard, None, None)?; + let id = Ulid::from_string(required_string(arguments, "dashboardId")?) + .map_err(|error| error.to_string())?; + serde_json::to_value( + dashboards::get_dashboard_internal(id, context.tenant_id) + .await + .map_err(|error| error.to_string())?, + ) + .map_err(|error| error.to_string()) + } + "list_dashboards" => { + authorize(context.session_key, Action::ListDashboard, None, None)?; + let limit = bounded_integer(arguments, "limit", 100, 1, 500) as usize; + Ok( + dashboards::list_dashboards_internal(limit, None, context.tenant_id) + .await + .into(), + ) + } + _ => unreachable!("saved object dispatcher only receives known tools"), + } +} + +async fn execute_oss_tool_inner( + name: &str, + arguments: &Value, + session_key: &SessionKey, + tenant_id: &Option, + request_headers: &HeaderMap, +) -> Result { + let context = OssToolContext { + session_key, + tenant_id, + request_headers, + }; + match name { + "list_datasets" | "resolve_dataset" | "resolve_time_range" => { + execute_dataset_resolution_tool(name, arguments, context) + } + "discover_datasets" => discover_datasets(arguments, context).await, + "get_dataset_schema" | "get_dataset_info" | "get_dataset_stats" => { + execute_dataset_metadata_tool(name, arguments, context).await + } + "sample_events" => sample_events(arguments, context).await, + "get_log_context" => get_log_context(arguments, context).await, + "query_sql" | "explain_query" => execute_sql_tool(name, arguments, context).await, + "list_alerts" | "get_alert" | "list_alert_tags" | "enable_alert" | "disable_alert" + | "evaluate_alert" | "create_alert" => execute_alert_tool(name, arguments, context).await, + "list_alert_targets" | "get_alert_target" | "create_alert_target" => { + execute_alert_target_tool(name, arguments, context).await + } + "get_traces" | "get_trace" => execute_trace_tool(name, arguments, context).await, + "ping" + | "get_cluster_status" + | "get_cluster_metrics" + | "get_retention" + | "get_hot_tier_config" => execute_system_tool(name, arguments, context).await, + "list_users" | "get_user" | "get_user_roles" | "list_roles" | "get_role" + | "get_default_role" => execute_identity_tool(name, arguments, context).await, + "list_filters" + | "get_filter" + | "list_dashboard_tags" + | "get_dashboard" + | "list_dashboards" => execute_saved_object_tool(name, arguments, context).await, + _ => Err(format!( + "tool `{name}` is registered but does not yet have an OSS internal executor" + )), + } +} + +/// Generic execution endpoint used by external MCP adapters. +pub async fn call_tool_registry( + req: HttpRequest, + web::Json(call): web::Json, +) -> HttpResponse { + if !call.arguments.is_object() { + return HttpResponse::BadRequest().json(ToolCallResult::error( + "tool arguments must be a JSON object", + )); + } + let session_key = match extract_session_key_from_req(&req) { + Ok(session_key) => session_key, + Err(error) => { + return HttpResponse::Unauthorized().json(ToolCallResult::error(error.to_string())); + } + }; + if !Users.session_exists(&session_key) { + return HttpResponse::Unauthorized().json(ToolCallResult::error( + "Your session has expired or is no longer valid. Please re-authenticate to access this resource.", + )); + } + let Some(spec) = oss_read_tool_specs() + .into_iter() + .find(|tool| tool.name == call.name) + else { + return HttpResponse::NotFound().json(ToolCallResult::error("unknown tool")); + }; + if let Err(error) = spec.validate_arguments(&call.arguments) { + return HttpResponse::BadRequest().json(ToolCallResult::error(error)); + } + let tenant_id = get_tenant_id_from_request(&req); + match execute_oss_tool_inner( + spec.name, + &call.arguments, + &session_key, + &tenant_id, + req.headers(), + ) + .await + { + Ok(result) => HttpResponse::Ok().json(ToolCallResult::success(result)), + Err(message) => HttpResponse::Ok().json(ToolCallResult::error(message)), + } +} + +#[cfg(test)] +mod tests { + use actix_web::{App, http::StatusCode, test, web}; + use chrono::{DateTime, Duration, Utc}; + use serde_json::{Value, json}; + + use super::{call_tool_registry, normalized_log_context_bounds}; + + #[actix_web::test] + async fn log_context_bounds_normalize_relative_times() { + let (start, end) = normalized_log_context_bounds( + &json!({ + "contextStartTime": "now-15m", + "contextEndTime": "now", + }), + Utc::now(), + ) + .expect("relative bounds must be accepted"); + let start = start.parse::>().expect("valid start time"); + let end = end.parse::>().expect("valid end time"); + + assert_eq!(end - start, Duration::minutes(15)); + } + + #[actix_web::test] + async fn invocation_endpoint_requires_authentication() { + let app = + test::init_service(App::new().route("/tools/call", web::post().to(call_tool_registry))) + .await; + let request = test::TestRequest::post() + .uri("/tools/call") + .set_json(json!({ "name": "list_datasets", "arguments": {} })) + .to_request(); + let response = test::call_service(&app, request).await; + + assert_eq!(response.status(), StatusCode::UNAUTHORIZED); + } + + #[actix_web::test] + async fn arguments_must_be_an_object() { + let app = + test::init_service(App::new().route("/tools/call", web::post().to(call_tool_registry))) + .await; + let request = test::TestRequest::post() + .uri("/tools/call") + .set_json(json!({ "name": "list_datasets", "arguments": [] })) + .to_request(); + let response: Value = test::call_and_read_body_json(&app, request).await; + + assert!( + response["isError"] + .as_bool() + .expect("isError must be a boolean") + ); + assert_eq!(response["content"][0]["type"], "text"); + assert_eq!( + response["content"][0]["text"], + "tool arguments must be a JSON object" + ); + } +}