DataPilot MCP Server

Instructions

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