DBT Core MCP Server

Instructions

Implementation Reference

• src/dbt_core_mcp/server.py:612-649 (handler)

  Main handler function that executes the tool logic: retrieves project info from manifest, adds paths and status, optionally runs 'dbt debug' for diagnostics and parses connection status.

  async def toolImpl_get_project_info(self, run_debug: bool = True) -> dict[str, Any]:
      """Implementation for get_project_info tool."""
      # Get project info from manifest
      info = self.manifest.get_project_info()  # type: ignore
      info["project_dir"] = str(self.project_dir)
      info["profiles_dir"] = self.profiles_dir
      info["status"] = "ready"
  
      # Run full dbt debug if requested (default behavior)
      if run_debug:
          debug_result_obj = await self.runner.invoke(["debug"])  # type: ignore
  
          # Convert DbtRunnerResult to dictionary
          debug_result = {
              "success": debug_result_obj.success,
              "output": debug_result_obj.stdout if debug_result_obj.stdout else "",
          }
  
          # Parse the debug output
          diagnostics: dict[str, Any] = {
              "command_run": "dbt debug",
              "success": debug_result.get("success", False),
              "output": debug_result.get("output", ""),
          }
  
          # Extract connection status from output
          output = str(debug_result.get("output", ""))
          if "Connection test: [OK connection ok]" in output or "Connection test: OK" in output:
              diagnostics["connection_status"] = "ok"
          elif "Connection test: [ERROR" in output or "Connection test: FAIL" in output:
              diagnostics["connection_status"] = "failed"
          else:
              diagnostics["connection_status"] = "unknown"
  
          info["diagnostics"] = diagnostics
  
      return info

• src/dbt_core_mcp/server.py:1277-1292 (registration)

  FastMCP tool registration decorator that binds the tool name 'get_project_info' to the implementation method, including docstring for tool description and parameters.

  @self.app.tool()
  async def get_project_info(
      ctx: Context,
      run_debug: bool = True,
  ) -> dict[str, Any]:
      """Get information about the dbt project with optional diagnostics.
  
      Args:
          run_debug: Run `dbt debug` to validate environment and test connection (default: True)
  
      Returns:
          Dictionary with project information and diagnostic results
      """
      await self._ensure_initialized_with_context(ctx)
      return await self.toolImpl_get_project_info(run_debug)

• src/dbt_core_mcp/dbt/manifest.py:358-383 (helper)

  Helper method in ManifestLoader that extracts and counts project metadata (name, version, adapter, resource counts) from the loaded manifest.json.

  def get_project_info(self) -> dict[str, Any]:
      """
      Get high-level project information from the manifest.
  
      Returns:
          Dictionary with project metadata
      """
      if not self._manifest:
          raise RuntimeError("Manifest not loaded. Call load() first.")
  
      metadata: dict[str, Any] = self._manifest.get("metadata", {})  # type: ignore[assignment]
  
      # Count resources directly from manifest
      nodes = self._manifest.get("nodes", {})
      model_count = sum(1 for node in nodes.values() if isinstance(node, dict) and node.get("resource_type") == "model")
      source_count = len(self._manifest.get("sources", {}))
  
      return {
          "project_name": metadata.get("project_name", ""),
          "dbt_version": metadata.get("dbt_version", ""),
          "adapter_type": metadata.get("adapter_type", ""),
          "generated_at": metadata.get("generated_at", ""),
          "model_count": model_count,
          "source_count": source_count,
      }