[{'Text': '        Returns:\n            Results starting with `query_id:<id>` on the first line, followed by CSV data.\n            The query_id can be passed to cancel_query() to cancel a running query,\n            or to get_query_status() to check its current state.\n            For multiple queries, each result section includes its own query_id.\n\n        Examples:\n            Single query: "SELECT COUNT(*) FROM table1"\n            Multiple queries: "SELECT COUNT(*) FROM table1; SELECT AVG(price) FROM table2"\n        """\n        from point_topic_mcp.connectors.snowflake import SnowflakeDB, non_empty_sql_statements\n\n        query_list = non_empty_sql_statements(sql_query)\n\n        # multi-query: use existing path (cancellation not supported per-query)\n        if len(query_list) > 1:\n            sf = SnowflakeDB()\n            sf.connect()\n            result = sf.execute_safe_queries(sql_query)\n            sf.close_connection()\n            return result\n\n        # single query: submit, emit query_id immediately via ctx, then fetch\n        sf = SnowflakeDB()\n        sf.connect()\n        try:\n            cursor, query_id = sf.submit_async_query(sql_query)\n        except ValueError as e:\n            sf.close_connection()\n            return f"ERROR: {e}"\n        except Exception as e:\n            sf.close_connection()\n            return f"DATABASE ERROR: {str(e)}"\n\n        # Emit query_id mid-execution via MCP notifications/progress (not logging).\n        # UPC Query Agent and other clients can subscribe to progress on tools/call.\n        if ctx:\n            report = getattr(ctx, "report_progress", None)\n            if report is None:\n                raise RuntimeError(\n                    "execute_query requires Context.report_progress (FastMCP progress API). "\n                    "Upgrade FastMCP or fix Context injection."\n                )\n            payload = json.dumps({"event": "query_executing", "query_id": query_id})\n            await report(progress=0, total=None, message=payload)\n\n        # run blocking poll loop in a thread so the event loop stays free\n        # to flush progress notifications to the client\n        import asyncio\n        result = await asyncio.to_thread(sf.poll_and_fetch, cursor, query_id)\n        sf.close_connection()\n        return result\n\n    def get_query_status(\n        query_id: str,\n        ctx: Optional[Context] = None\n    ) -> str:\n        """\n        Check the current status of a Snowflake query.\n\n        Useful for verifying a query is still running before cancelling it,\n        or confirming a cancellation completed.\n\n        Args:\n            query_id: The query ID returned in the first line of execute_query results.\n\n        Returns:\n            JSON with query_id and status: RUNNING, SUCCESS, CANCELLED, or ERROR:<message>\n        """\n        import json\n        from point_topic_mcp.connectors.snowflake import SnowflakeDB\n'}]