get_warehouse_status
Retrieve real-time status of warehouses, databases, and schemas to monitor performance and availability within Snowflake databases using AI-guided operations.
Instructions
Get current warehouse, database, and schema status
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
No arguments | |||
Implementation Reference
- src/main.py:254-269 (handler)The main MCP tool handler for get_warehouse_status, decorated with @mcp.tool(). It retrieves the Snowflake client and calls its get_warehouse_status method, handling errors and providing context updates.@mcp.tool() async def get_warehouse_status(ctx: Context) -> Dict[str, Any]: """Get current warehouse, database, and schema status""" await ctx.info("Getting current warehouse status...") try: client = await get_snowflake_client() status = await client.get_warehouse_status() await ctx.info("Retrieved warehouse status") return status except Exception as e: logger.error(f"Error getting warehouse status: {str(e)}") await ctx.error(f"Failed to get warehouse status: {str(e)}") return {}
- src/snowflake_client.py:193-203 (helper)Helper method in SnowflakeClient class that executes the SQL query to fetch current warehouse, database, and schema information.async def get_warehouse_status(self) -> Dict[str, Any]: """Get current warehouse status""" result = await self.execute_query("SELECT CURRENT_WAREHOUSE(), CURRENT_DATABASE(), CURRENT_SCHEMA()") if result.success and result.data: row = result.data[0] return { 'current_warehouse': row.get('CURRENT_WAREHOUSE()', ''), 'current_database': row.get('CURRENT_DATABASE()', ''), 'current_schema': row.get('CURRENT_SCHEMA()', '') } return {}