DBT Core MCP Server

Instructions

Implementation Reference

• src/dbt_core_mcp/server.py:650-786 (handler)

  Main handler function that executes the SQL query via the dbt runner's invoke_query method, handles errors, parses JSON results from dbt show, supports JSON/CSV/TSV output formats, and saves to file if specified.

  async def toolImpl_query_database(self, sql: str, output_file: str | None = None, output_format: str = "json") -> dict[str, Any]:
      """Implementation for query_database tool."""
      # Execute query using dbt show --inline
      result = await self.runner.invoke_query(sql)  # type: ignore
  
      if not result.success:
          error_msg = str(result.exception) if result.exception else "Unknown error"
          response = {
              "status": "failed",
              "error": error_msg,
          }
          # Include dbt output for debugging
          if result.stdout:
              response["dbt_output"] = result.stdout
          if result.stderr:
              response["stderr"] = result.stderr
          return response
  
      # Parse JSON output from dbt show
      import json
      import re
  
      output = result.stdout if hasattr(result, "stdout") else ""
  
      try:
          # dbt show --output json returns: {"show": [...rows...]}
          # Find the JSON object (look for {"show": pattern)
          json_match = re.search(r'\{\s*"show"\s*:\s*\[', output)
          if not json_match:
              return {
                  "status": "failed",
                  "error": "No JSON output found in dbt show response",
              }
  
          # Use JSONDecoder to parse just the first complete JSON object
          # This handles extra data after the JSON (like log lines)
          decoder = json.JSONDecoder()
          data, _ = decoder.raw_decode(output, json_match.start())
  
          if "show" in data:
              rows = data["show"]
              row_count = len(rows)
  
              # Handle different output formats
              if output_format in ("csv", "tsv"):
                  # Convert to CSV/TSV format
                  import csv
                  import io
  
                  delimiter = "\t" if output_format == "tsv" else ","
                  csv_buffer = io.StringIO()
  
                  if rows:
                      writer = csv.DictWriter(csv_buffer, fieldnames=rows[0].keys(), delimiter=delimiter)
                      writer.writeheader()
                      writer.writerows(rows)
                      csv_string = csv_buffer.getvalue()
                  else:
                      csv_string = ""
  
                  if output_file:
                      # Save to file
                      from pathlib import Path
  
                      output_path = Path(output_file)
                      output_path.parent.mkdir(parents=True, exist_ok=True)
  
                      with open(output_path, "w", newline="") as f:
                          f.write(csv_string)
  
                      # Get file size
                      file_size_bytes = output_path.stat().st_size
                      file_size_kb = file_size_bytes / 1024
  
                      return {
                          "status": "success",
                          "row_count": row_count,
                          "format": output_format,
                          "saved_to": str(output_path),
                          "file_size_kb": round(file_size_kb, 2),
                      }
                  else:
                      # Return CSV/TSV inline
                      return {
                          "status": "success",
                          "row_count": row_count,
                          "format": output_format,
                          output_format: csv_string,
                      }
              else:
                  # JSON format (default)
                  if output_file:
                      # Ensure directory exists
                      from pathlib import Path
  
                      output_path = Path(output_file)
                      output_path.parent.mkdir(parents=True, exist_ok=True)
  
                      # Write rows to file
                      with open(output_path, "w") as f:
                          json.dump(rows, f, indent=2)
  
                      # Get file size
                      file_size_bytes = output_path.stat().st_size
                      file_size_kb = file_size_bytes / 1024
  
                      # Return metadata with preview
                      return {
                          "status": "success",
                          "row_count": row_count,
                          "saved_to": str(output_path),
                          "file_size_kb": round(file_size_kb, 2),
                          "columns": list(rows[0].keys()) if rows else [],
                          "preview": rows[:3],  # First 3 rows as preview
                      }
                  else:
                      # Return all rows inline
                      return {
                          "status": "success",
                          "row_count": row_count,
                          "rows": rows,
                      }
          else:
              return {
                  "status": "failed",
                  "error": "Unexpected JSON format from dbt show",
                  "data": data,
              }
  
      except json.JSONDecodeError as e:
          return {
              "status": "error",
              "message": f"Failed to parse query results: {e}",
              "raw_output": output[:500],
          }
  
  async def toolImpl_run_models(

• src/dbt_core_mcp/server.py:1479-1525 (registration)

  Tool registration using @app.tool() decorator. Defines the input schema via parameters (sql: str, output_file: Optional[str], output_format: str='json') and comprehensive docstring with usage guidelines. Delegates to toolImpl_query_database.

  async def query_database(ctx: Context, sql: str, output_file: str | None = None, output_format: str = "json") -> dict[str, Any]:
      """Execute a SQL query against the dbt project's database.
  
      BEST PRACTICES:
      1. Before querying: Inspect schema using get_resource_info() with include_database_schema=True
      2. Always use {{ ref('model_name') }} for dbt models (never hard-code table paths)
      3. Always use {{ source('source_name', 'table_name') }} for source tables
      4. For non-dbt tables: Verify schema with user before querying
      5. After results: Report "Query Result: X rows retrieved" and summarize key findings
  
      QUERY EFFICIENCY:
      - Use aggregations (COUNT, SUM, AVG, etc.) instead of pulling raw data
      - Apply WHERE filters early to narrow scope before aggregation
      - Use LIMIT for exploratory queries to get representative samples
      - Calculate totals, ratios, and trends in SQL rather than returning all rows
      - Use GROUP BY for categorization within the query
      - Always ask: "Can SQL answer this question directly?" before returning data
  
      LARGE RESULT HANDLING:
      - For queries returning many rows (>100), use output_file parameter to save results to disk
      - This prevents context window overflow and improves performance
      - The tool returns metadata + preview instead of full results when output_file is used
      - Example: query_database(sql="SELECT * FROM large_table", output_file="temp_auto/results.json")
  
      OUTPUT FORMATS:
      - json (default): Returns data as JSON array of objects
      - csv: Returns comma-separated values with header row
      - tsv: Returns tab-separated values with header row
      - CSV/TSV formats use proper quoting (only when necessary) and are Excel-compatible
  
      Args:
          sql: SQL query with Jinja templating: {{ ref('model') }}, {{ source('src', 'table') }}
               For exploratory queries, include LIMIT. For aggregations/counts, omit it.
          output_file: Optional file path to save results. Recommended for large result sets (>100 rows).
                      If provided, only metadata is returned (no preview for CSV/TSV).
                      If omitted, all data is returned inline (may consume large context).
          output_format: Output format - "json" (default), "csv", or "tsv"
  
      Returns:
          JSON inline: {"status": "success", "row_count": N, "rows": [...]}
          JSON file: {"status": "success", "row_count": N, "saved_to": "path", "preview": [...]}
          CSV/TSV inline: {"status": "success", "row_count": N, "format": "csv", "csv": "..."}
          CSV/TSV file: {"status": "success", "row_count": N, "format": "csv", "saved_to": "path"}
      """
      await self._ensure_initialized_with_context(ctx)
      return await self.toolImpl_query_database(sql, output_file, output_format)