CrewAI 이벤트 리스너

CrewAI 이벤트 리스너 (Event Listeners)

크루 실행 중에 일어나는 다양한 사건(크루 시작, 에이전트 태스크 완료, 도구 사용 등)에 내 코드가 반응하게 하려면 이벤트 리스너를 만들면 돼요. CrewAI의 이벤트 시스템은 싱글턴 이벤트 버스를 기반으로 동작해서, 커스텀 통합·모니터링·로깅 같은 기능을 크루 내부 이벤트에 연결할 수 있습니다. 이 페이지에서 리스너 만드는 법과 등록 규칙, 그리고 어떤 이벤트 타입들이 있는지 정리합니다.

출처: 공식문서

본문

어떻게 동작하나

CrewAI는 실행 수명주기 전반에 걸쳐 이벤트를 방출하는 이벤트 버스 아키텍처를 사용합니다. 시스템은 다음 구성 요소로 이루어져 있어요.

CrewAIEventsBus — 이벤트 등록과 방출을 관리하는 싱글턴 이벤트 버스
BaseEvent — 시스템의 모든 이벤트가 상속받는 기반 클래스
BaseEventListener — 커스텀 이벤트 리스너를 만들기 위한 추상 기반 클래스

크루가 실행을 시작하거나, 에이전트가 태스크를 끝내거나, 도구가 사용되는 등 특정 동작이 일어나면 시스템이 해당 이벤트를 방출합니다. 이 이벤트에 핸들러를 등록해서 원하는 코드를 실행시킬 수 있어요.

커스텀 이벤트 리스너 만들기

커스텀 이벤트 리스너를 만들려면 다음 단계를 따릅니다.

BaseEventListener를 상속하는 클래스를 만든다
setup_listeners 메서드를 구현한다
관심 있는 이벤트에 핸들러를 등록한다
적절한 파일에서 리스너 인스턴스를 생성한다

간단한 커스텀 리스너 예시:

from crewai.events import (
    CrewKickoffStartedEvent,
    CrewKickoffCompletedEvent,
    AgentExecutionCompletedEvent,
)
from crewai.events 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}")

setup_listeners 안에서 crewai_event_bus.on(<EventClass>) 데코레이터로 특정 이벤트의 핸들러를 등록합니다.

리스너 등록하기 — 클래스 정의만으론 부족해요

리스너 클래스를 정의하는 것만으로는 충분하지 않습니다. 인스턴스를 만들고 애플리케이션에서 import해야 해요. 그래야 ① 이벤트 핸들러가 이벤트 버스에 등록되고, ② 리스너 인스턴스가 메모리에 유지되며(가비지 컬렉트되지 않음), ③ 이벤트가 방출될 때 리스너가 활성 상태가 됩니다.

Option 1: Crew 또는 Flow 구현 파일에서 import·인스턴스화 — 리스너 인스턴스를 Crew나 Flow를 정의·실행하는 파일 맨 위에 둡니다.

# 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=[...],
            # ...
        )

Flow 기반 애플리케이션도 마찬가지로 Flow 구현 파일 맨 위에 둡니다.

# 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):
        # ...

Option 2: 리스너용 패키지 만들기 — 리스너가 여러 개라면 더 구조화된 방식이 좋습니다.

리스너용 패키지를 만듭니다.

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

my_custom_listener.py에서 리스너 클래스를 정의하고 인스턴스를 만듭니다.

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

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

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

__init__.py에서 리스너 인스턴스들을 import해 로드되게 합니다.

# __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']

Crew 또는 Flow 파일에서 리스너 패키지를 import합니다.

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

class MyCustomCrew:
    # Your crew implementation...

이 방식이 CrewAI 코드베이스에서 서드파티 이벤트 리스너를 등록하는 방법입니다.

사용 가능한 이벤트 타입

CrewAI는 폭넓은 이벤트를 제공합니다. 주요 그룹별로 정리할게요.

Crew 이벤트

CrewKickoffStartedEvent: Crew가 실행을 시작할 때
CrewKickoffCompletedEvent: Crew가 실행을 완료할 때
CrewKickoffFailedEvent: Crew가 실행을 완료하지 못할 때
CrewTestStartedEvent / CrewTestCompletedEvent / CrewTestFailedEvent: Crew 테스트 시작·완료·실패
CrewTrainStartedEvent / CrewTrainCompletedEvent / CrewTrainFailedEvent: Crew 훈련 시작·완료·실패
CrewTestResultEvent: 테스트 결과가 나올 때. 품질 점수, 실행 시간, 사용 모델을 포함

Agent 이벤트

AgentExecutionStartedEvent / AgentExecutionCompletedEvent / AgentExecutionErrorEvent: 에이전트 태스크 실행 시작·완료·오류
LiteAgentExecutionStartedEvent / CompletedEvent / ErrorEvent: LiteAgent 실행 관련 (에이전트 정보, 도구, 메시지/출력/오류 포함)
AgentEvaluationStartedEvent / CompletedEvent / FailedEvent: 에이전트 평가 관련 (에이전트 ID·역할·태스크 ID·반복 번호, 완료 시 메트릭 카테고리·점수)

Task 이벤트

TaskStartedEvent / TaskCompletedEvent / TaskFailedEvent: 태스크 실행 시작·완료·실패
TaskEvaluationEvent: 태스크가 평가될 때

도구 사용 이벤트

ToolUsageStartedEvent / ToolUsageFinishedEvent / ToolUsageErrorEvent: 도구 실행 시작·완료·오류
ToolValidateInputErrorEvent: 도구 입력 검증 오류
ToolExecutionErrorEvent: 도구 실행 오류
ToolSelectionErrorEvent: 도구 선택 오류

