Emit skill usage events at runtime for observability (#6652)

* feat: emit skill usage events at runtime

* test: cover skill events via execute_task paths
This commit is contained in:
alex-clawd
2026-07-26 01:04:19 -07:00
committed by GitHub
parent 728183e420
commit 80fa0295c4
4 changed files with 255 additions and 0 deletions

View File

@@ -73,6 +73,7 @@ from crewai.events.types.memory_events import (
MemoryRetrievalFailedEvent, MemoryRetrievalFailedEvent,
MemoryRetrievalStartedEvent, MemoryRetrievalStartedEvent,
) )
from crewai.events.types.skill_events import SkillUsedEvent
from crewai.experimental.agent_executor import AgentExecutor from crewai.experimental.agent_executor import AgentExecutor
from crewai.knowledge.knowledge import Knowledge from crewai.knowledge.knowledge import Knowledge
from crewai.knowledge.source.base_knowledge_source import BaseKnowledgeSource from crewai.knowledge.source.base_knowledge_source import BaseKnowledgeSource
@@ -551,9 +552,37 @@ class Agent(BaseAgent):
The fully prepared task prompt. The fully prepared task prompt.
""" """
prepare_tools(self, tools, task) prepare_tools(self, tools, task)
self._emit_skill_usage(task)
return apply_training_data(self, task_prompt) 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: def _retrieve_memory_context(self, task: Task, task_prompt: str) -> str:
"""Retrieve memory context and append it to the task prompt. """Retrieve memory context and append it to the task prompt.

View File

@@ -137,6 +137,7 @@ if TYPE_CHECKING:
SkillEvent, SkillEvent,
SkillLoadFailedEvent, SkillLoadFailedEvent,
SkillLoadedEvent, SkillLoadedEvent,
SkillUsedEvent,
) )
from crewai.events.types.task_events import ( from crewai.events.types.task_events import (
TaskCompletedEvent, TaskCompletedEvent,
@@ -244,6 +245,7 @@ _LAZY_EVENT_MAPPING: dict[str, str] = {
"SkillEvent": "crewai.events.types.skill_events", "SkillEvent": "crewai.events.types.skill_events",
"SkillLoadFailedEvent": "crewai.events.types.skill_events", "SkillLoadFailedEvent": "crewai.events.types.skill_events",
"SkillLoadedEvent": "crewai.events.types.skill_events", "SkillLoadedEvent": "crewai.events.types.skill_events",
"SkillUsedEvent": "crewai.events.types.skill_events",
"TaskCompletedEvent": "crewai.events.types.task_events", "TaskCompletedEvent": "crewai.events.types.task_events",
"TaskEvaluationEvent": "crewai.events.types.task_events", "TaskEvaluationEvent": "crewai.events.types.task_events",
"TaskFailedEvent": "crewai.events.types.task_events", "TaskFailedEvent": "crewai.events.types.task_events",
@@ -376,6 +378,7 @@ __all__ = [
"SkillEvent", "SkillEvent",
"SkillLoadFailedEvent", "SkillLoadFailedEvent",
"SkillLoadedEvent", "SkillLoadedEvent",
"SkillUsedEvent",
"TaskCompletedEvent", "TaskCompletedEvent",
"TaskEvaluationEvent", "TaskEvaluationEvent",
"TaskFailedEvent", "TaskFailedEvent",

View File

@@ -60,3 +60,15 @@ class SkillLoadFailedEvent(SkillEvent):
type: Literal["skill_load_failed"] = "skill_load_failed" type: Literal["skill_load_failed"] = "skill_load_failed"
error: str 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

View File

@@ -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)