Connect a Mendix AI Agent to a Snowflake-Managed MCP Server

-- You can run this example under the Sysadmin role. For real production screnarios, use proper authorisation.
CREATE DATABASE IF NOT EXISTS SNOWFLAKE_MCP_DEMO;
CREATE SCHEMA   IF NOT EXISTS   SNOWFLAKE_MCP_DEMO.TOOLS;
CREATE SCHEMA   IF NOT EXISTS   SNOWFLAKE_MCP_DEMO.MCPSERVERS;
CREATE SCHEMA   IF NOT EXISTS   SNOWFLAKE_MCP_DEMO.TESTDATA;

CREATE OR REPLACE TABLE SNOWFLAKE_MCP_DEMO.TESTDATA.TICKETS (
TICKETID NUMBER AUTOINCREMENT START 1 INCREMENT 1,
PRIORITY VARCHAR(10),
TEXT VARCHAR(500)
);
      
INSERT INTO SNOWFLAKE_MCP_DEMO.TESTDATA.TICKETS (PRIORITY, TEXT)
VALUES
   ('High', 'Server is down in production environment'),
   ('Medium', 'User unable to reset password'),
   ('Low', 'Request for additional monitor'),
   ('High', 'Database connection timeout on checkout page'),
   ('Medium', 'Email notifications not being sent');    

-- You can run this example/demo under sysadmin role, for real production screnario's use proper authorisation
CREATE OR REPLACE PROCEDURE SNOWFLAKE_MCP_DEMO.TOOLS.GET_SCHEMA_METADATA(
    db_name VARCHAR,
    schema_name VARCHAR
)
RETURNS VARIANT
LANGUAGE PYTHON
RUNTIME_VERSION = '3.11'
PACKAGES = ('snowflake-snowpark-python')
HANDLER = 'run'
AS
$$
import json
def run(session, db_name, schema_name):
    rows = session.sql(f"""
        SELECT
            c.TABLE_CATALOG,
            c.TABLE_SCHEMA,
            c.TABLE_NAME,
            t.TABLE_TYPE,
            t.ROW_COUNT,
            t.COMMENT AS TABLE_COMMENT,
            c.COLUMN_NAME,
            c.ORDINAL_POSITION,
            c.DATA_TYPE,
            c.IS_NULLABLE,
            c.COLUMN_DEFAULT,
            c.CHARACTER_MAXIMUM_LENGTH,
            c.NUMERIC_PRECISION,
            c.NUMERIC_SCALE,
            c.COMMENT AS COLUMN_COMMENT
        FROM {db_name}.INFORMATION_SCHEMA.COLUMNS c
        JOIN {db_name}.INFORMATION_SCHEMA.TABLES t
            ON c.TABLE_CATALOG = t.TABLE_CATALOG
            AND c.TABLE_SCHEMA = t.TABLE_SCHEMA
            AND c.TABLE_NAME = t.TABLE_NAME
        WHERE c.TABLE_SCHEMA = '{schema_name}'
        ORDER BY c.TABLE_NAME, c.ORDINAL_POSITION
    """).collect()
        tables = {}
        for row in rows:
            tname = row["TABLE_NAME"]
              if tname not in tables:
                  tables[tname] = {
                      "database": row["TABLE_CATALOG"],
                      "schema": row["TABLE_SCHEMA"],
                      "table_type": row["TABLE_TYPE"],
                      "row_count": row["ROW_COUNT"],
                      "comment": row["TABLE_COMMENT"],
                      "columns": []
                  }
            tables[tname]["columns"].append({
                "name": row["COLUMN_NAME"],
                "position": row["ORDINAL_POSITION"],
                "data_type": row["DATA_TYPE"],
                "nullable": row["IS_NULLABLE"],
                "default": row["COLUMN_DEFAULT"],
                "max_length": row["CHARACTER_MAXIMUM_LENGTH"],
                "precision": row["NUMERIC_PRECISION"],
                "scale": row["NUMERIC_SCALE"],
                "comment": row["COLUMN_COMMENT"]
            })
        return tables
    $$;

