MCP Tools for dbt — Run artifacts, model status, test failures, source freshness, and job monitoring

dbt artifacts: the canonical health source

dbt produces four key artifact files in the target/ directory after each run. manifest.json describes the entire project — all models, tests, sources, seeds, and their DAG dependencies. run_results.json records the execution result of every node in the most recent run. sources.json records the result of dbt source freshness checks. catalog.json records schema information compiled from the data warehouse.

MCP tools should read artifact files from a shared location (S3 bucket, GCS, artifact store) rather than from the dbt runner's local filesystem. Most dbt Cloud setups upload artifacts to an S3 bucket after each run. For dbt Core running in CI/CD, upload the target/ directory as a build artifact. The metadata.generated_at field in each artifact file records when the run completed — if this timestamp is older than your expected run cadence × 1.5, the dbt job did not run on schedule.

async function loadDbtArtifacts(artifactBaseUrl) {
  // artifactBaseUrl: S3 pre-signed URL prefix, or local file path prefix
  const [runResults, sources, manifest] = await Promise.allSettled([
    fetch(`${artifactBaseUrl}/run_results.json`).then(r => {
      if (!r.ok) throw new Error(`run_results.json → ${r.status}`);
      return r.json();
    }),
    fetch(`${artifactBaseUrl}/sources.json`).then(r => {
      if (!r.ok) throw new Error(`sources.json → ${r.status}`);
      return r.json();
    }),
    fetch(`${artifactBaseUrl}/manifest.json`).then(r => {
      if (!r.ok) throw new Error(`manifest.json → ${r.status}`);
      return r.json();
    }),
  ]);

  return {
    runResults: runResults.status === 'fulfilled' ? runResults.value : null,
    sources: sources.status === 'fulfilled' ? sources.value : null,
    manifest: manifest.status === 'fulfilled' ? manifest.value : null,
    artifactErrors: [runResults, sources, manifest]
      .filter(r => r.status === 'rejected')
      .map(r => r.reason.message),
  };
}

Parsing run_results.json: models vs tests vs seeds

The results array in run_results.json contains one entry per node that was executed. Each result has a unique_id prefixed with the node type: model.project_name.model_name, test.project_name.test_name, seed.project_name.seed_name, snapshot.project_name.snapshot_name. Split on the first period to classify node type before assessing status.

dbt status values and their meaning: success (SQL executed and row count is within expectations), error (SQL execution failed — check message field for the database error), warn (for tests only — the test condition was violated but severity is configured to warn, not error; pipeline did not fail), skip (node was skipped because an upstream dependency failed), pass (for tests — the test condition was not violated).

