feat: agent skills

introduce the agent skills standard for packaging reusable instructions that agents can discover and activate at runtime.                                                             

- skills defined via SKILL.md with yaml frontmatter and markdown body
- three-level progressive disclosure: metadata, instructions, resources
- filesystem discovery with directory name validation                                                         
- skill lifecycle events (discovery, loaded, activated, failed)
- crew-level skills resolved once and shared across agents                                                    
- skill context injected into both task execution and standalone kickoff
This commit is contained in:
Greyson LaLonde
2026-03-24 07:03:35 -04:00
committed by GitHub
parent dd9ae02159
commit 555ee462a3
30 changed files with 1809 additions and 1 deletions

View File

@@ -42,6 +42,7 @@ dependencies = [
"mcp~=1.26.0",
"uv~=0.9.13",
"aiosqlite~=0.21.0",
"pyyaml~=6.0",
"lancedb>=0.29.2",
]

View File

@@ -3,6 +3,7 @@ from __future__ import annotations
import asyncio
from collections.abc import Callable, Coroutine, Sequence
import contextvars
from pathlib import Path
import shutil
import subprocess
import time
@@ -26,6 +27,7 @@ from typing_extensions import Self
from crewai.agent.planning_config import PlanningConfig
from crewai.agent.utils import (
ahandle_knowledge_retrieval,
append_skill_context,
apply_training_data,
build_task_prompt_with_schema,
format_task_with_context,
@@ -65,6 +67,8 @@ from crewai.mcp import MCPServerConfig
from crewai.mcp.tool_resolver import MCPToolResolver
from crewai.rag.embeddings.types import EmbedderConfig
from crewai.security.fingerprint import Fingerprint
from crewai.skills.loader import activate_skill, discover_skills
from crewai.skills.models import INSTRUCTIONS, Skill as SkillModel
from crewai.tools.agent_tools.agent_tools import AgentTools
from crewai.types.callback import SerializableCallable
from crewai.utilities.agent_utils import (
@@ -278,6 +282,8 @@ class Agent(BaseAgent):
if self.allow_code_execution:
self._validate_docker_installation()
self.set_skills()
# Handle backward compatibility: convert reasoning=True to planning_config
if self.reasoning and self.planning_config is None:
import warnings
@@ -321,6 +327,76 @@ class Agent(BaseAgent):
except (TypeError, ValueError) as e:
raise ValueError(f"Invalid Knowledge Configuration: {e!s}") from e
def set_skills(
self,
resolved_crew_skills: list[SkillModel] | None = None,
) -> None:
"""Resolve skill paths and activate skills to INSTRUCTIONS level.
Path entries trigger discovery and activation. Pre-loaded Skill objects
below INSTRUCTIONS level are activated. Crew-level skills are merged in
with event emission so observability is consistent regardless of origin.
Args:
resolved_crew_skills: Pre-resolved crew skills (already discovered
and activated). When provided, avoids redundant discovery per agent.
"""
from crewai.crew import Crew
from crewai.events.event_bus import crewai_event_bus
from crewai.events.types.skill_events import SkillActivatedEvent
if resolved_crew_skills is None:
crew_skills: list[Path | SkillModel] | None = (
self.crew.skills
if isinstance(self.crew, Crew) and isinstance(self.crew.skills, list)
else None
)
else:
crew_skills = list(resolved_crew_skills)
if not self.skills and not crew_skills:
return
needs_work = self.skills and any(
isinstance(s, Path)
or (isinstance(s, SkillModel) and s.disclosure_level < INSTRUCTIONS)
for s in self.skills
)
if not needs_work and not crew_skills:
return
seen: set[str] = set()
resolved: list[Path | SkillModel] = []
items: list[Path | SkillModel] = list(self.skills) if self.skills else []
if crew_skills:
items.extend(crew_skills)
for item in items:
if isinstance(item, Path):
discovered = discover_skills(item, source=self)
for skill in discovered:
if skill.name not in seen:
seen.add(skill.name)
resolved.append(activate_skill(skill, source=self))
elif isinstance(item, SkillModel):
if item.name not in seen:
seen.add(item.name)
activated = activate_skill(item, source=self)
if activated is item and item.disclosure_level >= INSTRUCTIONS:
crewai_event_bus.emit(
self,
event=SkillActivatedEvent(
from_agent=self,
skill_name=item.name,
skill_path=item.path,
disclosure_level=item.disclosure_level,
),
)
resolved.append(activated)
self.skills = resolved if resolved else None
def _is_any_available_memory(self) -> bool:
"""Check if unified memory is available (agent or crew)."""
if getattr(self, "memory", None):
@@ -442,6 +518,8 @@ class Agent(BaseAgent):
self.crew.query_knowledge if self.crew else lambda *a, **k: None,
)
task_prompt = append_skill_context(self, task_prompt)
prepare_tools(self, tools, task)
task_prompt = apply_training_data(self, task_prompt)
@@ -682,6 +760,8 @@ class Agent(BaseAgent):
self, task, task_prompt, knowledge_config
)
task_prompt = append_skill_context(self, task_prompt)
prepare_tools(self, tools, task)
task_prompt = apply_training_data(self, task_prompt)
@@ -1343,6 +1423,8 @@ class Agent(BaseAgent):
),
)
formatted_messages = append_skill_context(self, formatted_messages)
# Build the input dict for the executor
inputs: dict[str, Any] = {
"input": formatted_messages,

View File

@@ -210,6 +210,30 @@ def _combine_knowledge_context(agent: Agent) -> str:
return agent_ctx + separator + crew_ctx
def append_skill_context(agent: Agent, task_prompt: str) -> str:
"""Append activated skill context sections to the task prompt.
Args:
agent: The agent with optional skills.
task_prompt: The current task prompt.
Returns:
The task prompt with skill context appended.
"""
if not agent.skills:
return task_prompt
from crewai.skills.loader import format_skill_context
from crewai.skills.models import Skill
skill_sections = [
format_skill_context(s) for s in agent.skills if isinstance(s, Skill)
]
if skill_sections:
task_prompt += "\n\n" + "\n\n".join(skill_sections)
return task_prompt
def apply_training_data(agent: Agent, task_prompt: str) -> str:
"""Apply training data to the task prompt.

View File

@@ -3,6 +3,7 @@ from __future__ import annotations
from abc import ABC, abstractmethod
from copy import copy as shallow_copy
from hashlib import md5
from pathlib import Path
import re
from typing import Any, Final, Literal
import uuid
@@ -32,6 +33,7 @@ from crewai.memory.memory_scope import MemoryScope, MemorySlice
from crewai.memory.unified_memory import Memory
from crewai.rag.embeddings.types import EmbedderConfig
from crewai.security.security_config import SecurityConfig
from crewai.skills.models import Skill
from crewai.tools.base_tool import BaseTool, Tool
from crewai.types.callback import SerializableCallable
from crewai.utilities.config import process_config
@@ -217,6 +219,11 @@ class BaseAgent(BaseModel, ABC, metaclass=AgentMeta):
"If not set, falls back to crew memory."
),
)
skills: list[Path | Skill] | None = Field(
default=None,
description="Agent Skills. Accepts paths for discovery or pre-loaded Skill objects.",
min_length=1,
)
@model_validator(mode="before")
@classmethod
@@ -500,3 +507,6 @@ class BaseAgent(BaseModel, ABC, metaclass=AgentMeta):
def set_knowledge(self, crew_embedder: EmbedderConfig | None = None) -> None:
pass
def set_skills(self, resolved_crew_skills: list[Any] | None = None) -> None:
pass

View File

@@ -6,6 +6,7 @@ from concurrent.futures import Future
from copy import copy as shallow_copy
from hashlib import md5
import json
from pathlib import Path
import re
from typing import (
TYPE_CHECKING,
@@ -91,6 +92,7 @@ from crewai.rag.embeddings.types import EmbedderConfig
from crewai.rag.types import SearchResult
from crewai.security.fingerprint import Fingerprint
from crewai.security.security_config import SecurityConfig
from crewai.skills.models import Skill
from crewai.task import Task
from crewai.tasks.conditional_task import ConditionalTask
from crewai.tasks.task_output import TaskOutput
@@ -294,6 +296,11 @@ class Crew(FlowTrackable, BaseModel):
default=None,
description="Knowledge for the crew.",
)
skills: list[Path | Skill] | None = Field(
default=None,
description="Skill search paths or pre-loaded Skill objects applied to all agents in the crew.",
)
security_config: SecurityConfig = Field(
default_factory=SecurityConfig,
description="Security configuration for the crew, including fingerprinting.",
@@ -376,7 +383,7 @@ class Crew(FlowTrackable, BaseModel):
if self.embedder is not None:
from crewai.rag.embeddings.factory import build_embedder
embedder = build_embedder(self.embedder) # type: ignore[arg-type]
embedder = build_embedder(cast(dict[str, Any], self.embedder))
self._memory = Memory(embedder=embedder, root_scope=crew_root_scope)
elif self.memory:
# User passed a Memory / MemoryScope / MemorySlice instance

View File

@@ -4,6 +4,7 @@ from __future__ import annotations
import asyncio
from collections.abc import Callable, Coroutine, Iterable, Mapping
from pathlib import Path
from typing import TYPE_CHECKING, Any
from opentelemetry import baggage
@@ -11,6 +12,8 @@ from opentelemetry import baggage
from crewai.agents.agent_builder.base_agent import BaseAgent
from crewai.crews.crew_output import CrewOutput
from crewai.rag.embeddings.types import EmbedderConfig
from crewai.skills.loader import activate_skill, discover_skills
from crewai.skills.models import INSTRUCTIONS, Skill as SkillModel
from crewai.types.streaming import CrewStreamingOutput, FlowStreamingOutput
from crewai.utilities.file_store import store_files
from crewai.utilities.streaming import (
@@ -51,6 +54,30 @@ def enable_agent_streaming(agents: Iterable[BaseAgent]) -> None:
agent.llm.stream = True
def _resolve_crew_skills(crew: Crew) -> list[SkillModel] | None:
"""Resolve crew-level skill paths once so agents don't repeat the work."""
if not isinstance(crew.skills, list) or not crew.skills:
return None
resolved: list[SkillModel] = []
seen: set[str] = set()
for item in crew.skills:
if isinstance(item, Path):
for skill in discover_skills(item):
if skill.name not in seen:
seen.add(skill.name)
resolved.append(activate_skill(skill))
elif isinstance(item, SkillModel):
if item.name not in seen:
seen.add(item.name)
resolved.append(
activate_skill(item)
if item.disclosure_level < INSTRUCTIONS
else item
)
return resolved
def setup_agents(
crew: Crew,
agents: Iterable[BaseAgent],
@@ -67,9 +94,12 @@ def setup_agents(
function_calling_llm: Default function calling LLM for agents.
step_callback: Default step callback for agents.
"""
resolved_crew_skills = _resolve_crew_skills(crew)
for agent in agents:
agent.crew = crew
agent.set_knowledge(crew_embedder=embedder)
agent.set_skills(resolved_crew_skills=resolved_crew_skills)
if not agent.function_calling_llm: # type: ignore[attr-defined]
agent.function_calling_llm = function_calling_llm # type: ignore[attr-defined]
if not agent.step_callback: # type: ignore[attr-defined]

View File

@@ -88,6 +88,14 @@ from crewai.events.types.reasoning_events import (
AgentReasoningStartedEvent,
ReasoningEvent,
)
from crewai.events.types.skill_events import (
SkillActivatedEvent,
SkillDiscoveryCompletedEvent,
SkillDiscoveryStartedEvent,
SkillEvent,
SkillLoadFailedEvent,
SkillLoadedEvent,
)
from crewai.events.types.task_events import (
TaskCompletedEvent,
TaskEvaluationEvent,
@@ -186,6 +194,12 @@ __all__ = [
"MethodExecutionFinishedEvent",
"MethodExecutionStartedEvent",
"ReasoningEvent",
"SkillActivatedEvent",
"SkillDiscoveryCompletedEvent",
"SkillDiscoveryStartedEvent",
"SkillEvent",
"SkillLoadFailedEvent",
"SkillLoadedEvent",
"TaskCompletedEvent",
"TaskEvaluationEvent",
"TaskFailedEvent",

View File

@@ -0,0 +1,62 @@
"""Skill lifecycle events for the Agent Skills standard.
Events emitted during skill discovery, loading, and activation.
"""
from __future__ import annotations
from pathlib import Path
from typing import Any
from crewai.events.base_events import BaseEvent
class SkillEvent(BaseEvent):
"""Base event for skill operations."""
skill_name: str = ""
skill_path: Path | None = None
from_agent: Any | None = None
from_task: Any | None = None
def __init__(self, **data: Any) -> None:
super().__init__(**data)
self._set_agent_params(data)
self._set_task_params(data)
class SkillDiscoveryStartedEvent(SkillEvent):
"""Event emitted when skill discovery begins."""
type: str = "skill_discovery_started"
search_path: Path
class SkillDiscoveryCompletedEvent(SkillEvent):
"""Event emitted when skill discovery completes."""
type: str = "skill_discovery_completed"
search_path: Path
skills_found: int
skill_names: list[str]
class SkillLoadedEvent(SkillEvent):
"""Event emitted when a skill is loaded at metadata level."""
type: str = "skill_loaded"
disclosure_level: int = 1
class SkillActivatedEvent(SkillEvent):
"""Event emitted when a skill is activated (promoted to instructions level)."""
type: str = "skill_activated"
disclosure_level: int = 2
class SkillLoadFailedEvent(SkillEvent):
"""Event emitted when skill loading fails."""
type: str = "skill_load_failed"
error: str

View File

@@ -0,0 +1,17 @@
"""Agent Skills standard implementation for crewAI.
Provides filesystem-based skill packaging with progressive disclosure.
"""
from crewai.skills.loader import activate_skill, discover_skills
from crewai.skills.models import Skill, SkillFrontmatter
from crewai.skills.parser import SkillParseError
__all__ = [
"Skill",
"SkillFrontmatter",
"SkillParseError",
"activate_skill",
"discover_skills",
]

View File

@@ -0,0 +1,184 @@
"""Filesystem discovery and progressive loading for Agent Skills.
Provides functions to discover skills in directories, activate them
for agent use, and format skill context for prompt injection.
"""
from __future__ import annotations
import logging
from pathlib import Path
from typing import TYPE_CHECKING
from crewai.events.event_bus import crewai_event_bus
from crewai.events.types.skill_events import (
SkillActivatedEvent,
SkillDiscoveryCompletedEvent,
SkillDiscoveryStartedEvent,
SkillLoadFailedEvent,
SkillLoadedEvent,
)
from crewai.skills.models import INSTRUCTIONS, RESOURCES, Skill
from crewai.skills.parser import (
SKILL_FILENAME,
load_skill_instructions,
load_skill_metadata,
load_skill_resources,
)
if TYPE_CHECKING:
from crewai.agents.agent_builder.base_agent import BaseAgent
_logger = logging.getLogger(__name__)
def discover_skills(
search_path: Path,
source: BaseAgent | None = None,
) -> list[Skill]:
"""Scan a directory for skill directories containing SKILL.md.
Loads each discovered skill at METADATA disclosure level.
Args:
search_path: Directory to scan for skill subdirectories.
source: Optional event source (agent or crew) for event emission.
Returns:
List of Skill instances at METADATA level.
"""
if not search_path.is_dir():
msg = f"Skill search path does not exist or is not a directory: {search_path}"
raise FileNotFoundError(msg)
skills: list[Skill] = []
if source is not None:
crewai_event_bus.emit(
source,
event=SkillDiscoveryStartedEvent(
from_agent=source,
search_path=search_path,
),
)
for child in sorted(search_path.iterdir()):
if not child.is_dir():
continue
skill_md = child / SKILL_FILENAME
if not skill_md.is_file():
continue
try:
skill = load_skill_metadata(child)
skills.append(skill)
if source is not None:
crewai_event_bus.emit(
source,
event=SkillLoadedEvent(
from_agent=source,
skill_name=skill.name,
skill_path=skill.path,
disclosure_level=skill.disclosure_level,
),
)
except Exception as e:
_logger.warning("Failed to load skill from %s: %s", child, e)
if source is not None:
crewai_event_bus.emit(
source,
event=SkillLoadFailedEvent(
from_agent=source,
skill_name=child.name,
skill_path=child,
error=str(e),
),
)
if source is not None:
crewai_event_bus.emit(
source,
event=SkillDiscoveryCompletedEvent(
from_agent=source,
search_path=search_path,
skills_found=len(skills),
skill_names=[s.name for s in skills],
),
)
return skills
def activate_skill(
skill: Skill,
source: BaseAgent | None = None,
) -> Skill:
"""Promote a skill to INSTRUCTIONS disclosure level.
Idempotent: returns the skill unchanged if already at or above INSTRUCTIONS.
Args:
skill: Skill to activate.
source: Optional event source for event emission.
Returns:
Skill at INSTRUCTIONS level or higher.
"""
if skill.disclosure_level >= INSTRUCTIONS:
return skill
activated = load_skill_instructions(skill)
if source is not None:
crewai_event_bus.emit(
source,
event=SkillActivatedEvent(
from_agent=source,
skill_name=activated.name,
skill_path=activated.path,
disclosure_level=activated.disclosure_level,
),
)
return activated
def load_resources(skill: Skill) -> Skill:
"""Promote a skill to RESOURCES disclosure level.
Args:
skill: Skill to promote.
Returns:
Skill at RESOURCES level.
"""
return load_skill_resources(skill)
def format_skill_context(skill: Skill) -> str:
"""Format skill information for agent prompt injection.
At METADATA level: returns name and description only.
At INSTRUCTIONS level or above: returns full SKILL.md body.
Args:
skill: The skill to format.
Returns:
Formatted skill context string.
"""
if skill.disclosure_level >= INSTRUCTIONS and skill.instructions:
parts = [
f"## Skill: {skill.name}",
skill.description,
"",
skill.instructions,
]
if skill.disclosure_level >= RESOURCES and skill.resource_files:
parts.append("")
parts.append("### Available Resources")
for dir_name, files in sorted(skill.resource_files.items()):
if files:
parts.append(f"- **{dir_name}/**: {', '.join(files)}")
return "\n".join(parts)
return f"## Skill: {skill.name}\n{skill.description}"

View File

@@ -0,0 +1,175 @@
"""Pydantic data models for the Agent Skills standard.
Defines DisclosureLevel, SkillFrontmatter, and Skill models for
progressive disclosure of skill information.
"""
from __future__ import annotations
from pathlib import Path
from typing import Annotated, Any, Final, Literal
from pydantic import BaseModel, ConfigDict, Field, model_validator
from crewai.skills.validation import (
MAX_SKILL_NAME_LENGTH,
MIN_SKILL_NAME_LENGTH,
SKILL_NAME_PATTERN,
)
MAX_DESCRIPTION_LENGTH: Final[int] = 1024
ResourceDirName = Literal["scripts", "references", "assets"]
DisclosureLevel = Annotated[
Literal[1, 2, 3], "Progressive disclosure levels for skill loading."
]
METADATA: Final[
Annotated[
DisclosureLevel, "Only frontmatter metadata is loaded (name, description)."
]
] = 1
INSTRUCTIONS: Final[Annotated[DisclosureLevel, "Full SKILL.md body is loaded."]] = 2
RESOURCES: Final[
Annotated[
DisclosureLevel,
"Resource directories (scripts, references, assets) are cataloged.",
]
] = 3
class SkillFrontmatter(BaseModel):
"""YAML frontmatter from a SKILL.md file.
Attributes:
name: Unique skill identifier (1-64 chars, lowercase alphanumeric + hyphens).
description: Human-readable description (1-1024 chars).
license: Optional license name or reference.
compatibility: Optional compatibility information (max 500 chars).
metadata: Optional additional metadata as string key-value pairs.
allowed_tools: Optional space-delimited list of pre-approved tools.
"""
model_config = ConfigDict(frozen=True, populate_by_name=True)
name: str = Field(
min_length=MIN_SKILL_NAME_LENGTH,
max_length=MAX_SKILL_NAME_LENGTH,
pattern=SKILL_NAME_PATTERN,
)
description: str = Field(min_length=1, max_length=MAX_DESCRIPTION_LENGTH)
license: str | None = Field(
default=None,
description="SPDX license identifier or free-text license reference, e.g. 'MIT', 'Apache-2.0'.",
)
compatibility: str | None = Field(
default=None,
max_length=500,
description="Version or platform constraints for the skill, e.g. 'crewai >= 0.80'.",
)
metadata: dict[str, str] | None = Field(
default=None,
description="Arbitrary string key-value pairs for custom skill metadata.",
)
allowed_tools: list[str] | None = Field(
default=None,
alias="allowed-tools",
description="Pre-approved tool names the skill may use, parsed from a space-delimited string in frontmatter.",
)
@model_validator(mode="before")
@classmethod
def parse_allowed_tools(cls, values: dict[str, Any]) -> dict[str, Any]:
"""Parse space-delimited allowed-tools string into a list."""
key = "allowed-tools"
alt_key = "allowed_tools"
raw = values.get(key) or values.get(alt_key)
if isinstance(raw, str):
values[key] = raw.split()
return values
class Skill(BaseModel):
"""A loaded Agent Skill with progressive disclosure support.
Attributes:
frontmatter: Parsed YAML frontmatter.
instructions: Full SKILL.md body text (populated at INSTRUCTIONS level).
path: Filesystem path to the skill directory.
disclosure_level: Current disclosure level of the skill.
resource_files: Cataloged resource files (populated at RESOURCES level).
"""
frontmatter: SkillFrontmatter = Field(
description="Parsed YAML frontmatter from SKILL.md.",
)
instructions: str | None = Field(
default=None,
description="Full SKILL.md body text, populated at INSTRUCTIONS level.",
)
path: Path = Field(
description="Filesystem path to the skill directory.",
)
disclosure_level: DisclosureLevel = Field(
default=METADATA,
description="Current progressive disclosure level of the skill.",
)
resource_files: dict[ResourceDirName, list[str]] | None = Field(
default=None,
description="Cataloged resource files by directory, populated at RESOURCES level.",
)
@property
def name(self) -> str:
"""Skill name from frontmatter."""
return self.frontmatter.name
@property
def description(self) -> str:
"""Skill description from frontmatter."""
return self.frontmatter.description
@property
def scripts_dir(self) -> Path:
"""Path to the scripts directory."""
return self.path / "scripts"
@property
def references_dir(self) -> Path:
"""Path to the references directory."""
return self.path / "references"
@property
def assets_dir(self) -> Path:
"""Path to the assets directory."""
return self.path / "assets"
def with_disclosure_level(
self,
level: DisclosureLevel,
instructions: str | None = None,
resource_files: dict[ResourceDirName, list[str]] | None = None,
) -> Skill:
"""Create a new Skill at a different disclosure level.
Args:
level: The new disclosure level.
instructions: Optional instructions body text.
resource_files: Optional cataloged resource files.
Returns:
A new Skill instance at the specified disclosure level.
"""
return Skill(
frontmatter=self.frontmatter,
instructions=instructions
if instructions is not None
else self.instructions,
path=self.path,
disclosure_level=level,
resource_files=(
resource_files if resource_files is not None else self.resource_files
),
)

View File

@@ -0,0 +1,194 @@
"""SKILL.md file parsing for the Agent Skills standard.
Parses YAML frontmatter and markdown body from SKILL.md files,
and provides progressive loading functions for skill data.
"""
from __future__ import annotations
import logging
from pathlib import Path
import re
from typing import Any, Final
import yaml
from crewai.skills.models import (
INSTRUCTIONS,
METADATA,
RESOURCES,
ResourceDirName,
Skill,
SkillFrontmatter,
)
from crewai.skills.validation import validate_directory_name
_logger = logging.getLogger(__name__)
SKILL_FILENAME: Final[str] = "SKILL.md"
_CLOSING_DELIMITER: Final[re.Pattern[str]] = re.compile(r"\n---[ \t]*(?:\n|$)")
_MAX_BODY_CHARS: Final[int] = 50_000
class SkillParseError(ValueError):
"""Error raised when SKILL.md parsing fails."""
def parse_frontmatter(content: str) -> tuple[dict[str, Any], str]:
"""Split SKILL.md content into frontmatter dict and body text.
Args:
content: Raw SKILL.md file content.
Returns:
Tuple of (frontmatter dict, body text).
Raises:
SkillParseError: If frontmatter delimiters are missing or YAML is invalid.
"""
if not content.startswith("---"):
msg = "SKILL.md must start with '---' frontmatter delimiter"
raise SkillParseError(msg)
match = _CLOSING_DELIMITER.search(content, pos=3)
if match is None:
msg = "SKILL.md missing closing '---' frontmatter delimiter"
raise SkillParseError(msg)
yaml_content = content[3 : match.start()].strip()
body = content[match.end() :].strip()
try:
frontmatter = yaml.safe_load(yaml_content)
except yaml.YAMLError as e:
msg = f"Invalid YAML in frontmatter: {e}"
raise SkillParseError(msg) from e
if not isinstance(frontmatter, dict):
msg = "Frontmatter must be a YAML mapping"
raise SkillParseError(msg)
return frontmatter, body
def parse_skill_md(path: Path) -> tuple[SkillFrontmatter, str]:
"""Read and parse a SKILL.md file.
Args:
path: Path to the SKILL.md file.
Returns:
Tuple of (SkillFrontmatter, body text).
Raises:
FileNotFoundError: If the file does not exist.
SkillParseError: If parsing fails.
"""
content = path.read_text(encoding="utf-8")
frontmatter_dict, body = parse_frontmatter(content)
frontmatter = SkillFrontmatter(**frontmatter_dict)
return frontmatter, body
def load_skill_metadata(skill_dir: Path) -> Skill:
"""Load a skill at METADATA disclosure level.
Parses SKILL.md frontmatter only and validates directory name.
Args:
skill_dir: Path to the skill directory.
Returns:
Skill instance at METADATA level.
Raises:
FileNotFoundError: If SKILL.md is missing.
SkillParseError: If parsing fails.
ValueError: If directory name doesn't match skill name.
"""
skill_md_path = skill_dir / SKILL_FILENAME
frontmatter, body = parse_skill_md(skill_md_path)
validate_directory_name(skill_dir, frontmatter.name)
if len(body) > _MAX_BODY_CHARS:
_logger.warning(
"SKILL.md body for '%s' is %d chars (threshold: %d). "
"Large bodies may consume significant context window when injected into prompts.",
frontmatter.name,
len(body),
_MAX_BODY_CHARS,
)
return Skill(
frontmatter=frontmatter,
path=skill_dir,
disclosure_level=METADATA,
)
def load_skill_instructions(skill: Skill) -> Skill:
"""Promote a skill to INSTRUCTIONS disclosure level.
Reads the full SKILL.md body text.
Args:
skill: Skill at METADATA level.
Returns:
New Skill instance at INSTRUCTIONS level.
"""
if skill.disclosure_level >= INSTRUCTIONS:
return skill
skill_md_path = skill.path / SKILL_FILENAME
_, body = parse_skill_md(skill_md_path)
if len(body) > _MAX_BODY_CHARS:
_logger.warning(
"SKILL.md body for '%s' is %d chars (threshold: %d). "
"Large bodies may consume significant context window when injected into prompts.",
skill.name,
len(body),
_MAX_BODY_CHARS,
)
return skill.with_disclosure_level(
level=INSTRUCTIONS,
instructions=body,
)
def load_skill_resources(skill: Skill) -> Skill:
"""Promote a skill to RESOURCES disclosure level.
Catalogs available resource directories (scripts, references, assets).
Args:
skill: Skill at any level.
Returns:
New Skill instance at RESOURCES level.
"""
if skill.disclosure_level >= RESOURCES:
return skill
if skill.disclosure_level < INSTRUCTIONS:
skill = load_skill_instructions(skill)
resource_dirs: list[tuple[ResourceDirName, Path]] = [
("scripts", skill.scripts_dir),
("references", skill.references_dir),
("assets", skill.assets_dir),
]
resource_files: dict[ResourceDirName, list[str]] = {}
for dir_name, resource_dir in resource_dirs:
if resource_dir.is_dir():
resource_files[dir_name] = sorted(
str(f.relative_to(resource_dir))
for f in resource_dir.rglob("*")
if f.is_file()
)
return skill.with_disclosure_level(
level=RESOURCES,
instructions=skill.instructions,
resource_files=resource_files,
)

View File

@@ -0,0 +1,31 @@
"""Validation functions for Agent Skills specification constraints.
Validates skill names and directory structures per the Agent Skills standard.
"""
from __future__ import annotations
from pathlib import Path
import re
from typing import Final
MAX_SKILL_NAME_LENGTH: Final[int] = 64
MIN_SKILL_NAME_LENGTH: Final[int] = 1
SKILL_NAME_PATTERN: Final[re.Pattern[str]] = re.compile(r"^[a-z0-9]+(?:-[a-z0-9]+)*$")
def validate_directory_name(skill_dir: Path, skill_name: str) -> None:
"""Validate that a directory name matches the skill name.
Args:
skill_dir: Path to the skill directory.
skill_name: The declared skill name from frontmatter.
Raises:
ValueError: If the directory name does not match the skill name.
"""
dir_name = skill_dir.name
if dir_name != skill_name:
msg = f"Directory name '{dir_name}' does not match skill name '{skill_name}'"
raise ValueError(msg)

View File

View File

@@ -0,0 +1,4 @@
---
name: Invalid--Name
description: This skill has an invalid name.
---

View File

@@ -0,0 +1,4 @@
---
name: minimal-skill
description: A minimal skill with only required fields.
---

View File

@@ -0,0 +1,22 @@
---
name: valid-skill
description: A complete test skill with all optional directories.
license: Apache-2.0
compatibility: crewai>=0.1.0
metadata:
author: test
version: "1.0"
allowed-tools: web-search file-read
---
## Instructions
This skill provides comprehensive instructions for the agent.
### Usage
Follow these steps to use the skill effectively.
### Notes
Additional context for the agent.

View File

@@ -0,0 +1 @@
{"key": "value"}

View File

@@ -0,0 +1,3 @@
# Reference Guide
This is a reference document for the skill.

View File

@@ -0,0 +1,2 @@
#!/bin/bash
echo "setup"

View File

@@ -0,0 +1,78 @@
"""Integration tests for the skills system."""
from pathlib import Path
import pytest
from crewai.skills.loader import activate_skill, discover_skills, format_skill_context
from crewai.skills.models import INSTRUCTIONS, METADATA
def _create_skill_dir(parent: Path, name: str, body: str = "Body.") -> Path:
"""Helper to create a skill directory with SKILL.md."""
skill_dir = parent / name
skill_dir.mkdir()
(skill_dir / "SKILL.md").write_text(
f"---\nname: {name}\ndescription: Skill {name}\n---\n{body}"
)
return skill_dir
class TestSkillDiscoveryAndActivation:
"""End-to-end tests for discover + activate workflow."""
def test_discover_and_activate(self, tmp_path: Path) -> None:
_create_skill_dir(tmp_path, "my-skill", body="Use this skill.")
skills = discover_skills(tmp_path)
assert len(skills) == 1
assert skills[0].disclosure_level == METADATA
activated = activate_skill(skills[0])
assert activated.disclosure_level == INSTRUCTIONS
assert activated.instructions == "Use this skill."
context = format_skill_context(activated)
assert "## Skill: my-skill" in context
assert "Use this skill." in context
def test_filter_by_skill_names(self, tmp_path: Path) -> None:
_create_skill_dir(tmp_path, "alpha")
_create_skill_dir(tmp_path, "beta")
_create_skill_dir(tmp_path, "gamma")
all_skills = discover_skills(tmp_path)
wanted = {"alpha", "gamma"}
filtered = [s for s in all_skills if s.name in wanted]
assert {s.name for s in filtered} == {"alpha", "gamma"}
def test_full_fixture_skill(self) -> None:
fixtures = Path(__file__).parent / "fixtures"
valid_dir = fixtures / "valid-skill"
if not valid_dir.exists():
pytest.skip("Fixture not found")
skills = discover_skills(fixtures)
valid_skills = [s for s in skills if s.name == "valid-skill"]
assert len(valid_skills) == 1
skill = valid_skills[0]
assert skill.frontmatter.license == "Apache-2.0"
assert skill.frontmatter.allowed_tools == ["web-search", "file-read"]
activated = activate_skill(skill)
assert "Instructions" in (activated.instructions or "")
def test_multiple_search_paths(self, tmp_path: Path) -> None:
path_a = tmp_path / "a"
path_a.mkdir()
_create_skill_dir(path_a, "skill-a")
path_b = tmp_path / "b"
path_b.mkdir()
_create_skill_dir(path_b, "skill-b")
all_skills = []
for search_path in [path_a, path_b]:
all_skills.extend(discover_skills(search_path))
names = {s.name for s in all_skills}
assert names == {"skill-a", "skill-b"}

View File

@@ -0,0 +1,161 @@
"""Tests for skills/loader.py."""
from pathlib import Path
import pytest
from crewai.skills.loader import (
activate_skill,
discover_skills,
format_skill_context,
load_resources,
)
from crewai.skills.models import INSTRUCTIONS, METADATA, RESOURCES, Skill, SkillFrontmatter
from crewai.skills.parser import load_skill_metadata
def _create_skill_dir(parent: Path, name: str, body: str = "Body.") -> Path:
"""Helper to create a skill directory with SKILL.md."""
skill_dir = parent / name
skill_dir.mkdir()
(skill_dir / "SKILL.md").write_text(
f"---\nname: {name}\ndescription: Skill {name}\n---\n{body}"
)
return skill_dir
class TestDiscoverSkills:
"""Tests for discover_skills."""
def test_finds_valid_skills(self, tmp_path: Path) -> None:
_create_skill_dir(tmp_path, "alpha")
_create_skill_dir(tmp_path, "beta")
skills = discover_skills(tmp_path)
names = {s.name for s in skills}
assert names == {"alpha", "beta"}
def test_skips_dirs_without_skill_md(self, tmp_path: Path) -> None:
_create_skill_dir(tmp_path, "valid")
(tmp_path / "no-skill").mkdir()
skills = discover_skills(tmp_path)
assert len(skills) == 1
assert skills[0].name == "valid"
def test_skips_invalid_skills(self, tmp_path: Path) -> None:
_create_skill_dir(tmp_path, "good-skill")
bad_dir = tmp_path / "bad-skill"
bad_dir.mkdir()
(bad_dir / "SKILL.md").write_text(
"---\nname: Wrong-Name\ndescription: bad\n---\n"
)
skills = discover_skills(tmp_path)
assert len(skills) == 1
def test_empty_directory(self, tmp_path: Path) -> None:
skills = discover_skills(tmp_path)
assert skills == []
def test_nonexistent_path(self, tmp_path: Path) -> None:
with pytest.raises(FileNotFoundError):
discover_skills(tmp_path / "nonexistent")
def test_sorted_by_name(self, tmp_path: Path) -> None:
_create_skill_dir(tmp_path, "zebra")
_create_skill_dir(tmp_path, "alpha")
skills = discover_skills(tmp_path)
assert [s.name for s in skills] == ["alpha", "zebra"]
class TestActivateSkill:
"""Tests for activate_skill."""
def test_promotes_to_instructions(self, tmp_path: Path) -> None:
_create_skill_dir(tmp_path, "my-skill", body="Instructions.")
skill = load_skill_metadata(tmp_path / "my-skill")
activated = activate_skill(skill)
assert activated.disclosure_level == INSTRUCTIONS
assert activated.instructions == "Instructions."
def test_idempotent(self, tmp_path: Path) -> None:
_create_skill_dir(tmp_path, "my-skill")
skill = load_skill_metadata(tmp_path / "my-skill")
activated = activate_skill(skill)
again = activate_skill(activated)
assert again is activated
class TestLoadResources:
"""Tests for load_resources."""
def test_promotes_to_resources(self, tmp_path: Path) -> None:
skill_dir = _create_skill_dir(tmp_path, "my-skill")
(skill_dir / "scripts").mkdir()
(skill_dir / "scripts" / "run.sh").write_text("#!/bin/bash")
skill = load_skill_metadata(skill_dir)
full = load_resources(skill)
assert full.disclosure_level == RESOURCES
class TestFormatSkillContext:
"""Tests for format_skill_context."""
def test_metadata_level(self, tmp_path: Path) -> None:
fm = SkillFrontmatter(name="test-skill", description="A skill")
skill = Skill(
frontmatter=fm, path=tmp_path, disclosure_level=METADATA
)
ctx = format_skill_context(skill)
assert "## Skill: test-skill" in ctx
assert "A skill" in ctx
def test_instructions_level(self, tmp_path: Path) -> None:
fm = SkillFrontmatter(name="test-skill", description="A skill")
skill = Skill(
frontmatter=fm,
path=tmp_path,
disclosure_level=INSTRUCTIONS,
instructions="Do these things.",
)
ctx = format_skill_context(skill)
assert "## Skill: test-skill" in ctx
assert "Do these things." in ctx
def test_no_instructions_at_instructions_level(self, tmp_path: Path) -> None:
fm = SkillFrontmatter(name="test-skill", description="A skill")
skill = Skill(
frontmatter=fm,
path=tmp_path,
disclosure_level=INSTRUCTIONS,
instructions=None,
)
ctx = format_skill_context(skill)
assert ctx == "## Skill: test-skill\nA skill"
def test_resources_level(self, tmp_path: Path) -> None:
fm = SkillFrontmatter(name="test-skill", description="A skill")
skill = Skill(
frontmatter=fm,
path=tmp_path,
disclosure_level=RESOURCES,
instructions="Do things.",
resource_files={
"scripts": ["run.sh"],
"assets": ["data.json", "config.yaml"],
},
)
ctx = format_skill_context(skill)
assert "### Available Resources" in ctx
assert "**assets/**: data.json, config.yaml" in ctx
assert "**scripts/**: run.sh" in ctx
def test_resources_level_empty_files(self, tmp_path: Path) -> None:
fm = SkillFrontmatter(name="test-skill", description="A skill")
skill = Skill(
frontmatter=fm,
path=tmp_path,
disclosure_level=RESOURCES,
instructions="Do things.",
resource_files={},
)
ctx = format_skill_context(skill)
assert "### Available Resources" not in ctx

View File

@@ -0,0 +1,91 @@
"""Tests for skills/models.py."""
from pathlib import Path
import pytest
from crewai.skills.models import (
INSTRUCTIONS,
METADATA,
RESOURCES,
Skill,
SkillFrontmatter,
)
class TestDisclosureLevel:
"""Tests for DisclosureLevel constants."""
def test_ordering(self) -> None:
assert METADATA < INSTRUCTIONS
assert INSTRUCTIONS < RESOURCES
def test_values(self) -> None:
assert METADATA == 1
assert INSTRUCTIONS == 2
assert RESOURCES == 3
class TestSkillFrontmatter:
"""Tests for SkillFrontmatter model."""
def test_required_fields(self) -> None:
fm = SkillFrontmatter(name="my-skill", description="A test skill")
assert fm.name == "my-skill"
assert fm.description == "A test skill"
assert fm.license is None
assert fm.metadata is None
assert fm.allowed_tools is None
def test_all_fields(self) -> None:
fm = SkillFrontmatter(
name="web-search",
description="Search the web",
license="Apache-2.0",
compatibility="crewai>=0.1.0",
metadata={"author": "test"},
allowed_tools=["browser"],
)
assert fm.license == "Apache-2.0"
assert fm.metadata == {"author": "test"}
assert fm.allowed_tools == ["browser"]
def test_frozen(self) -> None:
fm = SkillFrontmatter(name="my-skill", description="desc")
with pytest.raises(Exception):
fm.name = "other" # type: ignore[misc]
def test_invalid_name_rejected(self) -> None:
with pytest.raises(ValueError):
SkillFrontmatter(name="Invalid--Name", description="bad")
class TestSkill:
"""Tests for Skill model."""
def test_properties(self, tmp_path: Path) -> None:
fm = SkillFrontmatter(name="test-skill", description="desc")
skill = Skill(frontmatter=fm, path=tmp_path / "test-skill")
assert skill.name == "test-skill"
assert skill.description == "desc"
assert skill.disclosure_level == METADATA
def test_resource_dirs(self, tmp_path: Path) -> None:
skill_dir = tmp_path / "test-skill"
skill_dir.mkdir()
fm = SkillFrontmatter(name="test-skill", description="desc")
skill = Skill(frontmatter=fm, path=skill_dir)
assert skill.scripts_dir == skill_dir / "scripts"
assert skill.references_dir == skill_dir / "references"
assert skill.assets_dir == skill_dir / "assets"
def test_with_disclosure_level(self, tmp_path: Path) -> None:
fm = SkillFrontmatter(name="test-skill", description="desc")
skill = Skill(frontmatter=fm, path=tmp_path)
promoted = skill.with_disclosure_level(
INSTRUCTIONS,
instructions="Do this.",
)
assert promoted.disclosure_level == INSTRUCTIONS
assert promoted.instructions == "Do this."
assert skill.disclosure_level == METADATA

View File

@@ -0,0 +1,167 @@
"""Tests for skills/parser.py."""
from pathlib import Path
import pytest
from crewai.skills.models import INSTRUCTIONS, METADATA, RESOURCES
from crewai.skills.parser import (
SkillParseError,
load_skill_instructions,
load_skill_metadata,
load_skill_resources,
parse_frontmatter,
parse_skill_md,
)
class TestParseFrontmatter:
"""Tests for parse_frontmatter."""
def test_valid_frontmatter_and_body(self) -> None:
content = "---\nname: test\ndescription: A test\n---\n\nBody text here."
fm, body = parse_frontmatter(content)
assert fm["name"] == "test"
assert fm["description"] == "A test"
assert body == "Body text here."
def test_empty_body(self) -> None:
content = "---\nname: test\ndescription: A test\n---"
fm, body = parse_frontmatter(content)
assert fm["name"] == "test"
assert body == ""
def test_missing_opening_delimiter(self) -> None:
with pytest.raises(SkillParseError, match="must start with"):
parse_frontmatter("name: test\n---\nBody")
def test_missing_closing_delimiter(self) -> None:
with pytest.raises(SkillParseError, match="missing closing"):
parse_frontmatter("---\nname: test\n")
def test_invalid_yaml(self) -> None:
with pytest.raises(SkillParseError, match="Invalid YAML"):
parse_frontmatter("---\n: :\n bad: [yaml\n---\nBody")
def test_triple_dash_in_body(self) -> None:
content = "---\nname: test\ndescription: desc\n---\n\nBody with --- inside."
fm, body = parse_frontmatter(content)
assert "---" in body
def test_inline_triple_dash_in_yaml_value(self) -> None:
content = '---\nname: test\ndescription: "Use---carefully"\n---\n\nBody.'
fm, body = parse_frontmatter(content)
assert fm["description"] == "Use---carefully"
assert body == "Body."
def test_unicode_content(self) -> None:
content = "---\nname: test\ndescription: Beschreibung\n---\n\nUnicode: \u00e4\u00f6\u00fc\u00df"
fm, body = parse_frontmatter(content)
assert fm["description"] == "Beschreibung"
assert "\u00e4\u00f6\u00fc\u00df" in body
def test_non_mapping_frontmatter(self) -> None:
with pytest.raises(SkillParseError, match="must be a YAML mapping"):
parse_frontmatter("---\n- item1\n- item2\n---\nBody")
class TestParseSkillMd:
"""Tests for parse_skill_md."""
def test_valid_file(self, tmp_path: Path) -> None:
skill_md = tmp_path / "SKILL.md"
skill_md.write_text(
"---\nname: my-skill\ndescription: desc\n---\nInstructions here."
)
fm, body = parse_skill_md(skill_md)
assert fm.name == "my-skill"
assert body == "Instructions here."
def test_file_not_found(self, tmp_path: Path) -> None:
with pytest.raises(FileNotFoundError):
parse_skill_md(tmp_path / "nonexistent" / "SKILL.md")
class TestLoadSkillMetadata:
"""Tests for load_skill_metadata."""
def test_valid_skill(self, tmp_path: Path) -> None:
skill_dir = tmp_path / "my-skill"
skill_dir.mkdir()
(skill_dir / "SKILL.md").write_text(
"---\nname: my-skill\ndescription: Test skill\n---\nBody"
)
skill = load_skill_metadata(skill_dir)
assert skill.name == "my-skill"
assert skill.disclosure_level == METADATA
assert skill.instructions is None
def test_directory_name_mismatch(self, tmp_path: Path) -> None:
skill_dir = tmp_path / "wrong-name"
skill_dir.mkdir()
(skill_dir / "SKILL.md").write_text(
"---\nname: my-skill\ndescription: Test skill\n---\n"
)
with pytest.raises(ValueError, match="does not match"):
load_skill_metadata(skill_dir)
class TestLoadSkillInstructions:
"""Tests for load_skill_instructions."""
def test_promotes_to_instructions(self, tmp_path: Path) -> None:
skill_dir = tmp_path / "my-skill"
skill_dir.mkdir()
(skill_dir / "SKILL.md").write_text(
"---\nname: my-skill\ndescription: Test\n---\nFull body."
)
skill = load_skill_metadata(skill_dir)
promoted = load_skill_instructions(skill)
assert promoted.disclosure_level == INSTRUCTIONS
assert promoted.instructions == "Full body."
def test_idempotent(self, tmp_path: Path) -> None:
skill_dir = tmp_path / "my-skill"
skill_dir.mkdir()
(skill_dir / "SKILL.md").write_text(
"---\nname: my-skill\ndescription: Test\n---\nBody."
)
skill = load_skill_metadata(skill_dir)
promoted = load_skill_instructions(skill)
again = load_skill_instructions(promoted)
assert again is promoted
class TestLoadSkillResources:
"""Tests for load_skill_resources."""
def test_catalogs_resources(self, tmp_path: Path) -> None:
skill_dir = tmp_path / "my-skill"
skill_dir.mkdir()
(skill_dir / "SKILL.md").write_text(
"---\nname: my-skill\ndescription: Test\n---\nBody."
)
(skill_dir / "scripts").mkdir()
(skill_dir / "scripts" / "run.sh").write_text("#!/bin/bash")
(skill_dir / "assets").mkdir()
(skill_dir / "assets" / "data.json").write_text("{}")
skill = load_skill_metadata(skill_dir)
full = load_skill_resources(skill)
assert full.disclosure_level == RESOURCES
assert full.instructions == "Body."
assert full.resource_files is not None
assert "scripts" in full.resource_files
assert "run.sh" in full.resource_files["scripts"]
assert "assets" in full.resource_files
assert "data.json" in full.resource_files["assets"]
def test_no_resource_dirs(self, tmp_path: Path) -> None:
skill_dir = tmp_path / "my-skill"
skill_dir.mkdir()
(skill_dir / "SKILL.md").write_text(
"---\nname: my-skill\ndescription: Test\n---\nBody."
)
skill = load_skill_metadata(skill_dir)
full = load_skill_resources(skill)
assert full.resource_files == {}

View File

@@ -0,0 +1,93 @@
"""Tests for skills validation."""
from pathlib import Path
import pytest
from crewai.skills.models import SkillFrontmatter
from crewai.skills.validation import (
MAX_SKILL_NAME_LENGTH,
validate_directory_name,
)
def _make(name: str) -> SkillFrontmatter:
"""Create a SkillFrontmatter with the given name."""
return SkillFrontmatter(name=name, description="desc")
class TestSkillNameValidation:
"""Tests for skill name constraints via SkillFrontmatter."""
def test_simple_name(self) -> None:
assert _make("web-search").name == "web-search"
def test_single_word(self) -> None:
assert _make("search").name == "search"
def test_numeric(self) -> None:
assert _make("tool3").name == "tool3"
def test_all_digits(self) -> None:
assert _make("123").name == "123"
def test_single_char(self) -> None:
assert _make("a").name == "a"
def test_max_length(self) -> None:
name = "a" * MAX_SKILL_NAME_LENGTH
assert _make(name).name == name
def test_multi_hyphen_segments(self) -> None:
assert _make("my-cool-skill").name == "my-cool-skill"
def test_empty_raises(self) -> None:
with pytest.raises(ValueError):
_make("")
def test_too_long_raises(self) -> None:
with pytest.raises(ValueError):
_make("a" * (MAX_SKILL_NAME_LENGTH + 1))
def test_uppercase_raises(self) -> None:
with pytest.raises(ValueError):
_make("MySkill")
def test_leading_hyphen_raises(self) -> None:
with pytest.raises(ValueError):
_make("-skill")
def test_trailing_hyphen_raises(self) -> None:
with pytest.raises(ValueError):
_make("skill-")
def test_consecutive_hyphens_raises(self) -> None:
with pytest.raises(ValueError):
_make("my--skill")
def test_underscore_raises(self) -> None:
with pytest.raises(ValueError):
_make("my_skill")
def test_space_raises(self) -> None:
with pytest.raises(ValueError):
_make("my skill")
def test_special_chars_raises(self) -> None:
with pytest.raises(ValueError):
_make("skill@v1")
class TestValidateDirectoryName:
"""Tests for validate_directory_name."""
def test_matching_names(self, tmp_path: Path) -> None:
skill_dir = tmp_path / "my-skill"
skill_dir.mkdir()
validate_directory_name(skill_dir, "my-skill")
def test_mismatched_names(self, tmp_path: Path) -> None:
skill_dir = tmp_path / "other-name"
skill_dir.mkdir()
with pytest.raises(ValueError, match="does not match"):
validate_directory_name(skill_dir, "my-skill")