mcp-snowflake-server

Instructions

Implementation Reference

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

  The handler function that executes the list_tables tool: validates inputs, queries Snowflake's information_schema.tables for the specified database and schema, filters excluded tables, formats output as YAML and optional JSON.

  async def handle_list_tables(arguments, db, *_, exclusion_config=None, exclude_json_results=False):
      if not arguments or "database" not in arguments or "schema" not in arguments:
          raise ValueError("Missing required 'database' and 'schema' parameters")
  
      database = arguments["database"]
      schema = arguments["schema"]
  
      query = f"""
          SELECT table_catalog, table_schema, table_name, comment 
          FROM {database}.information_schema.tables 
          WHERE table_schema = '{schema.upper()}'
      """
      data, data_id = await db.execute_query(query)
  
      # Filter out excluded tables
      if exclusion_config and "tables" in exclusion_config and exclusion_config["tables"]:
          filtered_data = []
          for item in data:
              table_name = item.get("TABLE_NAME", "")
              exclude = False
              for pattern in exclusion_config["tables"]:
                  if pattern.lower() in table_name.lower():
                      exclude = True
                      break
              if not exclude:
                  filtered_data.append(item)
          data = filtered_data
  
      output = {
          "type": "data",
          "data_id": data_id,
          "database": database,
          "schema": schema,
          "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:407-419 (registration)

  Registers the list_tables tool in the all_tools list, including name, description, input schema (requiring database and schema strings), and handler reference. The tools are later filtered into allowed_tools and exposed via list_tools().

  Tool(
      name="list_tables",
      description="List all tables in a specific database and schema",
      input_schema={
          "type": "object",
          "properties": {
              "database": {"type": "string", "description": "Database name"},
              "schema": {"type": "string", "description": "Schema name"},
          },
          "required": ["database", "schema"],
      },
      handler=handle_list_tables,
  ),

• src/mcp_snowflake_server/server.py:410-417 (schema)

  Input schema definition for list_tables tool: object with required 'database' and 'schema' string properties.

  input_schema={
      "type": "object",
      "properties": {
          "database": {"type": "string", "description": "Database name"},
          "schema": {"type": "string", "description": "Schema name"},
      },
      "required": ["database", "schema"],
  },

• src/mcp_snowflake_server/server.py:549-558 (helper)

  In the generic call_tool handler, special case for list_tables (and similar) to pass exclusion_config and exclude_json_results parameters.

  if name in ["list_databases", "list_schemas", "list_tables"]:
      return await handler(
          arguments,
          db,
          write_detector,
          allow_write,
          server,
          exclusion_config=exclusion_config,
          exclude_json_results=exclude_json_results,
      )