Agent Reference

/plugin marketplace add definite-app/claude-skills
/plugin install definite@definite

# Add Definite MCP server
claude mcp add definite --transport http https://api.definite.app/v3/mcp/http --header "Authorization: YOUR_API_KEY"

# Download this page as your local CLAUDE.md
curl -o CLAUDE.md https://docs.definite.app/mcp/agent-reference.md

run_sql_query(sql="SELECT table_schema, table_name FROM information_schema.tables WHERE table_catalog = 'LAKE' LIMIT 100")

run_cube_query(cube_query={
  "measures": ["sales.revenue"],
  "dimensions": ["sales.region"],
  "timeDimensions": [{"dimension": "sales.order_date", "dateRange": "last 90 days", "granularity": "month"}],
  "order": [["sales.revenue", "desc"]],
  "limit": 1000
})

version: 1
schemaVersion: "2025-01"
kind: dashboard

metadata:
  name: "Sales Dashboard"
  description: "Revenue by region"

layout:
  columns: 36
  tiles:
    - id: revenue_chart
      x: 0
      y: 0
      w: 18
      h: 10
      type: chart
      title: "Revenue by Region"
      source: revenue_data
    - id: total_kpi
      x: 18
      y: 0
      w: 9
      h: 6
      type: kpi
      source: total_revenue

datasets:
  revenue_data:
    engine: cube
    cube: sales
    measures:
      - sales.revenue
    dimensions:
      - sales.region
    viz:
      type: bar
  total_revenue:
    engine: sql
    sql: |
      SELECT SUM(amount) as total FROM LAKE.PUBLIC.sales
    viz:
      type: kpi

app.json (manifest)  -->  Platform (server-side fetch)  -->  DuckDB WASM (browser)  -->  App.tsx (client-side SQL)

apps-v2/my-app/
  app.json            # Manifest: declares data resources
  src/main.tsx        # Entry point (boilerplate)
  src/App.tsx         # UI code using runtime hooks
  dist/index.html     # Built artifact

{
  "version": 2,
  "name": "Sales Dashboard",
  "entry": "src/main.tsx",
  "resources": {
    "sales": {
      "kind": "dataset",
      "source": {
        "type": "sql",
        "sql": "SELECT region, STRFTIME(order_date, '%Y-%m-%d') AS orderDate, amount::DOUBLE AS amount FROM LAKE.PUBLIC.orders LIMIT 200000"
      },
      "public": false
    }
  }
}

import { useDataset, useSqlQuery, useTheme, AppShell, KpiCard } from "./definite-runtime";

export default function App() {
  const { theme, toggleTheme } = useTheme();
  const data = useDataset("sales");
  const kpis = useSqlQuery(data, data.tableRef
    ? `SELECT SUM(amount)::INTEGER AS revenue FROM ${data.tableRef}`
    : "", []);

  return (
    <AppShell title="Sales Dashboard" theme={theme} onToggleTheme={toggleTheme}>
      <KpiCard title="Revenue" value={kpis.data?.[0]?.revenue} format="currency" />
    </AppShell>
  );
}

version: 1
schemaVersion: "2025-01"
kind: dashboard
metadata:
  name: "Sales Dashboard"
datasets: {}
layout:
  columns: 36
  tiles:
    - id: app
      x: 0
      y: 0
      w: 36
      h: 22
      type: html
      fullScreen: true
      driveFile: "apps-v2/my-app/dist/index.html"

#!/usr/bin/env -S uv run
# /// script
# dependencies = ["definite-sdk", "duckdb"]
# ///
from definite_sdk import DefiniteClient
import os

client = DefiniteClient(os.environ["DEFINITE_API_KEY"], api_url=os.environ["DEFINITE_API_BASE_URL"])
sql = client.get_sql_client()

result = sql.execute("SELECT * FROM LAKE.PUBLIC.users WHERE created_at >= CURRENT_DATE - INTERVAL '1 day' LIMIT 2000")
rows = result.get("data", [])
print(f"Found {len(rows)} new users")

# 1. Call MCP tool: get_drive_upload_url(file_name="my_sync.py", folder="pipelines")
# 2. Upload using the returned signed URL:
curl -X PUT -T ./my_sync.py "SIGNED_UPLOAD_URL"

version: 1
schemaVersion: "2025-01"
kind: pipeline

metadata:
  name: "Daily User Sync"

datasets:
  prepare_staging:
    engine: sql
    sql: |
      CREATE OR REPLACE TABLE LAKE.STAGING.daily_snapshot AS
      SELECT * FROM LAKE.PUBLIC.users
      WHERE created_at >= CURRENT_DATE - INTERVAL '1 day'

  run_sync:
    engine: python
    command: "uv run /home/user/drive/pipelines/my_sync.py"
    dependsOn: [prepare_staging]
    fiEnvId: "<_THREAD_ID>"
    timeoutMs: 300000

automations:
  - trigger:
      type: schedule
      config:
        type: cron
        cron: "0 9 * * *"

automations:
  - trigger:
      type: schedule
      config:
        type: cron
        cron: "0 9 * * *"     # Daily at 9am UTC
    destination:               # Optional: send snapshot
      type: slack
      channel_id: "C1234..."

from definite_sdk import DefiniteClient
import os

# In pipelines, env vars are auto-injected
# For local use, set DEFINITE_API_KEY env var or pass directly
client = DefiniteClient(
    os.environ["DEFINITE_API_KEY"],
    api_url=os.environ.get("DEFINITE_API_BASE_URL", "https://api.definite.app")
)

sql_client = client.get_sql_client()
result = sql_client.execute("SELECT * FROM LAKE.PUBLIC.users LIMIT 10")

