"""Skill executor for executing skills.""" import logging import re from dataclasses import dataclass from typing import Any from loguru import logger from agents.skills.loader import Skill, SkillsLoader logger = logging.getLogger(__name__) @dataclass class SkillContext: """Execution context for a skill.""" skill_id: str skill_name: str input_data: dict[str, Any] user_message: str class SkillExecutor: """Executes skills based on user input.""" def __init__(self, skills_loader: SkillsLoader): """Initialize skill executor. Args: skills_loader: SkillsLoader instance for loading skills """ self.loader = skills_loader self._skills_prompt_cache: dict[str, str] = {} async def find_matching_skills(self, user_message: str) -> list[Skill]: """Find skills that match the user message. Args: user_message: User's input message Returns: List of matching skills (currently returns all active skills) """ # Get all active skills skills = await self.loader.list_skills() active_skills = [s for s in skills if s.status == "active"] return active_skills async def execute_skill( self, skill_id: str, context: SkillContext, ) -> str | None: """Execute a skill with given context. Args: skill_id: ID of skill to execute context: Execution context Returns: Execution result as string, or None if failed """ skill = await self.loader.load_skill_with_content(skill_id) if not skill: logger.warning(f"Skill not found: {skill_id}") return None if skill.status != "active": logger.warning(f"Skill is not active: {skill_id}") return None # Extract prompt/instructions from skill content prompt = self._extract_skill_prompt(skill) # Replace placeholders in prompt with context prompt = self._inject_context(prompt, context) return prompt def _extract_skill_prompt(self, skill: Skill) -> str: """Extract main prompt/instructions from skill content. Args: skill: Skill object with content Returns: Extracted prompt """ content = skill.content # Skip frontmatter if present lines = content.split("\n") start_line = 0 if content.startswith("---"): for i in range(1, len(lines)): if lines[i].strip() == "---": start_line = i + 1 break # Join remaining content main_content = "\n".join(lines[start_line:]) # Remove markdown headers but keep content prompt = re.sub(r"^#+\s+", "", main_content, flags=re.MULTILINE) return prompt.strip() def _inject_context(self, prompt: str, context: SkillContext) -> str: """Inject context into prompt template. Args: prompt: Prompt template context: Execution context Returns: Prompt with context injected """ # Replace common placeholders replacements = { "{{user_message}}": context.user_message, "{{skill_name}}": context.skill_name, "{{input}}": str(context.input_data), } result = prompt for placeholder, value in replacements.items(): result = result.replace(placeholder, value) return result async def get_skill_system_prompt(self, skill_id: str) -> str | None: """Get system prompt for a skill to be used in LLM context. Args: skill_id: Skill ID Returns: System prompt for the skill, or None if not found """ # Check cache if skill_id in self._skills_prompt_cache: return self._skills_prompt_cache[skill_id] skill = await self.loader.load_skill_with_content(skill_id) if not skill or skill.status != "active": return None # Extract prompt prompt = self._extract_skill_prompt(skill) # Cache it self._skills_prompt_cache[skill_id] = prompt return prompt def build_skills_context(self, skills: list[Skill]) -> str: """Build context string from multiple skills. Args: skills: List of skills Returns: Combined context string """ if not skills: return "" context_parts = ["## Available Skills\n"] for skill in skills: context_parts.append(f"### {skill.name}") context_parts.append(f"{skill.description}\n") return "\n".join(context_parts) def clear_cache(self) -> None: """Clear prompt cache.""" self._skills_prompt_cache.clear()