read_query
Execute SELECT queries to retrieve data from Snowflake databases, enabling data analysis and reporting.
Instructions
Execute a SELECT query.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| query | Yes | SELECT SQL query to execute |
Implementation Reference
- The handler function for the 'read_query' tool. Validates input, checks for write operations using write_detector, executes the SQL query on SnowflakeDB, formats results as YAML and optional JSON embedded resource.async def handle_read_query(arguments, db, write_detector, *_, exclude_json_results=False): if not arguments or "query" not in arguments: raise ValueError("Missing query argument") if write_detector.analyze_query(arguments["query"])["contains_write"]: raise ValueError("Calls to read_query should not contain write operations") data, data_id = await db.execute_query(arguments["query"]) output = { "type": "data", "data_id": data_id, "data": data, } yaml_output = to_yaml(output) json_output = to_json(output) results: list[ResponseType] = [types.TextContent(type="text", text=yaml_output)] if not exclude_json_results: results.append( types.EmbeddedResource( type="resource", resource=types.TextResourceContents( uri=f"data://{data_id}", text=json_output, mimeType="application/json" ), ) ) return results
- src/mcp_snowflake_server/server.py:435-444 (registration)Registers the 'read_query' tool in the all_tools list, providing name, description, input schema requiring a 'query' string, and linking to the handle_read_query handler.Tool( name="read_query", description="Execute a SELECT query.", input_schema={ "type": "object", "properties": {"query": {"type": "string", "description": "SELECT SQL query to execute"}}, "required": ["query"], }, handler=handle_read_query, ),
- Defines the input schema for the 'read_query' tool: an object with a required 'query' property of type string describing the SELECT SQL query.input_schema={ "type": "object", "properties": {"query": {"type": "string", "description": "SELECT SQL query to execute"}}, "required": ["query"], },