execute_query

Instructions

Implementation Reference

• server.py:322-363 (handler)

  The execute_query tool handler function that executes a native SQL query against a Metabase database. It takes database_id, query string, and optional native_parameters, constructs a Metabase API payload, and sends a POST request to /dataset endpoint. The function logs query execution info and handles errors via ToolError.

  @mcp.tool
  async def execute_query(
      database_id: int,
      query: str,
      ctx: Context,
      native_parameters: list[dict[str, Any]] | None = None
  ) -> dict[str, Any]:
      """
      Execute a native SQL query against a Metabase database.
  
      Args:
          database_id: The ID of the database to query.
          query: The SQL query to execute.
          native_parameters: Optional parameters for the query.
  
      Returns:
          Query execution results.
      """
      try:
          await ctx.info(f"Executing query on database {database_id}")
          await ctx.debug(f"Query: {query[:100]}...")  # Log first 100 chars
  
          payload = {
              "database": database_id,
              "type": "native",
              "native": {"query": query}
          }
  
          if native_parameters:
              payload["native"]["parameters"] = native_parameters
              await ctx.debug(f"Query parameters: {len(native_parameters)} parameters provided")
  
          result = await metabase_client.request("POST", "/dataset", json=payload)
  
          row_count = len(result.get("data", {}).get("rows", []))
          await ctx.info(f"Query executed successfully, returned {row_count} rows")
  
          return result
      except Exception as e:
          error_msg = f"Error executing query: {e}"
          await ctx.error(error_msg)
          raise ToolError(error_msg) from e

• server.py:322-328 (schema)

  The function signature defines the input schema for execute_query: database_id (int, required), query (str, required), native_parameters (list[dict] optional), and ctx (Context). The return type is dict[str, Any].

  @mcp.tool
  async def execute_query(
      database_id: int,
      query: str,
      ctx: Context,
      native_parameters: list[dict[str, Any]] | None = None
  ) -> dict[str, Any]:

• server.py:140-141 (registration)

  The tool is registered via the @mcp.tool decorator at line 140 (the first tool decorated similarly). All tools in this file use the @mcp.tool decorator pattern for registration with the FastMCP server instance 'mcp'. The execute_query tool specifically is registered at line 322 with @mcp.tool.

  @mcp.tool
  async def list_databases(ctx: Context) -> dict[str, Any]: