Simple Snowflake MCP

Instructions

Implementation Reference

• src/simple_snowflake_mcp/server.py:725-750 (handler)

  Handler logic for the 'execute-snowflake-sql' tool: parses arguments, executes SQL via helper, formats output in json/markdown/csv, returns TextContent.

  elif name == "execute-snowflake-sql":
      sql = args.get("sql")
      format_type = args.get("format", "json")
      if not sql:
          raise ValueError("'sql' parameter is required")
      
      result = _safe_snowflake_execute(sql, "SQL execution")
      if result["success"]:
          if format_type == "markdown" and isinstance(result["data"], list):
              output = _format_markdown_table(result["data"])
          elif format_type == "csv" and isinstance(result["data"], list):
              if result["data"]:
                  import csv, io
                  output_buffer = io.StringIO()
                  writer = csv.DictWriter(output_buffer, fieldnames=result["data"][0].keys())
                  writer.writeheader()
                  writer.writerows(result["data"])
                  output = output_buffer.getvalue()
              else:
                  output = "No data returned"
          else:
              output = json.dumps(result["data"], indent=2, default=str)
          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:588-606 (schema)

  Input JSON schema for the tool defining 'sql' (required string) and 'format' (optional enum: json/markdown/csv).

  inputSchema={
      "type": "object",
      "properties": {
          "sql": {
              "type": "string", 
              "description": "SQL query to execute",
              "minLength": 1,
              "examples": ["SELECT CURRENT_TIMESTAMP()", "SHOW DATABASES"]
          },
          "format": {
              "type": "string",
              "enum": ["json", "markdown", "csv"],
              "default": "json",
              "description": "Output format for results"
          }
      },
      "required": ["sql"],
      "additionalProperties": False
  },

• src/simple_snowflake_mcp/server.py:585-607 (registration)

  Tool registration in @server.list_tools() return list, including name, description, and input schema.

  types.Tool(
      name="execute-snowflake-sql",
      description="Execute a SQL query on Snowflake and return the result as JSON",
      inputSchema={
          "type": "object",
          "properties": {
              "sql": {
                  "type": "string", 
                  "description": "SQL query to execute",
                  "minLength": 1,
                  "examples": ["SELECT CURRENT_TIMESTAMP()", "SHOW DATABASES"]
              },
              "format": {
                  "type": "string",
                  "enum": ["json", "markdown", "csv"],
                  "default": "json",
                  "description": "Output format for results"
              }
          },
          "required": ["sql"],
          "additionalProperties": False
      },
  ),

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

  Helper function that performs the actual Snowflake connection, query execution, result fetching, and error handling used by the tool.

  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}