-- You can run this example/demo under sysadmin role, for real production screnario's use proper authorisation
CREATE OR REPLACE PROCEDURE SNOWFLAKE_MCP_DEMO.TOOLS.RETRIEVE_RECORDS(
    fully_qualified_table VARCHAR,
    filter_column VARCHAR,
    filter_value VARCHAR
)
RETURNS VARCHAR
LANGUAGE PYTHON
RUNTIME_VERSION = '3.11'
PACKAGES = ('snowflake-snowpark-python')
HANDLER = 'run'
AS
    $$
    import json
        def run(session, fully_qualified_table, filter_column, filter_value):
            parts = fully_qualified_table.split('.')
            if len(parts) != 3:
                return json.dumps({"status": "error", "message": "Table must be fully qualified: DATABASE.SCHEMA.TABLE"})
            try:
                if filter_column and filter_value:
                    escaped = filter_value.replace("'", "''")
                    sql = f"SELECT * FROM {fully_qualified_table} WHERE {filter_column} = '{escaped}'"
                else:
                    sql = f"SELECT * FROM {fully_qualified_table}"
            rows = session.sql(sql).collect()
            results = [row.as_dict() for row in rows]
                for r in results:
                    for k, v in r.items():
                        if not isinstance(v, (str, int, float, bool, type(None))):
                            r[k] = str(v)
                return json.dumps({"status": "success", "row_count": len(results), "data": results})
          except Exception as e:
              return json.dumps({"status": "error", "message": str(e)})
$$;

-- inputs can be for example:
-- fully_qualified_table = 'SNOWFLAKE_MCP_DEMO.TESTDATA.TICKETS'
-- column_values: '{"PRIORITY": "Low", "TEKST": "text here"}'
      
-- You can run this example/demo under sysadmin role, for real production screnario's use proper authorisation
    CREATE OR REPLACE PROCEDURE SNOWFLAKE_MCP_DEMO.TOOLS.INSERT_RECORD(
        fully_qualified_table VARCHAR,
          column_values VARCHAR
      )
      RETURNS VARCHAR
      LANGUAGE PYTHON
      RUNTIME_VERSION = '3.11'
      PACKAGES = ('snowflake-snowpark-python')
      HANDLER = 'run'
      AS
      $$
      import json
      def run(session, fully_qualified_table, column_values):
          parts = fully_qualified_table.split('.')
          if len(parts) != 3:
              return json.dumps({"status": "error", "message": "Table must be fully qualified: DATABASE.SCHEMA.TABLE"})
          db, schema, table = parts
          cols_rows = session.sql(f"""
              SELECT COLUMN_NAME, DATA_TYPE, IS_NULLABLE
              FROM {db}.INFORMATION_SCHEMA.COLUMNS
              WHERE TABLE_SCHEMA = '{schema}' AND TABLE_NAME = '{table}'
              ORDER BY ORDINAL_POSITION
          """).collect()
          if not cols_rows:
              return json.dumps({"status": "error", "message": f"Table {fully_qualified_table} not found or has no columns"})
          schema_info = {row["COLUMN_NAME"]: row["DATA_TYPE"] for row in cols_rows}
          try:
              values = json.loads(column_values)
          except json.JSONDecodeError as e:
              return json.dumps({"status": "error", "message": f"Invalid JSON in column_values: {str(e)}", "expected_columns": list(schema_info.keys())})
          for col_name in values:
              if col_name not in schema_info:
                  return json.dumps({"status": "error", "message": f"Column '{col_name}' does not exist in {fully_qualified_table}", "valid_columns": list(schema_info.keys())})
          col_names = list(values.keys())
          val_parts = []
          for col in col_names:
              val = values[col]
              if val is None:
                  val_parts.append("NULL")
              elif isinstance(val, (int, float)):
                  val_parts.append(str(val))
              else:
                  escaped = str(val).replace("'", "''")
                  val_parts.append(f"'{escaped}'")
          col_list = ", ".join(col_names)
          val_list = ", ".join(val_parts)
          sql = f"INSERT INTO {fully_qualified_table} ({col_list}) VALUES ({val_list})"
          try:
              session.sql(sql).collect()
              return json.dumps({"status": "success", "message": f"Record inserted into {fully_qualified_table}", "columns_inserted": col_names})
          except Exception as e:
              return json.dumps({"status": "error", "message": str(e)})
    $$;

