Panther MCP Server

Instructions

Implementation Reference

• src/mcp_panther/panther_mcp_core/tools/detections.py:121-426 (handler)

  Core handler for list_detections tool. Includes @mcp_tool decorator, input schema via Annotated[Field], validation helpers integrated, API interaction logic, client-side filtering, and structured response formatting.

  @mcp_tool(
      annotations={
          "permissions": all_perms(Permission.RULE_READ, Permission.POLICY_READ),
          "readOnlyHint": True,
      }
  )
  async def list_detections(
      detection_types: Annotated[
          list[str],
          Field(
              description="One or more detection types - rules, scheduled_rules, simple_rules, or policies.",
              examples=[
                  ["rules", "simple_rules", "scheduled_rules"],
                  ["policies"],
              ],
          ),
      ] = ["rules"],
      cursor: Annotated[
          str | None,
          Field(
              description="Optional cursor for pagination from a previous query (only supported for single detection type)"
          ),
      ] = None,
      limit: Annotated[
          int,
          Field(
              description="Maximum number of results to return per detection type",
              default=100,
              ge=1,
              le=1000,
          ),
      ] = 100,
      name_contains: Annotated[
          str | None, Field(description="Substring search by name (case-insensitive)")
      ] = None,
      state: Annotated[
          str,
          Field(
              description="Filter by state - 'enabled' or 'disabled'", default="enabled"
          ),
      ] = "",
      severity: Annotated[
          list[str],
          Field(
              description="Filter by severity levels - INFO, LOW, MEDIUM, HIGH, or CRITICAL.",
              examples=[
                  ["MEDIUM", "HIGH", "CRITICAL"],
                  ["INFO", "LOW"],
              ],
          ),
      ] = [],
      tag: Annotated[
          list[str],
          Field(
              description="A case-insensitive list of tags to filter by.",
              examples=[["Initial Access", "Persistence"]],
          ),
      ] = [],
      log_type: Annotated[
          list[str],
          Field(
              description="A list of log types to filter by (applies to rules and simple-rules only).",
              examples=[["AWS.CloudTrail", "GCP.AuditLog"]],
          ),
      ] = [],
      resource_type: Annotated[
          list[str],
          Field(
              description="Filter by resource types (applies to policies only) - list of resource type names",
              examples=[["AWS.S3.Bucket", "AWS.EC2.SecurityGroup"]],
          ),
      ] = [],
      compliance_status: Annotated[
          str | None,
          Field(
              description="Filter by compliance status (applies to policies only) - 'PASS', 'FAIL', or 'ERROR'"
          ),
      ] = None,
      created_by: Annotated[
          str | None, Field(description="Filter by creator user ID or actor ID")
      ] = None,
      last_modified_by: Annotated[
          str | None, Field(description="Filter by last modifier user ID or actor ID")
      ] = None,
      output_ids: Annotated[
          list[str],
          Field(
              description="Client-side filter by destination output IDs. Filters results after fetching from API to include only detections with at least one matching outputID.",
              examples=[["destination-id-123"], ["prod-slack", "prod-pagerduty"]],
          ),
      ] = [],
  ) -> dict[str, Any]:
      """List detections from your Panther instance with support for multiple detection types and filtering.
  
      Note: The output_ids filter is applied client-side after fetching all results from the API,
      as the Panther REST API does not support server-side filtering by outputID. For more efficient
      API-level filtering, consider using the 'tag' parameter if your detections are tagged by environment.
      """
      # Validate detection types
      validation_error = validate_detection_types(detection_types)
      if validation_error:
          return validation_error
  
      logger.info(f"Fetching {limit} detections per type for types: {detection_types}")
  
      # For multiple detection types, cursor pagination is not supported
      if len(detection_types) > 1 and cursor:
          return {
              "success": False,
              "message": "Cursor pagination is not supported when querying multiple detection types. Please query one type at a time for pagination.",
          }
  
      # Validate filtering parameters
      if state and state not in VALID_STATES:
          return {
              "success": False,
              "message": f"Invalid state value. Must be one of: {', '.join(VALID_STATES)}",
          }
  
      if severity:
          invalid_severities = [s for s in severity if s not in VALID_SEVERITIES]
          if invalid_severities:
              return {
                  "success": False,
                  "message": f"Invalid severity values: {invalid_severities}. Valid values are: {', '.join(VALID_SEVERITIES)}",
              }
  
      if compliance_status and compliance_status not in VALID_COMPLIANCE_STATUSES:
          return {
              "success": False,
              "message": f"Invalid compliance_status value. Must be one of: {', '.join(VALID_COMPLIANCE_STATUSES)}",
          }
  
      # Validate detection-type-specific parameters
      if log_type and not any(dt in ["rules", "simple_rules"] for dt in detection_types):
          return {
              "success": False,
              "message": "log_type parameter is only valid for 'rules' and 'simple_rules' detection types.",
          }
  
      if resource_type and "policies" not in detection_types:
          return {
              "success": False,
              "message": "resource_type parameter is only valid for 'policies' detection type.",
          }
  
      if compliance_status and "policies" not in detection_types:
          return {
              "success": False,
              "message": "compliance_status parameter is only valid for 'policies' detection type.",
          }
  
      # Use the centralized field mapping
      field_map = LIST_FIELD_MAP
  
      try:
          all_results = {}
          has_next_pages = {}
          next_cursors = {}
  
          async with get_rest_client() as client:
              for detection_type in detection_types:
                  # Build query parameters using helper function
                  params = build_detection_params(
                      limit,
                      cursor,
                      detection_types,
                      name_contains,
                      state,
                      severity,
                      tag,
                      created_by,
                      last_modified_by,
                      log_type,
                      resource_type,
                      compliance_status,
                      detection_type,
                  )
  
                  result, _ = await client.get(
                      get_endpoint_for_detection(detection_type), params=params
                  )
  
                  # Extract detections and pagination info
                  detections = result.get("results", [])
                  next_cursor = result.get("next")
  
                  # Store results for this detection type
                  all_results[detection_type] = detections
                  next_cursors[detection_type] = next_cursor
                  has_next_pages[detection_type] = bool(next_cursor)
  
          # Process results for each detection type
          response_data = {"success": True}
  
          for detection_type in detection_types:
              detections = all_results[detection_type]
  
              # Keep only specific fields for each detection to limit the amount of data returned
              if detection_type == "policies":
                  filtered_metadata = [
                      {
                          "id": item["id"],
                          "description": item.get("description"),
                          "displayName": item.get("displayName"),
                          "enabled": item.get("enabled", False),
                          "severity": item.get("severity"),
                          "resourceTypes": item.get("resourceTypes", []),
                          "tags": item.get("tags", []),
                          "reports": item.get("reports", {}),
                          "managed": item.get("managed", False),
                          "outputIDs": item.get("outputIDs", []),
                          "createdBy": item.get("createdBy"),
                          "createdAt": item.get("createdAt"),
                          "lastModified": item.get("lastModified"),
                      }
                      for item in detections
                  ]
              elif detection_type == "scheduled_rules":
                  filtered_metadata = [
                      {
                          "id": item["id"],
                          "description": item.get("description"),
                          "displayName": item.get("displayName"),
                          "enabled": item.get("enabled", False),
                          "severity": item.get("severity"),
                          "scheduledQueries": item.get("scheduledQueries", []),
                          "tags": item.get("tags", []),
                          "reports": item.get("reports", {}),
                          "managed": item.get("managed", False),
                          "outputIDs": item.get("outputIDs", []),
                          "threshold": item.get("threshold"),
                          "dedupPeriodMinutes": item.get("dedupPeriodMinutes"),
                          "createdBy": item.get("createdBy"),
                          "createdAt": item.get("createdAt"),
                          "lastModified": item.get("lastModified"),
                      }
                      for item in detections
                  ]
              else:  # rules and simple_rules
                  filtered_metadata = [
                      {
                          "id": item["id"],
                          "description": item.get("description"),
                          "displayName": item.get("displayName"),
                          "enabled": item.get("enabled"),
                          "severity": item.get("severity"),
                          "logTypes": item.get("logTypes"),
                          "tags": item.get("tags"),
                          "reports": item.get("reports", {}),
                          "managed": item.get("managed"),
                          "outputIDs": item.get("outputIDs", []),
                          "threshold": item.get("threshold"),
                          "dedupPeriodMinutes": item.get("dedupPeriodMinutes"),
                          "createdBy": item.get("createdBy"),
                          "createdAt": item.get("createdAt"),
                          "lastModified": item.get("lastModified"),
                      }
                      for item in detections
                  ]
  
              # Apply client-side output_ids filtering if requested
              if output_ids:
                  filtered_metadata = [
                      item
                      for item in filtered_metadata
                      if any(
                          output_id in item.get("outputIDs", [])
                          for output_id in output_ids
                      )
                  ]
                  logger.info(
                      f"Applied client-side output_ids filter for {detection_type}: {len(filtered_metadata)} results matched"
                  )
  
              # Add to response
              response_data[field_map[detection_type]] = filtered_metadata
              response_data[f"total_{field_map[detection_type]}"] = len(filtered_metadata)
  
              # Add pagination info (only for single detection type queries)
              if len(detection_types) == 1:
                  response_data["has_next_page"] = has_next_pages[detection_type]
                  response_data["next_cursor"] = next_cursors[detection_type]
              else:
                  response_data[f"{detection_type}_has_next_page"] = has_next_pages[
                      detection_type
                  ]
                  response_data[f"{detection_type}_next_cursor"] = next_cursors[
                      detection_type
                  ]
  
          # Add overall summary for multi-type queries
          if len(detection_types) > 1:
              total_detections = sum(len(all_results[dt]) for dt in detection_types)
              response_data["total_all_detections"] = total_detections
              response_data["detection_types_queried"] = detection_types
  
          logger.info(f"Successfully retrieved detections for types: {detection_types}")
          return response_data
      except Exception as e:
          logger.error(f"Failed to list detection types {detection_types}: {str(e)}")
          return {
              "success": False,
              "message": f"Failed to list detection types {detection_types}: {str(e)}",
          }

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

  Central point where register_all_tools(mcp) is called, which iterates over all @mcp_tool decorated functions (including list_detections) and registers them with the FastMCP server instance.

  # 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/tools/detections.py:127-212 (schema)

  Pydantic input schema defined via Annotated parameters with Field metadata for description, examples, validation (e.g., limit ge=1 le=1000), making it self-describing for MCP clients.

  async def list_detections(
      detection_types: Annotated[
          list[str],
          Field(
              description="One or more detection types - rules, scheduled_rules, simple_rules, or policies.",
              examples=[
                  ["rules", "simple_rules", "scheduled_rules"],
                  ["policies"],
              ],
          ),
      ] = ["rules"],
      cursor: Annotated[
          str | None,
          Field(
              description="Optional cursor for pagination from a previous query (only supported for single detection type)"
          ),
      ] = None,
      limit: Annotated[
          int,
          Field(
              description="Maximum number of results to return per detection type",
              default=100,
              ge=1,
              le=1000,
          ),
      ] = 100,
      name_contains: Annotated[
          str | None, Field(description="Substring search by name (case-insensitive)")
      ] = None,
      state: Annotated[
          str,
          Field(
              description="Filter by state - 'enabled' or 'disabled'", default="enabled"
          ),
      ] = "",
      severity: Annotated[
          list[str],
          Field(
              description="Filter by severity levels - INFO, LOW, MEDIUM, HIGH, or CRITICAL.",
              examples=[
                  ["MEDIUM", "HIGH", "CRITICAL"],
                  ["INFO", "LOW"],
              ],
          ),
      ] = [],
      tag: Annotated[
          list[str],
          Field(
              description="A case-insensitive list of tags to filter by.",
              examples=[["Initial Access", "Persistence"]],
          ),
      ] = [],
      log_type: Annotated[
          list[str],
          Field(
              description="A list of log types to filter by (applies to rules and simple-rules only).",
              examples=[["AWS.CloudTrail", "GCP.AuditLog"]],
          ),
      ] = [],
      resource_type: Annotated[
          list[str],
          Field(
              description="Filter by resource types (applies to policies only) - list of resource type names",
              examples=[["AWS.S3.Bucket", "AWS.EC2.SecurityGroup"]],
          ),
      ] = [],
      compliance_status: Annotated[
          str | None,
          Field(
              description="Filter by compliance status (applies to policies only) - 'PASS', 'FAIL', or 'ERROR'"
          ),
      ] = None,
      created_by: Annotated[
          str | None, Field(description="Filter by creator user ID or actor ID")
      ] = None,
      last_modified_by: Annotated[
          str | None, Field(description="Filter by last modifier user ID or actor ID")
      ] = None,
      output_ids: Annotated[
          list[str],
          Field(
              description="Client-side filter by destination output IDs. Filters results after fetching from API to include only detections with at least one matching outputID.",
              examples=[["destination-id-123"], ["prod-slack", "prod-pagerduty"]],
          ),
      ] = [],
  ) -> dict[str, Any]:

• src/mcp_panther/panther_mcp_core/tools/detections.py:46-62 (helper)

  Input validation helper for detection_types parameter used in list_detections.

  def validate_detection_types(detection_types: list[str]) -> dict[str, Any] | None:
      """Validate detection types and return error dict if invalid, None if valid."""
      if not detection_types:
          return {
              "success": False,
              "message": "At least one detection type must be specified.",
          }
  
      invalid_types = [dt for dt in detection_types if dt not in DETECTION_TYPES]
      if invalid_types:
          valid_types = ", ".join(DETECTION_TYPES.keys())
          return {
              "success": False,
              "message": f"Invalid detection_types {invalid_types}. Valid values are: {valid_types}",
          }
      return None

• src/mcp_panther/panther_mcp_core/tools/detections.py:72-118 (helper)

  Helper to build API request parameters, ensuring compatibility across detection types and validating param applicability.

  def build_detection_params(
      limit: int,
      cursor: str | None,
      detection_types: list[str],
      name_contains: str | None,
      state: str | None,
      severity: list[str] | None,
      tag: list[str] | None,
      created_by: str | None,
      last_modified_by: str | None,
      log_type: list[str] | None,
      resource_type: list[str] | None,
      compliance_status: str | None,
      detection_type: str,
  ) -> dict[str, Any]:
      """Build query parameters for detection API calls."""
      params = {"limit": limit}
  
      # Add cursor for single detection type queries only
      if cursor and len(detection_types) == 1:
          params["cursor"] = cursor
          logger.info(f"Using cursor for pagination: {cursor}")
  
      # Add common filtering parameters
      if name_contains:
          params["name-contains"] = name_contains
      if state:
          params["state"] = state
      if severity:
          params["severity"] = severity
      if tag:
          params["tag"] = tag
      if created_by:
          params["created-by"] = created_by
      if last_modified_by:
          params["last-modified-by"] = last_modified_by
  
      # Add detection-type-specific parameters
      if detection_type in ["rules", "simple_rules"] and log_type:
          params["log-type"] = log_type
      elif detection_type == "policies":
          if resource_type:
              params["resource-type"] = resource_type
          if compliance_status:
              params["compliance-status"] = compliance_status
  
      return params