get_warehouse_status
Check current Snowflake warehouse, database, and schema operational status to monitor system health and availability.
Instructions
Get current warehouse, database, and schema status
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
No arguments | |||
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
No arguments | |||
Implementation Reference
- src/main.py:254-269 (handler)The MCP tool handler for 'get_warehouse_status', decorated with @mcp.tool(). It initializes the Snowflake client and delegates to client.get_warehouse_status() to retrieve current warehouse, database, and schema.
@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-204 (helper)Supporting method in SnowflakeClient class that executes the SQL query to fetch and format the current warehouse, database, and schema status.
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 {} - src/main.py:254-254 (registration)The @mcp.tool() decorator registers the get_warehouse_status function as an MCP tool.
@mcp.tool()