Panther MCP Server

Instructions

Implementation Reference

• src/mcp_panther/panther_mcp_core/tools/metrics.py:28-142 (handler)

  The core handler implementation for the 'get_severity_alert_metrics' tool. This async function is decorated with @mcp_tool for automatic registration, defines the input schema using Pydantic Annotated types with validators and Field descriptions, executes a GraphQL query to fetch alert metrics grouped by severity, filters results based on provided alert_types and severities, handles default date ranges, and returns a dictionary with success status, filtered metrics, totals, and parameters.

  @mcp_tool(
      annotations={
          "permissions": all_perms(Permission.SUMMARY_READ),
          "readOnlyHint": True,
      }
  )
  async def get_severity_alert_metrics(
      start_date: Annotated[
          str | None,
          Field(
              description="Optional start date in ISO-8601 format. If provided, defaults to the start of the current day UTC.",
              examples=["2024-03-20T00:00:00Z"],
          ),
      ] = None,
      end_date: Annotated[
          str | None,
          Field(
              description="Optional end date in ISO-8601 format. If provided, defaults to the end of the current day UTC.",
              examples=["2024-03-20T00:00:00Z"],
          ),
      ] = None,
      alert_types: Annotated[
          list[str],
          BeforeValidator(_validate_alert_types),
          Field(
              description="The specific Panther alert types to get metrics for.",
              examples=[["Rule"], ["Policy"], ["Rule", "Policy"]],
          ),
      ] = ["Rule", "Policy"],
      severities: Annotated[
          list[str],
          BeforeValidator(_validate_severities),
          Field(
              description="The specific Panther alert severities to get metrics for.",
              examples=[
                  ["CRITICAL", "HIGH"],
                  ["MEDIUM", "LOW"],
                  ["CRITICAL", "HIGH", "MEDIUM", "LOW", "INFO"],
              ],
          ),
      ] = ["CRITICAL", "HIGH", "MEDIUM", "LOW"],
      interval_in_minutes: Annotated[
          int,
          BeforeValidator(_validate_interval),
          Field(
              description="How data points are aggregated over time, with smaller intervals providing more granular detail of when events occurred, while larger intervals show broader trends but obscure the precise timing of incidents.",
              examples=[15, 30, 60, 180, 360, 720, 1440],
          ),
      ] = 1440,
  ) -> dict[str, Any]:
      """Gets alert metrics grouped by severity for rule and policy alert types within a given time period. Use this tool to identify hot spots in your alerts, and use the list_alerts tool for specific details. Keep in mind that these metrics combine errors and alerts, so there may be inconsistencies from what list_alerts returns.
  
      Returns:
          Dict:
          - alerts_per_severity: List of series with breakdown by severity
          - total_alerts: Total number of alerts in the period
          - start_date: Start date of the period
          - end_date: End date of the period
          - interval_in_minutes: Grouping interval for the metrics
      """
      try:
          # If start or end date is missing, use week's date range
          if not start_date or not end_date:
              default_start_date, default_end_date = _get_week_date_range()
              if not start_date:
                  start_date = default_start_date
              if not end_date:
                  end_date = default_end_date
  
          logger.info(
              f"Fetching alerts per severity metrics from {start_date} to {end_date}"
          )
  
          # Prepare variables for GraphQL query
          variables = {
              "input": {
                  "fromDate": start_date,
                  "toDate": end_date,
                  "intervalInMinutes": interval_in_minutes,
              }
          }
  
          # Execute GraphQL query
          result = await _execute_query(METRICS_ALERTS_PER_SEVERITY_QUERY, variables)
  
          if not result or "metrics" not in result:
              logger.error(f"Could not find key 'metrics' in result: {result}")
              raise Exception("Failed to fetch metrics data")
  
          metrics_data = result["metrics"]
  
          # Filter metrics data by alert types and severities
          alerts_per_severity = [
              item
              for item in metrics_data["alertsPerSeverity"]
              if any(alert_type in item["label"] for alert_type in alert_types)
              and any(severity in item["label"] for severity in severities)
          ]
  
          return {
              "success": True,
              "alerts_per_severity": alerts_per_severity,
              "total_alerts": metrics_data["totalAlerts"],
              "start_date": start_date,
              "end_date": end_date,
              "interval_in_minutes": interval_in_minutes,
          }
  
      except Exception as e:
          logger.error(f"Failed to fetch alerts per severity metrics: {str(e)}")
          return {
              "success": False,
              "message": f"Failed to fetch alerts per severity metrics: {str(e)}",
          }

