MCP Server (AI Agent Integration)

Documentation Index

Fetch the complete documentation index at: https://docs.datris.ai/llms.txt

Use this file to discover all available pages before exploring further.

​

Resources

​

Available Tools

​

Pipeline Management

​

Vector Database Search

​

Database Queries

​

Metadata Discovery

​

AI

​

Taps

​

Discovery

​

Configuration

​

Managed Service

​

Monitoring Active Agents

​

Setup

​

Docker (automatic)

​

Local (for Claude Desktop / Claude Code)

uvx datris-mcp-server

​

Transport Modes

​

Configuring Claude

​

Environment Variables

​

Authentication

​

Example Agent Workflows

​

Ingest a CSV file directly

1. Check for an existing pipeline — list_pipelines. If one already fits the data, skip to step 3.
2. Create a pipeline — create_pipeline with the CSV as sample data. Schema is auto-detected. Keep the config simple; only pass codegen_rule / codegen_transform if the user explicitly asks for validation or transformation.
3. Upload — upload_data with the CSV content (base64).
4. Verify it landed — get_job_status with the pipelineToken returned from upload_data. Poll until rollup.allDone is true, then read rollup.status (success, warning, error). On failure, rollup.jobs[].lastError carries the failing process and message.

​

Onboard a new external data source via a tap

1. Check existing taps — list_taps. If a tap already covers this source, run or test it directly.
2. Discover (optional) — discover_source with a natural-language question like “what datasets are available in yfinance?”. Returns a structured catalog with a tapInstruction template per dataset.
3. Create credentials — if the source needs an API key, call create_tap_secret with the credential fields (they become env vars inside the script).
4. Create the tap — create_tap with an instruction (AI generates the script) or with your own Python fetch() function (faster and more reliable). Pass secret_name to bind the credentials from step 3. For PDFs/Word/HTML into a vector-store pipeline, pass tap_type="document".
5. Test — test_tap. If it fails, read the error, fix the script, and call create_tap again. Repeat until the test passes.
6. Run — run_tap. Read the response’s persisted field first:
   • persisted: true → capture publisherToken and continue to step 7.
   • persisted: false → read persistedReason (no_target_pipeline, test_mode, run_error, no_records, debounced), tell the user exactly why, and stop. (run_tap does not return records — call test_tap if you need to preview what the script produces.) debounced means the same tap was triggered within the last 5 seconds; the earlier run is still executing — do not retry, look it up via get_tap_logs and poll get_pipeline_status instead.
7. Verify it landed — get_pipeline_status(publisher_token=...) and poll until rollup.allDone is true. Then rollup.status is the outcome (success, warning, error) and rollup.jobs[].lastError carries any failure detail. The tap log only tells you the script ran; the publisher token is how you confirm records actually reached the destination.
8. Schedule (optional) — update_tap with a cron_expression. For each scheduled run, pick the corresponding entry from get_tap_logs to get its publisherToken, then verify with step 7.

​

Build and query a RAG knowledge base from external documents

1. Create a vector-store pipeline — create_pipeline with a Qdrant / Weaviate / Milvus / pgvector / Chroma destination and an unstructuredAttributes source.
2. Create a document tap — create_tap with tap_type="document" and an instruction describing the document source (e.g., “ingest every PDF under legal-contracts/2026/ in S3”). If the source needs credentials, call create_tap_secret first.
3. Test and run — test_tap, then run_tap. Read persisted / persistedReason exactly as above.
4. Verify every document landed — get_pipeline_status(publisher_token=...). Document taps fan out to N jobs (one per file) under one publisher token; poll until rollup.allDone is true. rollup.jobs[] has one entry per file with its own status and lastError.
5. Inspect the ledger (optional) — get_tap_ledger shows which files were discovered and processed. On subsequent runs, unchanged files are skipped automatically; pass clear_uri or clear_all=true to force a re-scan.
6. Search — search_qdrant (or the matching tool for your destination) to retrieve relevant chunks.
7. Answer — ai_answer with the retrieved chunks as context and the user’s question.

​

Discover and query existing data

1. List databases — list_postgres_databases.
2. List schemas — list_postgres_schemas for the target database.
3. List tables — list_postgres_tables (supports a vector-only filter).
4. Inspect columns — list_postgres_columns to understand structure.
5. Query — query_postgres with a SELECT, or query_natural to ask a plain-English question and have the AI generate + run the SQL.

​

Cross-modal analysis (structured + vector)

1. Vector search — search_pgvector (or another vector tool) for relevant document chunks.
2. Structured query — query_postgres or query_natural for related metrics.
3. Combine — merge the two result sets in the response.

​

Automated quality monitoring of scheduled taps

1. List taps — list_taps to find the tap of interest.
2. Read recent runs — get_tap_logs returns the last 50 entries with status, record count, duration, errors, and publisherToken for each run that submitted records.
3. Verify completion — for any run of interest, call get_pipeline_status(publisher_token=...) and read rollup.status to confirm records actually landed in the destination. The tap log tells you the script ran; the publisher token tells you the load finished.
4. Diagnose failures — on an errored job, read rollup.jobs[].lastError (processName + description) and either fix the tap (create_tap again with a corrected script) or retarget the pipeline (update_tap).

​

CLI Examples

# Install
brew tap datris/tap
brew install datris

# Ingest a CSV into PostgreSQL
datris ingest sales-data.csv --dest postgres

# Ingest with AI validation and transformation
datris ingest trades.csv --dest postgres \
  --ai-validate "all prices must be positive and dates must be YYYY-MM-DD" \
  --ai-transform "convert dates to YYYY/MM/DD and uppercase all ticker symbols"

# Ingest into a vector store for RAG
datris ingest manual.pdf --dest pgvector

# Ingest + analyze in one command
datris ingest trades.csv --dest postgres --ai-analyze "What are the top 5 stocks by volume?"
datris ingest report.pdf --dest pgvector --ai-analyze "What was the company's revenue?"

# Analyze existing data (PostgreSQL, MongoDB, or vector stores)
datris analyze "top 5 stocks by volume" --table trades
datris analyze "What is the return policy?" --table support_docs --dest pgvector

# Query PostgreSQL directly
datris query "SELECT * FROM public.sales LIMIT 10"

# Semantic search (raw results)
datris search "quarterly revenue" --store pgvector --collection financial_docs

# List pipelines, check health, get status
datris pipelines
datris health
datris status my_pipeline