diff --git a/.changelog/5188.yml b/.changelog/5188.yml new file mode 100644 index 0000000000..0fe176a836 --- /dev/null +++ b/.changelog/5188.yml @@ -0,0 +1,4 @@ +changes: + - description: Agentix Agents system instructions are now defined in a separate md file. + type: internal +pr_number: 5188 diff --git a/TestSuite/agentix_agent.py b/TestSuite/agentix_agent.py index cc78e40806..504149a134 100644 --- a/TestSuite/agentix_agent.py +++ b/TestSuite/agentix_agent.py @@ -1,35 +1,55 @@ from pathlib import Path from typing import Optional +from TestSuite.file import File from TestSuite.yml import YAML, yaml class AgentixAgent(YAML): def __init__(self, tmpdir: Path, name: str, repo): + # Create directory for the agent + self._tmpdir_agent_path = tmpdir / name + self._tmpdir_agent_path.mkdir(exist_ok=True) + # Save entities self.name = name self._repo = repo self.repo_path = repo.path - self.path = str(tmpdir) - super().__init__(tmp_path=tmpdir / f"{self.name}.yml", repo_path=str(repo.path)) + self.path = str(self._tmpdir_agent_path) + + super().__init__( + tmp_path=self._tmpdir_agent_path / f"{self.name}.yml", + repo_path=str(repo.path), + ) + + # Add system instructions file + self.system_instructions = File( + self._tmpdir_agent_path / f"{self.name}_systeminstructions.md", + self._repo.path, + ) def build( self, yml: Optional[dict] = None, + system_instructions: Optional[str] = None, ): """Writes not None objects to files.""" if yml is not None: self.write_dict(yml) + if system_instructions is not None: + self.system_instructions.write(system_instructions) def create_default_agentix_agent( self, name: str = "sample_agentix_agent", agent_id: str = "sample_agentix_agent_id", + system_instructions: str = "", ): """Creates a new agentix agent with basic data. Args: name: The name of the new agentix agent, default is "sample_agentix_agent". agent_id: The ID of the new agentix agent, default is "sample_agentix_agent_id". + system_instructions: The system instructions content, default is empty string. """ default_agentix_agent_dir = ( Path(__file__).parent / "assets" / "default_agentix_agent" @@ -38,4 +58,16 @@ def create_default_agentix_agent( yml = yaml.load(yml_file) yml["id"] = agent_id yml["name"] = name - self.build(yml=yml) + self.build(yml=yml, system_instructions=system_instructions) + + def update(self, update_obj: dict, key_dict_to_update: Optional[str] = None): + """Update the YAML content, handling systeminstructions specially.""" + # Extract systeminstructions if present and write to file + if "systeminstructions" in update_obj: + system_instructions = update_obj.pop("systeminstructions") + if system_instructions: + self.system_instructions.write(system_instructions) + + # Call parent update for remaining fields + if update_obj: + super().update(update_obj, key_dict_to_update) diff --git a/demisto_sdk/commands/content_graph/objects/agentix_agent.py b/demisto_sdk/commands/content_graph/objects/agentix_agent.py index 6c1b35d275..6252cb2742 100644 --- a/demisto_sdk/commands/content_graph/objects/agentix_agent.py +++ b/demisto_sdk/commands/content_graph/objects/agentix_agent.py @@ -1,7 +1,15 @@ +from functools import cached_property from pathlib import Path +from demisto_sdk.commands.common.constants import MarketplaceVersions from demisto_sdk.commands.content_graph.common import ContentType from demisto_sdk.commands.content_graph.objects.agentix_base import AgentixBase +from demisto_sdk.commands.content_graph.parsers.related_files import ( + SystemInstructionsRelatedFile, +) +from demisto_sdk.commands.prepare_content.agentix_agent_unifier import ( + AgentixAgentUnifier, +) class AgentixAgent(AgentixBase, content_type=ContentType.AGENTIX_AGENT): @@ -20,3 +28,33 @@ def match(_dict: dict, path: Path) -> bool: if "color" in _dict and path.suffix == ".yml": return True return False + + @cached_property + def system_instructions_file(self) -> SystemInstructionsRelatedFile: + """Get the system instructions related file.""" + return SystemInstructionsRelatedFile(self.path, git_sha=self.git_sha) + + def prepare_for_upload( + self, + current_marketplace: MarketplaceVersions = MarketplaceVersions.PLATFORM, + **kwargs, + ) -> dict: + """ + Prepare the AgentixAgent for upload by unifying system instructions. + + This method merges the system instructions from the separate file + into the YAML data during content creation. + + Args: + current_marketplace: Target marketplace (default: PLATFORM) + **kwargs: Additional arguments + + Returns: + Unified YAML dict with systeminstructions field populated from file + """ + if not kwargs.get("unify_only"): + data = super().prepare_for_upload(current_marketplace) + else: + data = self.data + data = AgentixAgentUnifier.unify(self.path, data, current_marketplace) + return data diff --git a/demisto_sdk/commands/content_graph/parsers/agentix_agent.py b/demisto_sdk/commands/content_graph/parsers/agentix_agent.py index baf49719d5..e7ed9ddbff 100644 --- a/demisto_sdk/commands/content_graph/parsers/agentix_agent.py +++ b/demisto_sdk/commands/content_graph/parsers/agentix_agent.py @@ -6,6 +6,9 @@ from demisto_sdk.commands.content_graph.common import ContentType from demisto_sdk.commands.content_graph.parsers.agentix_base import AgentixBaseParser from demisto_sdk.commands.content_graph.strict_objects.agentix_agent import AgentixAgent +from demisto_sdk.commands.prepare_content.agentix_agent_unifier import ( + AgentixAgentUnifier, +) class AgentixAgentParser(AgentixBaseParser, content_type=ContentType.AGENTIX_AGENT): @@ -22,7 +25,6 @@ def __init__( self.color: str = self.yml_data.get("color") # type: ignore self.visibility: str = self.yml_data.get("visibility") # type: ignore self.actionids: list[str] = self.yml_data.get("actionids", []) - self.systeminstructions: str = self.yml_data.get("systeminstructions", "") self.conversationstarters: list[str] = self.yml_data.get( "conversationstarters", [] ) @@ -34,6 +36,23 @@ def __init__( self.sharedwithroles: list[str] = self.yml_data.get("sharedwithroles", []) self.add_action_dependencies() + @property + def systeminstructions(self) -> str: + """Gets the agent system instructions. + + The system instructions are read from a separate file named + _systeminstructions.md in the agent's directory. + + Returns: + str: The agent system instructions. + """ + if not self.git_sha: + return AgentixAgentUnifier.get_system_instructions(self.path.parent) + else: + return AgentixAgentUnifier.get_system_instructions_with_sha( + self.path, self.git_sha + ) + def add_action_dependencies(self) -> None: """Collects the actions used in the agent as optional dependencies.""" if actions_ids := self.yml_data.get("actionids"): diff --git a/demisto_sdk/commands/content_graph/parsers/related_files.py b/demisto_sdk/commands/content_graph/parsers/related_files.py index 98151c5c33..04d9d9b696 100644 --- a/demisto_sdk/commands/content_graph/parsers/related_files.py +++ b/demisto_sdk/commands/content_graph/parsers/related_files.py @@ -40,6 +40,7 @@ class RelatedFileType(Enum): AUTHOR_IMAGE = "author_image_file" RELEASE_NOTE = "release_note" VERSION_CONFIG = "version_config" + SYSTEM_INSTRUCTIONS = "system_instructions" class RelatedFile(ABC): @@ -273,6 +274,23 @@ def get_optional_paths(self) -> List[Path]: ] +class SystemInstructionsRelatedFile(TextFiles): + """Related file for AgentixAgent system instructions.""" + + file_type = RelatedFileType.SYSTEM_INSTRUCTIONS + + def get_optional_paths(self) -> List[Path]: + """ + Get the path to the system instructions file. + + The file should be named: _systeminstructions.md + """ + return [ + self.main_file_path.parent + / f"{self.main_file_path.parts[-2]}_systeminstructions.md" + ] + + class ImageFiles(RelatedFile): def get_file_size(self): raise NotImplementedError diff --git a/demisto_sdk/commands/prepare_content/agentix_agent_unifier.py b/demisto_sdk/commands/prepare_content/agentix_agent_unifier.py new file mode 100644 index 0000000000..d207bc2809 --- /dev/null +++ b/demisto_sdk/commands/prepare_content/agentix_agent_unifier.py @@ -0,0 +1,179 @@ +import copy +from pathlib import Path +from typing import Optional + +from demisto_sdk.commands.common.constants import MarketplaceVersions +from demisto_sdk.commands.common.files import TextFile +from demisto_sdk.commands.common.logger import logger +from demisto_sdk.commands.prepare_content.unifier import Unifier + + +class AgentixAgentUnifier(Unifier): + """ + Unifier for AgentixAgent content items. + + This class handles merging system instructions from a separate file into the + agent's YAML during the content creation process. + + The system instructions file follows the naming convention: + `_systeminstructions.md` + + Directory structure: + AgentixAgents/AgentName/ + ├── AgentName.yml + └── AgentName_systeminstructions.md + """ + + # File suffix for system instructions + SYSTEM_INSTRUCTIONS_SUFFIX = "_systeminstructions.md" + + @staticmethod + def unify( + path: Path, + data: dict, + marketplace: Optional[MarketplaceVersions] = None, + **kwargs, + ) -> dict: + """ + Merges system instructions from a separate file into the YAML. + + Args: + path: Path to the agent YAML file + data: Parsed YAML data + marketplace: Target marketplace (unused for agents, kept for interface compatibility) + **kwargs: Additional arguments (unused) + + Returns: + Unified YAML dict with systeminstructions field populated from file + """ + logger.debug(f"Unifying AgentixAgent: {path}") + + package_path = path.parent + yml_unified = copy.deepcopy(data) + + # Find and insert system instructions + yml_unified = AgentixAgentUnifier.insert_system_instructions_to_yml( + package_path, yml_unified + ) + + logger.debug(f"Created unified AgentixAgent yml: {path.name}") + return yml_unified + + @staticmethod + def get_system_instructions_file(package_path: Path) -> Optional[Path]: + """ + Find the system instructions file in the package directory. + + The file should be named: _systeminstructions.md + + Args: + package_path: Path to the agent package directory + + Returns: + Path to the system instructions file if found, None otherwise + """ + # Get the folder name (which should match the agent name pattern) + folder_name = package_path.name + + # Build the expected system instructions file path + instructions_file = ( + package_path + / f"{folder_name}{AgentixAgentUnifier.SYSTEM_INSTRUCTIONS_SUFFIX}" + ) + + if instructions_file.exists(): + return instructions_file + + return None + + @staticmethod + def insert_system_instructions_to_yml( + package_path: Path, yml_unified: dict + ) -> dict: + """ + Read system instructions from file and add to YAML. + + Args: + package_path: Path to the agent package directory + yml_unified: The YAML dict to update + + Returns: + Updated YAML dict with systeminstructions field + """ + instructions_file = AgentixAgentUnifier.get_system_instructions_file( + package_path + ) + + if instructions_file: + try: + instructions_content = instructions_file.read_text(encoding="utf-8") + yml_unified["systeminstructions"] = instructions_content.strip() + logger.debug( + f"Inserted system instructions from '{instructions_file.name}'" + ) + except Exception as e: + logger.warning( + f"Failed to read system instructions file '{instructions_file}': {e}" + ) + else: + logger.debug(f"No system instructions file found in '{package_path}'") + + return yml_unified + + @staticmethod + def get_system_instructions(package_path: Path) -> str: + """ + Get system instructions content from the package directory. + + This method is used by the parser to read system instructions from the + separate file during content graph parsing. + + Args: + package_path: Path to the agent package directory + + Returns: + The system instructions content, or empty string if not found + """ + instructions_file = AgentixAgentUnifier.get_system_instructions_file( + package_path + ) + + if instructions_file: + try: + return instructions_file.read_text(encoding="utf-8").strip() + except Exception as e: + logger.warning( + f"Failed to read system instructions file '{instructions_file}': {e}" + ) + return "" + + @staticmethod + def get_system_instructions_with_sha(yml_path: Path, git_sha: str) -> str: + """ + Get system instructions content from a specific git commit. + + This method is used when comparing content between different versions + (e.g., for backward compatibility checks). + + Args: + yml_path: Path to the agent YAML file + git_sha: The git commit SHA to read from + + Returns: + The system instructions content, or empty string if not found + """ + # Build the expected system instructions file path + folder_name = yml_path.parent.name + instructions_file_path = str( + yml_path.parent + / f"{folder_name}{AgentixAgentUnifier.SYSTEM_INSTRUCTIONS_SUFFIX}" + ) + + try: + content = TextFile.read_from_git_path(instructions_file_path, tag=git_sha) + return content.strip() if content else "" + except Exception as e: + logger.debug( + f"Could not read system instructions from git sha {git_sha}: {e}" + ) + return "" diff --git a/demisto_sdk/commands/validate/sdk_validation_config.toml b/demisto_sdk/commands/validate/sdk_validation_config.toml index fbe1bb40d4..a49bc5c183 100644 --- a/demisto_sdk/commands/validate/sdk_validation_config.toml +++ b/demisto_sdk/commands/validate/sdk_validation_config.toml @@ -169,6 +169,7 @@ select = [ "AG107", "AG108", "AG109", + "AG110", "DS100", "DS101", "DS105", @@ -339,6 +340,7 @@ select = [ "AG107", "AG108", "AG109", + "AG110", "PA100", "PA101", "PA102", diff --git a/demisto_sdk/commands/validate/validators/AG_validators/AG110_is_system_instructions_file_exists.py b/demisto_sdk/commands/validate/validators/AG_validators/AG110_is_system_instructions_file_exists.py new file mode 100644 index 0000000000..77aad7af0e --- /dev/null +++ b/demisto_sdk/commands/validate/validators/AG_validators/AG110_is_system_instructions_file_exists.py @@ -0,0 +1,43 @@ +from __future__ import annotations + +from typing import Iterable, List + +from demisto_sdk.commands.content_graph.objects.agentix_agent import AgentixAgent +from demisto_sdk.commands.content_graph.parsers.related_files import RelatedFileType +from demisto_sdk.commands.validate.validators.base_validator import ( + BaseValidator, + ValidationResult, +) + +ContentTypes = AgentixAgent + + +class IsSystemInstructionsFileExistsValidator(BaseValidator[ContentTypes]): + error_code = "AG110" + description = "Checks if the AgentixAgent has a system instructions file." + error_message = ( + "The AgentixAgent '{0}' is missing a system instructions file. " + "Please create a file named '{1}_systeminstructions.md' in the agent's directory." + ) + related_field = "systeminstructions" + rationale = ( + "AgentixAgent system instructions should be stored in a separate file " + "for better maintainability and readability." + ) + related_file_type = [RelatedFileType.SYSTEM_INSTRUCTIONS] + + def obtain_invalid_content_items( + self, content_items: Iterable[ContentTypes] + ) -> List[ValidationResult]: + return [ + ValidationResult( + validator=self, + message=self.error_message.format( + content_item.display_name, + content_item.path.parent.name, + ), + content_object=content_item, + ) + for content_item in content_items + if not content_item.system_instructions_file.exist + ]