function parseRunResults(runResults) {
  if (!runResults || !runResults.results) {
    return { valid: false, reason: 'empty_or_missing_run_results' };
  }

  const metadata = runResults.metadata;
  const generatedAt = metadata?.generated_at ? new Date(metadata.generated_at) : null;
  const elapsedTime = runResults.elapsed_time;  // Total run duration in seconds

  // Classify results by node type
  const byType = { model: [], test: [], seed: [], snapshot: [], source: [], other: [] };

  for (const result of runResults.results) {
    const nodeType = result.unique_id.split('.')[0];
    const bucket = byType[nodeType] ?? byType.other;
    bucket.push({
      uniqueId: result.unique_id,
      status: result.status,        // 'success' | 'error' | 'warn' | 'skip' | 'pass'
      executionTimeMs: (result.execution_time ?? 0) * 1000,
      message: result.message ?? null,
      failures: result.failures,    // For tests: number of failing rows
      // Adapter response (row count, etc.)
      rowsAffected: result.adapter_response?.rows_affected ?? null,
    });
  }

  // Model health
  const failedModels = byType.model.filter(m => m.status === 'error');
  const skippedModels = byType.model.filter(m => m.status === 'skip');
  const slowModels = byType.model.filter(m => m.executionTimeMs > 300_000);  // > 5 minutes

  // Test health — distinguish errors (blocking) from warns (non-blocking)
  const failedTests = byType.test.filter(t => t.status === 'error');
  const warnedTests = byType.test.filter(t => t.status === 'warn');
  const passedTests = byType.test.filter(t => t.status === 'pass');

  // Cascading failure detection: skipped models imply upstream failures
  const cascadeDepth = skippedModels.length > 0 && failedModels.length > 0
    ? Math.round(skippedModels.length / failedModels.length)
    : 0;

  return {
    valid: true,
    generatedAt,
    elapsedTime,
    // Run-level status (dbt sets this from invocation return code)
    runStatus: runResults.results.some(r => r.status === 'error') ? 'error' : 'success',
    models: {
      total: byType.model.length,
      succeeded: byType.model.filter(m => m.status === 'success').length,
      failed: failedModels.length,
      skipped: skippedModels.length,
      slow: slowModels.length,
      failedDetails: failedModels.map(m => ({ id: m.uniqueId, message: m.message })),
    },
    tests: {
      total: byType.test.length,
      passed: passedTests.length,
      failed: failedTests.length,    // severity: error — blocking
      warned: warnedTests.length,    // severity: warn — non-blocking but degraded
      failedDetails: failedTests.map(t => ({ id: t.uniqueId, failures: t.failures, message: t.message })),
      warnedDetails: warnedTests.map(t => ({ id: t.uniqueId, failures: t.failures })),
    },
    cascadeDepth,
    seeds: { total: byType.seed.length, failed: byType.seed.filter(s => s.status === 'error').length },
    snapshots: { total: byType.snapshot.length, failed: byType.snapshot.filter(s => s.status === 'error').length },
  };
}

Source freshness: the leading indicator of pipeline staleness

Source freshness checks (dbt source freshness) verify that raw data is arriving from upstream systems on schedule. The sources.json artifact records, for each source table, the max_loaded_at timestamp (the most recent row timestamp in the table), the snapshotted_at timestamp (when dbt checked the source), and the status based on configured thresholds.

