dbt-get-source

Instructions

Implementation Reference

• src/tools/dbt-sources.ts:40-90 (handler)

  The main handler function for 'dbt-get-source'. It loads the manifest, resolves the source by uniqueId or (sourceName+tableName), fetches freshness results from sources.json, and returns a detailed object with columns, freshness criteria, and freshness result.

  export async function dbtGetSource(args: z.infer<typeof dbtGetSourceSchema>): Promise<unknown> {
    const manifest = loadManifest();
    let id = args.uniqueId;
    if (!id) {
      if (!args.sourceName || !args.tableName) {
        throw new Error("Provide uniqueId, or both sourceName and tableName");
      }
      const found = Object.values(manifest.sources).find(
        (s) => s.source_name === args.sourceName && s.name === args.tableName,
      );
      id = found?.unique_id;
    }
    if (!id) throw new Error(`Source not found: ${args.uniqueId ?? `${args.sourceName}.${args.tableName}`}`);
    const src = manifest.sources[id];
    if (!src) throw new Error(`Source not found in manifest: ${id}`);
  
    let freshnessResult: unknown = null;
    try {
      const sources = loadSources();
      const r = sources.results.find((res) => res.unique_id === id);
      if (r) {
        freshnessResult = {
          status: r.status,
          maxLoadedAt: r.max_loaded_at,
          snapshottedAt: r.snapshotted_at,
          ageInSeconds: r.max_loaded_at_time_ago_in_s,
          criteria: r.criteria,
          generatedAt: sources.metadata.generated_at,
        };
      }
    } catch {
      // sources.json may be absent
    }
  
    return {
      uniqueId: src.unique_id,
      sourceName: src.source_name,
      tableName: src.name,
      identifier: src.identifier ?? src.name,
      database: src.database,
      schema: src.schema,
      loader: src.loader,
      loadedAtField: src.loaded_at_field,
      description: src.description,
      sourceDescription: src.source_description,
      freshness: src.freshness,
      columns: src.columns ? Object.values(src.columns) : [],
      tags: src.tags ?? [],
      freshnessResult,
    };
  }

• src/tools/dbt-sources.ts:34-38 (schema)

  Zod schema for the 'dbt-get-source' tool, accepting optional uniqueId or both sourceName and tableName.

  export const dbtGetSourceSchema = z.object({
    uniqueId: z.string().optional().describe("dbt unique_id (e.g. 'source.proj.raw.users')"),
    sourceName: z.string().optional().describe("Source group name (with tableName)"),
    tableName: z.string().optional().describe("Source table name (with sourceName)"),
  });

• src/index.ts:82-82 (registration)

  Registration of the 'dbt-get-source' tool with the MCP server, linking its description, schema, and handler via wrapToolHandler.

  tool("dbt-get-source", "Get a single dbt source: freshness criteria, columns, latest freshness result from sources.json", dbtGetSourceSchema.shape, wrapToolHandler(dbtGetSource));

• src/tools/aggregations.ts:120-160 (helper)

  Helper usage of dbtGetSource within the 'incident-context' aggregation tool to fetch source details when sourceName.tableName is provided.

    const { model, source, failedTests, dqChecks } = await aggregate(
      {
        model: () =>
          args.modelName
            ? dbtGetModel({ name: args.modelName, includeCompiledSql: false })
            : Promise.resolve(null),
        source: () =>
          sourceParts && sourceParts.length === 2
            ? dbtGetSource({ sourceName: sourceParts[0], tableName: sourceParts[1] })
            : Promise.resolve(null),
        failedTests: () => dbtFailedTests({ recentRuns: 3 }),
        dqChecks: () =>
          dqConfigured() && dataset
            ? dqListChecks({ sinceHours: args.sinceHours, dataset, limit: 50 })
            : Promise.resolve(null),
      },
      caveats,
    );
  
    if (!dqConfigured()) caveats.push("DQ_RESULTS_TABLE not configured — quality category skipped");
  
    return {
      anchor: args.modelName ? { kind: "model", name: args.modelName } : { kind: "source", fqn: args.sourceFqn },
      window: { sinceHours: args.sinceHours },
      model,
      source,
      failedTests,
      dqChecks,
      caveats,
      notes: [
        "If you also have @us-all/airflow-mcp installed, call airflow-list-runs for the loading DAG to add run-time context.",
      ],
    };
  }

• src/index.ts:21-25 (registration)

  Import of dbtGetSourceSchema and dbtGetSource from the dbt-sources module.

  import {
    dbtListSourcesSchema, dbtListSources,
    dbtGetSourceSchema, dbtGetSource,
    dbtListExposuresSchema, dbtListExposures,
  } from "./tools/dbt-sources.js";