DBT Core MCP Server

Instructions

Implementation Reference

• src/dbt_core_mcp/server.py:521-583 (handler)

  Main execution logic for the get_resource_info tool: fetches from manifest, handles compilation for models, queries database schema, supports multiple matches.
  async def toolImpl_get_resource_info( self, name: str, resource_type: str | None = None, include_database_schema: bool = True, include_compiled_sql: bool = True, ) -> dict[str, Any]: """Implementation for get_resource_info tool.""" try: # Get resource info with manifest method (handles basic enrichment) result = self.manifest.get_resource_info( # type: ignore name, resource_type, include_database_schema=False, # We'll handle this below for database schema include_compiled_sql=include_compiled_sql, ) # Handle multiple matches case if result.get("multiple_matches"): # Enrich each match with database schema if requested if include_database_schema: matches = result.get("matches", []) for match in matches: node_type = match.get("resource_type") if node_type in ("model", "seed", "snapshot", "source"): resource_name = match.get("name") source_name = match.get("source_name") if node_type == "source" else None schema = await self._get_table_schema_from_db(resource_name, source_name) if schema: match["database_columns"] = schema return result # Single match - check if we need to trigger compilation node_type = result.get("resource_type") if include_compiled_sql and node_type == "model": # If compiled SQL requested but not available, trigger compilation if result.get("compiled_sql") is None and not result.get("compiled_sql_cached"): logger.info(f"Compiling model: {name}") compile_result = await self.runner.invoke_compile(name, force=False) # type: ignore if compile_result.success: # Reload manifest to get compiled code await self.manifest.load() # type: ignore # Re-fetch the resource to get updated compiled_code result = self.manifest.get_resource_info( # type: ignore name, resource_type, include_database_schema=False, include_compiled_sql=True, ) # Query database schema for applicable resource types if include_database_schema and node_type in ("model", "seed", "snapshot", "source"): resource_name = result.get("name", name) # For sources, pass source_name to use source() instead of ref() source_name = result.get("source_name") if node_type == "source" else None schema = await self._get_table_schema_from_db(resource_name, source_name) if schema: result["database_columns"] = schema return result
• src/dbt_core_mcp/server.py:1340-1384 (registration)

  FastMCP tool registration with @app.tool(), defining input schema via parameters/docstring and delegating to toolImpl_get_resource_info after initialization.
  async def get_resource_info( ctx: Context, name: str, resource_type: str | None = None, include_database_schema: bool = True, include_compiled_sql: bool = True, ) -> dict[str, Any]: """Get detailed information about any dbt resource (model, source, seed, snapshot, test, etc.). This unified tool works across all resource types, auto-detecting the resource or filtering by type. Designed for LLM consumption - returns complete data even when multiple matches exist. Args: name: Resource name. For sources, use "source_name.table_name" or just "table_name" resource_type: Optional filter to narrow search: - "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 - None: Auto-detect (searches all types) include_database_schema: If True (default), query actual database table schema for models/seeds/snapshots/sources and add as 'database_columns' field include_compiled_sql: If True (default), include compiled SQL with Jinja resolved ({{ ref() }}, {{ source() }} → actual table names). Only applicable to models. Will trigger dbt compile if not already compiled. Set to False to skip compilation. Returns: Resource information dictionary. If multiple matches found, returns: {"multiple_matches": True, "matches": [...], "message": "..."} Raises: ValueError: If resource not found Examples: get_resource_info("customers") -> auto-detect model or source get_resource_info("customers", "model") -> get model only get_resource_info("jaffle_shop.customers", "source") -> specific source get_resource_info("test_unique_customers") -> find test get_resource_info("customers", include_compiled_sql=True) -> include compiled SQL """ await self._ensure_initialized_with_context(ctx) return await self.toolImpl_get_resource_info(name, resource_type, include_database_schema, include_compiled_sql)
• src/dbt_core_mcp/dbt/manifest.py:304-357 (helper)

  Key helper in ManifestLoader that parses manifest.json for resource details, supports multiple matches, adds compiled_sql availability (without triggering compile).
  def get_resource_info( self, name: str, resource_type: str | None = None, include_database_schema: bool = True, include_compiled_sql: bool = True, ) -> dict[str, Any]: """Get detailed resource information with optional enrichments. This method extends get_resource_node() with optional enrichments: - include_database_schema: Query actual database schema - include_compiled_sql: Include compiled SQL (models only, requires compilation) Note: This method does NOT trigger compilation. If compiled SQL is requested but not available in the manifest, the 'compiled_sql' field will be None. The caller (e.g., server tool) is responsible for triggering compilation if needed. Args: name: Resource name resource_type: Optional resource type filter include_database_schema: Include database schema information (default: True) include_compiled_sql: Include compiled SQL for models (default: True) Returns: Resource dictionary with optional enrichments """ result = self.get_resource_node(name, resource_type) # Handle multiple matches case - return as-is if result.get("multiple_matches"): return result # Single match - enrich with additional data if requested node_type = result.get("resource_type") # Create a copy without heavy fields result_copy = dict(result) result_copy.pop("raw_code", None) result_copy.pop("compiled_code", None) # Include compiled SQL for models if requested and available if include_compiled_sql and node_type == "model": compiled_code = result.get("compiled_code") if compiled_code: result_copy["compiled_sql"] = compiled_code result_copy["compiled_sql_cached"] = True else: # Not compiled yet - set to None to indicate it's not available result_copy["compiled_sql"] = None result_copy["compiled_sql_cached"] = False return result_copy