Event Listeners

​

Overview

CrewAI provides a powerful event system that allows you to listen for and react to various events that occur during the execution of your Crew. This feature enables you to build custom integrations, monitoring solutions, logging systems, or any other functionality that needs to be triggered based on CrewAI’s internal events.

​

How It Works

CrewAI uses an event bus architecture to emit events throughout the execution lifecycle. The event system is built on the following components:

1. CrewAIEventsBus: A singleton event bus that manages event registration and emission
2. BaseEvent: Base class for all events in the system
3. BaseEventListener: Abstract base class for creating custom event listeners

When specific actions occur in CrewAI (like a Crew starting execution, an Agent completing a task, or a tool being used), the system emits corresponding events. You can register handlers for these events to execute custom code when they occur.

​

Creating a Custom Event Listener

To create a custom event listener, you need to:

1. Create a class that inherits from BaseEventListener
2. Implement the setup_listeners method
3. Register handlers for the events you’re interested in
4. Create an instance of your listener in the appropriate file

Here’s a simple example of a custom event listener class:

from crewai.utilities.events import (
    CrewKickoffStartedEvent,
    CrewKickoffCompletedEvent,
    AgentExecutionCompletedEvent,
)
from crewai.utilities.events.base_event_listener import BaseEventListener

class MyCustomListener(BaseEventListener):
    def __init__(self):
        super().__init__()

    def setup_listeners(self, crewai_event_bus):
        @crewai_event_bus.on(CrewKickoffStartedEvent)
        def on_crew_started(source, event):
            print(f"Crew '{event.crew_name}' has started execution!")

        @crewai_event_bus.on(CrewKickoffCompletedEvent)
        def on_crew_completed(source, event):
            print(f"Crew '{event.crew_name}' has completed execution!")
            print(f"Output: {event.output}")

        @crewai_event_bus.on(AgentExecutionCompletedEvent)
        def on_agent_execution_completed(source, event):
            print(f"Agent '{event.agent.role}' completed task")
            print(f"Output: {event.output}")

​

Properly Registering Your Listener

Simply defining your listener class isn’t enough. You need to create an instance of it and ensure it’s imported in your application. This ensures that:

1. The event handlers are registered with the event bus
2. The listener instance remains in memory (not garbage collected)
3. The listener is active when events are emitted

​

Option 1: Import and Instantiate in Your Crew or Flow Implementation

The most important thing is to create an instance of your listener in the file where your Crew or Flow is defined and executed:

​

For Crew-based Applications

Create and import your listener at the top of your Crew implementation file:

# In your crew.py file
from crewai import Agent, Crew, Task
from my_listeners import MyCustomListener

# Create an instance of your listener
my_listener = MyCustomListener()

class MyCustomCrew:
    # Your crew implementation...

    def crew(self):
        return Crew(
            agents=[...],
            tasks=[...],
            # ...
        )

​

For Flow-based Applications

Create and import your listener at the top of your Flow implementation file:

# In your main.py or flow.py file
from crewai.flow import Flow, listen, start
from my_listeners import MyCustomListener

# Create an instance of your listener
my_listener = MyCustomListener()

class MyCustomFlow(Flow):
    # Your flow implementation...

    @start()
    def first_step(self):
        # ...

This ensures that your listener is loaded and active when your Crew or Flow is executed.

​

Option 2: Create a Package for Your Listeners

For a more structured approach, especially if you have multiple listeners:

1. Create a package for your listeners:

my_project/
  ├── listeners/
  │   ├── __init__.py
  │   ├── my_custom_listener.py
  │   └── another_listener.py

2. In my_custom_listener.py, define your listener class and create an instance:

# my_custom_listener.py
from crewai.utilities.events.base_event_listener import BaseEventListener
# ... import events ...

class MyCustomListener(BaseEventListener):
    # ... implementation ...

# Create an instance of your listener
my_custom_listener = MyCustomListener()

3. In __init__.py, import the listener instances to ensure they’re loaded:

# __init__.py
from .my_custom_listener import my_custom_listener
from .another_listener import another_listener

# Optionally export them if you need to access them elsewhere
__all__ = ['my_custom_listener', 'another_listener']

