query_metrics

Instructions

Implementation Reference

• src/dbt_mcp/semantic_layer/tools.py:101-127 (handler)

  MCP tool handler function for 'query_metrics'. Decorated with @dbt_mcp_tool which defines the tool schema and metadata. Dispatches to SemanticLayerFetcher.

  @dbt_mcp_tool(
      description=get_prompt("semantic_layer/query_metrics"),
      title="Query Metrics",
      read_only_hint=True,
      destructive_hint=False,
      idempotent_hint=True,
  )
  async def query_metrics(
      context: SemanticLayerToolContext,
      metrics: list[str],
      group_by: list[GroupByParam] | None = None,
      order_by: list[OrderByParam] | None = None,
      where: str | None = None,
      limit: int | None = None,
  ) -> str:
      result = await context.semantic_layer_fetcher.query_metrics(
          metrics=metrics,
          group_by=group_by,
          order_by=order_by,
          where=where,
          limit=limit,
      )
      if isinstance(result, QueryMetricsSuccess):
          return result.result
      else:
          return result.error

• src/dbt_mcp/mcp/server.py:135-149 (registration)

  Registration of semantic layer tools (including query_metrics) to the FastMCP server instance.

  if config.semantic_layer_config_provider:
      logger.info("Registering semantic layer tools")
      register_sl_tools(
          dbt_mcp,
          config_provider=config.semantic_layer_config_provider,
          client_provider=DefaultSemanticLayerClientProvider(
              config_provider=config.semantic_layer_config_provider,
          ),
          disabled_tools=disabled_tools,
          enabled_tools=enabled_tools,
          enabled_toolsets=enabled_toolsets,
          disabled_toolsets=disabled_toolsets,
      )

• src/dbt_mcp/semantic_layer/client.py:374-417 (helper)

  Core implementation of query_metrics in SemanticLayerFetcher class, which performs validation, queries the dbt Semantic Layer via SDK, formats results or errors.

  async def query_metrics(
      self,
      metrics: list[str],
      group_by: list[GroupByParam] | None = None,
      order_by: list[OrderByParam] | None = None,
      where: str | None = None,
      limit: int | None = None,
      result_formatter: Callable[[pa.Table], str] | None = None,
  ) -> QueryMetricsResult:
      validation_error = await self.validate_query_metrics_params(
          metrics=metrics,
          group_by=group_by,
      )
      if validation_error:
          return QueryMetricsError(error=validation_error)
  
      try:
          query_error = None
          sl_client = await self.client_provider.get_client()
          with sl_client.session():
              # Catching any exception within the session
              # to ensure it is closed properly
              try:
                  parsed_order_by: list[OrderBySpec] = self._get_order_bys(
                      order_by=order_by, metrics=metrics, group_by=group_by
                  )
                  query_result = await asyncio.to_thread(
                      sl_client.query,
                      metrics=metrics,
                      group_by=group_by,  # type: ignore
                      order_by=parsed_order_by,  # type: ignore
                      where=[where] if where else None,
                      limit=limit,
                  )
              except Exception as e:
                  query_error = e
          if query_error:
              return self._format_query_failed_error(query_error)
          formatter = result_formatter or DEFAULT_RESULT_FORMATTER
          json_result = formatter(query_result)
          return QueryMetricsSuccess(result=json_result or "")
      except Exception as e:
          return self._format_query_failed_error(e)

• src/dbt_mcp/semantic_layer/types.py:1-77 (schema)

  Type definitions used for query_metrics tool parameters (e.g., OrderByParam, GroupByParam from external) and responses (QueryMetricsSuccess/Error). Input schema derived from handler function signature.

  from dataclasses import dataclass
  
  from dbtsl.models.dimension import DimensionType
  from dbtsl.models.entity import EntityType
  from dbtsl.models.metric import MetricType
  
  
  @dataclass
  class OrderByParam:
      name: str
      descending: bool
  
  
  @dataclass
  class MetricToolResponse:
      name: str
      type: MetricType
      label: str | None = None
      description: str | None = None
      metadata: str | None = None
  
  
  @dataclass
  class DimensionToolResponse:
      name: str
      type: DimensionType
      description: str | None = None
      label: str | None = None
      granularities: list[str] | None = None
  
  
  @dataclass
  class EntityToolResponse:
      name: str
      type: EntityType
      description: str | None = None
  
  
  @dataclass
  class SavedQueryToolResponse:
      name: str
      label: str | None = None
      description: str | None = None
      metrics: list[str] | None = None
      group_by: list[str] | None = None
      where: list[str] | None = None
  
  
  @dataclass
  class QueryMetricsSuccess:
      result: str
      error: None = None
  
  
  @dataclass
  class QueryMetricsError:
      error: str
      result: None = None
  
  
  QueryMetricsResult = QueryMetricsSuccess | QueryMetricsError
  
  
  @dataclass
  class GetMetricsCompiledSqlSuccess:
      sql: str
      error: None = None
  
  
  @dataclass
  class GetMetricsCompiledSqlError:
      error: str
      sql: None = None
  
  
  GetMetricsCompiledSqlResult = GetMetricsCompiledSqlSuccess | GetMetricsCompiledSqlError