Panther MCP Server

Instructions

Implementation Reference

• src/mcp_panther/panther_mcp_core/tools/data_lake.py:256-424 (handler)

  The primary handler function for the 'query_data_lake' tool. It validates the SQL query for time filters, processes reserved words, executes the query via GraphQL mutation, polls for completion, supports pagination with cursor and max_rows, handles timeouts and cancellations, and returns structured results including metadata and pagination info.

      annotations={
          "permissions": all_perms(Permission.DATA_ANALYTICS_READ),
          "readOnlyHint": True,
      }
  )
  async def query_data_lake(
      sql: Annotated[
          str,
          Field(
              description="The SQL query to execute. Must include a p_event_time filter condition after WHERE or AND. The query must be compatible with Snowflake SQL."
          ),
      ],
      database_name: str = "panther_logs.public",
      timeout: Annotated[
          int,
          Field(
              description="Timeout in seconds before the SQL query is cancelled. If the query fails due to timeout, the caller should consider a longer timeout."
          ),
      ] = 30,
      max_rows: Annotated[
          int,
          Field(
              description="Maximum number of result rows to return (prevents context overflow)",
              ge=1,
              le=999,
          ),
      ] = 100,
      cursor: Annotated[
          str | None,
          Field(
              description="Optional pagination cursor from previous query to fetch next page of results",
          ),
      ] = None,
  ) -> Dict[str, Any]:
      """Query Panther's security data lake using SQL for log analysis and threat hunting.
  
      REQUIRED: Include time filter with p_event_time (required for performance and partitioning)
  
      Panther Time Filter Macros (Recommended - optimized for performance):
      - p_occurs_since(timeOffset [, tableAlias[, column]])
        Examples: p_occurs_since('1 d'), p_occurs_since('6 h'), p_occurs_since('2 weeks'), p_occurs_since(3600)
        Time formats: '30 s', '15 m', '6 h', '2 d', '1 w' OR '2 weeks', '1 day' OR numeric seconds
  
      - p_occurs_between(startTime, endTime [, tableAlias[, column]])
        Examples: p_occurs_between('2024-01-01', '2024-01-02'), p_occurs_between('2024-03-20T00:00:00Z', '2024-03-20T23:59:59Z')
  
      - p_occurs_around(timestamp, timeOffset [, tableAlias[, column]])
        Example: p_occurs_around('2024-01-15T10:30:00Z', '1 h')  # ±1 hour around timestamp
  
      - p_occurs_after(timestamp [, tableAlias[, column]])
      - p_occurs_before(timestamp [, tableAlias[, column]])
  
      Alternative (manual): WHERE p_event_time >= '2024-01-01' AND p_event_time < '2024-01-02'
  
      Best Practices:
      - Always use time filters (macros preferred over manual p_event_time conditions)
      - Start with summary queries, then drill down to specific timeframes
      - Use p_any_* fields for faster correlation (p_any_ip_addresses, p_any_usernames, p_any_emails)
      - Query specific fields instead of SELECT * for better performance
  
      Pagination:
      - First call: No cursor parameter - returns first page with max_rows results
      - Subsequent calls: Use next_cursor from previous response to get next page
      - Continue until has_next_page is False
  
      Common Examples:
      - Recent failed logins: "SELECT * FROM panther_logs.public.aws_cloudtrail WHERE p_occurs_since('1 d') AND errorcode IS NOT NULL"
      - IP activity summary: "SELECT sourceippaddress, COUNT(*) FROM panther_logs.public.aws_cloudtrail WHERE p_occurs_since('6 h') GROUP BY sourceippaddress"
      - User correlation: "SELECT * FROM panther_logs.public.aws_cloudtrail WHERE p_occurs_since('2 h') AND ARRAY_CONTAINS('user@domain.com'::VARIANT, p_any_emails)"
      - Nested field access: "SELECT p_enrichment:ipinfo_privacy:\"context.ip_address\" FROM table WHERE p_occurs_since('1 h')"
  
      Query Syntax (Snowflake SQL):
      - Access nested JSON: column:field.subfield
      - Quote special characters: column:"field name" or p_enrichment:"context.ip_address"
      - Array searches: ARRAY_CONTAINS('value'::VARIANT, array_column)
  
      Returns:
          Dict with query results:
          - results: List of matching rows (paginated based on cursor parameter)
          - results_truncated: True if results were truncated (only for non-paginated requests)
          - total_rows_available: Total rows found (for non-paginated requests)
          - has_next_page: True if more results are available
          - next_cursor: Cursor for next page (use in subsequent call)
          - column_info: Column names and data types
          - stats: Query performance metrics (execution time, bytes scanned)
          - success/status/message: Query execution status
      """
      logger.info("Executing data lake query")
  
      start_time = time.time()
  
      # Validate that the query includes a time filter (p_event_time or Panther macros)
      sql_lower = sql.lower().replace("\n", " ")
      has_p_event_time = re.search(
          r"\b(where|and)\s+.*?(?:[\w.]+\.)?p_event_time\s*(>=|<=|=|>|<|between)",
          sql_lower,
      )
      has_panther_macros = re.search(
          r"p_occurs_(since|between|around|after|before)\s*\(",
          sql_lower,
      )
  
      if (not (has_p_event_time or has_panther_macros)) and re.search(
          r"\Wpanther_(views|signals|rule_matches|rule_errors|monitor|logs|cloudsecurity)\.",
          sql_lower,
      ):
          error_msg = "Query must include a time filter: either `p_event_time` condition or Panther macro (p_occurs_since, p_occurs_between, etc.)"
          logger.error(error_msg)
          return {
              "success": False,
              "message": error_msg,
              "query_id": None,
          }
  
      try:
          # Process reserved words in the SQL
          processed_sql = wrap_reserved_words(sql)
          logger.debug(f"Original SQL: {sql}")
          logger.debug(f"Processed SQL: {processed_sql}")
  
          # Prepare input variables
          variables = {"input": {"sql": processed_sql, "databaseName": database_name}}
  
          logger.debug(f"Query variables: {variables}")
  
          # Execute the query using shared client
          result = await _execute_query(EXECUTE_DATA_LAKE_QUERY, variables)
  
          # Get query ID from result
          query_id = result.get("executeDataLakeQuery", {}).get("id")
  
          if not query_id:
              raise ValueError("No query ID returned from execution")
  
          logger.info(f"Successfully executed query with ID: {query_id}")
  
          sleep_time = INITIAL_QUERY_SLEEP_SECONDS
          while True:
              await asyncio.sleep(sleep_time)
  
              result = await _get_data_lake_query_results(
                  query_id=query_id, max_rows=max_rows, cursor=cursor
              )
  
              if result.get("status") == "running":
                  if (time.time() - start_time) >= timeout:
                      await _cancel_data_lake_query(query_id=query_id)
                      return {
                          "success": False,
                          "status": "cancelled",
                          "message": "Query time exceeded timeout, and has been cancelled. A longer timout may be required. "
                          "Retrying may be faster due to caching, or you may need to reduce the duration of data being queried.",
                          "query_id": query_id,
                      }
              else:
                  return result
  
              if sleep_time <= MAX_QUERY_SLEEP_SECONDS:
                  sleep_time += 1
      except Exception as e:
          logger.error(f"Failed to execute data lake query: {str(e)}")
          # Try to get query_id if it was set before the error
          query_id = locals().get("query_id")
          return {
              "success": False,
              "message": f"Failed to execute data lake query: {str(e)}",
              "query_id": query_id,
          }

