list-snowflake-warehouses

Instructions

Implementation Reference

• src/simple_snowflake_mcp/server.py:751-763 (handler)

  The handler function that implements the core logic for the 'list-snowflake-warehouses' tool. It executes the 'SHOW WAREHOUSES' SQL command using the _safe_snowflake_execute helper and formats the output as detailed JSON or a simple list of names based on the 'include_details' parameter.

  elif name == "list-snowflake-warehouses":
      include_details = args.get("include_details", True)
      query = "SHOW WAREHOUSES"
      result = _safe_snowflake_execute(query, "List warehouses")
      if result["success"]:
          if include_details:
              output = json.dumps(result["data"], indent=2, default=str)
          else:
              warehouses = [row.get("name", "") for row in result["data"]]
              output = "\n".join(warehouses)
          return [types.TextContent(type="text", text=output)]
      else:
          return [types.TextContent(type="text", text=f"Snowflake error: {result['error']}")]

• src/simple_snowflake_mcp/server.py:608-622 (registration)

  The tool registration in the handle_list_tools() function, including the tool's description and input schema definition for JSON Schema validation.

  types.Tool(
      name="list-snowflake-warehouses",
      description="List available Data Warehouses (DWH) on Snowflake with detailed information",
      inputSchema={
          "type": "object",
          "properties": {
              "include_details": {
                  "type": "boolean",
                  "default": True,
                  "description": "Include detailed warehouse information"
              }
          },
          "additionalProperties": False
      },
  ),

• src/simple_snowflake_mcp/server.py:611-622 (schema)

  The input schema definition for the 'list-snowflake-warehouses' tool, specifying the optional 'include_details' boolean parameter.

      inputSchema={
          "type": "object",
          "properties": {
              "include_details": {
                  "type": "boolean",
                  "default": True,
                  "description": "Include detailed warehouse information"
              }
          },
          "additionalProperties": False
      },
  ),

• src/simple_snowflake_mcp/server.py:168-194 (helper)

  Supporting helper function used by the tool handler to safely connect to Snowflake, execute the query, fetch results as JSON-compatible dicts, and handle errors.

  def _safe_snowflake_execute(query: str, description: str = "Query") -> Dict[str, Any]:
      """
      Safely execute a Snowflake query with proper error handling and logging.
      """
      try:
          logger.info(f"Executing {description}: {query[:100]}...")
          ctx = snowflake.connector.connect(**SNOWFLAKE_CONFIG)
          cur = ctx.cursor()
          cur.execute(query)
          
          # Handle different query types
          if cur.description:
              rows = cur.fetchall()
              columns = [desc[0] for desc in cur.description]
              result = [dict(zip(columns, row)) for row in rows]
          else:
              result = {"status": "success", "rowcount": cur.rowcount}
              
          cur.close()
          ctx.close()
          logger.info(f"{description} completed successfully")
          return {"success": True, "data": result}
          
      except Exception as e:
          logger.error(f"{description} failed: {str(e)}")
          return {"success": False, "error": str(e), "data": None}