DBT Core MCP Server

Instructions

Implementation Reference

• src/dbt_core_mcp/dbt/manifest.py:90-194 (handler)

  Main handler logic that parses the dbt manifest.json and returns filtered list of resources with consistent structure across types (models, sources, seeds, etc.). Called by toolImpl_list_resources.
  def get_resources(self, resource_type: str | None = None) -> list[dict[str, Any]]: """ Get all resources from the manifest, optionally filtered by type. Returns simplified resource information across all types (models, sources, seeds, etc.). Designed for LLM consumption with consistent structure across resource types. Args: resource_type: Optional filter (model, source, seed, snapshot, test, analysis). If None, returns all resources. Returns: List of resource dictionaries with consistent structure: { "name": str, "unique_id": str, "resource_type": str, "schema": str (if applicable), "database": str (if applicable), "description": str, "tags": list[str], "package_name": str, ...additional type-specific fields } Raises: RuntimeError: If manifest not loaded ValueError: If invalid resource_type provided """ if not self._manifest: raise RuntimeError("Manifest not loaded. Call load() first.") # Validate resource_type if provided valid_types = {"model", "source", "seed", "snapshot", "test", "analysis"} if resource_type is not None and resource_type not in valid_types: raise ValueError(f"Invalid resource_type '{resource_type}'. Must be one of: {', '.join(sorted(valid_types))}") resources: list[dict[str, Any]] = [] # Collect from nodes (models, tests, seeds, snapshots, analyses) nodes = self._manifest.get("nodes", {}) for unique_id, node in nodes.items(): if not isinstance(node, dict): continue node_type = node.get("resource_type") # Filter by type if specified if resource_type is not None and node_type != resource_type: continue # Build consistent resource dict resource: dict[str, Any] = { "name": node.get("name", ""), "unique_id": unique_id, "resource_type": node_type, "package_name": node.get("package_name", ""), "description": node.get("description", ""), "tags": node.get("tags", []), } # Add common fields for materialized resources if node_type in ("model", "seed", "snapshot"): resource["schema"] = node.get("schema", "") resource["database"] = node.get("database", "") resource["alias"] = node.get("alias", "") # Add type-specific fields if node_type == "model": resource["materialization"] = node.get("config", {}).get("materialized", "") resource["file_path"] = node.get("original_file_path", "") elif node_type == "seed": resource["file_path"] = node.get("original_file_path", "") elif node_type == "snapshot": resource["file_path"] = node.get("original_file_path", "") elif node_type == "test": resource["test_metadata"] = node.get("test_metadata", {}) resource["column_name"] = node.get("column_name") resources.append(resource) # Collect from sources (if not filtered out) if resource_type is None or resource_type == "source": sources = self._manifest.get("sources", {}) for unique_id, source in sources.items(): if not isinstance(source, dict): continue resource = { "name": source.get("name", ""), "unique_id": unique_id, "resource_type": "source", "source_name": source.get("source_name", ""), "schema": source.get("schema", ""), "database": source.get("database", ""), "identifier": source.get("identifier", ""), "package_name": source.get("package_name", ""), "description": source.get("description", ""), "tags": source.get("tags", []), } resources.append(resource) logger.debug(f"Found {len(resources)} resources" + (f" of type '{resource_type}'" if resource_type else "")) return resources
• src/dbt_core_mcp/server.py:587-597 (handler)

  Direct handler implementation on DbtCoreMcpServer that delegates to ManifestLoader.get_resources()
  async def toolImpl_list_resources(self, resource_type: str | None = None) -> list[dict[str, Any]]: """Implementation for list_resources tool. Args: resource_type: Optional filter (model, source, seed, snapshot, test, analysis, macro) Returns: List of resource dictionaries with consistent structure """ return self.manifest.get_resources(resource_type) # type: ignore
• src/dbt_core_mcp/server.py:1294-1338 (registration)

  MCP tool registration with @app.tool() decorator, including schema (args/docstring) and dispatch to impl. Defines input schema: resource_type optional str filter.
  async def list_resources(ctx: Context, resource_type: str | None = None) -> list[dict[str, Any]]: """List all resources in the dbt project with optional filtering by type. This unified tool provides a consistent view across all dbt resource types. Returns simplified resource information optimized for LLM consumption. Args: resource_type: Optional filter to narrow results: - "model": Data transformation models - "source": External data sources - "seed": CSV reference data files - "snapshot": SCD Type 2 historical tables - "test": Data quality tests - "analysis": Ad-hoc analysis queries - "macro": Jinja macros (includes macros from installed packages) - None: Return all resources (default) Returns: List of resource dictionaries with consistent structure across types. Each resource includes: name, unique_id, resource_type, description, tags, etc. Package Discovery: Use resource_type="macro" to discover installed dbt packages. Macros follow the naming pattern: macro.{package_name}.{macro_name} Example - Check if dbt_utils is installed: macros = list_resources("macro") has_dbt_utils = any(m["unique_id"].startswith("macro.dbt_utils.") for m in macros) Example - List all installed packages: macros = list_resources("macro") packages = {m["unique_id"].split(".")[1] for m in macros if m["unique_id"].startswith("macro.") and m["unique_id"].split(".")[1] != "dbt"} Examples: list_resources() -> all resources list_resources("model") -> only models list_resources("source") -> only sources list_resources("test") -> only tests list_resources("macro") -> all macros (discover installed packages) """ await self._ensure_initialized_with_context(ctx) return await self.toolImpl_list_resources(resource_type=resource_type)