mcp-snowflake-server

Instructions

Implementation Reference

• src/mcp_snowflake_server/server.py:229-255 (handler)

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

• src/mcp_snowflake_server/server.py:438-442 (schema)

  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"],
  },