mcp-snowflake-server

Instructions

Implementation Reference

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

  Core handler function for the 'list_schemas' tool. Queries Snowflake INFORMATION_SCHEMA.SCHEMATA for the given database, filters out excluded schemas based on patterns, formats output as YAML text content and optional JSON embedded 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:395-404 (schema)

  Input schema definition for the 'list_schemas' tool, specifying the required 'database' parameter as a string.
  input_schema={ "type": "object", "properties": { "database": { "type": "string", "description": "Database name to list schemas from", }, }, "required": ["database"], },
• src/mcp_snowflake_server/server.py:392-406 (registration)

  Registration of the 'list_schemas' tool in the all_tools list, linking name, description, input schema, and 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:549-558 (helper)

  Helper logic in the generic call_tool handler to pass exclusion_config and exclude_json_results specifically to listing tools including 'list_schemas'.
  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, )