cube_query = {
    "measures": ["sales.revenue"],
    "timeDimensions": [{"dimension": "sales.date", "granularity": "month"}],
    "limit": 1000
}
result = sql_client.execute_cube_query(cube_query, integration_id="YOUR_CUBE_INTEGRATION_ID")

state = client.get_kv_store('my_pipeline')  # Unique name per pipeline
last_sync = state.get('last_sync_at')
state['last_sync_at'] = '2026-01-31T00:00:00'
state.commit()  # Persist changes

secrets = client.secret_store()
secrets.set_secret("external_api_key", "sk-...")
key = secrets.get_secret("external_api_key")

drive = client.get_drive_client()
sql = client.get_sql_client()

r = drive.write_temporary_file(parquet_bytes, name="events.parquet", ttl_days=7)
sql.execute(
    f"CREATE OR REPLACE TABLE LAKE.MY_SCHEMA.EVENTS AS "
    f"SELECT * FROM read_parquet('{r.gcs_path}')"
)

import duckdb
conn = duckdb.connect()
conn.execute(client.attach_ducklake())  # deprecated
df = conn.sql("SELECT * FROM lake.PUBLIC.users LIMIT 10").df()

integrations = list(client.integration_store().list_integrations(integration_type="slack"))
slack_id = integrations[0]["id"]

msg = client.get_message_client()

# Simple text message
msg.send_slack_message(
    integration_id="YOUR_SLACK_INTEGRATION_ID",
    channel_id="C1234567890",
    text="Hello from Definite!"
)

# Rich message with Slack Block Kit
msg.send_slack_message(
    integration_id="YOUR_SLACK_INTEGRATION_ID",
    channel_id="C1234567890",
    text="Daily report ready",  # Fallback for notifications
    blocks=[
        {"type": "header", "text": {"type": "plain_text", "text": "Daily Report"}},
        {"type": "section", "text": {"type": "mrkdwn", "text": "*New users today:* 42\n*Revenue:* $12,500"}},
        {"type": "actions", "elements": [
            {
                "type": "button",
                "text": {"type": "plain_text", "text": "View Dashboard"},
                "url": "https://ui.definite.app/docs/DOC_ID",
                "style": "primary"
            }
        ]}
    ],
    thread_ts="1234567890.123456"  # Optional: reply to thread
)

msg = client.get_message_client()
msg.send_email_message(
    to_emails=["team@example.com"],
    subject="Daily Report: 42 new users",
    body="<h2>New Users</h2><p>42 users signed up today.</p>"  # HTML supported
)

#!/usr/bin/env -S uv run
# /// script
# dependencies = ["definite-sdk", "duckdb"]
# ///
from definite_sdk import DefiniteClient
import os

client = DefiniteClient(os.environ["DEFINITE_API_KEY"], api_url=os.environ["DEFINITE_API_BASE_URL"])
sql = client.get_sql_client()

result = sql.execute(
    "SELECT email, created_at FROM LAKE.PUBLIC.users "
    "WHERE created_at >= CURRENT_DATE - INTERVAL '1 day' LIMIT 2000"
)
rows = result.get("data", [])

if rows:
    table_rows = "".join(
        f"<tr><td>{r['email']}</td><td>{r['created_at']}</td></tr>" for r in rows
    )
    body = (
        f"<h2>New Users: {len(rows)}</h2>"
        f"<table><tr><th>Email</th><th>Signed Up</th></tr>{table_rows}</table>"
    )
    client.get_message_client().send_email_message(
        to_emails=["team@example.com"],
        subject=f"Daily: {len(rows)} new users",
        body=body,
    )

#!/usr/bin/env -S uv run
# /// script
# dependencies = ["definite-sdk", "duckdb"]
# ///
from definite_sdk import DefiniteClient
import os

client = DefiniteClient(os.environ["DEFINITE_API_KEY"], api_url=os.environ["DEFINITE_API_BASE_URL"])
sql = client.get_sql_client()
state = client.get_kv_store('mrr_alert')

result = sql.execute(
    "SELECT SUM(amount) as mrr FROM LAKE.PUBLIC.subscriptions WHERE status = 'active' LIMIT 1"
)
current_mrr = float(result["data"][0]["mrr"])
previous_mrr = float(state.get('last_mrr', '0'))

if previous_mrr > 0:
    pct_change = (current_mrr - previous_mrr) / previous_mrr
    if abs(pct_change) >= 0.20:
        direction = "up" if pct_change > 0 else "down"
        client.get_message_client().send_slack_message(
            integration_id="YOUR_SLACK_INTEGRATION_ID",
            channel_id="C1234567890",
            text=f"MRR Alert: {direction} {abs(pct_change):.0%}",
            blocks=[
                {"type": "header", "text": {"type": "plain_text", "text": "MRR Alert"}},
                {
                    "type": "section",
                    "text": {
                        "type": "mrkdwn",
                        "text": f"MRR moved *{direction} {abs(pct_change):.0%}*\nFrom ${previous_mrr:,.0f} to ${current_mrr:,.0f}"
                    }
                },
            ],
        )

state['last_mrr'] = str(current_mrr)
state.commit()

version: 1
schemaVersion: "2025-01"
kind: pipeline

metadata:
  name: "MRR Alert"

datasets:
  check_mrr:
    engine: python
    command: "uv run /home/user/drive/alerts/mrr_alert.py"
    fiEnvId: "<_THREAD_ID>"
    timeoutMs: 600000

automations:
  - trigger:
      type: schedule
      config:
        type: cron
        cron: "0 9 * * *"

import os
with open(os.environ["DEFINITE_RESULT_TYPE_FILE"], "w") as f:
    f.write("html")
with open(os.environ["DEFINITE_RESULT_FILE"], "w") as f:
    f.write(html_content)