Datris

Ask the Datris AI to answer a question based on provided context. Ideal for RAG workflows: first retrieve relevant chunks using a search tool (search_qdrant, search_pgvector, etc.), then pass the retrieved text as context along with the user's question to get a synthesized answer.

Apply destination column types to an all-string pipeline. REQUIRES explicit user approval first. Landed data is migrated before the config changes (postgres retypes in place; snowflake/databricks validate then swap the table) — the migration locks or replaces the table, so do NOT apply while a run is in flight. Any landed value that will not cast fails the whole apply with the column named and NOTHING changed — then either re-apply with that field set back to string, or fix the data. `fields` must list EVERY destination column (same names as get_dest_types returned) with its intended type; keep a column as-is by passing type string. If this Datris instance's agent policy requires a person to approve this action, the call does NOT perform it: it returns status `pending_approval` with an `approvalId`. Tell the user it is waiting for approval in Datris, then poll `get_approval` with that id (pause with `wait_seconds` between polls) — do not re-issue the action.

Check which backend services are up, down, or not configured. Returns the health status of PostgreSQL, MongoDB, MinIO, ActiveMQ, Kafka, and any configured vector databases (Qdrant, Weaviate, Milvus, Chroma, pgvector). Call this before attempting search or query operations to know which services are available.