4. Import your listeners package in your Crew or Flow file:

# In your crew.py or flow.py file
import my_project.listeners  # This loads all your listeners

class MyCustomCrew:
    # Your crew implementation...

This is exactly how CrewAI’s built-in agentops_listener is registered. In the CrewAI codebase, you’ll find:

# src/crewai/utilities/events/third_party/__init__.py
from .agentops_listener import agentops_listener

This ensures the agentops_listener is loaded when the crewai.utilities.events package is imported.

​

Available Event Types

CrewAI provides a wide range of events that you can listen for:

​

Crew Events

• CrewKickoffStartedEvent: Emitted when a Crew starts execution
• CrewKickoffCompletedEvent: Emitted when a Crew completes execution
• CrewKickoffFailedEvent: Emitted when a Crew fails to complete execution
• CrewTestStartedEvent: Emitted when a Crew starts testing
• CrewTestCompletedEvent: Emitted when a Crew completes testing
• CrewTestFailedEvent: Emitted when a Crew fails to complete testing
• CrewTrainStartedEvent: Emitted when a Crew starts training
• CrewTrainCompletedEvent: Emitted when a Crew completes training
• CrewTrainFailedEvent: Emitted when a Crew fails to complete training

​

Agent Events

• AgentExecutionStartedEvent: Emitted when an Agent starts executing a task
• AgentExecutionCompletedEvent: Emitted when an Agent completes executing a task
• AgentExecutionErrorEvent: Emitted when an Agent encounters an error during execution

​

Task Events

• TaskStartedEvent: Emitted when a Task starts execution
• TaskCompletedEvent: Emitted when a Task completes execution
• TaskFailedEvent: Emitted when a Task fails to complete execution
• TaskEvaluationEvent: Emitted when a Task is evaluated

​

Tool Usage Events

• ToolUsageStartedEvent: Emitted when a tool execution is started
• ToolUsageFinishedEvent: Emitted when a tool execution is completed
• ToolUsageErrorEvent: Emitted when a tool execution encounters an error
• ToolValidateInputErrorEvent: Emitted when a tool input validation encounters an error
• ToolExecutionErrorEvent: Emitted when a tool execution encounters an error
• ToolSelectionErrorEvent: Emitted when there’s an error selecting a tool

​

Knowledge Events

• KnowledgeRetrievalStartedEvent: Emitted when a knowledge retrieval is started
• KnowledgeRetrievalCompletedEvent: Emitted when a knowledge retrieval is completed
• KnowledgeQueryStartedEvent: Emitted when a knowledge query is started
• KnowledgeQueryCompletedEvent: Emitted when a knowledge query is completed
• KnowledgeQueryFailedEvent: Emitted when a knowledge query fails
• KnowledgeSearchQueryFailedEvent: Emitted when a knowledge search query fails

​

LLM Guardrail Events

• LLMGuardrailStartedEvent: Emitted when a guardrail validation starts. Contains details about the guardrail being applied and retry count.
• LLMGuardrailCompletedEvent: Emitted when a guardrail validation completes. Contains details about validation success/failure, results, and error messages if any.

​

Flow Events

• FlowCreatedEvent: Emitted when a Flow is created
• FlowStartedEvent: Emitted when a Flow starts execution
• FlowFinishedEvent: Emitted when a Flow completes execution
• FlowPlotEvent: Emitted when a Flow is plotted
• MethodExecutionStartedEvent: Emitted when a Flow method starts execution
• MethodExecutionFinishedEvent: Emitted when a Flow method completes execution
• MethodExecutionFailedEvent: Emitted when a Flow method fails to complete execution

​

LLM Events

• LLMCallStartedEvent: Emitted when an LLM call starts
• LLMCallCompletedEvent: Emitted when an LLM call completes
• LLMCallFailedEvent: Emitted when an LLM call fails
• LLMStreamChunkEvent: Emitted for each chunk received during streaming LLM responses

​

Event Handler Structure

Each event handler receives two parameters:

1. source: The object that emitted the event
2. event: The event instance, containing event-specific data

