diff --git a/rust/crates/sift_cli/assets/skills/sift/SKILL.md b/rust/crates/sift_cli/assets/skills/sift/SKILL.md index 3a7c4fb71a..7be2260a29 100644 --- a/rust/crates/sift_cli/assets/skills/sift/SKILL.md +++ b/rust/crates/sift_cli/assets/skills/sift/SKILL.md @@ -49,8 +49,11 @@ Sift-entity request always wins. - **None:** lookups, explanations, how-to answers, and one-off numbers or short analysis. Do not wrap a chat-sized answer in a file. -- **Explore links:** plots and timeseries the user wants to inspect. Use native - Sift app visualizations instead of creating an intermediate image or HTML. +- **Declarative charts:** plots, charts, and timeseries the user wants to see. + Build or pass a spec to `create_declarative_chart`. +- **Explore links:** quick deep-links to specific assets, runs, or channels the + user wants to inspect in the Sift web app. Use native Sift app + visualizations instead of creating an intermediate image or HTML. - **Calculated channels:** reusable CEL transforms (unit conversions, rolling windows, derived signals, and simple filters) that should be plottable or usable by rules on later runs. List existing calculated channels first to @@ -109,6 +112,7 @@ tool. to enable them, then restart the MCP client. - **Data:** `get_data` writes channel data to a Parquet file. `sql` queries Parquet files. `upload_dataset` streams a Parquet dataset into Sift. +- **Charts:** `create_declarative_chart`. - **Links:** `explore_url`. - **Docs:** `search_docs`. - **Writes:** `create_rule`, `update_rule`, `archive_rule`, `unarchive_rule`, @@ -224,8 +228,10 @@ tool. listings without deleting its versions, links, or files. Find archived artifacts with `list_artifacts` and `include_archived=true`, then restore one with `unarchive_artifact`. Both writes need `--allow-destructive`. -- **Produce a chart.** Build a link with `explore_url`. When the user wants a - chart and numbers, do both and give the user both. +- **Produce a chart.** Use `create_declarative_chart` with a spec (see + references/declarative-charts.md). For quick links to Sift Explore, use + `explore_url` instead. When the user wants a chart and numbers, do both and + give the user both. - **Answer a question about how Sift works.** Call `search_docs`. Do not answer from memory, and cite the page you used. - **Create an asset.** There is no `create_asset` MCP tool because Sift creates @@ -254,9 +260,11 @@ tool. - **Stop on empty list results.** If any `list_*` tool returns no items, tell the user that nothing matched and ask how to proceed. Do not make subsequent tool calls using an empty or placeholder name or id from the missing result. -- **Surface URLs as plain text, in full.** The link from `explore_url` and the +- **Surface tool-returned URLs; never invent one.** The link from `explore_url` and the `View in Sift:` line from `sift-cli import` are deliverables. Never invent a - URL that a tool did not return. + URL that a tool did not return. Chart links from `create_declarative_chart` + follow the tool's `next_step` (render as a clickable markdown link with + descriptive text, mentioned once). - **Confirm every write before you run it.** Show the user the proposed change and its target, then wait for approval. - **Write tools are off by default.** Read-only is the default access mode. @@ -291,6 +299,7 @@ Read the file that matches the task in front of you. Do not read them all. | When the task is | Read | |---|---| | any `sift-cli` invocation: import, export, config | [references/cli.md](references/cli.md) | -| a chart, a plot, or an Explore link | [references/explore-links.md](references/explore-links.md) | +| a declarative chart spec | [references/declarative-charts.md](references/declarative-charts.md) | +| a quick Explore link | [references/explore-links.md](references/explore-links.md) | | install, update, or diagnose the Sift integration | [references/agent-setup.md](references/agent-setup.md) | | code written against Sift's libraries or REST API | [references/integration-code.md](references/integration-code.md) | diff --git a/rust/crates/sift_cli/assets/skills/sift/references/declarative-charts.md b/rust/crates/sift_cli/assets/skills/sift/references/declarative-charts.md new file mode 100644 index 0000000000..e0eed1702e --- /dev/null +++ b/rust/crates/sift_cli/assets/skills/sift/references/declarative-charts.md @@ -0,0 +1,111 @@ +# Declarative charts + +Use `create_declarative_chart` to validate a complete version 0 YAML or JSON spec and get a Sift link that opens it. +One chart is recommended: a spec with exactly one chart displays inline in Sift agent chat. +Specs with several charts or a top-level `layout` are allowed and still return a working Explore link. + +A valid result includes `exploreUrl`. Render it once as a clickable markdown link with descriptive text, such as what the chart plots. Do not paste the spec into your reply. + +## User supplies a spec + +If the user pastes or attaches a complete spec, pass it to `create_declarative_chart` verbatim. When it returns `invalid_spec` with reported paths, fix each path (or ask the user when the fix needs their intent) and resend the full spec. Repeat until valid or 3 attempts are exhausted, then explain the remaining issues to the user. + +## Plain-language request + +1. Resolve runs with Sift MCP `list_runs` and channels with `list_channels`. Resolve calculated channels with `list_calculated_channels` and bind them with `calculatedChannelId`, not `channel`. Never invent IDs or channel names. +2. Write a complete version 0 spec with `dataSources` and one `chart`. Prefer YAML. Use one chart unless the user asked for several. +3. Call `create_declarative_chart` with the full spec. +4. On `invalid_spec`, fix every reported path and resend the full corrected spec. +5. After 3 failed attempts, stop and summarize the remaining validation problems instead of retrying. +6. To edit a chart, resend the full edited spec. Each valid call returns a new `exploreUrl`. + +## Minimal spec + +Replace the illustrative ID and channel with values from Sift MCP. + +```yaml +version: 0 +dataSources: + flight: { run: { id: 7f3a2b10-0000-4000-8000-000000000001 } } +chart: + type: timeseries + series: [{ channel: pressure, source: flight, axis: L1 }] +``` + +## Checklist + +- [ ] IDs and channel names came from Sift MCP results. +- [ ] The complete version 0 spec has one chart unless the user asked for several or a `layout`. +- [ ] Each channel series has exactly one of `channel`, `channelId`, `calculatedChannelId`, or `match`. +- [ ] Encoding series have every required coordinate role. +- [ ] Asset-only charts include an explicit `timeInterval`. +- [ ] Every reported path is fixed before resending the full spec. +- [ ] After 3 failed attempts, explain the remaining problems instead of retrying. +- [ ] Optional `sampling`, `maxGap`, and axis `min`/`max` are omitted unless the user asked for them. + +## Schema reference + +Validation errors are the source of truth for allowed keys and constraints. When keys fail validation, read the reported paths and messages to understand what is allowed. The tool accepts `version: 0` YAML or JSON. A single chart uses `chart` or one entry under `charts` and displays inline in Sift agent chat. Several charts or a top-level `layout` are accepted and return an Explore link. + +## Run and inline expression example + +```yaml +version: 0 +dataSources: + flight: { run: { id: 7f3a2b10-0000-4000-8000-000000000001 } } +transforms: + power: + expr: '$1 * $2' + inputs: + $1: { channel: voltage } + $2: { channel: current } + unit: W +chart: + type: timeseries + title: Electrical power + sources: [flight] + timeMode: absolute + yAxis: + L1: { label: Power, scale: value } + series: + - channel: power + source: flight + axis: L1 + style: { line: { color: ruby } } +``` + +## Asset and explicit interval example + +```yaml +version: 0 +dataSources: + pad: { asset: { id: 5d10c4a0-0000-4000-8000-000000000002 } } +timeInterval: + start: '2024-06-01T00:00:00Z' + end: '2024-06-01T00:10:00Z' +chart: + type: timeseries + title: Pad temperature + timeMode: absolute + series: + - match: { name: { matches: '^temp_.*' } } + source: pad + axis: R1 + yAxis: + R1: { label: Temperature, scale: value } +``` + +## Calculated channel series example + +```yaml +version: 0 +dataSources: + flight: { run: { id: 7f3a2b10-0000-4000-8000-000000000001 } } +chart: + type: timeseries + series: + - calculatedChannelId: 1a2b3c4d-0000-4000-8000-000000000002 + source: flight + axis: L1 + label: Filtered pressure +``` diff --git a/rust/crates/sift_cli/assets/skills/sift/references/explore-links.md b/rust/crates/sift_cli/assets/skills/sift/references/explore-links.md index ab571b87d8..573b63a5ca 100644 --- a/rust/crates/sift_cli/assets/skills/sift/references/explore-links.md +++ b/rust/crates/sift_cli/assets/skills/sift/references/explore-links.md @@ -1,5 +1,7 @@ # Sift Explore links +To plot, chart, graph, or visualize data, use `create_declarative_chart` with a spec (see [references/declarative-charts.md](declarative-charts.md)). Use `explore_url` to build a quick deep-link that opens specific assets, runs, or channels. + Build the link with `explore_url`, then surface the URL to the user as plain text, in full. diff --git a/rust/crates/sift_mcp/src/server/mod.rs b/rust/crates/sift_mcp/src/server/mod.rs index cb36815006..0cb44df57c 100644 --- a/rust/crates/sift_mcp/src/server/mod.rs +++ b/rust/crates/sift_mcp/src/server/mod.rs @@ -48,7 +48,7 @@ pub(crate) const BASE_INSTRUCTIONS: &str = concat!( use crate::service::{ annotations::AnnotationService, artifacts::ArtifactService, assets::AssetService, calculated_channels::CalculatedChannelService, channels::ChannelService, data::DataService, - docs::DocsService, ingest::IngestService, ping::PingService, + declarative::DeclarativeService, docs::DocsService, ingest::IngestService, ping::PingService, report_templates::ReportTemplateService, reports::ReportService, rule_evaluation::RuleEvaluationService, rules::RuleService, runs::RunService, test_reports::TestReportService, url::UrlService, @@ -66,6 +66,7 @@ pub struct SiftMcpServer { pub calculated_channel_service: CalculatedChannelService, pub channel_service: ChannelService, pub data_service: DataService, + pub declarative_service: DeclarativeService, pub url_service: UrlService, pub ingest_service: IngestService, pub ping_service: PingService, @@ -273,6 +274,7 @@ impl SiftMcpServer { let annotation_service = AnnotationService::new(channel.clone(), retry_policy.clone()); let mut artifact_service = ArtifactService::new(channel.clone(), retry_policy.clone()); + let declarative_service = DeclarativeService::new(rest_config.clone(), &cli_version); if let Some(rest_config) = rest_config { artifact_service = artifact_service.with_uploader( crate::service::remote_files::RemoteFileUploader::new(rest_config, &cli_version), @@ -306,6 +308,7 @@ impl SiftMcpServer { calculated_channel_service, channel_service, data_service, + declarative_service, url_service, ingest_service, ping_service, diff --git a/rust/crates/sift_mcp/src/service/declarative/mod.rs b/rust/crates/sift_mcp/src/service/declarative/mod.rs new file mode 100644 index 0000000000..4ee3b31e03 --- /dev/null +++ b/rust/crates/sift_mcp/src/service/declarative/mod.rs @@ -0,0 +1,388 @@ +use std::time::Duration; + +use anyhow::{Context, Result, anyhow, bail}; +use reqwest::{StatusCode, header::USER_AGENT}; +use serde::{Deserialize, Serialize}; +use tokio::time::sleep; + +use crate::service::remote_files::RestConfig; + +const VALIDATE_PATH: &str = "/api/v1/declarative:validate"; +const MIGRATE_PATH: &str = "/api/v1/declarative:migrate"; +const SHARE_LINK_PATH: &str = "/api/v1/declarative:sharelink"; +/// Longest slice of an error body echoed back in a failure or invalid-spec +/// message. +const MAX_ERROR_BODY_CHARS: usize = 500; +const INVALID_SPEC_FALLBACK: &str = "declarative spec failed validation"; +const DEFAULT_MAX_ATTEMPTS: u32 = 3; +const DEFAULT_RETRY_DELAY: Duration = Duration::from_millis(250); +const DEFAULT_REQUEST_TIMEOUT: Duration = Duration::from_secs(10); + +/// One validation or migration finding, located by `path` within the spec. +#[derive(Clone, Debug, Default, PartialEq, Eq, Deserialize)] +pub struct Issue { + #[serde(default)] + pub path: String, + #[serde(default)] + pub message: String, +} + +#[derive(Clone, Debug, Default, PartialEq, Eq)] +pub struct ValidationResult { + pub valid: bool, + pub spec: String, + pub errors: Vec, + pub warnings: Vec, +} + +#[derive(Clone, Debug, Default, PartialEq, Eq)] +pub struct MigrationResult { + pub valid: bool, + pub original_spec: String, + pub migrated_spec: String, + pub source_version: String, + pub target_version: String, + pub errors: Vec, + pub warnings: Vec, +} + +/// Outcome of a share-link request. A spec the backend rejects is `Invalid`, +/// not an error; only transport, HTTP, and malformed-response failures are +/// `Err`. +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum ShareLinkResult { + Ok { short_link: String }, + Invalid { message: String }, +} + +/// Retry and timeout behavior shared by every declarative REST call. A call +/// is retried on transport errors, timeouts, and HTTP 408, 429, and >=500, +/// waiting `retry_delay * 2^(attempt - 1)` between attempts. The crate's +/// `with_retry` only understands gRPC codes, so REST has its own. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct RestRetry { + pub max_attempts: u32, + pub retry_delay: Duration, + pub request_timeout: Duration, +} + +impl Default for RestRetry { + fn default() -> Self { + Self { + max_attempts: DEFAULT_MAX_ATTEMPTS, + retry_delay: DEFAULT_RETRY_DELAY, + request_timeout: DEFAULT_REQUEST_TIMEOUT, + } + } +} + +#[derive(Serialize)] +struct ValidateRequest<'a> { + spec: &'a str, + strict: bool, +} + +#[derive(Serialize)] +struct MigrateRequest<'a> { + spec: &'a str, +} + +#[derive(Serialize)] +struct ShareLinkRequest<'a> { + spec: &'a str, +} + +// Proto3 JSON omits fields at their default value, so every field defaults. +#[derive(Deserialize, Default)] +#[serde(default, rename_all = "camelCase")] +struct ValidateResponse { + valid: bool, + spec: String, + errors: Vec, + warnings: Vec, +} + +#[derive(Deserialize, Default)] +#[serde(default, rename_all = "camelCase")] +struct MigrateResponse { + original_spec: String, + migrated_spec: String, + source_version: String, + target_version: String, + warnings: Vec, + errors: Vec, + valid: bool, +} + +#[derive(Deserialize, Default)] +#[serde(default, rename_all = "camelCase")] +struct ShareLinkResponse { + short_link: String, +} + +#[derive(Deserialize)] +struct InvalidSpecBody { + message: Option, +} + +/// Validates and migrates declarative specs through the backend's REST +/// gateway. The declarative protos are not part of `sift_rs`, so there is no +/// gRPC client. The backend is the authority on the schema: specs are passed +/// through verbatim and never parsed here. +#[derive(Clone)] +pub struct DeclarativeService { + // Absent only when the server runs without a REST endpoint (some tests). + client: Option, + retry: RestRetry, +} + +#[derive(Clone)] +struct RestClient { + http: reqwest::Client, + base_uri: String, + api_key: String, + user_agent: String, +} + +impl DeclarativeService { + pub fn new(rest_config: Option, cli_version: &str) -> Self { + let client = rest_config.map(|config| RestClient { + http: reqwest::Client::new(), + base_uri: config.rest_uri.trim_end_matches('/').to_owned(), + api_key: config.api_key, + user_agent: format!("{}/{cli_version}", config.client_name.as_str()), + }); + Self { + client, + retry: RestRetry::default(), + } + } + + /// Overrides the attempt count, backoff base, and per-request timeout. + #[allow(dead_code)] // Only tests override the defaults today. + pub fn with_retry(mut self, retry: RestRetry) -> Self { + self.retry = retry; + self + } + + /// Validates `spec` in strict mode. A spec that fails validation is an + /// `Ok` result with `valid == false` and its `errors`, including an + /// HTTP 400 (an empty spec), which yields one whole-spec error. Only + /// transport and other HTTP failures are `Err`. + pub async fn validate(&self, spec: &str) -> Result { + let resp: ValidateResponse = match self + .post( + VALIDATE_PATH, + &ValidateRequest { spec, strict: true }, + "validate", + ) + .await? + { + Ok(resp) => resp, + Err(message) => { + return Ok(ValidationResult { + errors: vec![whole_spec_issue(message)], + ..Default::default() + }); + } + }; + Ok(ValidationResult { + valid: resp.valid, + spec: resp.spec, + errors: resp.errors, + warnings: resp.warnings, + }) + } + + /// Migrates `spec` to the latest schema version. As with `validate`, an + /// HTTP 400 is an `Ok` result with `valid == false` and one whole-spec + /// error. + pub async fn migrate(&self, spec: &str) -> Result { + let resp: MigrateResponse = match self + .post(MIGRATE_PATH, &MigrateRequest { spec }, "migrate") + .await? + { + Ok(resp) => resp, + Err(message) => { + return Ok(MigrationResult { + errors: vec![whole_spec_issue(message)], + ..Default::default() + }); + } + }; + Ok(MigrationResult { + valid: resp.valid, + original_spec: resp.original_spec, + migrated_spec: resp.migrated_spec, + source_version: resp.source_version, + target_version: resp.target_version, + errors: resp.errors, + warnings: resp.warnings, + }) + } + + /// Creates a share link for `spec`. HTTP 400 means the backend rejected + /// the spec and is returned as `ShareLinkResult::Invalid`. + pub async fn create_share_link(&self, spec: &str) -> Result { + let action = "share link"; + let response = self + .send( + SHARE_LINK_PATH, + &ShareLinkRequest { spec }, + action, + Some(StatusCode::BAD_REQUEST), + ) + .await?; + + if response.status() == StatusCode::BAD_REQUEST { + let body = truncate(response.text().await.unwrap_or_default()); + return Ok(ShareLinkResult::Invalid { + message: invalid_spec_message(&body), + }); + } + + let resp: ShareLinkResponse = decode(response, action).await?; + if resp.short_link.is_empty() { + bail!("declarative share link response did not include shortLink"); + } + Ok(ShareLinkResult::Ok { + short_link: resp.short_link, + }) + } + + /// POSTs `body` and decodes the response. HTTP 400 is the backend + /// rejecting an empty or malformed request, so it comes back as + /// `Err(message)` for the caller to report as an invalid spec. + async fn post( + &self, + path: &str, + body: &B, + action: &str, + ) -> Result> + where + B: Serialize, + R: for<'de> Deserialize<'de>, + { + let response = self + .send(path, body, action, Some(StatusCode::BAD_REQUEST)) + .await?; + if response.status() == StatusCode::BAD_REQUEST { + let body = truncate(response.text().await.unwrap_or_default()); + return Ok(Err(invalid_spec_message(&body))); + } + decode(response, action).await.map(Ok) + } + + /// POSTs `body` under the retry policy. Returns the first successful + /// response, or one whose status is `accepted_status`. Any other status + /// is an error, retried only when transient. + async fn send( + &self, + path: &str, + body: &B, + action: &str, + accepted_status: Option, + ) -> Result { + let client = self.client.as_ref().context( + "this server was started without a REST endpoint, so declarative specs cannot be checked", + )?; + let url = format!("{}{path}", client.base_uri); + let max_attempts = self.retry.max_attempts.max(1); + + let mut attempt = 1; + loop { + let result = client + .http + .post(&url) + .timeout(self.retry.request_timeout) + .bearer_auth(&client.api_key) + .header(USER_AGENT, &client.user_agent) + .json(body) + .send() + .await; + + let error = match result { + Err(error) if error.is_timeout() => anyhow!(error).context(format!( + "declarative {action} request timed out after {}ms", + self.retry.request_timeout.as_millis() + )), + Err(error) => anyhow!(error) + .context(format!("failed to reach the declarative {action} endpoint")), + Ok(response) => { + let status = response.status(); + if status.is_success() || Some(status) == accepted_status { + return Ok(response); + } + let detail = truncate(response.text().await.unwrap_or_default()); + let error = anyhow!("declarative {action} returned HTTP {status}: {detail}"); + if !is_retryable_status(status) { + return Err(error); + } + error + } + }; + + if attempt >= max_attempts { + return Err(error.context(format!( + "declarative {action} failed after {max_attempts} attempts" + ))); + } + sleep(backoff(self.retry.retry_delay, attempt)).await; + attempt += 1; + } + } +} + +async fn decode(response: reqwest::Response, action: &str) -> Result +where + R: for<'de> Deserialize<'de>, +{ + response + .json() + .await + .with_context(|| format!("failed to decode the declarative {action} response")) +} + +/// An issue that applies to the spec as a whole (empty path). +fn whole_spec_issue(message: String) -> Issue { + Issue { + path: String::new(), + message, + } +} + +fn is_retryable_status(status: StatusCode) -> bool { + status == StatusCode::REQUEST_TIMEOUT + || status == StatusCode::TOO_MANY_REQUESTS + || status.as_u16() >= 500 +} + +/// `base * 2^(attempt - 1)`, where `attempt` is the one that just failed. +fn backoff(base: Duration, attempt: u32) -> Duration { + let factor = 1u32.checked_shl(attempt - 1).unwrap_or(u32::MAX); + base.saturating_mul(factor) +} + +fn truncate(body: String) -> String { + match body.char_indices().nth(MAX_ERROR_BODY_CHARS) { + Some((end, _)) => body[..end].to_owned(), + None => body, + } +} + +/// Prefers the backend's JSON `message`, then the raw (already truncated) +/// body, then a generic fallback. +fn invalid_spec_message(body: &str) -> String { + if body.is_empty() { + return INVALID_SPEC_FALLBACK.to_owned(); + } + match serde_json::from_str::(body) { + Ok(InvalidSpecBody { + message: Some(message), + }) if !message.is_empty() => message, + _ => body.to_owned(), + } +} + +#[cfg(test)] +pub(crate) mod test; diff --git a/rust/crates/sift_mcp/src/service/declarative/test.rs b/rust/crates/sift_mcp/src/service/declarative/test.rs new file mode 100644 index 0000000000..8e413c6eba --- /dev/null +++ b/rust/crates/sift_mcp/src/service/declarative/test.rs @@ -0,0 +1,622 @@ +use std::time::Duration; + +use tokio::{ + io::{AsyncReadExt, AsyncWriteExt}, + net::TcpListener, + task::JoinHandle, +}; + +use super::{DeclarativeService, Issue, RestRetry, ShareLinkResult}; +use crate::client_event::start_http_server; +use crate::service::remote_files::RestConfig; + +pub(crate) fn http_response(status: &str, body: &str) -> Vec { + format!( + "HTTP/1.1 {status}\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{body}", + body.len() + ) + .into_bytes() +} + +/// Tiny backoff so retry tests finish quickly. +pub(crate) fn fast_retry() -> RestRetry { + RestRetry { + max_attempts: 3, + retry_delay: Duration::from_millis(1), + request_timeout: Duration::from_secs(5), + } +} + +fn service(rest_uri: String) -> DeclarativeService { + service_with_retry(rest_uri, fast_retry()) +} + +fn service_with_retry(rest_uri: String, retry: RestRetry) -> DeclarativeService { + DeclarativeService::new(Some(RestConfig::new(rest_uri, "test-key".into())), "1.2.3") + .with_retry(retry) +} + +/// How the sequence server answers one connection. +pub(crate) enum Reply { + Respond(Vec), + /// Read the request, then never answer, so the client times out. + Hang, +} + +/// Serves one `Reply` per incoming connection, in order, then returns every +/// request it read. `start_http_server` handles a single connection, which is +/// not enough to observe a retry. +pub(crate) async fn start_sequence_server( + replies: Vec, +) -> (String, JoinHandle>>) { + fn request_complete(request: &[u8]) -> bool { + let Some(header_end) = request.windows(4).position(|window| window == b"\r\n\r\n") else { + return false; + }; + let headers = String::from_utf8_lossy(&request[..header_end]); + let content_length = headers.lines().find_map(|line| { + let (name, value) = line.split_once(':')?; + name.eq_ignore_ascii_case("content-length") + .then(|| value.trim().parse::().ok()) + .flatten() + }); + content_length.is_none_or(|length| request.len() >= header_end + 4 + length) + } + + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + let server = tokio::spawn(async move { + let mut requests = Vec::new(); + for reply in replies { + let (mut stream, _) = listener.accept().await.unwrap(); + let mut request = Vec::new(); + while !request_complete(&request) { + let mut buffer = [0; 1024]; + let count = stream.read(&mut buffer).await.unwrap(); + assert!( + count > 0, + "the client closed before the request was complete" + ); + request.extend_from_slice(&buffer[..count]); + } + requests.push(request); + match reply { + Reply::Respond(response) => stream.write_all(&response).await.unwrap(), + Reply::Hang => { + // Keep the connection open without answering. The task is + // dropped with the test runtime. + tokio::spawn(async move { + let _stream = stream; + tokio::time::sleep(Duration::from_secs(60)).await; + }); + } + } + } + requests + }); + + (format!("http://{address}"), server) +} + +fn split_request(request: Vec) -> (String, serde_json::Value) { + let request = String::from_utf8(request).unwrap(); + let (headers, body) = request.split_once("\r\n\r\n").unwrap(); + (headers.to_string(), serde_json::from_str(body).unwrap()) +} + +#[tokio::test] +async fn validate_sends_the_spec_verbatim_in_strict_mode() { + let (rest_uri, server) = start_http_server(http_response( + "200 OK", + r#"{"valid":true,"spec":"normalized: spec","warnings":[{"path":"charts[0]","message":"deprecated"}]}"#, + )) + .await; + + let result = service(rest_uri) + .validate("charts:\n - name: \"a\"\n") + .await + .unwrap(); + + let (headers, body) = split_request(server.await.unwrap()); + assert!(headers.starts_with("POST /api/v1/declarative:validate HTTP/1.1")); + assert!( + headers + .lines() + .any(|line| line.eq_ignore_ascii_case("authorization: Bearer test-key")) + ); + assert!( + headers + .lines() + .any(|line| line.eq_ignore_ascii_case("user-agent: sift_mcp/1.2.3")) + ); + assert_eq!( + body, + serde_json::json!({"spec": "charts:\n - name: \"a\"\n", "strict": true}) + ); + + assert!(result.valid); + assert_eq!(result.spec, "normalized: spec"); + assert!(result.errors.is_empty()); + assert_eq!( + result.warnings, + vec![Issue { + path: "charts[0]".into(), + message: "deprecated".into() + }] + ); +} + +#[tokio::test] +async fn validate_returns_errors_for_an_invalid_spec() { + let (rest_uri, server) = start_http_server(http_response( + "200 OK", + r#"{"spec":"bad","errors":[{"path":"charts[0].name","message":"required"},{"path":"","message":"unknown field"}]}"#, + )) + .await; + + let result = service(rest_uri).validate("bad").await.unwrap(); + server.await.unwrap(); + + assert!(!result.valid); + assert_eq!(result.errors.len(), 2); + assert_eq!(result.errors[0].path, "charts[0].name"); + assert_eq!(result.errors[0].message, "required"); + assert_eq!(result.errors[1].path, ""); + assert!(result.warnings.is_empty()); +} + +#[tokio::test] +async fn validate_treats_omitted_default_fields_as_empty() { + let (rest_uri, server) = start_http_server(http_response("200 OK", "{}")).await; + + let result = service(rest_uri).validate("x").await.unwrap(); + server.await.unwrap(); + + assert!(!result.valid); + assert_eq!(result.spec, ""); + assert!(result.errors.is_empty()); + assert!(result.warnings.is_empty()); +} + +#[tokio::test] +async fn migrate_returns_the_migrated_spec_and_versions() { + let (rest_uri, server) = start_http_server(http_response( + "200 OK", + r#"{"originalSpec":"old","migratedSpec":"new","sourceVersion":"DECLARATIVE_SCHEMA_VERSION_V0","targetVersion":"DECLARATIVE_SCHEMA_VERSION_V1","warnings":[{"path":"a","message":"renamed"}],"valid":true}"#, + )) + .await; + + let result = service(rest_uri).migrate("old").await.unwrap(); + + let (headers, body) = split_request(server.await.unwrap()); + assert!(headers.starts_with("POST /api/v1/declarative:migrate HTTP/1.1")); + assert_eq!(body, serde_json::json!({"spec": "old"})); + + assert!(result.valid); + assert_eq!(result.original_spec, "old"); + assert_eq!(result.migrated_spec, "new"); + assert_eq!(result.source_version, "DECLARATIVE_SCHEMA_VERSION_V0"); + assert_eq!(result.target_version, "DECLARATIVE_SCHEMA_VERSION_V1"); + assert_eq!(result.warnings.len(), 1); + assert!(result.errors.is_empty()); +} + +#[tokio::test] +async fn migrate_with_errors_and_missing_default_fields() { + let (rest_uri, server) = start_http_server(http_response( + "200 OK", + r#"{"originalSpec":"old","errors":[{"path":"b","message":"cannot migrate"}]}"#, + )) + .await; + + let result = service(rest_uri).migrate("old").await.unwrap(); + server.await.unwrap(); + + assert!(!result.valid); + assert_eq!(result.migrated_spec, ""); + assert_eq!(result.source_version, ""); + assert_eq!(result.target_version, ""); + assert!(result.warnings.is_empty()); + assert_eq!(result.errors[0].message, "cannot migrate"); +} + +#[tokio::test] +async fn a_non_2xx_response_reports_the_status_and_detail() { + let (rest_uri, server) = start_http_server(http_response( + "403 Forbidden", + r#"{"message":"permission denied"}"#, + )) + .await; + + let error = service(rest_uri).validate("x").await.unwrap_err(); + server.await.unwrap(); + + let message = format!("{error:#}"); + assert!(message.contains("403"), "{message}"); + assert!(message.contains("permission denied"), "{message}"); +} + +#[tokio::test] +async fn a_migrate_400_is_an_invalid_result_with_the_backend_message() { + let (rest_uri, server) = start_sequence_server(vec![Reply::Respond(http_response( + "400 Bad Request", + r#"{"message":"spec must not be empty"}"#, + ))]) + .await; + + let result = service(rest_uri).migrate(" ").await.unwrap(); + let requests = server.await.unwrap(); + + assert!(!result.valid); + assert_eq!(result.migrated_spec, ""); + assert_eq!( + result.errors, + vec![Issue { + path: String::new(), + message: "spec must not be empty".into() + }] + ); + // A 400 is final: no retry. + assert_eq!(requests.len(), 1); +} + +#[tokio::test] +async fn a_validate_400_is_an_invalid_result_with_the_backend_message() { + let (rest_uri, server) = + start_http_server(http_response("400 Bad Request", "spec must not be empty")).await; + + let result = service(rest_uri).validate(" ").await.unwrap(); + server.await.unwrap(); + + assert!(!result.valid); + assert_eq!( + result.errors, + vec![Issue { + path: String::new(), + message: "spec must not be empty".into() + }] + ); +} + +#[tokio::test] +async fn a_non_2xx_migrate_response_is_an_error() { + let (rest_uri, server) = start_http_server(http_response( + "404 Not Found", + r#"{"message":"no such endpoint"}"#, + )) + .await; + + let error = service(rest_uri).migrate("x").await.unwrap_err(); + server.await.unwrap(); + + let message = format!("{error:#}"); + assert!(message.contains("migrate"), "{message}"); + assert!(message.contains("404"), "{message}"); +} + +#[tokio::test] +async fn a_long_error_body_is_truncated_to_500_chars() { + let body = "x".repeat(800); + let (rest_uri, server) = start_http_server(http_response("403 Forbidden", &body)).await; + + let error = service(rest_uri).validate("x").await.unwrap_err(); + server.await.unwrap(); + + let message = format!("{error:#}"); + assert!(message.contains(&"x".repeat(500)), "{message}"); + assert!(!message.contains(&"x".repeat(501)), "{message}"); +} + +#[tokio::test] +async fn a_transient_status_is_retried_until_it_succeeds() { + let (rest_uri, server) = start_sequence_server(vec![ + Reply::Respond(http_response( + "503 Service Unavailable", + r#"{"message":"try later"}"#, + )), + Reply::Respond(http_response("200 OK", r#"{"valid":true,"spec":"ok"}"#)), + ]) + .await; + + let result = service(rest_uri).validate("x").await.unwrap(); + let requests = server.await.unwrap(); + + assert!(result.valid); + assert_eq!(requests.len(), 2); + // Every attempt resends the same request. + assert_eq!(requests[0], requests[1]); +} + +#[tokio::test] +async fn a_429_and_a_408_are_retried() { + let (rest_uri, server) = start_sequence_server(vec![ + Reply::Respond(http_response("429 Too Many Requests", "{}")), + Reply::Respond(http_response("408 Request Timeout", "{}")), + Reply::Respond(http_response( + "200 OK", + r#"{"migratedSpec":"new","valid":true}"#, + )), + ]) + .await; + + let result = service(rest_uri).migrate("old").await.unwrap(); + let requests = server.await.unwrap(); + + assert_eq!(result.migrated_spec, "new"); + assert_eq!(requests.len(), 3); +} + +#[tokio::test] +async fn a_non_retryable_status_fails_without_retrying() { + let (rest_uri, server) = start_sequence_server(vec![Reply::Respond(http_response( + "404 Not Found", + r#"{"message":"no such endpoint"}"#, + ))]) + .await; + + let error = service(rest_uri).validate("x").await.unwrap_err(); + let requests = server.await.unwrap(); + + let message = format!("{error:#}"); + assert!(message.contains("404"), "{message}"); + assert!(message.contains("no such endpoint"), "{message}"); + assert!(!message.contains("attempts"), "{message}"); + assert_eq!(requests.len(), 1); +} + +#[tokio::test] +async fn a_persistent_5xx_gives_up_after_three_attempts() { + let (rest_uri, server) = start_sequence_server(vec![ + Reply::Respond(http_response( + "500 Internal Server Error", + r#"{"message":"boom"}"#, + )), + Reply::Respond(http_response( + "500 Internal Server Error", + r#"{"message":"boom"}"#, + )), + Reply::Respond(http_response( + "500 Internal Server Error", + r#"{"message":"boom"}"#, + )), + ]) + .await; + + let error = service(rest_uri).migrate("x").await.unwrap_err(); + let requests = server.await.unwrap(); + + let message = format!("{error:#}"); + assert!(message.contains("migrate"), "{message}"); + assert!(message.contains("failed after 3 attempts"), "{message}"); + assert!(message.contains("500"), "{message}"); + assert!(message.contains("boom"), "{message}"); + assert_eq!(requests.len(), 3); +} + +#[tokio::test] +async fn max_attempts_is_configurable() { + let (rest_uri, server) = start_sequence_server(vec![ + Reply::Respond(http_response("502 Bad Gateway", "{}")), + Reply::Respond(http_response("502 Bad Gateway", "{}")), + ]) + .await; + let retry = RestRetry { + max_attempts: 2, + ..fast_retry() + }; + + let error = service_with_retry(rest_uri, retry) + .validate("x") + .await + .unwrap_err(); + let requests = server.await.unwrap(); + + assert!( + format!("{error:#}").contains("failed after 2 attempts"), + "{error:#}" + ); + assert_eq!(requests.len(), 2); +} + +#[tokio::test] +async fn a_timed_out_request_is_retried() { + let (rest_uri, server) = start_sequence_server(vec![ + Reply::Hang, + Reply::Respond(http_response("200 OK", r#"{"valid":true}"#)), + ]) + .await; + let retry = RestRetry { + request_timeout: Duration::from_millis(100), + ..fast_retry() + }; + + let result = service_with_retry(rest_uri, retry) + .validate("x") + .await + .unwrap(); + let requests = server.await.unwrap(); + + assert!(result.valid); + assert_eq!(requests.len(), 2); +} + +#[tokio::test] +async fn a_request_that_always_times_out_reports_the_timeout() { + let (rest_uri, server) = + start_sequence_server(vec![Reply::Hang, Reply::Hang, Reply::Hang]).await; + let retry = RestRetry { + request_timeout: Duration::from_millis(50), + ..fast_retry() + }; + + let error = service_with_retry(rest_uri, retry) + .validate("x") + .await + .unwrap_err(); + let requests = server.await.unwrap(); + + let message = format!("{error:#}"); + assert!(message.contains("timed out after 50ms"), "{message}"); + assert!(message.contains("failed after 3 attempts"), "{message}"); + assert_eq!(requests.len(), 3); +} + +#[tokio::test] +async fn share_link_returns_the_short_link() { + let (rest_uri, server) = + start_http_server(http_response("200 OK", r#"{"shortLink":"aB3dE9x"}"#)).await; + + let result = service(rest_uri) + .create_share_link("charts:\n - name: \"a\"\n") + .await + .unwrap(); + + let (headers, body) = split_request(server.await.unwrap()); + assert!(headers.starts_with("POST /api/v1/declarative:sharelink HTTP/1.1")); + assert!( + headers + .lines() + .any(|line| line.eq_ignore_ascii_case("authorization: Bearer test-key")) + ); + assert_eq!( + body, + serde_json::json!({"spec": "charts:\n - name: \"a\"\n"}) + ); + assert_eq!( + result, + ShareLinkResult::Ok { + short_link: "aB3dE9x".into() + } + ); +} + +#[tokio::test] +async fn share_link_400_is_an_invalid_spec_not_an_error() { + let (rest_uri, server) = start_sequence_server(vec![Reply::Respond(http_response( + "400 Bad Request", + "charts[0].name is required", + ))]) + .await; + + let result = service(rest_uri).create_share_link("bad").await.unwrap(); + let requests = server.await.unwrap(); + + assert_eq!( + result, + ShareLinkResult::Invalid { + message: "charts[0].name is required".into() + } + ); + // A 400 is final: no retry. + assert_eq!(requests.len(), 1); +} + +#[tokio::test] +async fn share_link_400_prefers_the_json_message() { + let (rest_uri, server) = start_http_server(http_response( + "400 Bad Request", + r#"{"code":3,"message":"unknown field"}"#, + )) + .await; + + let result = service(rest_uri).create_share_link("bad").await.unwrap(); + server.await.unwrap(); + + assert_eq!( + result, + ShareLinkResult::Invalid { + message: "unknown field".into() + } + ); +} + +#[tokio::test] +async fn share_link_400_message_is_truncated_to_500_chars() { + let body = "y".repeat(800); + let (rest_uri, server) = start_http_server(http_response("400 Bad Request", &body)).await; + + let result = service(rest_uri).create_share_link("bad").await.unwrap(); + server.await.unwrap(); + + assert_eq!( + result, + ShareLinkResult::Invalid { + message: "y".repeat(500) + } + ); +} + +#[tokio::test] +async fn share_link_400_with_an_empty_body_uses_a_fallback_message() { + let (rest_uri, server) = start_http_server(http_response("400 Bad Request", "")).await; + + let result = service(rest_uri).create_share_link("bad").await.unwrap(); + server.await.unwrap(); + + assert_eq!( + result, + ShareLinkResult::Invalid { + message: "declarative spec failed validation".into() + } + ); +} + +#[tokio::test] +async fn share_link_without_a_short_link_is_an_error() { + for body in ["{}", r#"{"shortLink":""}"#] { + let (rest_uri, server) = start_http_server(http_response("200 OK", body)).await; + + let error = service(rest_uri).create_share_link("x").await.unwrap_err(); + server.await.unwrap(); + + assert!( + format!("{error:#}").contains("did not include shortLink"), + "{body}: {error:#}" + ); + } +} + +#[tokio::test] +async fn share_link_retries_a_transient_failure() { + let (rest_uri, server) = start_sequence_server(vec![ + Reply::Respond(http_response("503 Service Unavailable", "{}")), + Reply::Respond(http_response("200 OK", r#"{"shortLink":"Zk4mQ7p"}"#)), + ]) + .await; + + let result = service(rest_uri).create_share_link("x").await.unwrap(); + let requests = server.await.unwrap(); + + assert_eq!( + result, + ShareLinkResult::Ok { + short_link: "Zk4mQ7p".into() + } + ); + assert_eq!(requests.len(), 2); +} + +#[tokio::test] +async fn a_transport_error_is_reported() { + // Bind then drop a listener so the port is free and refuses connections. + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let rest_uri = format!("http://{}", listener.local_addr().unwrap()); + drop(listener); + + let error = service(rest_uri).validate("x").await.unwrap_err(); + + assert!( + format!("{error:#}").contains("failed to reach the declarative validate endpoint"), + "{error:#}" + ); +} + +#[tokio::test] +async fn without_a_rest_endpoint_calls_fail_clearly() { + let error = DeclarativeService::new(None, "1.2.3") + .validate("x") + .await + .unwrap_err(); + + assert!(format!("{error:#}").contains("without a REST endpoint")); +} diff --git a/rust/crates/sift_mcp/src/service/mod.rs b/rust/crates/sift_mcp/src/service/mod.rs index 8ccaa277c6..462bd5250d 100644 --- a/rust/crates/sift_mcp/src/service/mod.rs +++ b/rust/crates/sift_mcp/src/service/mod.rs @@ -4,6 +4,7 @@ pub mod assets; pub mod calculated_channels; pub mod channels; pub mod data; +pub mod declarative; pub mod docs; pub mod ingest; pub mod ping; diff --git a/rust/crates/sift_mcp/src/service/url/mod.rs b/rust/crates/sift_mcp/src/service/url/mod.rs index e6a21bf692..0d96814af2 100644 --- a/rust/crates/sift_mcp/src/service/url/mod.rs +++ b/rust/crates/sift_mcp/src/service/url/mod.rs @@ -138,6 +138,11 @@ impl UrlService { Ok(format!("{host}/explore?{query}")) } + pub fn build_share_url(&self, short_link: &str) -> Result { + let host = self.app_host()?; + Ok(format!("{host}/share/{}", encode_value(short_link))) + } + pub fn build_report_url(&self, report_id: &str) -> Result { let host = self.app_host()?; Ok(format!("{host}/reports/{}", encode_value(report_id))) diff --git a/rust/crates/sift_mcp/src/service/url/test.rs b/rust/crates/sift_mcp/src/service/url/test.rs index eb7c247c0d..5929c90461 100644 --- a/rust/crates/sift_mcp/src/service/url/test.rs +++ b/rust/crates/sift_mcp/src/service/url/test.rs @@ -186,6 +186,12 @@ fn report_url_uses_configured_app_uri() { assert_eq!(url, "https://app.siftstack.com/reports/rep-123"); } +#[test] +fn share_url_uses_configured_app_uri() { + let url = service().build_share_url("abc123").unwrap(); + assert_eq!(url, "https://app.siftstack.com/share/abc123"); +} + #[test] fn rule_url_uses_configured_app_uri() { let url = service().build_rule_url("rule-123").unwrap(); diff --git a/rust/crates/sift_mcp/src/tool/data/mod.rs b/rust/crates/sift_mcp/src/tool/data/mod.rs index 0ddf837187..641d17829f 100644 --- a/rust/crates/sift_mcp/src/tool/data/mod.rs +++ b/rust/crates/sift_mcp/src/tool/data/mod.rs @@ -193,17 +193,17 @@ impl SiftMcpServer { Narrow `channel_regex`, pass explicit `channel_names` or `channel_ids`, or split the request. Guidance: - - If the user's intent is to view/plot/graph/visualize the data in a UI, call `explore_url` first - instead — it returns a Sift Explore deep-link and skips the download entirely. Use `get_data` only - when the bytes are needed locally for SQL, custom analysis, or a static artifact the user explicitly - asked for. + - When the user wants to plot, chart, graph, or visualize data, call `create_declarative_chart` + with a spec instead — it validates the chart and skips download. For quick links to Sift Explore, + call `explore_url` instead. Use `get_data` only when the bytes are needed locally for SQL, custom + analysis, or a static artifact the user explicitly asked for. - Data is buffered in memory until size/row thresholds are hit, so very large time ranges or wide channel sets can be slow or memory-heavy. For large pulls, split the time range into successive calls with disjoint `[start, end)` windows. A long call can also hit the client's request timeout and lose everything it fetched. Each shorter window keeps its own result. - Analysis needs `sample_ms = 0`. Reserve `sample_ms > 0` for data used exclusively to draw a - picture, and prefer `explore_url` over `get_data` entirely when that is the goal. An overview, - a summary and a quick look are all analysis: they end in numbers, and numbers taken off + picture, and prefer `create_declarative_chart` to render a chart instead of downloading decimated data. + An overview, a summary and a quick look are all analysis: they end in numbers, and numbers taken off decimated data are wrong. - The result reports `sample_ms_applied` and `decimated`. Check them before quoting any statistic: a non-zero applied rate means means, standard deviations and trends from that file are not valid. diff --git a/rust/crates/sift_mcp/src/tool/explore/mod.rs b/rust/crates/sift_mcp/src/tool/explore/mod.rs index f7a6cdaa7f..5f3e7f5a86 100644 --- a/rust/crates/sift_mcp/src/tool/explore/mod.rs +++ b/rust/crates/sift_mcp/src/tool/explore/mod.rs @@ -5,12 +5,28 @@ use rmcp::{ tool, tool_router, }; use serde::Deserialize; +use serde_json::{Value, json}; -use crate::{error, server::SiftMcpServer, service::url::ExploreUrlRequest}; +use crate::{ + error, + server::SiftMcpServer, + service::{ + declarative::{Issue, ShareLinkResult}, + url::ExploreUrlRequest, + }, +}; #[cfg(test)] mod test; +/// Failed attempts the agent gets to repair a spec before it must report the problem to the user. +const MAX_REPAIR_ATTEMPTS: u32 = 3; + +#[derive(Debug, Deserialize, JsonSchema)] +pub struct CreateDeclarativeChartParams { + spec: String, +} + #[derive(Debug, Deserialize, JsonSchema)] pub struct ExploreUrlParams { asset_ids: Option>, @@ -27,9 +43,10 @@ impl SiftMcpServer { #[tool( name = "explore_url", description = " - Build a Sift Explore deep-link URL for the given asset/run/channel selection. Pure URL construction — no - API call is made. Hand the returned URL to the user verbatim, rendered inline as a clickable link, so - they can open the view in the Sift web app. + Build a quick Sift Explore deep-link URL that opens specific assets, runs, or channels. Pure URL + construction — no API call is made. Hand the returned URL to the user verbatim, rendered inline as a + clickable link, so they can open the view in the Sift web app. To plot, chart, graph, or visualize + data, call `create_declarative_chart` instead. Output: - `{ \"url\": \"\", \"next_step\": \"...\", \"start_time_unix_nanos\": ?, @@ -64,9 +81,10 @@ impl SiftMcpServer { known set, or if `end_time_unix_nanos < start_time_unix_nanos`. Guidance: - - Reach for this tool when the user asks to \"see\", \"view\", \"graph\", \"plot\", \"visualize\", or - \"open\" data in Sift. Pair it with `get_data` only when the user also wants the data locally for - SQL or further processing. + - Reach for this tool when the user asks to \"open\" or \"link to\" specific assets, runs, or channels + in Sift. Requests to \"plot\", \"chart\", \"graph\", or \"visualize\" data belong to + `create_declarative_chart`. Pair this tool with `get_data` only when the user also wants the data + locally for SQL or further processing. - Do not add the asset alongside a run to be thorough. A run is already scoped to its asset, so the asset adds a second, wider source to the view. Send both only on an explicit request for both. @@ -137,6 +155,192 @@ impl SiftMcpServer { result.content = vec![ContentBlock::text(next_step)]; Ok(result) } + + #[tool( + name = "create_declarative_chart", + description = " + Validate a declarative chart spec (YAML or JSON) against the Sift schema, migrating it to the latest + version first, then create a shareable Sift Explore link for it. A valid result hands back the + migrated spec and the link. + + Output: + - Valid: `{ \"valid\": true, \"spec\": \"\", \"shortLink\", \"exploreUrl\", + \"warnings\": [{\"path\", \"message\"}], \"sourceVersion\", \"targetVersion\", \"next_step\" }`. + `exploreUrl` is the link that opens the chart in Sift. + - Invalid: `{ \"valid\": false, \"reason\": \"invalid_spec\", \"issues\": [{\"path\", \"message\"}], + \"next_step\" }`. `path` locates the problem in the spec; `/` means the whole spec. + - Unavailable: `{ \"valid\": false, \"reason\": \"service_unavailable\", \"issues\": [...], + \"next_step\" }` when the validation or link service could not be reached. + + Parameters: + - `spec`: the complete spec as a YAML or JSON string. The Sift backend decides what is valid; the + tool does not pre-check the spec, so multi-chart specs and a top-level `layout` are accepted. + + Errors: + - A bad spec is not an MCP error. It returns `valid: false` with `issues` so you can fix and retry. + - `INVALID_PARAMS` is returned only when `spec` is missing or not a string. + + Guidance: + - Resolve run IDs and channel names with `list_runs`, `list_channels`, and + `list_calculated_channels` before writing the spec. Never invent them. + - Bind a calculated channel with `calculatedChannelId`, not by name. + - A spec with exactly one chart displays inline in Sift agent chat. Specs with several charts or a + `layout` still return a working Explore link. + - Prefer YAML. JSON is accepted. + - On `invalid_spec`, fix every listed path and resend the complete spec. After 3 failed attempts, + stop and summarize the problem to the user instead of retrying. + - Omit `sampling`, `maxGap`, and axis `min`/`max` unless the user asked for them. + - On a valid result, render `exploreUrl` once as a clickable markdown link with descriptive text, such + as what the chart plots. Do not paste the spec into your reply. + ", + annotations( + title = "explore/create_declarative_chart", + read_only_hint = false, + destructive_hint = false, + idempotent_hint = false + ) + )] + pub async fn create_declarative_chart( + &self, + params: Parameters, + ) -> error::McpResult { + let Parameters(CreateDeclarativeChartParams { spec }) = params; + + let migration = match self.declarative_service.migrate(&spec).await { + Ok(migration) => migration, + Err(error) => return Ok(service_unavailable_result(&error)), + }; + if !migration.valid || !migration.errors.is_empty() { + return Ok(invalid_spec_result(issues_or( + &migration.errors, + "the spec could not be migrated", + ))); + } + let migrated_spec = if migration.migrated_spec.is_empty() { + spec + } else { + migration.migrated_spec + }; + + let validation = match self.declarative_service.validate(&migrated_spec).await { + Ok(validation) => validation, + Err(error) => return Ok(service_unavailable_result(&error)), + }; + if !validation.valid || !validation.errors.is_empty() { + return Ok(invalid_spec_result(issues_or( + &validation.errors, + "the spec failed validation", + ))); + } + + let short_link = match self + .declarative_service + .create_share_link(&migrated_spec) + .await + { + Ok(ShareLinkResult::Ok { short_link }) => short_link, + Ok(ShareLinkResult::Invalid { message }) => { + return Ok(invalid_spec_result(vec![issue("/", &message)])); + } + Err(error) => return Ok(service_unavailable_result(&error)), + }; + let explore_url = self.url_service.build_share_url(&short_link)?; + + let mut warnings = issues_json(&migration.warnings); + warnings.extend(issues_json(&validation.warnings)); + + let next_step = format!( + "The chart spec validated and a Sift link was created: {explore_url}\n\nRender this URL \ + in your response as a clickable markdown link with descriptive text naming what the \ + chart plots, never the word \"link\" or the bare URL. Mention it once. Do not paste \ + the spec into your response." + ); + let mut result = CallToolResult::structured(json!({ + "valid": true, + "spec": migrated_spec, + "shortLink": short_link, + "exploreUrl": explore_url, + "warnings": warnings, + "sourceVersion": migration.source_version, + "targetVersion": migration.target_version, + "next_step": next_step, + })); + result.content = vec![ContentBlock::text(next_step)]; + Ok(result) + } +} + +fn issue(path: &str, message: &str) -> Value { + json!({ "path": path, "message": message }) +} + +fn issues_json(issues: &[Issue]) -> Vec { + issues + .iter() + .map(|found| { + let path = if found.path.is_empty() { + "/" + } else { + found.path.as_str() + }; + issue(path, &found.message) + }) + .collect() +} + +/// The service's issues, or one whole-spec issue carrying `fallback` when it listed none. +fn issues_or(issues: &[Issue], fallback: &str) -> Vec { + if issues.is_empty() { + vec![issue("/", fallback)] + } else { + issues_json(issues) + } +} + +/// A non-error result the client can act on: `reason`, the `issues`, and a `next_step` that is +/// also the text content. +fn failure_result(reason: &str, issues: Vec, next_step: String) -> CallToolResult { + let listing = issues + .iter() + .map(|found| { + let path = found["path"].as_str().unwrap_or("/"); + let message = found["message"].as_str().unwrap_or_default(); + format!("{path}: {message}") + }) + .collect::>() + .join("\n"); + let mut result = CallToolResult::structured(json!({ + "valid": false, + "reason": reason, + "issues": issues, + "next_step": next_step, + })); + result.content = vec![ContentBlock::text(format!("{listing}\n\n{next_step}"))]; + result +} + +fn invalid_spec_result(issues: Vec) -> CallToolResult { + failure_result( + "invalid_spec", + issues, + format!( + "The spec is not valid yet. Fix every listed path and resend the complete spec. After \ + {MAX_REPAIR_ATTEMPTS} failed attempts, stop retrying and summarize the problem to the user." + ), + ) +} + +// TODO: Non-retryable 4xx responses (401, 403, 404: bad API key, missing permission, or a backend +// without the declarative endpoint) also land here and tell the agent to retry. Tag them in +// DeclarativeService with a typed error and map them to a no-retry, check-credentials result. +fn service_unavailable_result(error: &anyhow::Error) -> CallToolResult { + failure_result( + "service_unavailable", + vec![issue("/", &format!("{error:#}"))], + "The spec could not be checked because the validation service failed, which is not a problem \ + with the spec. Retry once; if it fails again, tell the user the chart could not be validated." + .to_string(), + ) } /// Explore URLs carry ISO 8601 timestamps with millisecond precision, so the URL service drops diff --git a/rust/crates/sift_mcp/src/tool/explore/test.rs b/rust/crates/sift_mcp/src/tool/explore/test.rs index 5eec2f861d..14c3f20d17 100644 --- a/rust/crates/sift_mcp/src/tool/explore/test.rs +++ b/rust/crates/sift_mcp/src/tool/explore/test.rs @@ -3,7 +3,16 @@ use serde_json::Value; use sift_test_util::grpc::memory_sift_channel; use super::*; -use crate::server::SiftMcpServer; +use crate::{ + server::SiftMcpServer, + service::{ + declarative::{ + DeclarativeService, + test::{Reply, fast_retry, http_response, start_sequence_server}, + }, + remote_files::RestConfig, + }, +}; const APP_URI: &str = "https://app.siftstack.com"; @@ -219,3 +228,523 @@ async fn handler_keeps_both_source_types_when_the_opt_in_is_set() { Some("https://app.siftstack.com/explore?method=single&assets=asset-id&runs=run-id") ); } + +const YAML_CHART: &str = "chart:\n type: timeseries\n title: Motor current\n series:\n - channel: a\n - channel: b\n"; +const MIGRATED_CHART: &str = "chart:\n type: timeseries\n title: Motor current\n series: []\n"; + +/// Serves one canned response per connection, in order, and returns every request as text. +async fn serve(responses: Vec>) -> (String, tokio::task::JoinHandle>) { + let (rest_uri, server) = + start_sequence_server(responses.into_iter().map(Reply::Respond).collect()).await; + let requests = tokio::spawn(async move { + server + .await + .unwrap() + .into_iter() + .map(|request| String::from_utf8(request).unwrap()) + .collect() + }); + (rest_uri, requests) +} + +fn request_body(request: &str) -> Value { + serde_json::from_str(request.split_once("\r\n\r\n").unwrap().1).unwrap() +} + +/// A successful migrate response with no warnings. +fn migrate_ok(migrated_spec: &str) -> Vec { + http_response( + "200 OK", + &migrate_body(migrated_spec, serde_json::json!([])), + ) +} + +fn migrate_body(migrated_spec: &str, warnings: Value) -> String { + serde_json::json!({ + "originalSpec": "ignored", + "migratedSpec": migrated_spec, + "sourceVersion": "DECLARATIVE_SCHEMA_VERSION_V0", + "targetVersion": "DECLARATIVE_SCHEMA_VERSION_V1", + "warnings": warnings, + "valid": true, + }) + .to_string() +} + +/// A successful share-link response carrying `short_link`. +fn share_link_ok(short_link: &str) -> Vec { + http_response( + "200 OK", + &serde_json::json!({ "shortLink": short_link }).to_string(), + ) +} + +async fn server_with_rest(rest_uri: String) -> SiftMcpServer { + let mut server = server_for_explore(APP_URI).await; + server.declarative_service = + DeclarativeService::new(Some(RestConfig::new(rest_uri, "test-key".into())), "1.2.3") + .with_retry(fast_retry()); + server +} + +async fn create_chart(server: &SiftMcpServer, spec: &str) -> rmcp::model::CallToolResult { + server + .create_declarative_chart(Parameters(CreateDeclarativeChartParams { + spec: spec.to_string(), + })) + .await + .expect("invalid specs are results, not MCP errors") +} + +fn structured(result: &rmcp::model::CallToolResult) -> Value { + result + .structured_content + .clone() + .expect("expected structured content") +} + +fn assert_invalid_spec(result: &rmcp::model::CallToolResult, path: &str, message: &str) { + let structured = structured(result); + assert_eq!(structured["valid"], false, "{structured}"); + assert_eq!(structured["reason"], "invalid_spec", "{structured}"); + let issues = structured["issues"].as_array().unwrap(); + assert_eq!(issues.len(), 1, "{structured}"); + assert_eq!(issues[0]["path"], path, "{structured}"); + let actual = issues[0]["message"].as_str().unwrap(); + assert!(actual.contains(message), "{actual}"); + let next_step = structured["next_step"].as_str().unwrap(); + assert!( + next_step.contains("resend the complete spec"), + "{next_step}" + ); + assert!(next_step.contains('3'), "{next_step}"); + assert_eq!(result.content.len(), 1, "content should carry the issues"); +} + +#[test] +fn create_declarative_chart_schema_is_one_flat_spec_string() { + let schema = serde_json::to_value(schemars::schema_for!(CreateDeclarativeChartParams)).unwrap(); + let properties = schema["properties"].as_object().unwrap(); + + assert_eq!(properties.len(), 1); + assert_eq!(properties["spec"]["type"], "string"); +} + +#[test] +fn create_declarative_chart_is_registered_as_an_additive_write() { + let tools = SiftMcpServer::explore_router().list_all(); + let tool = tools + .iter() + .find(|tool| tool.name == "create_declarative_chart") + .expect("create_declarative_chart is routed"); + let annotations = tool.annotations.as_ref().expect("annotations are set"); + + assert_eq!( + annotations.title.as_deref(), + Some("explore/create_declarative_chart") + ); + assert_eq!(annotations.read_only_hint, Some(false)); + assert_eq!(annotations.destructive_hint, Some(false)); + assert_eq!(annotations.idempotent_hint, Some(false)); + + let description = tool.description.as_deref().expect("has a description"); + for needle in [ + "list_runs", + "list_channels", + "list_calculated_channels", + "calculatedChannelId", + "exactly one chart displays inline", + "After 3 failed attempts", + "`maxGap`", + "`exploreUrl`", + "Do not paste the spec", + ] { + assert!(description.contains(needle), "missing `{needle}`"); + } +} + +#[test] +fn explore_url_description_defers_charting_requests() { + let tools = SiftMcpServer::explore_router().list_all(); + let tool = tools + .iter() + .find(|tool| tool.name == "explore_url") + .expect("explore_url is routed"); + let description = tool.description.as_deref().expect("has a description"); + + assert!(description.contains("call `create_declarative_chart` instead")); + assert!(description.contains("belong to")); +} + +#[test] +fn create_declarative_chart_has_a_client_event() { + assert_eq!( + crate::client_event::event_for_tool("create_declarative_chart"), + Some("CLIENT_EVENT_USER_CALLED_MCP_TOOL_CREATE_DECLARATIVE_CHART") + ); +} + +#[tokio::test] +async fn a_migrate_error_is_an_invalid_spec_result() { + let (rest_uri, server_task) = serve(vec![http_response( + "200 OK", + r#"{"errors":[{"path":"/chart/series/0","message":"cannot migrate"},{"path":"","message":"unknown version"}]}"#, + )]) + .await; + let server = server_with_rest(rest_uri).await; + + let result = create_chart(&server, YAML_CHART).await; + let requests = server_task.await.unwrap(); + + assert_eq!(requests.len(), 1, "validate ran after a failed migrate"); + assert!(requests[0].starts_with("POST /api/v1/declarative:migrate ")); + let structured = structured(&result); + assert_eq!(structured["valid"], false); + assert_eq!(structured["reason"], "invalid_spec"); + assert_eq!( + structured["issues"], + serde_json::json!([ + {"path": "/chart/series/0", "message": "cannot migrate"}, + {"path": "/", "message": "unknown version"}, + ]) + ); + assert!( + structured["next_step"] + .as_str() + .unwrap() + .contains("resend the complete spec") + ); +} + +#[tokio::test] +async fn a_validate_error_is_an_invalid_spec_result() { + let (rest_uri, server_task) = serve(vec![ + migrate_ok(MIGRATED_CHART), + http_response( + "200 OK", + r#"{"errors":[{"path":"/chart/series/0/channel","message":"is required"}]}"#, + ), + ]) + .await; + let server = server_with_rest(rest_uri).await; + + let result = create_chart(&server, YAML_CHART).await; + let requests = server_task.await.unwrap(); + + assert!(requests[1].starts_with("POST /api/v1/declarative:validate ")); + assert_eq!( + request_body(&requests[1]), + serde_json::json!({"spec": MIGRATED_CHART, "strict": true}), + "validate must receive the migrated spec in strict mode" + ); + assert_invalid_spec(&result, "/chart/series/0/channel", "is required"); +} + +#[tokio::test] +async fn a_migrate_400_is_an_invalid_spec_result() { + let (rest_uri, server_task) = serve(vec![http_response( + "400 Bad Request", + r#"{"message":"spec must not be empty"}"#, + )]) + .await; + let server = server_with_rest(rest_uri).await; + + let result = create_chart(&server, " ").await; + let requests = server_task.await.unwrap(); + + assert_eq!(requests.len(), 1, "a 400 must not be retried"); + assert_invalid_spec(&result, "/", "spec must not be empty"); +} + +#[tokio::test] +async fn a_validate_400_is_an_invalid_spec_result() { + let (rest_uri, server_task) = serve(vec![ + migrate_ok(MIGRATED_CHART), + http_response("400 Bad Request", r#"{"message":"spec must not be empty"}"#), + ]) + .await; + let server = server_with_rest(rest_uri).await; + + let result = create_chart(&server, YAML_CHART).await; + let requests = server_task.await.unwrap(); + + assert_eq!(requests.len(), 2, "a 400 must not be retried"); + assert!(requests[1].starts_with("POST /api/v1/declarative:validate ")); + assert_invalid_spec(&result, "/", "spec must not be empty"); +} + +#[tokio::test] +async fn a_service_error_is_a_service_unavailable_result() { + // The retry policy makes three attempts, so each needs an answer. + let (rest_uri, server_task) = serve(vec![ + http_response( + "503 Service Unavailable", + r#"{"message":"down"}"# + ); + 3 + ]) + .await; + let server = server_with_rest(rest_uri).await; + + let result = create_chart(&server, YAML_CHART).await; + server_task.await.unwrap(); + + let structured = structured(&result); + assert_eq!(structured["valid"], false); + assert_eq!(structured["reason"], "service_unavailable"); + let issues = structured["issues"].as_array().unwrap(); + assert_eq!(issues.len(), 1); + assert_eq!(issues[0]["path"], "/"); + assert!(issues[0]["message"].as_str().unwrap().contains("503")); + assert!(structured["next_step"].as_str().is_some()); + assert_eq!(result.content.len(), 1); +} + +#[tokio::test] +async fn a_validate_service_error_is_a_service_unavailable_result() { + let mut responses = vec![migrate_ok(MIGRATED_CHART)]; + responses.extend(vec![ + http_response( + "500 Internal Server Error", + r#"{"message":"boom"}"# + ); + 3 + ]); + let (rest_uri, server_task) = serve(responses).await; + let server = server_with_rest(rest_uri).await; + + let result = create_chart(&server, YAML_CHART).await; + server_task.await.unwrap(); + + assert_eq!(structured(&result)["reason"], "service_unavailable"); +} + +#[tokio::test] +async fn a_share_link_service_error_is_a_service_unavailable_result() { + let mut responses = vec![ + migrate_ok(MIGRATED_CHART), + http_response("200 OK", r#"{"valid":true}"#), + ]; + responses.extend(vec![ + http_response( + "502 Bad Gateway", + r#"{"message":"upstream"}"# + ); + 3 + ]); + let (rest_uri, server_task) = serve(responses).await; + let server = server_with_rest(rest_uri).await; + + let result = create_chart(&server, YAML_CHART).await; + let requests = server_task.await.unwrap(); + + assert!(requests[2].starts_with("POST /api/v1/declarative:sharelink ")); + let structured = structured(&result); + assert_eq!(structured["valid"], false, "{structured}"); + assert_eq!(structured["reason"], "service_unavailable"); + assert_eq!(structured["issues"][0]["path"], "/"); + assert!( + structured["issues"][0]["message"] + .as_str() + .unwrap() + .contains("502") + ); + assert!(structured.get("shortLink").is_none()); + assert!(structured.get("exploreUrl").is_none()); +} + +#[tokio::test] +async fn a_rejected_share_link_is_an_invalid_spec_result() { + let (rest_uri, server_task) = serve(vec![ + migrate_ok(MIGRATED_CHART), + http_response("200 OK", r#"{"valid":true}"#), + http_response("400 Bad Request", r#"{"message":"spec cannot be shared"}"#), + ]) + .await; + let server = server_with_rest(rest_uri).await; + + let result = create_chart(&server, YAML_CHART).await; + let requests = server_task.await.unwrap(); + + assert!(requests[2].starts_with("POST /api/v1/declarative:sharelink ")); + assert_invalid_spec(&result, "/", "spec cannot be shared"); + let structured = structured(&result); + assert_eq!( + structured["issues"], + serde_json::json!([{"path": "/", "message": "spec cannot be shared"}]) + ); + assert!(structured.get("shortLink").is_none()); + assert!(structured.get("exploreUrl").is_none()); +} + +#[tokio::test] +async fn a_server_without_a_rest_endpoint_reports_service_unavailable() { + let server = server_for_explore(APP_URI).await; + + let result = create_chart(&server, YAML_CHART).await; + + let structured = structured(&result); + assert_eq!(structured["reason"], "service_unavailable"); + assert!( + structured["issues"][0]["message"] + .as_str() + .unwrap() + .contains("without a REST endpoint") + ); +} + +const JSON_CHART: &str = r#"{"chart": {"type": "timeseries", "series": [{"channel": "a"}]}}"#; +const MULTI_CHART: &str = "charts:\n a:\n type: timeseries\n b:\n type: stat\n"; +const LAYOUT_CHART: &str = "layout:\n rows: []\nchart:\n type: timeseries\n"; + +/// Keys of a success result, sorted. Chart metadata is not among them. +const SUCCESS_KEYS: [&str; 8] = [ + "exploreUrl", + "next_step", + "shortLink", + "sourceVersion", + "spec", + "targetVersion", + "valid", + "warnings", +]; + +/// Runs `spec` against a backend that accepts it and migrates it to `migrated`. Returns the +/// structured result and the three requests the backend saw. +async fn forward(spec: &str, migrated: &str) -> (Value, Vec) { + let (rest_uri, server_task) = serve(vec![ + migrate_ok(migrated), + http_response("200 OK", r#"{"valid":true}"#), + share_link_ok("aB3dE9x"), + ]) + .await; + let server = server_with_rest(rest_uri).await; + + let result = create_chart(&server, spec).await; + (structured(&result), server_task.await.unwrap()) +} + +/// The backend, not the tool, judges `spec`: migrate gets it byte-for-byte, then validate and +/// the share link get the migrated spec, and the result has the success shape. +fn assert_forwarded(spec: &str, migrated: &str, structured: &Value, requests: &[String]) { + assert_eq!(requests.len(), 3, "{requests:?}"); + assert!(requests[0].starts_with("POST /api/v1/declarative:migrate ")); + assert!(requests[1].starts_with("POST /api/v1/declarative:validate ")); + assert!(requests[2].starts_with("POST /api/v1/declarative:sharelink ")); + assert_eq!( + request_body(&requests[0]), + serde_json::json!({"spec": spec}), + "migrate must receive the spec unchanged" + ); + assert_eq!( + request_body(&requests[1]), + serde_json::json!({"spec": migrated, "strict": true}), + "validate must receive the migrated spec" + ); + assert_eq!( + request_body(&requests[2]), + serde_json::json!({"spec": migrated}), + "the share link must be created from the migrated spec" + ); + + assert_eq!(structured["valid"], true, "{structured}"); + assert_eq!(structured["spec"], migrated); + assert_eq!(structured["shortLink"], "aB3dE9x"); + let mut keys: Vec<&str> = structured + .as_object() + .unwrap() + .keys() + .map(String::as_str) + .collect(); + keys.sort_unstable(); + assert_eq!(keys, SUCCESS_KEYS); +} + +#[tokio::test] +async fn yaml_and_json_specs_are_forwarded_unchanged_with_the_same_success_shape() { + let (yaml_result, yaml_requests) = forward(YAML_CHART, MIGRATED_CHART).await; + let (json_result, json_requests) = forward(JSON_CHART, MIGRATED_CHART).await; + + assert_forwarded(YAML_CHART, MIGRATED_CHART, &yaml_result, &yaml_requests); + assert_forwarded(JSON_CHART, MIGRATED_CHART, &json_result, &json_requests); + assert_eq!(yaml_result, json_result); +} + +#[tokio::test] +async fn a_multi_chart_spec_is_forwarded_not_rejected_locally() { + let migrated = "charts:\n a:\n type: timeseries\n b:\n type: stat\nversion: 0\n"; + let (result, requests) = forward(MULTI_CHART, migrated).await; + + assert_forwarded(MULTI_CHART, migrated, &result, &requests); +} + +#[tokio::test] +async fn a_spec_with_layout_is_forwarded_not_rejected_locally() { + let migrated = "layout:\n rows: []\nchart:\n type: timeseries\nversion: 0\n"; + let (result, requests) = forward(LAYOUT_CHART, migrated).await; + + assert_forwarded(LAYOUT_CHART, migrated, &result, &requests); +} + +#[tokio::test] +async fn a_success_result_carries_warnings_the_link_and_a_next_step() { + let (rest_uri, server_task) = serve(vec![ + http_response( + "200 OK", + &migrate_body( + MIGRATED_CHART, + serde_json::json!([{"path": "/chart/series", "message": "renamed"}]), + ), + ), + http_response( + "200 OK", + r#"{"valid":true,"spec":"normalized","warnings":[{"path":"/chart/title","message":"long title"}]}"#, + ), + share_link_ok("aB3dE9x"), + ]) + .await; + let server = server_with_rest(rest_uri).await; + + let result = create_chart(&server, YAML_CHART).await; + let requests = server_task.await.unwrap(); + + assert_eq!( + request_body(&requests[0]), + serde_json::json!({"spec": YAML_CHART}) + ); + let structured = structured(&result); + assert_eq!(structured["valid"], true, "{structured}"); + assert_eq!(structured["spec"], MIGRATED_CHART); + assert_eq!(structured["sourceVersion"], "DECLARATIVE_SCHEMA_VERSION_V0"); + assert_eq!(structured["targetVersion"], "DECLARATIVE_SCHEMA_VERSION_V1"); + assert_eq!( + structured["warnings"], + serde_json::json!([ + {"path": "/chart/series", "message": "renamed"}, + {"path": "/chart/title", "message": "long title"}, + ]) + ); + let next_step = structured["next_step"].as_str().unwrap(); + assert!(next_step.contains("validated"), "{next_step}"); + assert_eq!(structured["shortLink"], "aB3dE9x"); + assert_eq!( + structured["exploreUrl"], + "https://app.siftstack.com/share/aB3dE9x" + ); + assert!( + next_step.contains("https://app.siftstack.com/share/aB3dE9x"), + "{next_step}" + ); + assert!(next_step.contains("clickable markdown link"), "{next_step}"); + assert!(next_step.contains("Mention it once"), "{next_step}"); + assert!(next_step.contains("Do not paste the spec"), "{next_step}"); + assert!(requests[2].starts_with("POST /api/v1/declarative:sharelink ")); + assert_eq!( + request_body(&requests[2]), + serde_json::json!({"spec": MIGRATED_CHART}), + "the share link must be created from the migrated spec" + ); + assert_eq!(result.content.len(), 1); + let content = serde_json::to_value(&result.content[0]).unwrap(); + assert_eq!(content["text"], next_step); +} diff --git a/rust/crates/sift_mcp/src/tool_events.json b/rust/crates/sift_mcp/src/tool_events.json index 7ceb2c4d90..dfa3b1b446 100644 --- a/rust/crates/sift_mcp/src/tool_events.json +++ b/rust/crates/sift_mcp/src/tool_events.json @@ -10,6 +10,7 @@ "create_annotation": "CLIENT_EVENT_USER_CALLED_MCP_TOOL_CREATE_ANNOTATION", "create_artifact": "CLIENT_EVENT_USER_CALLED_MCP_TOOL_CREATE_ARTIFACT", "create_calculated_channel": "CLIENT_EVENT_USER_CALLED_MCP_TOOL_CREATE_CALCULATED_CHANNEL", + "create_declarative_chart": "CLIENT_EVENT_USER_CALLED_MCP_TOOL_CREATE_DECLARATIVE_CHART", "create_report": "CLIENT_EVENT_USER_CALLED_MCP_TOOL_CREATE_REPORT", "create_report_template": "CLIENT_EVENT_USER_CALLED_MCP_TOOL_CREATE_REPORT_TEMPLATE", "create_rule": "CLIENT_EVENT_USER_CALLED_MCP_TOOL_CREATE_RULE",