Create a pipeline. FOUR destination categories: STRUCTURED (postgres, mongodb, snowflake, databricks) — send a TINY sample (via content_text) and the schema is auto-detected; snowflake additionally REQUIRES credentialsSecret (a Platform secret with account/user/privateKey or password — discover via list_platform_secrets), warehouse, and database; databricks additionally REQUIRES credentialsSecret (a Platform secret with host plus clientId/clientSecret or token), warehouse (the SQL warehouse ID), and database (the Unity Catalog name). OBJECTSTORE (objectstore — writes Parquet files, ORC files, or an Iceberg table to MinIO or AWS S3) — same shape as structured (CSV-only, sample required for schema detection), plus objectStore-specific knobs (bucket, prefix, fileFormat, partitionBy, writeMode; keyFields for iceberg+merge; provider+credentialsSecret for S3). VECTOR (pgvector, qdrant, weaviate, milvus, chroma) — no schema; pass ONLY pipeline + destination (and optionally filename for the file-extension hint), no sample at all. SCRATCH (scratch) — same sample-required shape as structured, nothing is landed; the run's rows come back on the status rollup (`resultPreview`) and via get_pipeline_result; results expire — use it when the user wants an answer now, a validation result, or a transformed view with no reason to keep the rows (see the KEEP-OR-SCRATCH RULE). SAMPLE-SIZE RULE: the sample exists ONLY for schema detection — send the header row plus 3-5 representative rows, NEVER a full dataset. Composing hundreds of rows here wastes minutes of generation time; the real data arrives later via the tap or upload_data. Prefer content_text (plain text) over content (base64) — the server encodes it for you. If this Datris instance's agent policy requires a person to approve this action, the call does NOT perform it: it returns status `pending_approval` with an `approvalId`. Tell the user it is waiting for approval in Datris, then poll `get_approval` with that id (pause with `w…

Create a tap — a fetcher that pulls data from an external source and pushes it into a pipeline. Two kinds: a Python script the platform executes (default — provide a plain-English `instruction` to have AI generate the script, or supply your own `script` directly), or `kind: "http"` — a user-hosted HTTP endpoint (any language) that Datris POSTs the run context to on each run and which responds with the tap envelope; pass `endpoint_url` and see the tap-http-contract doc. HTTP taps run no code on the platform; AI script generation and the DATRIS_PLATFORM_* platform-data callback do NOT apply to them, so keep any tap whose fetch logic reads platform data as a Python tap. If the user wants the tap to feed a pipeline, pass `target_pipeline` now. Without it, `run_tap` will fetch but not persist (response will show `persisted: false, persistedReason: "no_target_pipeline"`) — you'd then need to call update_tap to wire a pipeline. For a fetch-only tap (the user wants the rows back, not landed), target a pipeline whose destination is scratch — see the KEEP-OR-SCRATCH RULE in the server instructions. If the user mentioned ANY recurrence (nightly, daily, hourly, every morning, market open, etc.), pass `cron_expression` NOW — the platform's scheduler will run the tap on that cadence automatically. This is the canonical way to make a tap recurring; do NOT respond with shell commands or external schedulers for the user to run themselves. See the SCHEDULING RULE in the server instructions. AFTER creating, call `test_tap` to validate the script BEFORE any `run_tap` or before relying on a scheduled cron run — see the VALIDATION RULE. Setting a cron on a never-tested script is a guaranteed-bad nightly run waiting to happen. The platform enforces this: passing `cron_expression` for a script (or endpoint) that has not passed `test_tap` returns HTTP 409 with the remedy — create the tap without `cron_expression`, call `test_tap` until it succeeds, then `update_tap` with the cron. Replacin…

Create or update a secret for a tap to use. The secret's fields are injected as environment variables into the tap's Python script at runtime. Use this before create_tap when the tap needs credentials (API key, DB password, etc.). FIRST call list_tap_secrets to check whether a suitable secret already exists — if so, reuse it instead of creating a duplicate. By default, fails if a secret with this name already exists — pass overwrite=true to replace it (ask the user first). Agents can only overwrite secrets that were also created by an agent (tagged _type=tap); secrets owned by a human user must be updated via the UI. If this Datris instance's agent policy requires a person to approve this action, the call does NOT perform it: it returns status `pending_approval` with an `approvalId`. Tell the user it is waiting for approval in Datris, then poll `get_approval` with that id (pause with `wait_seconds` between polls) — do not re-issue the action.

Delete a pipeline. This is DESTRUCTIVE: by default it removes BOTH the pipeline configuration AND all data already written to the destination (MongoDB collection rows, Postgres table rows, vector-store entries). It also wipes document-tap ledgers and staged files for any tap that targets this pipeline, so a recreate gets a clean re-ingest. The platform deliberately does NOT support deleting just the config and orphaning the data — that creates ghost state. If you want to keep the config but wipe the destination data ("reset"), pass keep_config=true; the config survives, the data does not. ALWAYS confirm with the user before calling this tool. If this Datris instance's agent policy requires a person to approve this action, the call does NOT perform it: it returns status `pending_approval` with an `approvalId`. Tell the user it is waiting for approval in Datris, then poll `get_approval` with that id (pause with `wait_seconds` between polls) — do not re-issue the action.

Delete a tap and its stored script. If this Datris instance's agent policy requires a person to approve this action, the call does NOT perform it: it returns status `pending_approval` with an `approvalId`. Tell the user it is waiting for approval in Datris, then poll `get_approval` with that id (pause with `wait_seconds` between polls) — do not re-issue the action.

Delete a tap secret. Only secrets created by an agent (tagged _type=tap) can be deleted via this tool; secrets owned by a human user must be removed from the Secrets tab. Use this to clean up after a tap is no longer needed. If this Datris instance's agent policy requires a person to approve this action, the call does NOT perform it: it returns status `pending_approval` with an `approvalId`. Tell the user it is waiting for approval in Datris, then poll `get_approval` with that id (pause with `wait_seconds` between polls) — do not re-issue the action.

Compare two versions of a pipeline's definition. Returns a server-computed field-by-field config diff. Read-only. `version` is the newer/selected snapshot; `against` is the baseline.

Compare two versions of a tap's definition. Returns a server-computed field-by-field config diff and a line-level diff of the script. Read-only. `version` is the newer/selected snapshot; `against` is the baseline to compare it to.

Find datasets by meaning: ranks the pipelines your key can read against a natural-language query over their names, descriptions, tags, catalogs, destination field names, and source hosts. Each hit returns where the data lives, how fresh it is, provenance handles (latest run id, config version, script sha), lineage, and a `howToQuery` hint naming the existing query/search tool with pre-filled arguments. Discovery only — nothing is executed for you: make the query call yourself, then pass its provenance handles to ai_answer as `sources` so the answer carries them.

Read this Datris instance's agent policy: for each action (e.g. tap:delete, pipeline:update:dest-types) whether an agent may do it on its own (auto), must wait for a person to approve it (approve), or is refused (deny). Call this before a delete, schema migration, or other consequential change to know whether it will run immediately or queue for approval, and tell the user accordingly. Returns {enabled, policy:{actions, overrides, limits}, actions:[all policy-able action keys], pendingCount}. When enabled is false, every action is auto.

Poll one queued approval by the `approvalId` a mutating tool returned with status pending_approval. Returns its current state: pending (still waiting for a person), executed (approved and performed — `resultStatus` / `resultBody` carry the outcome of the original call), failed (approved but the replay failed), rejected, or expired. Poll with `wait_seconds` between calls, backing off to 30-60s; a decision can take minutes or hours, so after a few polls tell the user you will check back and stop polling unless they ask you to wait.

Propose real column types for a pipeline whose destination columns are all stored as text (the default for agent-created pipelines). Stateless: samples up to 1000 rows that already landed in the destination, infers types deterministically, and returns per-column evidence — a few sample values, plus (for columns kept as string by a dirty value) the offending value and the type it blocked. Only for postgres, snowflake, and databricks destinations; `eligible: false` with a `reason` of destination-not-supported, already-typed, or no-landed-rows (run the pipeline once first) otherwise. Nothing is stored or changed by this call.

Read one recovery-agent incident by id — its trigger, classification, proposal, step-by-step narrative, approvals it is waiting on, and outcome. When the operator asks about a failure the platform is already working, explain THIS record instead of re-diagnosing from scratch.

Get job status for an upload_data submission. Pass `pipeline_token` (returned from upload_data) for the recommended path. Pass `pipeline_name` instead for a paginated summary of recent jobs for that pipeline. When queried by `pipeline_token`, the response is `{rollup: {allDone, status, jobs: [...]}, events: [...]}` — poll every few seconds until `rollup.allDone` is true, then read `rollup.status` (`success` | `warning` | `error`) for the outcome. Per-job detail is in `rollup.jobs[]` with `pipelineToken`, `pipeline`, `filename`, `status`, `startedAt`, `lastEventAt`, `elapsed`, and `lastError` (populated on failure with `processName` and `description`). `events[]` is the raw begin/info/end audit trail; the rollup is the source of truth for completion. When queried by `pipeline_name`, the response is a paginated array of summary rows; the most recent job is index 0 and its `status` field is `success` | `processing` | `error`. Do NOT proceed to query/search until the job is in a terminal state.

Traverse the lineage graph from one node: everything upstream (what feeds it) and/or downstream (what depends on it) of a source, tap, pipeline, dataset, or catalog, with the edges between them, freshness for the pipeline involved, and optionally the most recent recorded runs (what each run read and wrote, per destination). Datasets a pipeline used to land into under an earlier configuration are marked `historical`. Use it for impact analysis before changing or deleting something — 'what is downstream of this tap?' — or to see which datasets a pipeline actually wrote. Node names: a tap or pipeline name; a dataset's `kind:coordinates` id as returned by find_data or list_pipelines lineage; a source host; a catalog name.

Get a specific pipeline configuration by name. Returns the full JSON config including source, dataQuality, transformation, preprocessor, and destination sections.

Read the rows a scratch pipeline produced. Only pipelines whose destination is scratch have a result — anything else is a 404 (an error body of `Not Found` / `status: 404` — rather than the endpoint's own `Only scratch pipelines have a result` message — for a pipeline you know is scratch means the server predates scratch results: report the version mismatch and stop, do not retry). The first rows are already on the `get_pipeline_status` rollup as `resultPreview`, so call this ONLY when the rollup's `resultTruncated` is true and you need the rest. Paging with `offset`/`limit` never re-runs the source — it reads the stored result. `limit` is clamped server-side, so the next offset is always `offset + returnedCount`, never `offset + limit`. Results expire after the retention window, so read them promptly; a 410 means the result is gone and the pipeline must be run again. Pass `publisher_token` (from run_tap) or `pipeline_token` (a single ingestion job); exactly one of the two must be supplied. Response shape: `{records: [...], rowCount, returnedCount, offset, truncated, resultUri, resultExpiresAt}` — `rowCount` is the whole result, `returnedCount` this page.

Read pipeline ingestion status. Use this after `run_tap` to watch a tap-submitted load progress. Pass `publisher_token` to see every job the tap run submitted (the recommended option — works for both structured and document taps). Pass `pipeline_token` for a single ingestion job. Exactly one of the two must be supplied. Response shape: `{rollup: {allDone, status, jobs: [...]}, events: [...]}`. Poll every few seconds until `rollup.allDone` is true, then read `rollup.status` (`success` | `warning` | `error`) for the outcome. Per-job detail is in `rollup.jobs[]` — each entry has `pipelineToken`, `pipeline`, `filename`, `status`, `startedAt`, `lastEventAt`, `elapsed`, and `lastError` (populated on failure with `processName` and `description`). `events[]` is the raw begin/info/end audit trail if you need it; the rollup is the source of truth for completion.

View one historical snapshot of a pipeline's definition: the full config as it was at that version. Read-only; does not change the live pipeline. Get version numbers from `list_pipeline_versions`.

Return the FIELD NAMES (keys only — never values) of an existing platform secret. Use this after list_platform_secrets to verify a candidate secret has the keys a destination config requires (e.g. an S3 credentialsSecret must contain accessKey, secretKey, region). Secret values are intentionally NOT returned. If the named secret is tap-tagged, this tool refuses — use get_tap_secret_fields for those.

Resolve a stamped `_datris_run_id` value back to its origin: the pipeline run, the tap run that fed it, the script commit, the pipeline config version, and the declared source. Use it when rows or search results carry `_datris_*` provenance fields (pipelines with provenance stamping enabled). Returns one document walking the whole chain.

Get a tap's static definition: configuration, schedule, target pipeline, and the generated Python script content. This is config-only — it returns the SAME data on every call and tells you NOTHING about run state. Do NOT call this to check whether a run is finished or to poll for completion — repeatedly calling get_tap after run_tap is a bug, the response will never change to reflect ingestion progress. For run status: call `get_pipeline_status(publisher_token=...)` (token comes from the run_tap response or from a get_tap_logs entry). For run history: call `get_tap_logs`.

For a document tap: return the ledger of discovered documents (URI, filename, status, hashes, first/last seen timestamps). The ledger is what tells the platform which documents have already been processed so re-runs skip unchanged files. Pass 'clear_uri' to delete one entry (forces that document to be re-processed on the next run) or 'clear_all=true' to wipe the entire ledger (forces a full re-scan).

Get the run history for a tap. Returns the last 50 run log entries sorted by most recent first, including status, record count, duration, errors, logs, and `publisherToken` for each run that submitted records to a pipeline. Works for both manual runs (triggered by `run_tap`) and scheduled runs (triggered by the platform's cron scheduler) — they share the same log. Use this to verify whether a scheduled run fired, whether any recent run's script succeeded, and to recover the `publisherToken` for any run if you didn't keep the original `run_tap` response in context. To verify the actual destination ingestion outcome — not just that the script ran — pick the relevant entry and call `get_pipeline_status(publisher_token=entry.publisherToken)`. The tap log only records what the script did; the publisher token is how you trace a run through to whether the data actually landed in the destination.

Return the FIELD NAMES (keys only — never values) of an existing tap secret. Use this after list_tap_secrets to verify a candidate secret has the keys your tap script will need (e.g. API_KEY, USER_AGENT). Secret values are intentionally NOT returned and are never visible to the agent.

Get a tap's incremental-sync state — the bookmark/cursor its script saved after the last successful run (injected into the next run as the DATRIS_TAP_STATE env var). Returns `{tap, state, updatedAt, updatedBy}`; `state` is null when the tap has never committed state (non-incremental tap, or no successful run yet). Use this to see where an incremental tap will resume from, or to debug why a scheduled tap is re-fetching or skipping a window.

View one historical snapshot of a tap's definition: the full config and the pinned Python script as they were at that version. Read-only; does not change the live tap. Get version numbers from `list_tap_versions`.

Get the Datris server version.

Kill a running pipeline job by its pipeline token. The job thread will be interrupted and the job marked as cancelled. If this Datris instance's agent policy requires a person to approve this action, the call does NOT perform it: it returns status `pending_approval` with an `approvalId`. Tell the user it is waiting for approval in Datris, then poll `get_approval` with that id (pause with `wait_seconds` between polls) — do not re-issue the action.

List all collections in the Chroma vector database. Use this to discover available collections before running search_chroma.

List the platform's recovery-agent incidents, newest first. An incident is opened by the platform itself when a scheduled tap finally fails, a pipeline job errors, a tap goes stale, or a pipeline's volume swings; the recovery agent diagnoses it and — within the agent policy's recovery mode — repairs and verifies it. Each entry carries kind, resource, state (open | diagnosing | proposed | awaiting_approval | executing | verifying | resolved | failed | abandoned), classification, and a step-by-step narrative. Use state=open for what is being worked right now. Returns {enabled:false, incidents:[]} when the recovery agent is off.

List all collections in the Milvus vector database. Use this to discover available collections before running search_milvus.

List MongoDB collections. If database is specified, lists collections in that database. If omitted, lists all collections across all databases in 'db.collection' format.

List all MongoDB databases available in the Datris platform. Use this as the first step when exploring what data has been ingested into MongoDB destinations.

List the actions this agent queued for human approval under the agent policy, newest first. Each entry has id, action, resource, state (pending | approved | rejected | expired | executed | failed), createdAt, expiresAt and — once decided — decidedBy and the result. Use `state` to filter (default: all).

List all pgvector tables (tables with an embedding column) in PostgreSQL. Use this to discover available collections before running search_pgvector.

List the change HISTORY of a pipeline's definition, newest first. Each entry has `version`, `createdAt`, `createdBy`, and `changeNote`. The platform snapshots a pipeline's full config every time it is created or updated. Read-only. IMPORTANT: an EMPTY result does NOT mean the pipeline has no version — it means it has not been edited since versioning was enabled. The CURRENT version number is the `version` field on the pipeline itself (from `list_pipelines` or `get_pipeline`), at least 1 for every pipeline. Use this tool for 'what changed and when'; use `list_pipelines`/`get_pipeline` for 'what version is it on now'. See also `get_pipeline_version`, `diff_pipeline_versions`, `restore_pipeline_version`.

List all registered pipeline configurations. Each pipeline defines a complete data processing flow: source format and schema, AI-powered data quality and transformations, and destination (database, message queue, or vector store). CALL THIS FIRST on any data-related user request — before suggesting external sources, before asking scope questions. The user almost always cares more about what's already in their Datris environment than about a generic options menu.

List the names of PLATFORM secrets (all secrets NOT tagged _type=tap — the Platform tab in the UI's Secrets section). These are human-owned credentials for destinations and infrastructure (e.g. S3 destination credentials, Postgres connection, MongoDB connection, embedding/vector-store endpoints). The agent can READ these (to look up their names and field shape via get_platform_secret_fields and reference them in pipeline configs) but cannot create, update, or delete them — those operations are the user's responsibility via the Secrets tab. Use this whenever a pipeline's destination needs a credentialsSecret reference (e.g. objectStore with provider=s3) and you need to discover which secrets the user already has.

List all columns and their data types for a specific PostgreSQL table. Use this to understand table structure before writing a query_postgres SQL query.

List all PostgreSQL databases available in the Datris platform. Use this as the first step when exploring what data has been ingested into PostgreSQL destinations.

List all schemas in a PostgreSQL database. Schemas organize tables within a database (e.g., 'public', 'analytics'). Use after list_postgres_databases to drill into a specific database.

List all tables in a PostgreSQL schema. Set vector_only=true to show only pgvector embedding tables, or false (default) to show regular data tables. Use after list_postgres_schemas.

List all collections in the Qdrant vector database. Use this to discover available collections before running search_qdrant.

List the names of tap secrets that already exist (secrets tagged _type=tap — both agent-created and human-owned ones surfaced via the UI's Tap Secrets section). ALWAYS call this before create_tap_secret: if a suitable secret already exists, prefer reusing it by passing its name as secret_name to create_tap. Only ask the user to provide credentials when no existing secret covers the need.

List the change HISTORY of a tap's definition, newest first. Each entry has `version`, `createdAt`, `createdBy`, and `changeNote`. The platform snapshots a tap's config + script every time it is created or updated. Read-only. IMPORTANT: an EMPTY result does NOT mean the tap has no version — it means the tap has not been edited since versioning was enabled, so no change snapshots exist yet. The tap's CURRENT version number is the `version` field on the tap itself (from `list_taps` or `get_tap`), which is at least 1 for every tap. Use this tool for 'what changed and when'; use `list_taps`/`get_tap` for 'what version is it on now'. See also `get_tap_version`, `diff_tap_versions`, `restore_tap_version`.

List all taps with their status, target pipeline, schedule, and last run info. CALL THIS FIRST (alongside list_pipelines) on any data-related user request — the user's existing taps often reveal what's already being pulled and where, which short-circuits the entire 'should I suggest an external API?' conversation.