• src/mcp_panther/panther_mcp_core/tools/data_lake.py:262-289 (schema)

  Pydantic-based input schema definition using Annotated and Field for parameters: sql (required), database_name, timeout, max_rows, cursor.

      sql: Annotated[
          str,
          Field(
              description="The SQL query to execute. Must include a p_event_time filter condition after WHERE or AND. The query must be compatible with Snowflake SQL."
          ),
      ],
      database_name: str = "panther_logs.public",
      timeout: Annotated[
          int,
          Field(
              description="Timeout in seconds before the SQL query is cancelled. If the query fails due to timeout, the caller should consider a longer timeout."
          ),
      ] = 30,
      max_rows: Annotated[
          int,
          Field(
              description="Maximum number of result rows to return (prevents context overflow)",
              ge=1,
              le=999,
          ),
      ] = 100,
      cursor: Annotated[
          str | None,
          Field(
              description="Optional pagination cursor from previous query to fetch next page of results",
          ),
      ] = None,
  ) -> Dict[str, Any]:

• src/mcp_panther/server.py:72-72 (registration)

  Registers all tools decorated with @mcp_tool, including query_data_lake, with the FastMCP server instance.

  register_all_tools(mcp)

• src/mcp_panther/panther_mcp_core/tools/data_lake.py:94-131 (helper)

  Helper function to process SQL for Snowflake reserved words by converting single-quoted reserved identifiers to double-quoted.

  def wrap_reserved_words(sql: str) -> str:
      """
      Simple function to wrap reserved words in SQL using sqlparse.
  
      This function:
      1. Parses the SQL using sqlparse
      2. Identifies string literals that match reserved words
      3. Converts single-quoted reserved words to double-quoted ones
  
      Args:
          sql: The SQL query string to process
  
      Returns:
          The SQL with reserved words properly quoted
      """
      try:
          # Parse the SQL
          parsed = sqlparse.parse(sql)[0]
  
          # Convert the parsed SQL back to string, but process tokens
          result = []
          for token in parsed.flatten():
              if token.ttype is sqlparse.tokens.Literal.String.Single:
                  # Remove quotes and check if it's a reserved word
                  value = token.value.strip("'")
                  if value.upper() in SNOWFLAKE_RESERVED_WORDS:
                      # Convert to double-quoted identifier
                      result.append(f'"{value}"')
                  else:
                      result.append(token.value)
              else:
                  result.append(token.value)
  
          return "".join(result)
      except Exception as e:
          logger.warning(f"Failed to parse SQL for reserved words: {e}")
          return sql

