mcp-snowflake-server

Instructions

Implementation Reference

• src/mcp_snowflake_server/server.py:91-131 (handler)

  Core handler function that lists schemas for a given database by querying Snowflake's INFORMATION_SCHEMA.SCHEMATA, applies exclusion filters if configured, and returns YAML text content with optional embedded JSON resource.

  async def handle_list_schemas(arguments, db, *_, exclusion_config=None, exclude_json_results=False):
      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 = 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:392-406 (registration)

  Registration of the list_schemas tool in the all_tools list, including name, description, input schema definition, and reference to the handler function.

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

• src/mcp_snowflake_server/server.py:395-404 (schema)

  Input schema defining the required 'database' parameter as a string for the list_schemas tool.

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