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