The structure of the event object depends on the event type, but all events inherit from BaseEvent and include:

• timestamp: The time when the event was emitted
• type: A string identifier for the event type

Additional fields vary by event type. For example, CrewKickoffCompletedEvent includes crew_name and output fields.

​

Real-World Example: Integration with AgentOps

CrewAI includes an example of a third-party integration with AgentOps, a monitoring and observability platform for AI agents. Here’s how it’s implemented:

from typing import Optional

from crewai.utilities.events import (
    CrewKickoffCompletedEvent,
    ToolUsageErrorEvent,
    ToolUsageStartedEvent,
)
from crewai.utilities.events.base_event_listener import BaseEventListener
from crewai.utilities.events.crew_events import CrewKickoffStartedEvent
from crewai.utilities.events.task_events import TaskEvaluationEvent

try:
    import agentops
    AGENTOPS_INSTALLED = True
except ImportError:
    AGENTOPS_INSTALLED = False

class AgentOpsListener(BaseEventListener):
    tool_event: Optional["agentops.ToolEvent"] = None
    session: Optional["agentops.Session"] = None

    def __init__(self):
        super().__init__()

    def setup_listeners(self, crewai_event_bus):
        if not AGENTOPS_INSTALLED:
            return

        @crewai_event_bus.on(CrewKickoffStartedEvent)
        def on_crew_kickoff_started(source, event: CrewKickoffStartedEvent):
            self.session = agentops.init()
            for agent in source.agents:
                if self.session:
                    self.session.create_agent(
                        name=agent.role,
                        agent_id=str(agent.id),
                    )

        @crewai_event_bus.on(CrewKickoffCompletedEvent)
        def on_crew_kickoff_completed(source, event: CrewKickoffCompletedEvent):
            if self.session:
                self.session.end_session(
                    end_state="Success",
                    end_state_reason="Finished Execution",
                )

        @crewai_event_bus.on(ToolUsageStartedEvent)
        def on_tool_usage_started(source, event: ToolUsageStartedEvent):
            self.tool_event = agentops.ToolEvent(name=event.tool_name)
            if self.session:
                self.session.record(self.tool_event)

        @crewai_event_bus.on(ToolUsageErrorEvent)
        def on_tool_usage_error(source, event: ToolUsageErrorEvent):
            agentops.ErrorEvent(exception=event.error, trigger_event=self.tool_event)

This listener initializes an AgentOps session when a Crew starts, registers agents with AgentOps, tracks tool usage, and ends the session when the Crew completes.

The AgentOps listener is registered in CrewAI’s event system through the import in src/crewai/utilities/events/third_party/__init__.py:

from .agentops_listener import agentops_listener

This ensures the agentops_listener is loaded when the crewai.utilities.events package is imported.

​

Advanced Usage: Scoped Handlers

For temporary event handling (useful for testing or specific operations), you can use the scoped_handlers context manager:

from crewai.utilities.events import crewai_event_bus, CrewKickoffStartedEvent

with crewai_event_bus.scoped_handlers():
    @crewai_event_bus.on(CrewKickoffStartedEvent)
    def temp_handler(source, event):
        print("This handler only exists within this context")

    # Do something that emits events

# Outside the context, the temporary handler is removed

​

Use Cases

Event listeners can be used for a variety of purposes:

1. Logging and Monitoring: Track the execution of your Crew and log important events
2. Analytics: Collect data about your Crew’s performance and behavior
3. Debugging: Set up temporary listeners to debug specific issues
4. Integration: Connect CrewAI with external systems like monitoring platforms, databases, or notification services
5. Custom Behavior: Trigger custom actions based on specific events

​

Best Practices

1. Keep Handlers Light: Event handlers should be lightweight and avoid blocking operations
2. Error Handling: Include proper error handling in your event handlers to prevent exceptions from affecting the main execution
3. Cleanup: If your listener allocates resources, ensure they’re properly cleaned up
4. Selective Listening: Only listen for events you actually need to handle
5. Testing: Test your event listeners in isolation to ensure they behave as expected

By leveraging CrewAI’s event system, you can extend its functionality and integrate it seamlessly with your existing infrastructure.