Snowflake MCP Server

Instructions

Implementation Reference

• src/mcp_snowflake_server/server.py:98-134 (handler)

  Core handler function that executes the logic to list schemas from a specified Snowflake database. It runs a SQL query, applies exclusion filters if configured, formats results as YAML and JSON, and returns them as TextContent and EmbeddedResource.

  async def handle_list_schemas(arguments, db, *_, exclusion_config=None):
      if not arguments or "database" not in arguments:
          raise ValueError("Missing required 'database' parameter")
  
      database = arguments["database"]
      query = f"SELECT SCHEMA_NAME FROM {database.upper()}.INFORMATION_SCHEMA.SCHEMATA"
      data, data_id = await db.execute_query(query)
  
      # Filter out excluded schemas
      if exclusion_config and "schemas" in exclusion_config and exclusion_config["schemas"]:
          filtered_data = []
          for item in data:
              schema_name = item.get("SCHEMA_NAME", "")
              exclude = False
              for pattern in exclusion_config["schemas"]:
                  if pattern.lower() in schema_name.lower():
                      exclude = True
                      break
              if not exclude:
                  filtered_data.append(item)
          data = filtered_data
  
      output = {
          "type": "data",
          "data_id": data_id,
          "database": database,
          "data": data,
      }
      yaml_output = data_to_yaml(output)
      json_output = json.dumps(output)
      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:439-448 (schema)

  Input schema validation for the tool, requiring a single 'database' string parameter.

  input_schema={
      "type": "object",
      "properties": {
          "database": {
              "type": "string",
              "description": "Database name to list schemas from",
          },
      },
      "required": ["database"],
  },

• src/mcp_snowflake_server/server.py:436-450 (registration)

  Registers the 'list_schemas' tool in the all_tools list, which is used by handle_list_tools() to expose the tool to MCP clients and by handle_call_tool() to dispatch calls to the handler.

  Tool(
      name="list_schemas",
      description="List all schemas in a database",
      input_schema={
          "type": "object",
          "properties": {
              "database": {
                  "type": "string",
                  "description": "Database name to list schemas from",
              },
          },
          "required": ["database"],
      },
      handler=handle_list_schemas,
  ),