-- You can run this example/demo under sysadmin role, for real production screnario's use proper authorisation
CREATE OR REPLACE MCP SERVER SNOWFLAKE_MCP_DEMO.MCPSERVERS.DEMO_MCP_SERVER
    FROM SPECIFICATION $$
        tools:
        - title: "Get Schema Metadata"
              identifier: "SNOWFLAKE_MCP_DEMO.TOOLS.GET_SCHEMA_METADATA"
              name: "get_schema_metadata"
              type: "GENERIC"
              description: "Returns metadata for all tables and columns in a given database schema, including data types, nullability, row counts, and comments."
              config:
                type: "procedure"
                warehouse: "MYWAREHOUSE"
                input_schema:
                  type: "object"
                  properties:
                    db_name:
                      description: "The database name to inspect"
                      type: "string"
                    schema_name:
                      description: "The schema name to inspect"
                      type: "string"
            - title: "Insert Record"
              identifier: "SNOWFLAKE_MCP_DEMO.TOOLS.INSERT_RECORD"
              name: "insert_record"
              type: "GENERIC"
              description: "Inserts a single record into a specified table. Accepts a fully qualified table name and a JSON string of column-value pairs."
              config:
                type: "procedure"
                warehouse: "MYWAREHOUSE"
                input_schema:
                  type: "object"
                  properties:
                    fully_qualified_table:
                      description: "Fully qualified table name in DATABASE.SCHEMA.TABLE format"
                      type: "string"
                    column_values:
                      description: "JSON string of column names and values to insert, e.g. {\"PRIORITY\": \"High\", \"TEKST\": \"New ticket\"}"
                      type: "string"
            - title: "Retrieve Records"
              identifier: "SNOWFLAKE_MCP_DEMO.TOOLS.RETRIEVE_RECORDS"
              name: "retrieve_records"
              type: "GENERIC"
              description: "Retrieves records from a specified table. Optionally filter by a single column value. Returns all rows if no filter is provided."
              config:
                type: "procedure"
                warehouse: "MYWAREHOUSE"
                input_schema:
                  type: "object"
                  properties:
                    fully_qualified_table:
                      description: "Fully qualified table name in DATABASE.SCHEMA.TABLE format"
                      type: "string"
                    filter_column:
                      description: "Optional column name to filter on. Pass empty string for no filter."
                      type: "string"
                    filter_value:
                      description: "Optional value to match in the filter column. Pass empty string for no filter."
                      type: "string"
    $$;

You are a support ticket management assistant connected to a Snowflake database via MCP tools.
         
You help users create, retrieve, and inspect support tickets stored in the SNOWFLAKE_MCP_DEMO.TESTDATA schema.
         
## Available Tools
         
You have access to the following MCP tools:
         
1. **get_schema_metadata** - Retrieves table and column metadata for a database  - Use this FIRST when you need to understand the data structure before performing other operations.
- Parameters: db_name (e.g. "SNOWFLAKE_MCP_DEMO"), schema_name (e.g. "TESTDATA")
         
2. **insert_record** - Inserts a single record into a table.
- Parameters: fully_qualified_table (e.g. "SNOWFLAKE_MCP_DEMO.TESTDATA.TICKETS"), column_values (JSON string)
- The value for the TICKETID column is automatically generated in the database, so omit this column when inserting.
- Always provide both PRIORITY and TEXT columns.
- PRIORITY must be one of: "High", "Medium", "Low"
         
3. **retrieve_records** - Retrieves records from a table with optional filtering.
- Parameters: fully_qualified_table, filter_column (optional), filter_value (optional)
- Pass empty strings for filter_column and filter_value to retrieve all records.
         
## Data Model
         
The primary table is SNOWFLAKE_MCP_DEMO.TESTDATA.TICKETS:
- TICKETID (NUMBER, auto-increment) - Unique ticket identifier
- PRIORITY (VARCHAR) - Ticket priority: High, Medium, or Low
- TEXT (VARCHAR) - Description of the ticket/issue
         
## Guidelines
         
- When a user asks to create a ticket, extract the priority and description from their request. If priority is not specified, ask for it.
- When a user asks to view or search tickets, use retrieve_records. Use the filter parameters when they want to filter by a specific column.
- If a user asks about the data structure or available tables, use get_schema_metadata.
- Always confirm successful operations by showing the user what was created or retrieved.
- Use fully qualified table names (DATABASE.SCHEMA.TABLE) in all tool calls.
- If a tool call returns an error, explain the issue clearly and suggest a correction.