Snowflake MCP Server

Instructions

Implementation Reference

• src/mcp_snowflake_server/server.py:224-246 (handler)

  The handler function that implements the logic for the read_query tool. It validates the input, checks for write operations using write_detector, executes the query on the Snowflake database, formats the output as YAML and JSON, and returns it as TextContent and EmbeddedResource.

  async def handle_read_query(arguments, db, write_detector, *_):
      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 = data_to_yaml(output)
      json_output = json.dumps(output, default=data_json_serializer)
      return [
          types.TextContent(type="text", text=yaml_output),
          types.EmbeddedResource(
              type="resource",
              resource=types.TextResourceContents(uri=f"data://{data_id}", text=json_output, mimeType="application/json"),
          ),
      ]

• src/mcp_snowflake_server/server.py:482-486 (schema)

  The input schema for the read_query tool, defining it as 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"],
  },

• src/mcp_snowflake_server/server.py:479-488 (registration)

  The registration of the read_query tool in the all_tools list, specifying its name, description, input schema, and handler function.

  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,
  ),