fix: improve skills typing, path validation, and event tracing

This commit is contained in:
Greyson Lalonde
2026-03-06 00:07:23 -05:00
parent d84a7f32bd
commit 01c1bf4bcc
5 changed files with 35 additions and 11 deletions

View File

@@ -268,6 +268,8 @@ class Agent(BaseAgent):
if self.allow_code_execution: if self.allow_code_execution:
self._validate_docker_installation() self._validate_docker_installation()
self.set_skills()
return self return self
def _setup_agent_executor(self) -> None: def _setup_agent_executor(self) -> None:

View File

@@ -9,12 +9,18 @@ from crewai.skills.loader import (
format_skill_context, format_skill_context,
load_resources, load_resources,
) )
from crewai.skills.models import DisclosureLevel, Skill, SkillFrontmatter from crewai.skills.models import (
DisclosureLevel,
ResourceDirName,
Skill,
SkillFrontmatter,
)
from crewai.skills.parser import SkillParseError, parse_skill_md from crewai.skills.parser import SkillParseError, parse_skill_md
__all__ = [ __all__ = [
"DisclosureLevel", "DisclosureLevel",
"ResourceDirName",
"Skill", "Skill",
"SkillFrontmatter", "SkillFrontmatter",
"SkillParseError", "SkillParseError",

View File

@@ -7,7 +7,11 @@ for agent use, and format skill context for prompt injection.
from __future__ import annotations from __future__ import annotations
from pathlib import Path from pathlib import Path
from typing import Any from typing import TYPE_CHECKING
if TYPE_CHECKING:
from crewai.agents.agent_builder.base_agent import BaseAgent
from crewai.events.event_bus import crewai_event_bus from crewai.events.event_bus import crewai_event_bus
from crewai.events.types.skill_events import ( from crewai.events.types.skill_events import (
@@ -28,7 +32,7 @@ from crewai.skills.parser import (
def discover_skills( def discover_skills(
search_path: Path, search_path: Path,
source: Any | None = None, source: BaseAgent | None = None,
) -> list[Skill]: ) -> list[Skill]:
"""Scan a directory for skill directories containing SKILL.md. """Scan a directory for skill directories containing SKILL.md.
@@ -51,6 +55,7 @@ def discover_skills(
crewai_event_bus.emit( crewai_event_bus.emit(
source, source,
event=SkillDiscoveryStartedEvent( event=SkillDiscoveryStartedEvent(
from_agent=source,
search_path=search_path, search_path=search_path,
), ),
) )
@@ -68,6 +73,7 @@ def discover_skills(
crewai_event_bus.emit( crewai_event_bus.emit(
source, source,
event=SkillLoadedEvent( event=SkillLoadedEvent(
from_agent=source,
skill_name=skill.name, skill_name=skill.name,
skill_path=skill.path, skill_path=skill.path,
disclosure_level=skill.disclosure_level.value, disclosure_level=skill.disclosure_level.value,
@@ -78,6 +84,7 @@ def discover_skills(
crewai_event_bus.emit( crewai_event_bus.emit(
source, source,
event=SkillLoadFailedEvent( event=SkillLoadFailedEvent(
from_agent=source,
skill_name=child.name, skill_name=child.name,
skill_path=child, skill_path=child,
error=str(e), error=str(e),
@@ -88,6 +95,7 @@ def discover_skills(
crewai_event_bus.emit( crewai_event_bus.emit(
source, source,
event=SkillDiscoveryCompletedEvent( event=SkillDiscoveryCompletedEvent(
from_agent=source,
search_path=search_path, search_path=search_path,
skills_found=len(skills), skills_found=len(skills),
skill_names=[s.name for s in skills], skill_names=[s.name for s in skills],
@@ -99,7 +107,7 @@ def discover_skills(
def activate_skill( def activate_skill(
skill: Skill, skill: Skill,
source: Any | None = None, source: BaseAgent | None = None,
) -> Skill: ) -> Skill:
"""Promote a skill to INSTRUCTIONS disclosure level. """Promote a skill to INSTRUCTIONS disclosure level.
@@ -121,6 +129,7 @@ def activate_skill(
crewai_event_bus.emit( crewai_event_bus.emit(
source, source,
event=SkillActivatedEvent( event=SkillActivatedEvent(
from_agent=source,
skill_name=activated.name, skill_name=activated.name,
skill_path=activated.path, skill_path=activated.path,
disclosure_level=activated.disclosure_level.value, disclosure_level=activated.disclosure_level.value,

View File

@@ -8,7 +8,7 @@ from __future__ import annotations
from enum import IntEnum from enum import IntEnum
from pathlib import Path from pathlib import Path
from typing import Any from typing import Any, Literal
from pydantic import BaseModel, Field, model_validator from pydantic import BaseModel, Field, model_validator
@@ -20,6 +20,7 @@ from crewai.skills.validation import (
MAX_DESCRIPTION_LENGTH: int = 1024 MAX_DESCRIPTION_LENGTH: int = 1024
ResourceDirName = Literal["scripts", "references", "assets"]
class DisclosureLevel(IntEnum): class DisclosureLevel(IntEnum):
@@ -88,7 +89,7 @@ class Skill(BaseModel):
instructions: str | None = None instructions: str | None = None
path: Path path: Path
disclosure_level: DisclosureLevel = Field(default=DisclosureLevel.METADATA) disclosure_level: DisclosureLevel = Field(default=DisclosureLevel.METADATA)
resource_files: dict[str, list[str]] | None = None resource_files: dict[ResourceDirName, list[str]] | None = None
@property @property
def name(self) -> str: def name(self) -> str:
@@ -119,7 +120,7 @@ class Skill(BaseModel):
self, self,
level: DisclosureLevel, level: DisclosureLevel,
instructions: str | None = None, instructions: str | None = None,
resource_files: dict[str, list[str]] | None = None, resource_files: dict[ResourceDirName, list[str]] | None = None,
) -> Skill: ) -> Skill:
"""Create a new Skill at a different disclosure level. """Create a new Skill at a different disclosure level.

View File

@@ -12,7 +12,12 @@ from typing import Any
import yaml import yaml
from crewai.skills.models import DisclosureLevel, Skill, SkillFrontmatter from crewai.skills.models import (
DisclosureLevel,
ResourceDirName,
Skill,
SkillFrontmatter,
)
from crewai.skills.validation import validate_directory_name from crewai.skills.validation import validate_directory_name
@@ -145,12 +150,13 @@ def load_skill_resources(skill: Skill) -> Skill:
if skill.disclosure_level < DisclosureLevel.INSTRUCTIONS: if skill.disclosure_level < DisclosureLevel.INSTRUCTIONS:
skill = load_skill_instructions(skill) skill = load_skill_instructions(skill)
resource_files: dict[str, list[str]] = {} resource_dirs: list[tuple[ResourceDirName, Path]] = [
for dir_name, resource_dir in (
("scripts", skill.scripts_dir), ("scripts", skill.scripts_dir),
("references", skill.references_dir), ("references", skill.references_dir),
("assets", skill.assets_dir), ("assets", skill.assets_dir),
): ]
resource_files: dict[ResourceDirName, list[str]] = {}
for dir_name, resource_dir in resource_dirs:
if resource_dir.is_dir(): if resource_dir.is_dir():
resource_files[dir_name] = sorted( resource_files[dir_name] = sorted(
str(f.relative_to(resource_dir)) str(f.relative_to(resource_dir))