MCP 이벤트

MCPConnectionStartedEvent / CompletedEvent / FailedEvent: MCP 서버 연결 시작·성공·실패 (서버 이름, URL, 전송 타입, 연결 타임아웃, 재연결 여부 / 지속 시간(ms) / 오류 메시지·오류 타입 포함)
MCPToolExecutionStartedEvent / CompletedEvent / FailedEvent: MCP 도구 실행 관련 (서버 이름, 도구 이름, 인자 / 결과·실행 시간(ms) / 오류 메시지·오류 타입)
MCPConfigFetchFailedEvent: MCP 서버 설정을 가져오지 못할 때 (slug, 오류 메시지, 오류 타입 not_connected/api_error/connection_failed)

Knowledge 이벤트

KnowledgeRetrievalStartedEvent / CompletedEvent: 지식 검색 시작·완료
KnowledgeQueryStartedEvent / CompletedEvent / FailedEvent: 지식 쿼리 시작·완료·실패
KnowledgeSearchQueryFailedEvent: 지식 검색 쿼리 실패

LLM 가드레일 이벤트

LLMGuardrailStartedEvent / CompletedEvent / FailedEvent: 가드레일 검증 시작·완료·실패 (적용 중인 가드레일·재시도 횟수 / 성공·실패·결과·오류 메시지 / 오류 메시지·재시도 횟수)

Flow 이벤트

FlowCreatedEvent: Flow 생성 시
FlowStartedEvent: Flow 실행 시작 시
FlowFinishedEvent: Flow 실행 완료 시
FlowFailedEvent: Flow 실행 실패 시. Flow 이름과 실행을 끝낸 예외를 포함
FlowPausedEvent: Flow가 사람 피드백을 기다리며 멈출 때. Flow 이름, Flow ID, 메서드 이름, 현재 상태, 피드백 요청 메시지, 라우팅용 가능한 결과 목록(선택)을 포함
FlowPlotEvent: Flow가 플롯될 때
MethodExecutionStartedEvent / FinishedEvent / FailedEvent / PausedEvent: Flow 메서드 실행 시작·완료·실패·일시정지

Human In The Loop 이벤트

FlowInputRequestedEvent: Flow가 Flow.ask()로 사용자 입력을 요청할 때. Flow 이름, 메서드 이름, 사용자에게 보여지는 질문, 선택적 메타데이터 포함
FlowInputReceivedEvent: Flow.ask() 후 사용자 입력을 받았을 때. 응답(타임아웃 시 None)과 선택적 응답 메타데이터 포함
HumanFeedbackRequestedEvent: @human_feedback 데코레이터 메서드가 사람 검토자 입력을 요구할 때
HumanFeedbackReceivedEvent: @human_feedback 메서드에 대한 피드백을 사람이 제공했을 때. 원시 텍스트 피드백과 (emit 지정 시) 축약된 결과 문자열 포함

LLM 이벤트

LLMCallStartedEvent / CompletedEvent / FailedEvent: LLM 호출 시작·완료·실패
LLMStreamChunkEvent: 스트리밍 LLM 응답에서 각 청크 수신 시
LLMThinkingChunkEvent: 씽킹 모델에서 reasoning 청크를 받을 때. 청크 텍스트와 선택적 응답 ID 포함

Memory 이벤트

MemoryQueryStartedEvent / CompletedEvent / FailedEvent: 메모리 쿼리 관련 (쿼리, limit, score threshold / 결과·실행 시간 / 오류)
MemorySaveStartedEvent / CompletedEvent / FailedEvent: 메모리 저장 관련 (저장 값, 메타데이터, 에이전트 역할)
MemoryRetrievalStartedEvent / CompletedEvent / FailedEvent: 태스크 프롬프트용 메모리 검색 관련

Reasoning 이벤트

AgentReasoningStartedEvent / CompletedEvent / FailedEvent: 에이전트 추론 관련 (에이전트 역할, 태스크 ID, 시도 번호 / 생성된 계획·진행 준비 여부 / 오류 메시지)

Observation 이벤트

StepObservationStartedEvent / CompletedEvent / FailedEvent: Planner가 단계 결과를 관찰할 때 (에이전트 역할, 단계 번호, 단계 설명 / 성공 여부·배운 핵심 정보·계획 유효성·전체 재계획 필요 여부·제안된 개선 / 오류 시 계속 진행)
PlanRefinementEvent: Planner가 전체 재계획 없이 예정된 단계 설명을 개선할 때
PlanReplanTriggeredEvent: Planner가 남은 계획이 근본적으로 잘못됐다고 판단해 전체 재계획을 트리거할 때
GoalAchievedEarlyEvent: Planner가 목표가 조기 달성됐다고 감지해 남은 단계를 건너뛸 때

A2A (Agent-to-Agent) 이벤트 — 위임(Delegation) 관련

A2ADelegationStartedEvent / CompletedEvent: A2A 위임 시작·완료 (엔드포인트 URL, 태스크 설명, 에이전트 ID, 컨텍스트 ID, 멀티턴 여부, 턴 번호, 에이전트 카드 메타데이터, 프로토콜 버전, 프로바이더 정보, 선택적 스킬 ID / 완료 상태, 결과, 오류 메시지)
A2AParallelDelegationStartedEvent: 여러 A2A 에이전트에 병렬 위임이 시작될 때

더 알아보기

CrewAI Flows — Flow 실행도 이벤트로 관찰
CrewAI Memory — 메모리 이벤트 타입
Event Listeners 공식 문서