• src/mcp_panther/panther_mcp_core/tools/data_lake.py:426-575 (helper)

  Internal helper to retrieve query results, supporting pagination and formatting the response structure used by query_data_lake.

  async def _get_data_lake_query_results(
      query_id: Annotated[
          str,
          Field(
              description="The ID of the query to get results for",
              examples=["01be5f14-0206-3c48-000d-9eff005dd176"],
          ),
      ],
      max_rows: int = 100,
      cursor: str | None = None,
  ) -> Dict[str, Any]:
      """Get the results of a previously executed data lake query.
  
      Returns:
          Dict containing:
          - success: Boolean indicating if the query was successful
          - status: Status of the query (e.g., "succeeded", "running", "failed", "cancelled")
          - message: Error message if unsuccessful
          - results: List of query result rows
          - column_info: Dict containing column names and types
          - stats: Dict containing stats about the query
          - has_next_page: Boolean indicating if there are more results available
          - next_cursor: Cursor for fetching the next page of results, or null if no more pages
      """
      logger.info(f"Fetching data lake query results for query ID: {query_id}")
  
      try:
          # Prepare input variables for pagination
          variables = {
              "id": query_id,
              "root": False,
              "resultsInput": {
                  "pageSize": max_rows,
                  "cursor": cursor,
              },
          }
  
          if cursor:
              logger.info(f"Using pagination cursor: {cursor}")
  
          logger.debug(f"Query variables: {variables}")
  
          # Execute the query using shared client
          result = await _execute_query(GET_DATA_LAKE_QUERY, variables)
  
          # Get query data
          query_data = result.get("dataLakeQuery", {})
  
          if not query_data:
              logger.warning(f"No query found with ID: {query_id}")
              return {
                  "success": False,
                  "message": f"No query found with ID: {query_id}",
                  "query_id": query_id,
              }
  
          # Get query status
          status = query_data.get("status")
          if status == "running":
              return {
                  "success": True,
                  "status": "running",
                  "message": "Query is still running",
                  "query_id": query_id,
              }
          elif status == "failed":
              return {
                  "success": False,
                  "status": "failed",
                  "message": query_data.get("message", "Query failed"),
                  "query_id": query_id,
              }
          elif status == "cancelled":
              return {
                  "success": False,
                  "status": "cancelled",
                  "message": "Query was cancelled",
                  "query_id": query_id,
              }
  
          # Get results data
          results = query_data.get("results", {})
          edges = results.get("edges", [])
          column_info = results.get("columnInfo", {})
          stats = results.get("stats", {})
  
          # Extract results from edges
          query_results = [edge["node"] for edge in edges]
  
          # Check pagination info
          page_info = results.get("pageInfo", {})
          has_next_page = page_info.get("hasNextPage", False)
          end_cursor = page_info.get("endCursor")
  
          # Track pagination state
          original_count = len(query_results)
  
          # For cursor-based requests, we don't truncate since GraphQL handles pagination
          was_truncated = False
          if cursor:
              logger.info(
                  f"Retrieved page of {len(query_results)} results using cursor pagination"
              )
          else:
              # For initial requests without cursor, apply legacy truncation if needed
              was_truncated = original_count > max_rows
              if was_truncated:
                  query_results = query_results[:max_rows]
                  logger.info(
                      f"Query results truncated from {original_count} to {max_rows} rows for context window protection"
                  )
              else:
                  logger.info(
                      f"Retrieved {len(query_results)} results (no truncation needed)"
                  )
  
          logger.info(
              f"Successfully retrieved {len(query_results)} results for query ID: {query_id}"
          )
  
          # Format the response
          return {
              "success": True,
              "status": status,
              "results": query_results,
              "results_truncated": was_truncated,
              "total_rows_available": original_count,
              "rows_returned": len(query_results),
              "column_info": {
                  "order": column_info.get("order", []),
                  "types": column_info.get("types", {}),
              },
              "stats": {
                  "bytes_scanned": stats.get("bytesScanned", 0),
                  "execution_time": stats.get("executionTime", 0),
                  "row_count": stats.get("rowCount", 0),
              },
              "has_next_page": has_next_page,
              "next_cursor": end_cursor,
              "message": query_data.get("message", "Query executed successfully"),
              "query_id": query_id,
          }
      except Exception as e:
          logger.error(f"Failed to fetch data lake query results: {str(e)}")
          return {
              "success": False,
              "message": f"Failed to fetch data lake query results: {str(e)}",
              "query_id": query_id,
          }