[Part 1 of 3]: Replace hand-rolled Pydantic schemas with dbt-artifacts-parser (#745)

## Summary

Part 1 of 3 to replace `get_job_run_artifacts` with an in-memory DuckDB
store that lets
LLMs query and run full-text search over dbt job run artifacts.

PR sequence:
1. **Artifact parsing infrastructure** (this PR)
2. ArtifactStore + extraction layer
3. Tools + MCP wiring

## What Changed

- Added `dbt-artifacts-parser~=0.13.2` dependency — handles dbt schema
version differences internally
- Replaced hand-rolled schemas in `config.py` with a new `artifacts/`
subpackage:
- Per-artifact `parse()` functions (`manifest.py`, `catalog.py`,
`run_results.py`, `sources.py`)
  - `ARTIFACT_PARSERS` dispatch dict for the upcoming extraction layer
- `LenientXxx` Pydantic fallback models for Fusion/preview builds that
deviate from published schemas
- Split `config.py` into `schemas/job_run.py` (Admin API shapes) and
`schemas/output.py` (tool output contract)
- Refactored `ErrorFetcher` / `WarningFetcher` to use the new artifact
modules — no behavior change

## Related Issues

Related to #413

## Checklist
- [x] I have performed a self-review of my code
- [x] I have added tests that prove my fix is effective or that my
feature works
- [x] New and existing unit tests pass locally with my changes
- [ ] I have made corresponding changes to the documentation (in
https://github.com/dbt-labs/docs.getdbt.com) if required

## Additional Notes

`dbt-artifacts-parser` has type annotations on its `parse_*()` functions
but lacks a
[py.typed PEP 561 marker file](https://peps.python.org/pep-0561/).
Without it, mypy treats
the package as untyped, requiring `# type: ignore[import-untyped]` at
every import site.

NOTE: Addressed the above in
yu-iskw/dbt-artifacts-parser#228 which was
included in [release
0.13.2](https://github.com/yu-iskw/dbt-artifacts-parser/releases/tag/v0.13.2)

---------

Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: jairus-m

.changes/unreleased/Under the Hood-20260430-125234.yaml

@@ -0,0 +1,3 @@
+kind: Under the Hood
+body: '[Part 1 of 3]: Use dbt-artifacts-parser schemas for artifact schema parsing'
+time: 2026-04-30T12:52:34.933451-10:00

pyproject.toml

@@ -44,10 +44,11 @@ dependencies = [
   "httpx~=0.28.1",
   "filelock~=3.20.3",
   "starlette~=0.50.0",
+  "dbt-artifacts-parser>=0.13.2",
 ]
 [tool.uv]
 exclude-newer = "7 days"
-exclude-newer-package = { dbt-protos = false, dbt-sl-sdk = false, dbtlabs-vortex = false }
+exclude-newer-package = { dbt-protos = false, dbt-sl-sdk = false, dbtlabs-vortex = false, dbt-artifacts-parser = false}
 
 [dependency-groups]
 dev = [

src/dbt_mcp/dbt_admin/run_artifacts/__init__.py

@@ -1,3 +0,0 @@
-from .parser import ErrorFetcher, WarningFetcher
-
-__all__ = ["ErrorFetcher", "WarningFetcher"]

src/dbt_mcp/dbt_admin/run_artifacts/artifacts/__init__.py

@@ -0,0 +1 @@
+"""Artifact parsing modules for dbt Cloud job run artifacts."""

src/dbt_mcp/dbt_admin/run_artifacts/artifacts/catalog.py

@@ -0,0 +1,32 @@
+"""Parsing and mapping for catalog.json artifacts."""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+from dbt_artifacts_parser.parser import parse_catalog
+from dbt_artifacts_parser.parsers.catalog.catalog_v1 import CatalogV1
+
+from dbt_mcp.dbt_admin.run_artifacts.artifacts.lenient import LenientCatalog
+
+logger = logging.getLogger(__name__)
+
+CatalogParsed = CatalogV1 | LenientCatalog
+
+
+def parse(raw: dict[str, Any]) -> CatalogParsed:
+    """Parse catalog.json using dbt-artifacts-parser (version-aware).
+
+    Falls back to ``LenientCatalog`` when strict Pydantic validation fails.
+    """
+    try:
+        return parse_catalog(catalog=raw)
+    except Exception as e:
+        logger.warning(
+            "Strict catalog parsing failed (%s: %s); "
+            "falling back to lenient dict-based parsing.",
+            type(e).__name__,
+            str(e)[:200],
+        )
+        return LenientCatalog.model_validate(raw)

src/dbt_mcp/dbt_admin/run_artifacts/artifacts/lenient.py

@@ -0,0 +1,99 @@
+"""Lenient Pydantic schemas used as fallbacks when dbt-artifacts-parser fails.
+
+dbt-artifacts-parser uses Pydantic internally with strict enum validation.
+It fails on real-world artifacts that deviate from the published schema —
+e.g. a ``"reused"`` status from incremental builds, or preview dbt versions
+that emit extra fields.
+
+These schemas are maximally permissive:
+- ``extra="allow"`` — unknown fields don't cause failures
+- All non-essential fields are optional with safe defaults
+- ``status`` is ``str | None`` (not an enum) — accepts any value dbt may emit
+
+The ``parse()`` functions in each artifact module always return a Pydantic
+``BaseModel`` — either the strict dbt-artifacts-parser model (happy path) or
+one of these lenient models (fallback). Downstream extractors receive a typed
+object in both cases.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from pydantic import BaseModel, ConfigDict, Field, field_validator
+
+
+class LenientRunResultsResult(BaseModel):
+    model_config = ConfigDict(extra="allow")
+
+    status: str | None = None
+    unique_id: str | None = None
+    relation_name: str | None = None
+    message: str | None = None
+    compiled_code: str | None = None
+    compiled_sql: str | None = None  # older dbt versions used compiled_sql
+
+
+class LenientRunResultsArgs(BaseModel):
+    model_config = ConfigDict(extra="allow")
+
+    target: str | None = None
+
+
+class LenientRunResults(BaseModel):
+    model_config = ConfigDict(extra="allow")
+
+    results: list[LenientRunResultsResult] = Field(default_factory=list)
+    args: LenientRunResultsArgs | None = None
+
+    @field_validator("results", mode="before")
+    @classmethod
+    def coerce_results(cls, v: Any) -> list[Any]:
+        return v if isinstance(v, list) else []
+
+
+class LenientSourceResult(BaseModel):
+    model_config = ConfigDict(extra="allow")
+
+    status: str | None = None
+    unique_id: str | None = None
+    max_loaded_at_time_ago_in_s: float | None = None
+
+
+class LenientSources(BaseModel):
+    model_config = ConfigDict(extra="allow")
+
+    results: list[LenientSourceResult] = Field(default_factory=list)
+
+    @field_validator("results", mode="before")
+    @classmethod
+    def coerce_results(cls, v: Any) -> list[Any]:
+        return v if isinstance(v, list) else []
+
+
+class LenientCatalog(BaseModel):
+    """Minimal lenient catalog schema — nodes/sources dicts for PR 2/3 extraction."""
+
+    model_config = ConfigDict(extra="allow")
+
+    nodes: dict[str, Any] = Field(default_factory=dict)
+    sources: dict[str, Any] = Field(default_factory=dict)
+
+    @field_validator("nodes", "sources", mode="before")
+    @classmethod
+    def coerce_dict(cls, v: Any) -> dict[str, Any]:
+        return v if isinstance(v, dict) else {}
+
+
+class LenientManifest(BaseModel):
+    """Minimal lenient manifest schema — nodes/sources dicts for PR 2/3 extraction."""
+
+    model_config = ConfigDict(extra="allow")
+
+    nodes: dict[str, Any] = Field(default_factory=dict)
+    sources: dict[str, Any] = Field(default_factory=dict)
+
+    @field_validator("nodes", "sources", mode="before")
+    @classmethod
+    def coerce_dict(cls, v: Any) -> dict[str, Any]:
+        return v if isinstance(v, dict) else {}

src/dbt_mcp/dbt_admin/run_artifacts/artifacts/manifest.py

@@ -0,0 +1,61 @@
+"""Parsing and mapping for manifest.json artifacts."""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+from dbt_artifacts_parser.parser import parse_manifest
+from dbt_artifacts_parser.parsers.manifest.manifest_v1 import ManifestV1
+from dbt_artifacts_parser.parsers.manifest.manifest_v2 import ManifestV2
+from dbt_artifacts_parser.parsers.manifest.manifest_v3 import ManifestV3
+from dbt_artifacts_parser.parsers.manifest.manifest_v4 import ManifestV4
+from dbt_artifacts_parser.parsers.manifest.manifest_v5 import ManifestV5
+from dbt_artifacts_parser.parsers.manifest.manifest_v6 import ManifestV6
+from dbt_artifacts_parser.parsers.manifest.manifest_v7 import ManifestV7
+from dbt_artifacts_parser.parsers.manifest.manifest_v8 import ManifestV8
+from dbt_artifacts_parser.parsers.manifest.manifest_v9 import ManifestV9
+from dbt_artifacts_parser.parsers.manifest.manifest_v10 import ManifestV10
+from dbt_artifacts_parser.parsers.manifest.manifest_v11 import ManifestV11
+from dbt_artifacts_parser.parsers.manifest.manifest_v12 import ManifestV12
+
+from dbt_mcp.dbt_admin.run_artifacts.artifacts.lenient import LenientManifest
+
+logger = logging.getLogger(__name__)
+
+ManifestParsed = (
+    ManifestV1
+    | ManifestV2
+    | ManifestV3
+    | ManifestV4
+    | ManifestV5
+    | ManifestV6
+    | ManifestV7
+    | ManifestV8
+    | ManifestV9
+    | ManifestV10
+    | ManifestV11
+    | ManifestV12
+    | LenientManifest
+)
+
+
+def parse(raw: dict[str, Any]) -> ManifestParsed:
+    """Parse manifest.json using dbt-artifacts-parser (version-aware).
+
+    Falls back to ``LenientManifest`` when strict Pydantic validation fails.
+    This covers preview / unreleased dbt versions that
+    emit a manifest claiming a published schema version (e.g. v12) but
+    containing additional fields not yet in that schema.
+    """
+    try:
+        return parse_manifest(manifest=raw)
+    except Exception as e:
+        logger.warning(
+            "Strict manifest parsing failed (%s: %s); "
+            "falling back to lenient dict-based parsing.  "
+            "This typically occurs with dbt preview builds.",
+            type(e).__name__,
+            str(e)[:200],
+        )
+        return LenientManifest.model_validate(raw)

src/dbt_mcp/dbt_admin/run_artifacts/artifacts/parsers.py

@@ -0,0 +1,63 @@
+"""ArtifactType enum and ARTIFACT_PARSERS dispatch table.
+
+ARTIFACT_PARSERS always returns a plain ``dict[str, Any]``:
+- Happy path: strict dbt-artifacts-parser Pydantic model → ``.model_dump(mode="json")``
+  which normalises enums to strings and aliases (e.g. ``schema_``) to their JSON
+  keys (e.g. ``"schema"``).
+- Fallback: raw dict passed through as-is — same JSON shape, just unvalidated.
+
+Downstream extractors can therefore use ``.get()`` uniformly on every path.
+
+Note: the ``parse()`` helpers in the sibling artifact modules (manifest.py, catalog.py,
+run_results.py, sources.py) are a separate API used by the job error/warning fetcher in
+parser.py and are intentionally left unchanged.
+"""
+
+from __future__ import annotations
+
+import logging
+from collections.abc import Callable
+from enum import Enum
+from typing import Any
+
+from dbt_artifacts_parser.parser import (
+    parse_catalog,
+    parse_manifest,
+    parse_run_results,
+    parse_sources,
+)
+
+logger = logging.getLogger(__name__)
+
+
+class ArtifactType(str, Enum):
+    RUN_RESULTS = "run_results.json"
+    SOURCES = "sources.json"
+    MANIFEST = "manifest.json"
+    CATALOG = "catalog.json"
+
+
+def _to_dict(raw: dict[str, Any], strict_parse_fn: Callable[[], Any]) -> dict[str, Any]:
+    """Try strict parsing and dump to a plain dict; fall back to raw on any error."""
+    try:
+        return strict_parse_fn().model_dump(mode="json")
+    except Exception as exc:
+        logger.warning(
+            "Strict artifact parsing failed (%s: %s); falling back to raw dict. "
+            "This is expected for dbt Fusion or preview builds.",
+            type(exc).__name__,
+            str(exc)[:200],
+        )
+        return raw
+
+
+ARTIFACT_PARSERS: dict[ArtifactType, Callable[[dict[str, Any]], dict[str, Any]]] = {
+    ArtifactType.MANIFEST: lambda raw: _to_dict(
+        raw, lambda: parse_manifest(manifest=raw)
+    ),
+    ArtifactType.CATALOG: lambda raw: _to_dict(raw, lambda: parse_catalog(catalog=raw)),
+    ArtifactType.RUN_RESULTS: lambda raw: _to_dict(
+        raw, lambda: parse_run_results(run_results=raw)
+    ),
+    ArtifactType.SOURCES: lambda raw: _to_dict(raw, lambda: parse_sources(sources=raw)),
+}