Support inline skill definitions (#6396)

* Support inline skill definitions

This commit adds inline skill loading without a need for a file. It also
DRYs the skill loading feature.

* Address code review suggestions

lib/crewai/src/crewai/agent/core.py

@@ -73,7 +73,6 @@ from crewai.events.types.memory_events import (
     MemoryRetrievalFailedEvent,
     MemoryRetrievalStartedEvent,
 )
-from crewai.events.types.skill_events import SkillActivatedEvent
 from crewai.experimental.agent_executor import AgentExecutor
 from crewai.knowledge.knowledge import Knowledge
 from crewai.knowledge.source.base_knowledge_source import BaseKnowledgeSource
@@ -82,8 +81,8 @@ from crewai.llms.base_llm import BaseLLM
 from crewai.mcp.config import MCPServerConfig
 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.skills.loader import load_skills
+from crewai.skills.models import Skill as SkillModel
 from crewai.state.checkpoint_config import CheckpointConfig, apply_checkpoint
 from crewai.tools.agent_tools.agent_tools import AgentTools
 from crewai.types.callback import SerializableCallable
@@ -429,13 +428,12 @@ class Agent(BaseAgent):
         self,
         resolved_crew_skills: list[SkillModel] | None = None,
     ) -> None:
-        """Resolve skill paths while preserving explicit disclosure levels.
+        """Load configured skills while preserving explicit disclosure levels.
 
-        Path entries trigger discovery and activation because directory-based
-        skills opt into eager loading. Pre-loaded Skill objects keep their
-        current disclosure level so callers can attach METADATA-only skills and
-        progressively activate them later. Crew-level skills are merged in with
-        event emission so observability is consistent regardless of origin.
+        Path strings, Path objects, inline SKILL.md strings, and registry refs
+        are loaded through the shared skill loader. Pre-loaded Skill objects
+        keep their current disclosure level so callers can attach METADATA-only
+        skills and progressively activate them later.
 
         Args:
             resolved_crew_skills: Pre-resolved crew skills. When provided,
@@ -444,7 +442,7 @@ class Agent(BaseAgent):
         from crewai.crew import Crew
 
         if resolved_crew_skills is None:
-            crew_skills: list[Path | SkillModel | str] | None = (
+            crew_skills = (
                 self.crew.skills
                 if isinstance(self.crew, Crew) and isinstance(self.crew.skills, list)
                 else None
@@ -455,58 +453,14 @@ class Agent(BaseAgent):
         if not self.skills and not crew_skills:
             return
 
-        needs_work = self.skills and any(
-            isinstance(s, (Path, str))
-            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 | str] = []
-        items: list[Path | SkillModel | str] = list(self.skills) if self.skills else []
-
+        items = list(self.skills) if self.skills else []
         if crew_skills:
             items.extend(crew_skills)
 
-        for item in items:
-            if isinstance(item, str):
-                from crewai.experimental.skills.registry import (
-                    is_registry_ref,
-                    parse_registry_ref,
-                    resolve_registry_ref,
-                )
-
-                if is_registry_ref(item):
-                    skill = resolve_registry_ref(item, source=self)
-                    org, _ = parse_registry_ref(item)
-                    dedup_key = f"{org}/{skill.name}"
-                    if dedup_key not in seen:
-                        seen.add(dedup_key)
-                        resolved.append(skill)
-            elif 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)
-                    if 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(item)
-
-        self.skills = resolved if resolved else None
+        self.skills = cast(
+            list[Path | SkillModel | str] | None,
+            load_skills(items, source=self) or None,
+        )
 
     def _is_any_available_memory(self) -> bool:
         """Check if unified memory is available (agent or crew)."""

lib/crewai/src/crewai/agents/agent_builder/base_agent.py

@@ -388,7 +388,7 @@ class BaseAgent(BaseModel, ABC, metaclass=AgentMeta):
     )
     skills: list[Path | Skill | str] | None = Field(
         default=None,
-        description="Agent Skills. Accepts paths for discovery, pre-loaded Skill objects, or '@org/name' registry refs.",
+        description="Agent Skills. Accepts paths for discovery, inline SKILL.md strings, pre-loaded Skill objects, or '@org/name' registry refs.",
         min_length=1,
     )
     execution_context: ExecutionContext | None = Field(default=None)
