rust(feature): mcp tools for download/uploading datasets (#586)

Author: solidiquis

Cargo.toml

@@ -26,8 +26,9 @@ license = "MIT"
 
 [workspace.dependencies]
 anyhow = "1.0"
-arrow-array = "58.1.0"
-arrow-schema = "58.1.0"
+arrow = "58.3.0"
+arrow-array = "58.3.0"
+arrow-schema = "58.3.0"
 async-channel = "2.2"
 async-trait = "^0.1"
 mockall = "0.14.0"
@@ -49,11 +50,12 @@ hyper = { version = "1.8", features = ["server", "http1"] }
 hyper-util = { version = "0.1.20", features = ["service", "server", "tokio"] }
 indicatif = "0.18"
 indoc = "2.0"
-parquet = "58.0"
+parquet = "58.3.0"
 prost = "^0.14"
 prost-types = "^0.14"
 pbjson = "^0.9"
 pbjson-types = "^0.9"
+polars = { version = "0.53.0", features = ["sql", "lazy", "parquet"] }
 pyo3 = "0.28"
 pyo3-async-runtimes = { version = "0.28", features = ["tokio-runtime"] }
 pyo3-stub-gen = "0.10" 

rust/crates/sift_cli/Cargo.toml

@@ -9,8 +9,7 @@ repository.workspace = true
 keywords.workspace = true
 readme.workspace = true
 license.workspace = true
-description = "CLI to streamline programmatic workflows with Sift's API"
-changelog = "CHANGELOG.md"
+description = "Sift CLI"
 
 [[bin]]
 name = "sift-cli"
@@ -31,6 +30,7 @@ parquet = { workspace = true }
 pbjson-types = { workspace = true }
 reqwest = { workspace = true }
 serde_json = { workspace = true }
+sift_mcp.workspace = true
 sift_pbfs = { workspace = true }
 tdms = { workspace = true }
 hdf5 = { workspace = true }

rust/crates/sift_cli/src/cli/mod.rs

@@ -53,6 +53,9 @@ pub enum Cmd {
     #[command(subcommand)]
     Import(ImportCmd),
 
+    /// Start the Sift MCP server
+    Mcp,
+
     /// Ping the Sift API to verify credentials and connectivity
     Ping,
 }

rust/crates/sift_cli/src/cmd/mcp.rs

@@ -0,0 +1,17 @@
+use std::process::ExitCode;
+
+use anyhow::Result;
+use sift_rs::Credentials;
+
+use crate::cmd::Context;
+
+pub async fn run(ctx: Context) -> Result<ExitCode> {
+    let credentials = Credentials::Config {
+        uri: ctx.grpc_uri,
+        apikey: ctx.api_key,
+    };
+    match sift_mcp::run(credentials, !ctx.disable_tls).await {
+        Ok(_) => Ok(ExitCode::SUCCESS),
+        Err(err) => Err(err),
+    }
+}

rust/crates/sift_cli/src/cmd/mod.rs

@@ -8,6 +8,7 @@ pub mod completions;
 pub mod config;
 pub mod export;
 pub mod import;
+pub mod mcp;
 pub mod ping;
 
 pub struct Context {

rust/crates/sift_cli/src/main.rs

@@ -40,6 +40,18 @@ where
     runtime.block_on(fut)
 }
 
+fn run_future_mt<F>(fut: F) -> Result<ExitCode>
+where
+    F: Future<Output = Result<ExitCode>> + 'static,
+{
+    let runtime = runtime::Builder::new_multi_thread()
+        .enable_all()
+        .build()
+        .context("failed to initialize Tokio runtime")?;
+
+    runtime.block_on(fut)
+}
+
 fn run(clargs: cli::Args) -> Result<ExitCode> {
     // These commands don't require `Context`
     match clargs.cmd {
@@ -56,11 +68,17 @@ fn run(clargs: cli::Args) -> Result<ExitCode> {
         _ => (),
     }
 
+    let ctx = Context::new(clargs.profile.clone(), clargs.disable_tls)?;
+
+    // Mcp Server
+    if let Cmd::Mcp = clargs.cmd {
+        return run_future_mt(cmd::mcp::run(ctx));
+    }
+
     let profile = clargs
         .profile
         .as_ref()
         .map_or_else(|| "default".to_string().cyan(), |s| s.clone().cyan());
-    let ctx = Context::new(clargs.profile, clargs.disable_tls)?;
 
     Output::new()
         .line(format!("{} profile '{profile}'", "Using".green()))

rust/crates/sift_mcp/CLAUDE.md

@@ -0,0 +1,53 @@
+# sift_mcp — guidance for Claude
+
+## Writing tool descriptions
+
+Tool descriptions in this crate are read by other agents at call time. They are the *only* documentation the calling LLM gets, so optimize for an agent making a decision under context pressure, not for a human reading the source.
+
+### Structure
+
+Use this section ordering. Skip a section only when it has no content; do not reorder.
+
+1. **One-line purpose.** First sentence states what the tool does and where the output goes ("Retrieve X and write to Y", "List Z filtered by W"). The agent should be able to match intent from this line alone.
+2. **Output schema.** When the tool returns structured data or writes a file, describe the shape — column names, types, what null means, where metadata lives. The agent will consume this output; don't make it guess.
+3. **Parameters.** One bullet per parameter, in declaration order. Spell out:
+   - Whether matching is exact or pattern-based.
+   - Conditional requirements ("required when X is omitted").
+   - Sentinel values and their meaning (e.g. `sample_ms = 0` → raw samples).
+   - Mutually exclusive choices (e.g. `Names` vs `Regex` variants).
+   - Side effects on the filesystem or external state (truncate mode, idempotency).
+4. **Errors.** Name the actual `ErrorData` variants the tool returns (`RESOURCE_NOT_FOUND`, `INVALID_PARAMS`, etc.) and the condition that triggers each. The agent can then recover with different inputs instead of treating every failure as terminal.
+5. **Guidance.** Performance characteristics, recommended call patterns, when to chunk, when to prefer one parameter shape over another. Keep this to load-bearing advice — the agent doesn't need general SQL/Arrow background.
+
+### Style rules
+
+- Write in direct voice. "Retrieve …", not "This tool retrieves …".
+- Use backticks for parameter names, enum variants, and field names so they survive Markdown rendering on the client.
+- Escape inner double quotes as `\"` — the description is a Rust string literal inside the `#[tool(...)]` attribute.
+- Prefer bullets over paragraphs for multi-fact sections (output schema, parameters, errors). Paragraphs hide structure.
+- Don't restate the obvious from the type signature. The parameter's name and type already tell the agent it's a `String` or `Option<i64>`; the description adds what isn't in the type — semantics, constraints, defaults.
+- No marketing or filler ("powerful", "easy to use"). Every line should change what the agent does.
+- Cap length around 30–40 lines. Beyond that, agents start truncating mentally; trim the guidance section first.
+
+### Reference
+
+`tool/data/mod.rs::get_data` is the canonical example. Mirror its layout when adding a new tool.
+
+### list_router tools — sourcing from protos
+
+Tools in `tool/list/` (`list_assets`, `list_runs`, `list_channels`, etc.) are thin wrappers over `sift_rs::<service>::<version>::List<Resource>Request`. Their parameters and per-parameter semantics MUST be derived from the proto comments on that message, not invented.
+
+When you add or update a list-router tool:
+
+1. **Open the matching proto.** Path pattern: `protos/sift/<service>/<version>/<service>.proto`. Find the `message List<Resource>Request { ... }` block. Examples:
+   - `list_assets` → `protos/sift/assets/v1/assets.proto::ListAssetsRequest`
+   - `list_runs` → `protos/sift/runs/v2/runs.proto::ListRunsRequest`
+   - `list_channels` → `protos/sift/channels/v3/channels.proto::ListChannelsRequest`
+2. **Copy the field comments verbatim into the tool description.** The proto authors curate the filterable/orderable field lists, default sort, page-size caps, and metadata syntax. Re-stating those in your own words risks drift; quoting from the proto keeps the tool spec aligned with the API.
+3. **Map every wrapped field to a bullet** under the `Parameters:` section of the description, using the structure in `### Structure` above. Include:
+   - Filter: list every filterable field named in the proto's `filter` comment. Preserve metadata-key syntax notes (`metadata.{key}`) and CEL helper notes (`duration(...)`).
+   - Order-by: list every orderable field, the default sort if the field is empty (assets/runs default to `created_date desc`; channels defaults to `created_date` ascending — these differ, do not assume), and the `\"FIELD_NAME[ desc],...\"` format.
+   - Limit: describe the `1..=1000` cap behavior of `service::common::paging` (different from the proto's raw `page_size`, which caps higher for some services).
+4. **Re-read the proto whenever the resource changes.** If a new filterable or orderable field is added to the proto, update the tool description in the same change. Stale descriptions are worse than missing ones because agents will trust them.
+
+If the proto's comments are themselves wrong or incomplete, fix the proto first and regenerate — the tool description is downstream of it.

rust/crates/sift_mcp/Cargo.toml

@@ -21,7 +21,15 @@ tokio.workspace = true
 tonic.workspace = true
 anyhow.workspace = true
 clap = { workspace = true, features = ["cargo"] }
+pbjson-types.workspace = true
+prost.workspace = true
+arrow.workspace = true
+parquet.workspace = true
+polars = { workspace = true, features = ["lazy", "parquet", "sql"] }
+tokio-stream.workspace = true
 
 [dev-dependencies]
 sift_test_util.workspace = true
 tokio-stream.workspace = true
+bytes.workspace = true
+tempdir.workspace = true

rust/crates/sift_mcp/src/error/mod.rs

@@ -1,27 +1,16 @@
 use rmcp::model::{CallToolResult, ErrorCode};
-use serde_json::json;
 use tonic::{Code, Status};
 
 pub type McpResult = Result<CallToolResult, rmcp::ErrorData>;
 
-pub fn from_grpc_status(status: Status) -> rmcp::ErrorData {
-    let code = from_grpc_code(status.code());
-    let message = status.message().to_string();
-    let data = Some(json!({
-        "grpc_code": status.code().to_string(),
-    }));
-
-    rmcp::ErrorData {
-        code,
-        message: message.into(),
-        data,
-    }
-}
-
 pub fn from_anyhow(error: anyhow::Error) -> rmcp::ErrorData {
-    let code = ErrorCode::INTERNAL_ERROR;
+    let mut code = ErrorCode::INTERNAL_ERROR;
     let message = format!("{error:?}");
 
+    if let Ok(grpc_status) = error.downcast::<Status>() {
+        code = from_grpc_code(grpc_status.code());
+    }
+
     rmcp::ErrorData {
         code,
         message: message.into(),

rust/crates/sift_mcp/src/lib.rs

@@ -3,12 +3,12 @@ use clap::{crate_name, crate_version};
 use rmcp::{ServiceExt, transport::stdio};
 use sift_rs::{Credentials, SiftChannelBuilder};
 
-pub(crate) mod server;
+mod server;
 use server::SiftMcpServer;
 
-pub mod tool;
-
 mod error;
+mod service;
+mod tool;
 
 pub async fn run(credentials: Credentials, use_tls: bool) -> Result<()> {
     let channel = SiftChannelBuilder::new(credentials)