execute-snowflake-sql
Run SQL queries on Snowflake and retrieve results directly through the Simple Snowflake MCP Server, designed to operate behind corporate proxies.
Instructions
Execute a SQL query on Snowflake and return the result.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| sql | Yes | SQL query to execute |
Implementation Reference
- 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']}")]
- 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 }, ),
- 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}