diff --git a/lib/crewai/src/crewai/agent/core.py b/lib/crewai/src/crewai/agent/core.py index 4c20693a8..eb7d7834b 100644 --- a/lib/crewai/src/crewai/agent/core.py +++ b/lib/crewai/src/crewai/agent/core.py @@ -73,6 +73,7 @@ from crewai.events.types.memory_events import ( MemoryRetrievalFailedEvent, MemoryRetrievalStartedEvent, ) +from crewai.events.types.skill_events import SkillUsedEvent from crewai.experimental.agent_executor import AgentExecutor from crewai.knowledge.knowledge import Knowledge from crewai.knowledge.source.base_knowledge_source import BaseKnowledgeSource @@ -551,9 +552,37 @@ class Agent(BaseAgent): The fully prepared task prompt. """ prepare_tools(self, tools, task) + self._emit_skill_usage(task) return apply_training_data(self, task_prompt) + def _emit_skill_usage(self, task: Task) -> None: + """Emit one SkillUsedEvent per skill injected into this task's prompt. + + Skills are agent-scoped and rendered into the prompt on every execution, + so this is the runtime usage signal traces need — attributing each skill + to the agent and task that used it. + + Args: + task: The task whose prompt the skills are being applied to. + """ + if not self.skills: + return + + for skill in self.skills: + if not isinstance(skill, SkillModel): + continue + crewai_event_bus.emit( + self, + event=SkillUsedEvent( + from_agent=self, + from_task=task, + skill_name=skill.name, + skill_path=skill.path, + disclosure_level=skill.disclosure_level, + ), + ) + def _retrieve_memory_context(self, task: Task, task_prompt: str) -> str: """Retrieve memory context and append it to the task prompt. diff --git a/lib/crewai/src/crewai/events/__init__.py b/lib/crewai/src/crewai/events/__init__.py index ceb6271e9..8a31e5397 100644 --- a/lib/crewai/src/crewai/events/__init__.py +++ b/lib/crewai/src/crewai/events/__init__.py @@ -137,6 +137,7 @@ if TYPE_CHECKING: SkillEvent, SkillLoadFailedEvent, SkillLoadedEvent, + SkillUsedEvent, ) from crewai.events.types.task_events import ( TaskCompletedEvent, @@ -244,6 +245,7 @@ _LAZY_EVENT_MAPPING: dict[str, str] = { "SkillEvent": "crewai.events.types.skill_events", "SkillLoadFailedEvent": "crewai.events.types.skill_events", "SkillLoadedEvent": "crewai.events.types.skill_events", + "SkillUsedEvent": "crewai.events.types.skill_events", "TaskCompletedEvent": "crewai.events.types.task_events", "TaskEvaluationEvent": "crewai.events.types.task_events", "TaskFailedEvent": "crewai.events.types.task_events", @@ -376,6 +378,7 @@ __all__ = [ "SkillEvent", "SkillLoadFailedEvent", "SkillLoadedEvent", + "SkillUsedEvent", "TaskCompletedEvent", "TaskEvaluationEvent", "TaskFailedEvent", diff --git a/lib/crewai/src/crewai/events/types/skill_events.py b/lib/crewai/src/crewai/events/types/skill_events.py index aab625dda..820ff862c 100644 --- a/lib/crewai/src/crewai/events/types/skill_events.py +++ b/lib/crewai/src/crewai/events/types/skill_events.py @@ -60,3 +60,15 @@ class SkillLoadFailedEvent(SkillEvent): type: Literal["skill_load_failed"] = "skill_load_failed" error: str + + +class SkillUsedEvent(SkillEvent): + """Event emitted when an agent uses a skill during task execution. + + Discovery/load/activation events describe setup. This one is the runtime + signal: it fires each time a skill's context is injected into an agent's + prompt for a task, so traces can attribute skill usage to an agent and task. + """ + + type: Literal["skill_used"] = "skill_used" + disclosure_level: int = 1 diff --git a/lib/crewai/tests/events/test_skill_usage_events.py b/lib/crewai/tests/events/test_skill_usage_events.py new file mode 100644 index 000000000..04044c19e --- /dev/null +++ b/lib/crewai/tests/events/test_skill_usage_events.py @@ -0,0 +1,211 @@ +"""Tests for runtime skill usage events. + +Discovery/load/activation events cover setup. `SkillUsedEvent` is the runtime +signal: it fires whenever a skill's context is injected into an agent's prompt +for a task, so observability can attribute usage to an agent and task. +""" + +from pathlib import Path + +import pytest + +from crewai import Agent, Task +from crewai.events import crewai_event_bus +from crewai.events.types.skill_events import SkillUsedEvent + + +# Handlers run on the event bus thread pool, so tests must flush before +# asserting — otherwise assertions race the handler and pass/fail on timing. + + +def _create_skill_dir(parent: Path, name: str, body: str = "Body.") -> Path: + """Create a skill directory containing a minimal 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 + + +def _agent_with_skills(search_path: Path) -> Agent: + # discover_skills scans the SUBdirectories of the given path. + return Agent( + role="Analyst", + goal="Analyze things", + backstory="Experienced", + skills=[str(search_path)], + llm="gpt-4o-mini", + ) + + +def _task_for(agent: Agent) -> Task: + return Task(description="Do the thing", expected_output="text", agent=agent) + + +class TestSkillUsedEvent: + def test_emits_one_event_per_skill_with_agent_and_task_attribution( + self, tmp_path: Path + ) -> None: + _create_skill_dir(tmp_path, "alpha") + agent = _agent_with_skills(tmp_path) + task = _task_for(agent) + + received: list[SkillUsedEvent] = [] + with crewai_event_bus.scoped_handlers(): + + @crewai_event_bus.on(SkillUsedEvent) + def _handler(source, event: SkillUsedEvent) -> None: # noqa: ARG001 + received.append(event) + + agent._finalize_task_prompt("PROMPT", [], task) + assert crewai_event_bus.flush(timeout=10) + + assert len(received) == 1 + event = received[0] + assert event.type == "skill_used" + assert event.skill_name == "alpha" + assert event.skill_path is not None + # Attribution is what makes the event useful in traces. + assert event.agent_role == "Analyst" + assert event.agent_id == str(agent.id) + assert event.task_id == str(task.id) + + def test_emits_for_every_skill(self, tmp_path: Path) -> None: + _create_skill_dir(tmp_path, "alpha") + _create_skill_dir(tmp_path, "beta") + agent = _agent_with_skills(tmp_path) + task = _task_for(agent) + + received: list[SkillUsedEvent] = [] + with crewai_event_bus.scoped_handlers(): + + @crewai_event_bus.on(SkillUsedEvent) + def _handler(source, event: SkillUsedEvent) -> None: # noqa: ARG001 + received.append(event) + + agent._finalize_task_prompt("PROMPT", [], task) + assert crewai_event_bus.flush(timeout=10) + + assert sorted(e.skill_name for e in received) == ["alpha", "beta"] + + def test_re_emits_on_each_execution(self, tmp_path: Path) -> None: + """Usage is per-execution, unlike activation which is idempotent.""" + _create_skill_dir(tmp_path, "alpha") + agent = _agent_with_skills(tmp_path) + task = _task_for(agent) + + received: list[SkillUsedEvent] = [] + with crewai_event_bus.scoped_handlers(): + + @crewai_event_bus.on(SkillUsedEvent) + def _handler(source, event: SkillUsedEvent) -> None: # noqa: ARG001 + received.append(event) + + agent._finalize_task_prompt("PROMPT", [], task) + agent._finalize_task_prompt("PROMPT", [], task) + assert crewai_event_bus.flush(timeout=10) + + assert len(received) == 2 + + def test_reports_disclosure_level(self, tmp_path: Path) -> None: + """Path-loaded skills are activated, so they report INSTRUCTIONS level.""" + _create_skill_dir(tmp_path, "alpha") + agent = _agent_with_skills(tmp_path) + task = _task_for(agent) + + received: list[SkillUsedEvent] = [] + with crewai_event_bus.scoped_handlers(): + + @crewai_event_bus.on(SkillUsedEvent) + def _handler(source, event: SkillUsedEvent) -> None: # noqa: ARG001 + received.append(event) + + agent._finalize_task_prompt("PROMPT", [], task) + assert crewai_event_bus.flush(timeout=10) + + assert received[0].disclosure_level >= 2 + + def test_agent_without_skills_emits_nothing(self) -> None: + agent = Agent( + role="Analyst", + goal="Analyze things", + backstory="Experienced", + llm="gpt-4o-mini", + ) + task = _task_for(agent) + + received: list[SkillUsedEvent] = [] + with crewai_event_bus.scoped_handlers(): + + @crewai_event_bus.on(SkillUsedEvent) + def _handler(source, event: SkillUsedEvent) -> None: # noqa: ARG001 + received.append(event) + + agent._finalize_task_prompt("PROMPT", [], task) + assert crewai_event_bus.flush(timeout=10) + + assert received == [] + + def test_event_is_exported_from_events_package(self) -> None: + from crewai.events import SkillUsedEvent as Exported + + assert Exported is SkillUsedEvent + + +class TestSkillUsedEventThroughExecution: + """Coverage through the public execution entry points. + + The helper-level tests above pin per-skill details; these prove the event + actually reaches the bus during a real `execute_task` / `aexecute_task` + run. The agent executor is stubbed so no LLM call is made. + """ + + def test_emitted_during_execute_task( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + _create_skill_dir(tmp_path, "alpha") + agent = _agent_with_skills(tmp_path) + task = _task_for(agent) + monkeypatch.setattr( + Agent, "_execute_without_timeout", lambda self, prompt, task: "done" + ) + + received: list[SkillUsedEvent] = [] + with crewai_event_bus.scoped_handlers(): + + @crewai_event_bus.on(SkillUsedEvent) + def _handler(source, event: SkillUsedEvent) -> None: # noqa: ARG001 + received.append(event) + + agent.execute_task(task) + assert crewai_event_bus.flush(timeout=10) + + assert [e.skill_name for e in received] == ["alpha"] + assert received[0].task_id == str(task.id) + + @pytest.mark.asyncio + async def test_emitted_during_aexecute_task( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + _create_skill_dir(tmp_path, "alpha") + agent = _agent_with_skills(tmp_path) + task = _task_for(agent) + + async def _fake_execute(self, prompt: str, task: Task) -> str: # noqa: ANN001 + return "done" + + monkeypatch.setattr(Agent, "_aexecute_without_timeout", _fake_execute) + + received: list[SkillUsedEvent] = [] + with crewai_event_bus.scoped_handlers(): + + @crewai_event_bus.on(SkillUsedEvent) + def _handler(source, event: SkillUsedEvent) -> None: # noqa: ARG001 + received.append(event) + + await agent.aexecute_task(task) + assert crewai_event_bus.flush(timeout=10) + + assert [e.skill_name for e in received] == ["alpha"] + assert received[0].task_id == str(task.id)