list_warehouses
Retrieve available Snowflake warehouses for database operations and resource management through the DataPilot MCP Server.
Instructions
List all warehouses available to the user
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
No arguments | |||
Implementation Reference
- src/main.py:237-252 (handler)The main tool handler for 'list_warehouses', decorated with @mcp.tool() for registration. It gets the Snowflake client, calls its list_warehouses method, and returns the list of warehouses.@mcp.tool() async def list_warehouses(ctx: Context) -> List[Dict[str, Any]]: """List all warehouses available to the user""" await ctx.info("Retrieving list of warehouses...") try: client = await get_snowflake_client() warehouses = await client.list_warehouses() await ctx.info(f"Found {len(warehouses)} warehouses") return warehouses except Exception as e: logger.error(f"Error listing warehouses: {str(e)}") await ctx.error(f"Failed to list warehouses: {str(e)}") return []
- src/snowflake_client.py:188-192 (helper)Helper method in SnowflakeClient class that executes 'SHOW WAREHOUSES' query and returns the results.async def list_warehouses(self) -> List[Dict[str, Any]]: """List all warehouses user has access to""" result = await self.execute_query("SHOW WAREHOUSES") return result.data if result.success else []
- src/main.py:237-252 (registration)The @mcp.tool() decorator registers this function as an MCP tool named 'list_warehouses'.@mcp.tool() async def list_warehouses(ctx: Context) -> List[Dict[str, Any]]: """List all warehouses available to the user""" await ctx.info("Retrieving list of warehouses...") try: client = await get_snowflake_client() warehouses = await client.list_warehouses() await ctx.info(f"Found {len(warehouses)} warehouses") return warehouses except Exception as e: logger.error(f"Error listing warehouses: {str(e)}") await ctx.error(f"Failed to list warehouses: {str(e)}") return []