Panther MCP Server

Instructions

Implementation Reference

• src/mcp_panther/panther_mcp_core/tools/alerts.py:35-322 (handler)

  The primary handler function for the 'list_alerts' MCP tool. It validates parameters using Pydantic annotations (serving as schema), applies business logic for defaults and limits, constructs query parameters for the Panther /alerts REST API endpoint, handles pagination, errors, and formats the response.
  @mcp_tool( annotations={ "permissions": all_perms(Permission.ALERT_READ), "readOnlyHint": True, } ) async def list_alerts( start_date: Annotated[ str | None, BeforeValidator(_validate_iso_date), 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, BeforeValidator(_validate_iso_date), 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, severities: Annotated[ list[str], BeforeValidator(_validate_severities), Field( description="Optional list of severities to filter by", examples=[["CRITICAL", "HIGH", "MEDIUM", "LOW", "INFO"]], ), ] = [], statuses: Annotated[ list[str], BeforeValidator(_validate_statuses), Field( description="Optional list of statuses to filter by", examples=[ ["OPEN", "TRIAGED", "RESOLVED", "CLOSED"], ["RESOLVED", "CLOSED"], ["OPEN", "TRIAGED"], ], ), ] = [], cursor: Annotated[ str | None, Field( min_length=1, description="Optional cursor for pagination returned from a previous call", ), ] = None, detection_id: Annotated[ str | None, Field( min_length=1, description="Optional detection ID to filter alerts by; if not provided, default date range (7days) is applied", ), ] = None, event_count_max: Annotated[ int | None, Field( ge=1, description="Optional maximum number of events an alert may contain" ), ] = None, event_count_min: Annotated[ int, Field( ge=1, description="Optional minimum number of events an alert must contain" ), ] = 1, log_sources: Annotated[ list[str], Field(description="Optional list of log‑source IDs to filter alerts by"), ] = [], log_types: Annotated[ list[str], Field(description="Optional list of log‑type names to filter alerts by"), ] = [], name_contains: Annotated[ str | None, Field( min_length=1, description="Optional substring to match within alert titles" ), ] = None, page_size: Annotated[ int, Field( description="Number of results per page (max 50, default 25)", ge=1, le=50, ), ] = 25, resource_types: Annotated[ list[str], Field( description="Optional list of AWS resource‑type names to filter alerts by" ), ] = [], subtypes: Annotated[ list[str], BeforeValidator(_validate_subtypes), Field( description="Optional list of alert subtypes (valid values depend on alert_type)", examples=[ ["RULE"], # Python rules only ["SCHEDULED_RULE"], # Scheduled queries only ["POLICY"], # Cloud policies only ["RULE", "SCHEDULED_RULE"], # Both rule types (when alert_type=ALERT) [ "RULE_ERROR", "SCHEDULED_RULE_ERROR", ], # When alert_type=DETECTION_ERROR ], ), ] = [], alert_type: Annotated[ str, BeforeValidator(_validate_alert_api_types), Field( description="Type of alerts to return", examples=["ALERT", "DETECTION_ERROR", "SYSTEM_ERROR"], ), ] = "ALERT", ) -> dict[str, Any]: """List alerts from Panther with comprehensive filtering options Args: start_date: Optional start date in ISO 8601 format (e.g. "2024-03-20T00:00:00Z") end_date: Optional end date in ISO 8601 format (e.g. "2024-03-21T00:00:00Z") severities: Optional list of severities to filter by (e.g. ["CRITICAL", "HIGH", "MEDIUM", "LOW", "INFO"]) statuses: Optional list of statuses to filter by (e.g. ["OPEN", "TRIAGED", "RESOLVED", "CLOSED"]) cursor: Optional cursor for pagination from a previous query detection_id: Optional detection ID to filter alerts by. If not provided, default date range (7days) is applied. event_count_max: Optional maximum number of events that returned alerts must have event_count_min: Optional minimum number of events that returned alerts must have log_sources: Optional list of log source IDs to filter alerts by log_types: Optional list of log type names to filter alerts by name_contains: Optional string to search for in alert titles page_size: Number of results per page (default: 25, maximum: 50) resource_types: Optional list of AWS resource type names to filter alerts by subtypes: Optional list of alert subtypes. Valid values depend on alert_type: - When alert_type="ALERT": ["POLICY", "RULE", "SCHEDULED_RULE"] - When alert_type="DETECTION_ERROR": ["RULE_ERROR", "SCHEDULED_RULE_ERROR"] - When alert_type="SYSTEM_ERROR": subtypes are not allowed alert_type: Type of alerts to return (default: "ALERT"). One of: - "ALERT": Regular detection alerts - "DETECTION_ERROR": Alerts from detection errors - "SYSTEM_ERROR": System error alerts """ logger.info("Fetching alerts from Panther") try: # Validate page size if page_size < 1: raise ValueError("page_size must be greater than 0") if page_size > 50: logger.warning( f"page_size {page_size} exceeds maximum of 50, using 50 instead" ) page_size = 50 # Validate alert_type and subtypes combination valid_alert_types = ["ALERT", "DETECTION_ERROR", "SYSTEM_ERROR"] if alert_type not in valid_alert_types: raise ValueError(f"alert_type must be one of {valid_alert_types}") if subtypes: valid_subtypes = { "ALERT": ["POLICY", "RULE", "SCHEDULED_RULE"], "DETECTION_ERROR": ["RULE_ERROR", "SCHEDULED_RULE_ERROR"], "SYSTEM_ERROR": [], } if alert_type == "SYSTEM_ERROR": raise ValueError( "subtypes are not allowed when alert_type is SYSTEM_ERROR" ) allowed_subtypes = valid_subtypes[alert_type] invalid_subtypes = [st for st in subtypes if st not in allowed_subtypes] if invalid_subtypes: raise ValueError( f"Invalid subtypes {invalid_subtypes} for alert_type={alert_type}. " f"Valid subtypes are: {allowed_subtypes}" ) # Prepare query parameters params = { "type": alert_type, "limit": page_size, "sort-dir": "desc", } # Handle the required filter: either detection-id OR date range if detection_id: params["detection-id"] = detection_id logger.info(f"Filtering by detection ID: {detection_id}") # Add a default date filter (7days) if no detection_id if not detection_id and not (start_date or end_date): start_date, end_date = _get_week_date_range() if start_date: params["created-after"] = start_date if end_date: params["created-before"] = end_date # Add optional filters if cursor: if not isinstance(cursor, str): raise ValueError( "Cursor must be a string value from previous response's next" ) params["cursor"] = cursor logger.info(f"Using cursor for pagination: {cursor}") if severities: params["severity"] = severities logger.info(f"Filtering by severities: {severities}") if statuses: params["status"] = statuses logger.info(f"Filtering by statuses: {statuses}") if event_count_max is not None: params["event-count-max"] = event_count_max logger.info(f"Filtering by max event count: {event_count_max}") if event_count_min is not None: params["event-count-min"] = event_count_min logger.info(f"Filtering by min event count: {event_count_min}") if log_sources: params["log-source"] = log_sources logger.info(f"Filtering by log sources: {log_sources}") if log_types: params["log-type"] = log_types logger.info(f"Filtering by log types: {log_types}") if name_contains: params["name-contains"] = name_contains logger.info(f"Filtering by name contains: {name_contains}") if resource_types: params["resource-type"] = resource_types logger.info(f"Filtering by resource types: {resource_types}") if subtypes: params["sub-type"] = subtypes logger.info(f"Filtering by subtypes: {subtypes}") logger.debug(f"Query parameters: {params}") # Execute the REST API call async with get_rest_client() as client: result, status = await client.get( "/alerts", params=params, expected_codes=[200, 400] ) if status == 400: logger.error("Bad request when fetching alerts") return { "success": False, "message": "Bad request when fetching alerts", } # Log the raw result for debugging logger.debug(f"Raw API result: {result}") # Process results alerts = result.get("results", []) next_cursor = result.get("next") logger.info(f"Successfully retrieved {len(alerts)} alerts") # Format the response return { "success": True, "alerts": alerts, "total_alerts": len(alerts), "has_next_page": next_cursor is not None, "has_previous_page": cursor is not None, "end_cursor": next_cursor, "start_cursor": cursor, } except Exception as e: logger.error(f"Failed to fetch alerts: {str(e)}") return {"success": False, "message": f"Failed to fetch alerts: {str(e)}"}
• src/mcp_panther/server.py:72-76 (registration)

  The central registration call in the MCP server startup that registers all @mcp_tool-decorated functions, including list_alerts, with the FastMCP instance.
  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/tools/__init__.py:26-26 (registration)

  Import of alerts.py module which triggers execution of the @mcp_tool decorator on list_alerts, adding it to the tool registry.
  alerts,
• src/mcp_panther/panther_mcp_core/tools/alerts.py:42-157 (schema)

  Pydantic-based input schema defined via Annotated types, validators, and Field metadata in the function signature.
  start_date: Annotated[ str | None, BeforeValidator(_validate_iso_date), 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, BeforeValidator(_validate_iso_date), 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, severities: Annotated[ list[str], BeforeValidator(_validate_severities), Field( description="Optional list of severities to filter by", examples=[["CRITICAL", "HIGH", "MEDIUM", "LOW", "INFO"]], ), ] = [], statuses: Annotated[ list[str], BeforeValidator(_validate_statuses), Field( description="Optional list of statuses to filter by", examples=[ ["OPEN", "TRIAGED", "RESOLVED", "CLOSED"], ["RESOLVED", "CLOSED"], ["OPEN", "TRIAGED"], ], ), ] = [], cursor: Annotated[ str | None, Field( min_length=1, description="Optional cursor for pagination returned from a previous call", ), ] = None, detection_id: Annotated[ str | None, Field( min_length=1, description="Optional detection ID to filter alerts by; if not provided, default date range (7days) is applied", ), ] = None, event_count_max: Annotated[ int | None, Field( ge=1, description="Optional maximum number of events an alert may contain" ), ] = None, event_count_min: Annotated[ int, Field( ge=1, description="Optional minimum number of events an alert must contain" ), ] = 1, log_sources: Annotated[ list[str], Field(description="Optional list of log‑source IDs to filter alerts by"), ] = [], log_types: Annotated[ list[str], Field(description="Optional list of log‑type names to filter alerts by"), ] = [], name_contains: Annotated[ str | None, Field( min_length=1, description="Optional substring to match within alert titles" ), ] = None, page_size: Annotated[ int, Field( description="Number of results per page (max 50, default 25)", ge=1, le=50, ), ] = 25, resource_types: Annotated[ list[str], Field( description="Optional list of AWS resource‑type names to filter alerts by" ), ] = [], subtypes: Annotated[ list[str], BeforeValidator(_validate_subtypes), Field( description="Optional list of alert subtypes (valid values depend on alert_type)", examples=[ ["RULE"], # Python rules only ["SCHEDULED_RULE"], # Scheduled queries only ["POLICY"], # Cloud policies only ["RULE", "SCHEDULED_RULE"], # Both rule types (when alert_type=ALERT) [ "RULE_ERROR", "SCHEDULED_RULE_ERROR", ], # When alert_type=DETECTION_ERROR ], ), ] = [], alert_type: Annotated[ str, BeforeValidator(_validate_alert_api_types), Field( description="Type of alerts to return", examples=["ALERT", "DETECTION_ERROR", "SYSTEM_ERROR"], ), ] = "ALERT", ) -> dict[str, Any]:
• src/mcp_panther/panther_mcp_core/tools/alerts.py:12-15 (helper)

  Imports helpers like get_rest_client from client.py used in the API call.
  _execute_query, _get_week_date_range, get_rest_client, )