Panther MCP Server

Instructions

Implementation Reference

• src/mcp_panther/panther_mcp_core/tools/metrics.py:144-247 (handler)

  Core handler function for get_rule_alert_metrics tool. Executes GraphQL query for metrics, handles date ranges, filters by rule_ids, and returns formatted results. Includes input schema definitions and @mcp_tool decorator for registration.

  @mcp_tool(
      annotations={
          "permissions": all_perms(Permission.SUMMARY_READ),
          "readOnlyHint": True,
      }
  )
  async def get_rule_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,
      interval_in_minutes: Annotated[
          int,
          BeforeValidator(_validate_interval),
          Field(
              description="Intervals for aggregating data points. Smaller intervals provide 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],
          ),
      ] = 15,
      rule_ids: Annotated[
          list[str],
          BeforeValidator(_validate_rule_ids),
          Field(
              description="A valid JSON list of Panther rule IDs to get metrics for",
              examples=[["AppOmni.Alert.Passthrough", "Auth0.MFA.Policy.Disabled"]],
          ),
      ] = [],
  ) -> dict[str, Any]:
      """Gets alert metrics grouped by detection rule for ALL alert types, including alerts, detection errors, and system errors within a given time period. Use this tool to identify hot spots in alerts and use list_alerts for specific alert details.
  
      Returns:
          Dict:
          - alerts_per_rule: List of series with entityId, label, and value
          - 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
          - rule_ids: List of rule IDs if provided
      """
      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 rule metrics from {start_date} to {end_date}")
  
          # Prepare variables
          variables = {
              "input": {
                  "fromDate": start_date,
                  "toDate": end_date,
                  "intervalInMinutes": interval_in_minutes,
              }
          }
  
          # Execute query
          result = await _execute_query(METRICS_ALERTS_PER_RULE_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 by rule IDs if provided
          if rule_ids:
              alerts_per_rule = [
                  item
                  for item in metrics_data["alertsPerRule"]
                  if item["entityId"] in rule_ids
              ]
          else:
              alerts_per_rule = metrics_data["alertsPerRule"]
  
          return {
              "success": True,
              "alerts_per_rule": alerts_per_rule,
              "total_alerts": len(alerts_per_rule),
              "start_date": start_date,
              "end_date": end_date,
              "interval_in_minutes": interval_in_minutes,
              "rule_ids": rule_ids if rule_ids else None,
          }
  
      except Exception as e:
          logger.error(f"Failed to fetch rule alert metrics: {str(e)}")
          return {
              "success": False,
              "message": f"Failed to fetch rule alert metrics: {str(e)}",
          }

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

  Registration of all MCP tools (including get_rule_alert_metrics) to the FastMCP server instance via register_all_tools(mcp).

  # Create the MCP server with lifespan context for shared HTTP client management
  # Note: Dependencies are declared in fastmcp.json for FastMCP v2.14.0+
  mcp = FastMCP(MCP_SERVER_NAME, lifespan=lifespan)
  
  # Register all tools with MCP using the registry
  register_all_tools(mcp)
  # Register all prompts with MCP using the registry
  register_all_prompts(mcp)
  # Register all resources with MCP using the registry
  register_all_resources(mcp)

• src/mcp_panther/panther_mcp_core/queries.py:191-202 (helper)

  GraphQL query definition used by the get_rule_alert_metrics handler to fetch alerts per rule metrics.

  METRICS_ALERTS_PER_RULE_QUERY = gql("""
  query Metrics($input: MetricsInput!) {
      metrics(input: $input) {
          alertsPerRule {
              entityId
              label
              value
          }
          totalAlerts
      }
  }
  """)

• src/mcp_panther/panther_mcp_core/validators.py:82-90 (helper)

  Pydantic validator for rule_ids parameter used in the tool's input schema.

  def _validate_rule_ids(v: list[str]) -> list[str]:
      """Validate rule IDs don't contain problematic characters."""
      problematic_chars = re.compile(r"[@\s#]")
      for rule_id in v:
          if problematic_chars.search(rule_id):
              raise ValueError(
                  f"Invalid rule ID '{rule_id}'. Rule IDs cannot contain '@', spaces, or '#' characters"
              )
      return v