• src/mcp_panther/panther_mcp_core/validators.py:12-21 (helper)

  The _validate_severities validator function used in the tool's input schema to validate severity parameters. Similar validators like _validate_alert_types are in this file, providing schema validation helpers.

  def _validate_severities(v: list[str]) -> list[str]:
      """Validate severities are valid."""
      valid_severities = {"CRITICAL", "HIGH", "MEDIUM", "LOW", "INFO"}
      for severity in v:
          if severity not in valid_severities:
              raise ValueError(
                  f"Invalid severity '{severity}'. Must be one of: {', '.join(sorted(valid_severities))}"
              )
      return v

• src/mcp_panther/panther_mcp_core/queries.py:21-178 (helper)

  The GraphQL query METRICS_ALERTS_PER_SEVERITY_QUERY used by the handler to fetch raw metrics data from Panther.

                      logProcessingRole
                      logStreamType
                      logStreamTypeOptions {
                          jsonArrayEnvelopeField
                      }
                      managedBucketNotifications
                      s3Bucket
                      s3Prefix
                      s3PrefixLogTypes {
                          prefix
                          logTypes
                          excludedPrefixes
                      }
                      stackName
                  }
              }
          }
          pageInfo {
              hasNextPage
              hasPreviousPage
              startCursor
              endCursor
          }
      }
  }
  """)
  
  # Data Lake Queries
  EXECUTE_DATA_LAKE_QUERY = gql("""
  mutation ExecuteDataLakeQuery($input: ExecuteDataLakeQueryInput!) {
      executeDataLakeQuery(input: $input) {
          id
      }
  }
  """)
  
  GET_DATA_LAKE_QUERY = gql("""
  query GetDataLakeQuery($id: ID!, $root: Boolean = false, $resultsInput: DataLakeQueryResultsInput) {
      dataLakeQuery(id: $id, root: $root) {
          id
          status
          message
          sql
          startedAt
          completedAt
          results(input: $resultsInput) {
              edges {
                  node
              }
              pageInfo {
                  hasNextPage
                  endCursor
              }
              columnInfo {
                  order
                  types
              }
              stats {
                  bytesScanned
                  executionTime
                  rowCount
              }
          }
      }
  }
  """)
  
  LIST_DATABASES_QUERY = gql("""
  query ListDatabases {
      dataLakeDatabases {
          name
          description
      }
  }
  """)
  
  LIST_TABLES_QUERY = gql("""
  query ListTables($databaseName: String!, $pageSize: Int, $cursor: String) {
    dataLakeDatabaseTables(
      input: {
        databaseName: $databaseName
        pageSize: $pageSize
        cursor: $cursor
      }
    ) {
      edges {
        node {
          name
          description
          logType
        }
      }
      pageInfo {
        hasNextPage
        endCursor
      }
    }
  }
  """)
  
  GET_COLUMNS_FOR_TABLE_QUERY = gql("""
  query GetColumnDetails($databaseName: String!, $tableName: String!) {
    dataLakeDatabaseTable(input: { databaseName: $databaseName, tableName: $tableName }) {
      name,
      displayName,
      description,
      logType,
      columns {
        name,
        type,
        description
      }
    }
  }
  """)
  
  LIST_SCHEMAS_QUERY = gql("""
  query ListSchemas($input: SchemasInput!) {
      schemas(input: $input) {
          edges {
              node {
                  name
                  description
                  revision
                  isArchived
                  isManaged
                  referenceURL
                  createdAt
                  updatedAt
              }
          }
      }
  }
  """)
  
  CREATE_OR_UPDATE_SCHEMA_MUTATION = gql("""
  mutation CreateOrUpdateSchema($input: CreateOrUpdateSchemaInput!) {
      createOrUpdateSchema(input: $input) {
          schema {
              name
              description
              spec
              version
              revision
              isArchived
              isManaged
              isFieldDiscoveryEnabled
              referenceURL
              discoveredSpec
              createdAt
              updatedAt
          }
      }
  }
  """)
  
  # Metrics Queries
  METRICS_ALERTS_PER_SEVERITY_QUERY = gql("""

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

  The call to register_all_tools(mcp) which registers all @mcp_tool decorated functions, including get_severity_alert_metrics, with the MCP server instance.

  register_all_tools(mcp)