feat: add --result-format CLI arg for compact query results (csv)

README.md

@@ -171,6 +171,24 @@ export SNOWFLAKE_MCP_ENDPOINT="/my-mcp"
 uvx snowflake-labs-mcp --service-config-file config.yaml --transport streamable-http
 ```
 
+## Result Format
+
+By default, query results are returned as a JSON array of objects, where column names repeat for every row. For large result sets this can be verbose and token-intensive. The `--result-format` option lets you select a more compact representation.
+
+| Parameter | CLI Argument | Environment Variable | Default | Options |
+|-----------|--------------|---------------------|---------|---------|
+| Result Format | --result-format | SNOWFLAKE_MCP_RESULT_FORMAT | json | `json`, `csv` |
+
+**CSV format** returns results as a CSV string with a header row. This can significantly reduce token usage for large result sets since column names are only written once.
+
+```bash
+# Use CSV format for more compact query results
+uvx snowflake-labs-mcp --service-config-file config.yaml --result-format csv
+```
+
+> [!NOTE]
+> CSV format applies to SQL query results from all query tools. It does not apply to Cortex Search or Cortex Agent responses. CSV output loses type information — numbers, booleans, and nulls are all represented as strings.
+
 # Using with MCP Clients
 
 The MCP server is client-agnostic and will work with most MCP Clients that support basic functionality for MCP tools and (optionally) resources. Below are examples for local installation. For connecting to containerized deployments, see [Connecting MCP Clients to Containers](#connecting-mcp-clients-to-containers).

mcp_server_snowflake/query_manager/tools.py

@@ -5,7 +5,7 @@
 from pydantic import Field
 
 from mcp_server_snowflake.query_manager.prompts import query_tool_prompt
-from mcp_server_snowflake.utils import SnowflakeException
+from mcp_server_snowflake.utils import SnowflakeException, results_to_csv
 
 
 def run_query(statement: str, snowflake_service):
@@ -41,7 +41,8 @@ def run_query(statement: str, snowflake_service):
             cur,
         ):
             cur.execute(statement)
-            return cur.fetchall()
+            results = cur.fetchall()
+            return results_to_csv(results) if snowflake_service.result_format == "csv" and results else results
     except Exception as e:
         raise SnowflakeException(
             tool="query_manager",

mcp_server_snowflake/server.py

@@ -107,6 +107,7 @@ def __init__(
         transport: str,
         connection_params: dict,
         endpoint: str = "/mcp",
+        result_format: str = "json",
     ):
         if service_config_file is None:
             raise ValueError(
@@ -129,6 +130,7 @@ def __init__(
         self.query_manager = False
         self.semantic_manager = False
         self.default_session_parameters: Dict[str, Any] = {}
+        self.result_format = result_format
         self.query_tag = query_tag if query_tag is not None else None
         self.tag_major_version = (
             tag_major_version if tag_major_version is not None else None
@@ -508,6 +510,13 @@ def parse_arguments():
         help="Enable verbose/debug logging",
         default=False,
     )
+    parser.add_argument(
+        "--result-format",
+        required=False,
+        choices=["json", "csv"],
+        default=os.getenv("SNOWFLAKE_MCP_RESULT_FORMAT", "json"),
+        help="Format for query result sets: json (default) or csv",
+    )
 
     return parser.parse_args()
 
@@ -542,6 +551,7 @@ async def create_snowflake_service(
                 transport=args.transport,
                 connection_params=connection_params,
                 endpoint=endpoint or args.endpoint,
+                result_format=args.result_format,
             )
 
             # Initialize tools and resources now that we have the service

mcp_server_snowflake/tests/test_result_format.py

@@ -0,0 +1,113 @@
+# Copyright 2025 Snowflake Inc.
+# SPDX-License-Identifier: Apache-2.0
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+# http://www.apache.org/licenses/LICENSE-2.0
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import sys
+from unittest.mock import MagicMock, patch
+
+import pytest
+import yaml
+
+from mcp_server_snowflake.server import SnowflakeService, parse_arguments
+from mcp_server_snowflake.utils import results_to_csv
+
+
+# --- results_to_csv unit tests ---
+
+
+def test_results_to_csv_returns_string():
+    data = [{"col1": 1, "col2": "a"}, {"col1": 2, "col2": "b"}]
+    assert results_to_csv(data) == "col1,col2\r\n1,a\r\n2,b\r\n"
+
+
+def test_results_to_csv_single_row():
+    data = [{"name": "Alice", "score": 99}]
+    assert results_to_csv(data) == "name,score\r\nAlice,99\r\n"
+
+
+def test_results_to_csv_preserves_column_order():
+    data = [{"z": 1, "a": 2, "m": 3}]
+    first_line = results_to_csv(data).split("\r\n")[0]
+    assert first_line == "z,a,m"
+
+
+# --- SnowflakeService result_format attribute tests ---
+
+
+@pytest.fixture
+def minimal_config_file(tmp_path):
+    config = {"search_services": []}
+    config_file = tmp_path / "config.yaml"
+    config_file.write_text(yaml.dump(config))
+    return config_file
+
+
+@pytest.fixture
+def mock_snowflake_connect():
+    with (
+        patch("mcp_server_snowflake.server.connect") as mock_connect,
+        patch("mcp_server_snowflake.server.Root") as mock_root,
+    ):
+        mock_connect.return_value = MagicMock()
+        mock_root.return_value = MagicMock()
+        yield mock_connect
+
+
+def test_snowflake_service_default_result_format(
+    mock_snowflake_connect, minimal_config_file
+):
+    service = SnowflakeService(
+        service_config_file=str(minimal_config_file),
+        transport="stdio",
+        connection_params={},
+    )
+    assert service.result_format == "json"
+
+
+def test_snowflake_service_csv_result_format(
+    mock_snowflake_connect, minimal_config_file
+):
+    service = SnowflakeService(
+        service_config_file=str(minimal_config_file),
+        transport="stdio",
+        connection_params={},
+        result_format="csv",
+    )
+    assert service.result_format == "csv"
+
+
+# --- CLI argument tests ---
+
+
+def test_parse_arguments_default_result_format(monkeypatch):
+    monkeypatch.delenv("SNOWFLAKE_MCP_RESULT_FORMAT", raising=False)
+    with patch("sys.argv", ["prog"]):
+        args = parse_arguments()
+    assert args.result_format == "json"
+
+
+def test_parse_arguments_csv_result_format():
+    with patch("sys.argv", ["prog", "--result-format", "csv"]):
+        args = parse_arguments()
+    assert args.result_format == "csv"
+
+
+def test_parse_arguments_invalid_result_format():
+    with patch("sys.argv", ["prog", "--result-format", "toon"]):
+        with pytest.raises(SystemExit):
+            parse_arguments()
+
+
+def test_parse_arguments_result_format_from_env(monkeypatch):
+    monkeypatch.setenv("SNOWFLAKE_MCP_RESULT_FORMAT", "csv")
+    with patch("sys.argv", ["prog"]):
+        args = parse_arguments()
+    assert args.result_format == "csv"

mcp_server_snowflake/utils.py

@@ -9,6 +9,8 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
+import csv
+import io
 import json
 import os
 import re
@@ -55,6 +57,15 @@ def warn_deprecated_params() -> None:
         logger.info(f"Deprecated parameters: {', '.join(deprecated_found)}")
 
 
+def results_to_csv(data: list[dict]) -> str:
+    """Convert a non-empty list of row dicts to a CSV string with a header row."""
+    buf = io.StringIO()
+    writer = csv.DictWriter(buf, fieldnames=data[0].keys())
+    writer.writeheader()
+    writer.writerows(data)
+    return buf.getvalue()
+
+
 def execute_query(statement: str, snowflake_service, bindvars: list[str] = []):
     """Execute a Snowflake query and return the results using Python connector dictionary cursor."""
     with snowflake_service.get_connection(
@@ -65,7 +76,8 @@ def execute_query(statement: str, snowflake_service, bindvars: list[str] = []):
         cur,
     ):
         cur.execute(statement, bindvars)
-        return cur.fetchall()
+        results = cur.fetchall()
+        return results_to_csv(results) if snowflake_service.result_format == "csv" and results else results
 
 
 def sanitize_tool_name(service_name: str) -> str:
@@ -111,7 +123,7 @@ class AnalystResponse(BaseModel):
 
     text: str
     sql: Optional[str] = None
-    results: Optional[Union[dict, list]] = None
+    results: Optional[Union[dict, list, str]] = None
 
 
 class AgentResponse(BaseModel):
@@ -214,7 +226,8 @@ def fetch_results(self, statement: str, service, **kwargs):
             cur,
         ):
             cur.execute(statement)
-            return cur.fetchall()
+            results = cur.fetchall()
+            return results_to_csv(results) if service.result_format == "csv" and results else results
 
     def parse_analyst_response(
         self, response: requests.Response | dict, service, **kwargs