Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
38 changes: 35 additions & 3 deletions TestSuite/agentix_agent.py
Original file line number Diff line number Diff line change
@@ -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"
Expand All @@ -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)
38 changes: 38 additions & 0 deletions demisto_sdk/commands/content_graph/objects/agentix_agent.py
Original file line number Diff line number Diff line change
@@ -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):
Expand All @@ -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
21 changes: 20 additions & 1 deletion demisto_sdk/commands/content_graph/parsers/agentix_agent.py
Original file line number Diff line number Diff line change
Expand Up @@ -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):
Expand All @@ -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", []
)
Expand All @@ -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
<agent_folder_name>_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"):
Expand Down
18 changes: 18 additions & 0 deletions demisto_sdk/commands/content_graph/parsers/related_files.py
Original file line number Diff line number Diff line change
Expand Up @@ -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):
Expand Down Expand Up @@ -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: <agent_folder_name>_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
Expand Down
179 changes: 179 additions & 0 deletions demisto_sdk/commands/prepare_content/agentix_agent_unifier.py
Original file line number Diff line number Diff line change
@@ -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:
`<agent_folder_name>_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"<green>Created unified AgentixAgent yml: {path.name}</green>")
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: <agent_folder_name>_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 ""
Loading
Loading