Source freshness is the most important leading indicator because it identifies data pipeline breaks upstream of dbt — if raw data stops arriving, all downstream dbt models will succeed (they'll just produce stale results from cached data). A source freshness warn that is not investigated often becomes an error on the next run, followed by all dependent models producing silently stale output for hours before anyone notices.

function parseSourceFreshness(sourcesArtifact) {
  if (!sourcesArtifact || !sourcesArtifact.results) {
    return { valid: false, reason: 'no_source_freshness_artifact' };
  }

  const now = new Date();
  const snapshotTime = new Date(sourcesArtifact.metadata?.generated_at ?? now);

  const sources = sourcesArtifact.results.map(r => {
    const maxLoadedAt = r.max_loaded_at ? new Date(r.max_loaded_at) : null;
    const snappedAt = r.snapshotted_at ? new Date(r.snapshotted_at) : snapshotTime;

    // Age of source data at time of freshness check
    const ageSecs = maxLoadedAt ? (snappedAt.getTime() - maxLoadedAt.getTime()) / 1000 : null;

    // dbt freshness criteria
    const warnAfterSecs = criteriaToSeconds(r.criteria?.warn_after);
    const errorAfterSecs = criteriaToSeconds(r.criteria?.error_after);

    return {
      uniqueId: r.unique_id,
      sourceName: r.unique_id.split('.').slice(1).join('.'),
      status: r.status,             // 'pass' | 'warn' | 'error'
      maxLoadedAt,
      snappedAt,
      ageSecs,
      ageHuman: ageSecs !== null ? formatDuration(ageSecs) : null,
      warnAfterSecs,
      errorAfterSecs,
      // How far past the warn threshold (negative = within threshold)
      warnOverageSecs: ageSecs !== null && warnAfterSecs !== null ? ageSecs - warnAfterSecs : null,
    };
  });

  const fresh = sources.filter(s => s.status === 'pass');
  const staleWarn = sources.filter(s => s.status === 'warn');
  const staleError = sources.filter(s => s.status === 'error');

  // Find sources that are deeply stale (warn for more than 2x warn threshold)
  const criticallyStale = staleWarn.filter(s =>
    s.warnOverageSecs !== null && s.warnAfterSecs !== null && s.warnOverageSecs > s.warnAfterSecs
  );

  return {
    valid: true,
    total: sources.length,
    fresh: fresh.length,
    staleWarn: staleWarn.length,
    staleError: staleError.length,
    criticallyStale: criticallyStale.length,
    sources,
    errorDetails: staleError.map(s => ({ id: s.sourceName, ageSecs: s.ageSecs, ageHuman: s.ageHuman })),
    warnDetails: staleWarn.map(s => ({ id: s.sourceName, ageSecs: s.ageSecs, ageHuman: s.ageHuman, warnOverageSecs: s.warnOverageSecs })),
    healthy: staleError.length === 0 && criticallyStale.length === 0,
  };
}

function criteriaToSeconds(criteria) {
  if (!criteria || !criteria.count || !criteria.period) return null;
  const periods = { minute: 60, hour: 3600, day: 86400 };
  return criteria.count * (periods[criteria.period] ?? 3600);
}

function formatDuration(secs) {
  if (secs < 60) return `${Math.round(secs)}s`;
  if (secs < 3600) return `${Math.round(secs / 60)}m`;
  if (secs < 86400) return `${(secs / 3600).toFixed(1)}h`;
  return `${(secs / 86400).toFixed(1)}d`;
}

dbt Cloud API: job monitoring without artifact access

When artifacts are not directly accessible, the dbt Cloud Admin API provides job run status. Authenticate with a service token (Settings → Account → Service Tokens, permission scope: "Read Jobs"). The API returns run status as an integer: 1=queued, 2=starting, 3=running, 10=success, 20=error, 30=cancelled. A run with status 10 (success) may still have individual model failures if they were configured as --no-fail-fast — always parse the run artifacts for the full picture.

async function dbtCloudJobHealth(accountId, jobId, apiToken) {
  const baseUrl = 'https://cloud.getdbt.com/api/v2';
  const headers = {
    'Authorization': `Token ${apiToken}`,
    'Content-Type': 'application/json',
  };

  // Get last 5 runs for this job
  const runsRes = await fetch(
    `${baseUrl}/accounts/${accountId}/runs/?job_definition_id=${jobId}&order_by=-id&limit=5&include_related=["run_steps"]`,
    { headers, signal: AbortSignal.timeout(15_000) }
  );

  if (!runsRes.ok) {
    throw new Error(`dbt Cloud API → ${runsRes.status}: ${await runsRes.text()}`);
  }

  const { data: runs } = await runsRes.json();

  const STATUS_LABELS = {
    1: 'queued', 2: 'starting', 3: 'running',
    10: 'success', 20: 'error', 30: 'cancelled',
  };

  const parsedRuns = runs.map(run => ({
    id: run.id,
    status: STATUS_LABELS[run.status] ?? `unknown(${run.status})`,
    statusCode: run.status,
    isRunning: run.status === 3,
    isSuccess: run.status === 10,
    isError: run.status === 20,
    isCancelled: run.status === 30,
    startedAt: run.started_at,
    finishedAt: run.finished_at,
    durationMs: run.duration_humanized
      ? null  // Parse from run.duration_humanized or compute from timestamps
      : (run.finished_at && run.started_at
        ? new Date(run.finished_at) - new Date(run.started_at)
        : null),
    triggeredBy: run.trigger?.cause ?? 'unknown',
    gitBranch: run.trigger?.git_branch ?? null,
    hasStepErrors: (run.run_steps ?? []).some(s => s.status_humanized === 'Error'),
  }));

  const lastRun = parsedRuns[0];
  const recentErrorRate = parsedRuns.filter(r => r.isError).length / parsedRuns.length;

  // Duration anomaly: last run more than 1.5x the median of recent successful runs
  const successfulDurations = parsedRuns
    .filter(r => r.isSuccess && r.durationMs !== null)
    .map(r => r.durationMs)
    .sort((a, b) => a - b);
  const medianDurationMs = successfulDurations.length > 0
    ? successfulDurations[Math.floor(successfulDurations.length / 2)]
    : null;
  const lastRunDuration = lastRun?.durationMs;
  const durationAnomaly = medianDurationMs && lastRunDuration
    ? lastRunDuration > medianDurationMs * 1.5
    : false;

  return {
    jobId,
    lastRun,
    recentErrorRate,
    medianDurationMs,
    durationAnomaly,
    recentRuns: parsedRuns,
    healthy: lastRun ? (lastRun.isSuccess && !durationAnomaly) : false,
    alerts: [
      ...(lastRun?.isError ? ['Last dbt run failed'] : []),
      ...(lastRun?.isCancelled ? ['Last dbt run was cancelled'] : []),
      ...(recentErrorRate > 0.4 ? [`${Math.round(recentErrorRate * 100)}% of recent runs failed`] : []),
      ...(durationAnomaly ? [`Last run duration anomaly: ${Math.round(lastRunDuration / 60000)}m vs median ${Math.round(medianDurationMs / 60000)}m`] : []),
    ],
  };
}

AliveMCP integration for dbt pipeline monitoring

Register three separate AliveMCP checks for a dbt pipeline. First: an artifact freshness probe that checks whether new artifacts have been written since the last expected run. Second: a run results probe that parses run_results.json for model failures and blocking test failures. Third: a source freshness probe that checks sources.json for stale raw data. Each probe reads artifacts from a shared store (S3, GCS, or the dbt Cloud artifact API).

async function dbtPipelineHealthProbe(artifactBaseUrl, expectedRunIntervalMs = 86_400_000) {
  const artifacts = await loadDbtArtifacts(artifactBaseUrl);

  const alerts = [];

  // Probe 1: Artifact freshness — did the dbt run fire on schedule?
  if (artifacts.runResults) {
    const generatedAt = artifacts.runResults.metadata?.generated_at
      ? new Date(artifacts.runResults.metadata.generated_at)
      : null;
    const ageMs = generatedAt ? Date.now() - generatedAt.getTime() : null;
    if (ageMs !== null && ageMs > expectedRunIntervalMs * 1.5) {
      alerts.push(`run_results.json is ${formatDuration(ageMs / 1000)} old — dbt may not have run on schedule`);
    }
  } else {
    alerts.push('run_results.json not found — artifact store may be unavailable or dbt has never run');
  }

  // Probe 2: Model and test health
  let runHealth = null;
  if (artifacts.runResults) {
    runHealth = parseRunResults(artifacts.runResults);
    if (runHealth.models.failed > 0) {
      alerts.push(`${runHealth.models.failed} dbt model(s) failed`);
    }
    if (runHealth.models.skipped > 0) {
      alerts.push(`${runHealth.models.skipped} model(s) skipped (upstream failures cascading)`);
    }
    if (runHealth.tests.failed > 0) {
      alerts.push(`${runHealth.tests.failed} dbt test(s) failed (blocking severity)`);
    }
    if (runHealth.tests.warned > 0) {
      // Warn but don't block
      alerts.push(`${runHealth.tests.warned} dbt test(s) in warn state (non-blocking)`);
    }
  }

  // Probe 3: Source freshness
  let freshnessHealth = null;
  if (artifacts.sources) {
    freshnessHealth = parseSourceFreshness(artifacts.sources);
    if (freshnessHealth.staleError > 0) {
      alerts.push(`${freshnessHealth.staleError} source(s) past error freshness threshold`);
    }
    if (freshnessHealth.criticallyStale > 0) {
      alerts.push(`${freshnessHealth.criticallyStale} source(s) >2x past warn freshness threshold`);
    }
  }

  return {
    healthy: alerts.length === 0,
    alerts,
    runHealth,
    freshnessHealth,
    artifactErrors: artifacts.artifactErrors,
  };
}

Related guides

• MCP tools for Apache Spark — SparkUI REST API, stage health, executor state, and shuffle spill
• MCP tools for Apache Flink — streaming job health, checkpoint failures, and backpressure detection
• MCP tools for Apache Airflow — DAG health, task instance status, and scheduler heartbeat
• MCP tools for ClickHouse — ReplicatedMergeTree health, system.parts, and replication queue monitoring
• MCP server health check patterns — composite endpoints and failure classification