@@ -494,20 +494,6 @@ class BaseAgent(BaseModel, ABC, metaclass=AgentMeta):
     def process_model_config(cls, values: Any) -> dict[str, Any]:
         return process_config(values, cls)
 
-    @field_validator("skills", mode="before")
-    @classmethod
-    def coerce_skill_strings(cls, skills: Any) -> Any:
-        """Coerce plain path strings to Path objects; keep @-prefixed refs as str."""
-        if not isinstance(skills, list):
-            return skills
-        result = []
-        for item in skills:
-            if isinstance(item, str) and not item.startswith("@"):
-                result.append(Path(item))
-            else:
-                result.append(item)
-        return result
-
     @field_validator("tools")
     @classmethod
     def validate_tools(cls, tools: list[Any]) -> list[BaseTool]:

lib/crewai/src/crewai/crew.py

@@ -361,7 +361,7 @@ class Crew(FlowTrackable, BaseModel):
     )
     skills: list[Path | Skill | str] | None = Field(
         default=None,
-        description="Skill search paths, pre-loaded Skill objects, or '@org/name' registry refs applied to all agents in the crew.",
+        description="Skill search paths, inline SKILL.md strings, pre-loaded Skill objects, or '@org/name' registry refs applied to all agents in the crew.",
     )
 
     security_config: SecurityConfig = Field(
@@ -574,20 +574,6 @@ class Crew(FlowTrackable, BaseModel):
         if max_seq > 0:
             set_emission_counter(max_seq)
 
-    @field_validator("skills", mode="before")
-    @classmethod
-    def coerce_skill_strings(cls, skills: Any) -> Any:
-        """Coerce plain path strings to Path objects; keep @-prefixed refs as str."""
-        if not isinstance(skills, list):
-            return skills
-        result = []
-        for item in skills:
-            if isinstance(item, str) and not item.startswith("@"):
-                result.append(Path(item))
-            else:
-                result.append(item)
-        return result
-
     @field_validator("id", mode="before")
     @classmethod
     def _deny_user_set_id(cls, v: UUID4 | None, info: Any) -> UUID4 | None:

lib/crewai/src/crewai/crews/utils.py

@@ -4,7 +4,6 @@ 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
@@ -13,7 +12,7 @@ from crewai.agents.agent_builder.base_agent import BaseAgent
 from crewai.crews.crew_output import CrewOutput
 from crewai.llms.base_llm import BaseLLM
 from crewai.rag.embeddings.types import EmbedderConfig
-from crewai.skills.loader import activate_skill, discover_skills
+from crewai.skills.loader import activate_skill, load_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
@@ -60,23 +59,13 @@ def _resolve_crew_skills(crew: Crew) -> list[SkillModel] | None:
     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
+    resolved = load_skills(crew.skills)
+    if not resolved:
+        return None
+    return [
+        activate_skill(skill) if skill.disclosure_level < INSTRUCTIONS else skill
+        for skill in resolved
+    ]
 
 
 def setup_agents(

lib/crewai/src/crewai/skills/__init__.py

@@ -3,7 +3,12 @@
 Provides filesystem-based skill packaging with progressive disclosure.
 """
 
-from crewai.skills.loader import activate_skill, discover_skills
+from crewai.skills.loader import (
+    activate_skill,
+    discover_skills,
+    load_skill,
+    load_skills,
+)
 from crewai.skills.models import Skill, SkillFrontmatter
 from crewai.skills.parser import SkillParseError
 
@@ -14,4 +19,6 @@ __all__ = [
     "SkillParseError",
     "activate_skill",
     "discover_skills",
+    "load_skill",
+    "load_skills",
 ]

lib/crewai/src/crewai/skills/loader.py

@@ -6,6 +6,7 @@ for agent use, and format skill context for prompt injection.
 
 from __future__ import annotations
 
+from collections.abc import Iterable
 import logging
 from pathlib import Path
 from typing import TYPE_CHECKING
@@ -18,12 +19,13 @@ from crewai.events.types.skill_events import (
     SkillLoadFailedEvent,
     SkillLoadedEvent,
 )
-from crewai.skills.models import INSTRUCTIONS, RESOURCES, Skill
+from crewai.skills.models import INSTRUCTIONS, RESOURCES, Skill, SkillFrontmatter
 from crewai.skills.parser import (
     SKILL_FILENAME,
     load_skill_instructions,
     load_skill_metadata,
     load_skill_resources,
+    parse_frontmatter,
 )
 
 
@@ -143,6 +145,72 @@ def activate_skill(
     return activated
 
 
+def load_skill(
+    skill: Path | Skill | str,
+    source: BaseAgent | None = None,
+) -> list[Skill]:
+    """Load one skill input into Skill objects.
+
+    Accepts a pre-loaded Skill object, skill search path, inline SKILL.md
+    string, or '@org/name' registry reference. Path inputs can expand to many
+    skills. Path and inline inputs are activated immediately; pre-loaded Skill
+    objects keep their disclosure level.
+    """
+    if isinstance(skill, Skill):
+        return [skill]
+    if isinstance(skill, Path):
+        return [
+            activate_skill(s, source=source)
+            for s in discover_skills(skill, source=source)
+        ]
+    if isinstance(skill, str) and skill.startswith("@"):
+        from crewai.experimental.skills.registry import resolve_registry_ref
+
+        return [resolve_registry_ref(skill, source=source)]
+    if isinstance(skill, str) and skill.lstrip().startswith("---\n"):
+        frontmatter_dict, body = parse_frontmatter(skill.strip())
+        return [
+            Skill(
+                frontmatter=SkillFrontmatter(**frontmatter_dict),
+                instructions=body,
+                path=Path("."),
+                disclosure_level=INSTRUCTIONS,
+            )
+        ]
+    if isinstance(skill, str):
+        return [
+            activate_skill(s, source=source)
+            for s in discover_skills(Path(skill), source=source)
+        ]
+
+    msg = f"Unsupported skill input: {skill!r}"
+    raise TypeError(msg)
+
+
+def load_skills(
+    skills: Iterable[Path | Skill | str],
+    source: BaseAgent | None = None,
+) -> list[Skill]:
+    """Load skill inputs into de-duplicated Skill objects.
+
+    Preserves first-seen order when multiple inputs resolve to the same skill
+    name. Registry refs are scoped by org so different orgs can publish skills
+    that share a frontmatter name.
+    """
+    loaded: dict[str, Skill] = {}
+    for skill_input in skills:
+        for skill in load_skill(skill_input, source=source):
+            dedup_key = skill.name
+            if isinstance(skill_input, str) and skill_input.startswith("@"):
+                from crewai.experimental.skills.registry import parse_registry_ref
+
+                org, _ = parse_registry_ref(skill_input)
+                dedup_key = f"{org}/{skill.name}"
+            if dedup_key not in loaded:
+                loaded[dedup_key] = skill
+    return list(loaded.values())
+
+
 def load_resources(skill: Skill) -> Skill:
     """Promote a skill to RESOURCES disclosure level.
 

lib/crewai/tests/skills/test_integration.py

@@ -4,8 +4,13 @@ from pathlib import Path
 
 import pytest
 
-from crewai import Agent
-from crewai.skills.loader import activate_skill, discover_skills, format_skill_context
+from crewai import Agent, Crew, Task
+from crewai.crews.utils import _resolve_crew_skills
+from crewai.skills.loader import (
+    activate_skill,
+    discover_skills,
+    format_skill_context,
+)
 from crewai.skills.models import INSTRUCTIONS, METADATA
 from crewai.utilities.prompts import Prompts
 
@@ -100,3 +105,104 @@ class TestSkillDiscoveryAndActivation:
         assert "Skill travel" in system
         # METADATA-level skills must not leak full instructions into the prompt
         assert "Use this skill for travel planning." not in system
+
+    def test_agent_accepts_inline_skill_string(self) -> None:
+        agent = Agent(
+            role="Reviewer",
+            goal="Review changes.",
+            backstory="An experienced reviewer.",
+            skills=[
+                "---\n"
+                "name: inline-review\n"
+                "description: Inline review guidance\n"
+                "---\n"
+                "Focus on behavior and missing tests."
+            ],
+        )
+
+        assert agent.skills is not None
+        assert [skill.name for skill in agent.skills] == ["inline-review"]
+        assert [skill.disclosure_level for skill in agent.skills] == [INSTRUCTIONS]
+        assert [skill.instructions for skill in agent.skills] == [
+            "Focus on behavior and missing tests."
+        ]
+
+        result = Prompts(agent=agent, has_tools=False, use_system_prompt=True).task_execution()
+        system = getattr(result, "system", "") or result.prompt
+        assert '<skill name="inline-review">' in system
+        assert "Focus on behavior and missing tests." in system
+
+    def test_agent_treats_plain_skill_string_as_path(self, tmp_path: Path) -> None:
+        _create_skill_dir(tmp_path, "path-skill", body="Use the path skill.")
+
+        agent = Agent(
+            role="Reviewer",
+            goal="Review changes.",
+            backstory="An experienced reviewer.",
+            skills=[str(tmp_path)],
+        )
+
+        assert agent.skills is not None
+        assert [skill.name for skill in agent.skills] == ["path-skill"]
+        assert [skill.instructions for skill in agent.skills] == ["Use the path skill."]
+
+    def test_crew_resolves_inline_skill_string(self) -> None:
+        agent = Agent(
+            role="Reviewer",
+            goal="Review changes.",
+            backstory="An experienced reviewer.",
+        )
+        task = Task(
+            description="Review the diff.",
+            expected_output="Findings.",
+            agent=agent,
+        )
+        crew = Crew(
+            agents=[agent],
+            tasks=[task],
+            skills=[
+                "---\n"
+                "name: crew-inline-review\n"
+                "description: Crew-level inline review guidance\n"
+                "---\n"
+                "Apply this to every agent."
+            ],
+        )
+
+        skills = _resolve_crew_skills(crew)
+
+        assert skills is not None
+        assert [skill.name for skill in skills] == ["crew-inline-review"]
+        assert [skill.instructions for skill in skills] == ["Apply this to every agent."]
+
+    def test_crew_activates_preloaded_metadata_skill(self, tmp_path: Path) -> None:
+        _create_skill_dir(
+            tmp_path,
+            "crew-preloaded",
+            body="Apply this crew-level guidance to every agent.",
+        )
+        metadata_skill = discover_skills(tmp_path)[0]
+        agent = Agent(
+            role="Reviewer",
+            goal="Review changes.",
+            backstory="An experienced reviewer.",
+        )
+        task = Task(
+            description="Review the diff.",
+            expected_output="Findings.",
+            agent=agent,
+        )
+        crew = Crew(
+            agents=[agent],
+            tasks=[task],
+            skills=[metadata_skill],
+        )
+
+        skills = _resolve_crew_skills(crew)
+
+        assert skills is not None
+        assert [skill.name for skill in skills] == ["crew-preloaded"]
+        assert [skill.disclosure_level for skill in skills] == [INSTRUCTIONS]
+        assert [skill.instructions for skill in skills] == [
+            "Apply this crew-level guidance to every agent."
+        ]

lib/crewai/tests/skills/test_loader.py

@@ -9,9 +9,11 @@ from crewai.skills.loader import (
     discover_skills,
     format_skill_context,
     load_resources,
+    load_skill,
+    load_skills,
 )
 from crewai.skills.models import INSTRUCTIONS, METADATA, RESOURCES, Skill, SkillFrontmatter
-from crewai.skills.parser import load_skill_metadata
+from crewai.skills.parser import SkillParseError, load_skill_metadata
 
 
 def _create_skill_dir(parent: Path, name: str, body: str = "Body.") -> Path:
@@ -84,6 +86,126 @@ class TestActivateSkill:
         assert again is activated
 
 
+class TestLoadSkill:
+    """Tests for load_skill."""
+
+    @pytest.mark.parametrize("as_string", [False, True])
+    def test_loads_path_input(self, tmp_path: Path, as_string: bool) -> None:
+        _create_skill_dir(tmp_path, "first-skill", body="First.")
+        _create_skill_dir(tmp_path, "second-skill", body="Second.")
+        path = str(tmp_path) if as_string else tmp_path
+
+        skills = load_skill(path)
+
+        assert [skill.name for skill in skills] == ["first-skill", "second-skill"]
+        assert [skill.disclosure_level for skill in skills] == [
+            INSTRUCTIONS,
+            INSTRUCTIONS,
+        ]
+        assert [skill.instructions for skill in skills] == ["First.", "Second."]
+
+    def test_loads_preloaded_skill(self, tmp_path: Path) -> None:
+        preloaded = Skill(
+            frontmatter=SkillFrontmatter(
+                name="preloaded-skill",
+                description="Preloaded skill",
+            ),
+            path=tmp_path / "preloaded-skill",
+        )
+
+        skills = load_skill(preloaded)
+
+        assert skills == [preloaded]
+
+    def test_loads_inline_skill(self) -> None:
+        inline_skill = (
+            "---\n"
+            "name: inline-skill\n"
+            "description: Inline guidance\n"
+            "---\n"
+            "Follow these instructions."
+        )
+
+        skills = load_skill(inline_skill)
+
+        assert [skill.name for skill in skills] == ["inline-skill"]
+        assert [skill.disclosure_level for skill in skills] == [INSTRUCTIONS]
+        assert [skill.instructions for skill in skills] == [
+            "Follow these instructions."
+        ]
+
+    def test_invalid_inline_skill_raises_parse_error(self) -> None:
+        with pytest.raises(SkillParseError, match="missing closing"):
+            load_skill("---\nname: inline-skill\n")
+
+    def test_missing_path_raises_file_not_found(self, tmp_path: Path) -> None:
+        with pytest.raises(FileNotFoundError):
+            load_skill(tmp_path / "missing")
+
+    def test_unsupported_input_raises_type_error(self) -> None:
+        with pytest.raises(TypeError, match="Unsupported skill input"):
+            load_skill(object())  # type: ignore[arg-type]
+
+    def test_load_skills_deduplicates_by_name(self, tmp_path: Path) -> None:
+        first = Skill(
+            frontmatter=SkillFrontmatter(
+                name="duplicate-skill",
+                description="First skill",
+            ),
+            path=tmp_path / "first",
+        )
+        second = Skill(
+            frontmatter=SkillFrontmatter(
+                name="duplicate-skill",
+                description="Second skill",
+            ),
+            path=tmp_path / "second",
+        )
+
+        skills = load_skills([first, second])
+
+        assert skills == [first]
+
+    def test_load_skills_keeps_registry_refs_from_different_orgs(
+        self,
+        monkeypatch: pytest.MonkeyPatch,
+        tmp_path: Path,
+    ) -> None:
+        first = Skill(
+            frontmatter=SkillFrontmatter(
+                name="shared-skill",
+                description="First registry skill",
+            ),
+            path=tmp_path / "first",
+            disclosure_level=INSTRUCTIONS,
+            instructions="First instructions.",
+        )
+        second = Skill(
+            frontmatter=SkillFrontmatter(
+                name="shared-skill",
+                description="Second registry skill",
+            ),
+            path=tmp_path / "second",
+            disclosure_level=INSTRUCTIONS,
+            instructions="Second instructions.",
+        )
+
+        def resolve_registry_ref(ref: str, source: object = None) -> Skill:
+            return {
+                "@first/shared-skill": first,
+                "@second/shared-skill": second,
+            }[ref]
+
+        monkeypatch.setattr(
+            "crewai.experimental.skills.registry.resolve_registry_ref",
+            resolve_registry_ref,
+        )
+
+        skills = load_skills(["@first/shared-skill", "@second/shared-skill"])
+
+        assert skills == [first, second]
+
+
 class TestLoadResources:
     """Tests for load_resources."""