Skip to main content

CrewAI Integration

CrewAIEventListener subclasses CrewAI's BaseEventListener to capture all crew lifecycle events — crews, agents, tasks, tools, LLM calls, knowledge retrieval, memory operations, guardrails, and flows — with start/end pairing and execution_time_ms computation.

Installation​

pip install briefcase-ai
pip install crewai>=0.80.0
# or
pip install briefcase-ai[crewai]

Quick Setup with auto()​

The simplest way to instrument CrewAI is one line:

import briefcase_ai

briefcase_ai.auto("crewai")
# That's it — all CrewAI crew events are now traced.

Manual Setup​

Instantiating CrewAIEventListener automatically registers it with the global CrewAI event bus via BaseEventListener.__init_subclass__.

from briefcase_ai.integrations.frameworks import CrewAIEventListener

listener = CrewAIEventListener(context_version="v2.1")

Constructor​

CrewAIEventListener(
    context_version: Optional[str] = None,
    async_capture: bool = True,
    capture_crews: bool = True,
    capture_agents: bool = True,
    capture_tasks: bool = True,
    capture_tools: bool = True,
    capture_llm: bool = True,
    capture_knowledge: bool = True,
    capture_memory: bool = True,
    capture_guardrails: bool = True,
    capture_flows: bool = True,
    max_input_chars: int = 10000,
    max_output_chars: int = 10000,
    exporter: Optional[BaseExporter] = None,
)
ParameterDefaultDescription
context_versionNoneVersion tag added to all decision records
async_captureTrueExport in a background daemon thread
capture_crewsTrueCapture crew kickoff / test / train events
capture_agentsTrueCapture agent execution events
capture_tasksTrueCapture task lifecycle events
capture_toolsTrueCapture tool usage events
capture_llmTrueCapture LLM call events
capture_knowledgeTrueCapture knowledge retrieval events
capture_memoryTrueCapture memory query / save events
capture_guardrailsTrueCapture guardrail evaluation events
capture_flowsTrueCapture flow and method execution events
max_input_chars10000Maximum length of captured input strings
max_output_chars10000Maximum length of captured output strings
exporterNonePer-instance BaseExporter override

Basic Usage​

from crewai import Crew, Agent, Task
from briefcase_ai.integrations.frameworks import CrewAIEventListener

listener = CrewAIEventListener(context_version="v2.1")

researcher = Agent(
    role="Research Analyst",
    goal="Find comprehensive information on the given topic.",
    backstory="You are an expert researcher.",
)

task = Task(
    description="Research AI safety regulations in 2026.",
    expected_output="A summary of key regulations.",
    agent=researcher,
)

crew = Crew(agents=[researcher], tasks=[task])
result = crew.kickoff()

# Inspect captured records
for record in listener.get_records():
    print(f"{record['decision_type']}: {record.get('execution_time_ms', 'N/A')}ms")

Record Format​

Each record is a Dict[str, Any] with these fields:

{
    "decision_id": "uuid-string",
    "decision_type": "crew_kickoff",     # see table below
    "event_key": "crew1",
    "started_at": "2026-02-26T10:00:00Z",
    "ended_at": "2026-02-26T10:00:05Z",
    "execution_time_ms": 5000.0,
    "inputs": {"crew_id": "crew1", "name": "MyCrew"},
    "outputs": {"output": "result text"},
    "context_version": "v2.1",           # if set
    "error": "..."                        # only on failure
}

Event Types​

decision_typeCaptured whenStart/End paired
crew_kickoffCrewKickoffStarted/Completed/FailedYes
agent_executionAgentExecutionStarted/Completed/ErrorYes
taskTaskStarted/Completed/FailedYes
tool_usageToolUsageStarted/Finished/ErrorYes
llm_callLLMCallStarted/Completed/FailedYes
knowledge_retrievalKnowledgeRetrievalStarted/CompletedYes
knowledge_queryKnowledgeQueryStarted/Completed/FailedYes
knowledge_search_failedKnowledgeSearchQueryFailedNo
memory_queryMemoryQueryStarted/Completed/FailedYes
memory_saveMemorySaveStarted/Completed/FailedYes
memory_retrievalMemoryRetrievalStarted/CompletedYes
llm_guardrailLLMGuardrailStarted/CompletedYes
flowFlowStarted/FinishedYes
flow_createdFlowCreatedEventNo
flow_plotFlowPlotEventNo
method_executionMethodExecutionStarted/Finished/FailedYes

Capture Flags​

Disable event categories you don't need to reduce overhead:

listener = CrewAIEventListener(
    capture_knowledge=False,
    capture_memory=False,
    capture_flows=False,
)

Export on Crew Completion​

When a crew kickoff completes, the record is automatically passed to the configured BriefcaseConfig.exporter:

from briefcase_ai.config import setup
from briefcase_ai.exporters import OTelExporter

setup(exporter=OTelExporter(endpoint="http://localhost:4317"))

listener = CrewAIEventListener(async_capture=True)

# Run crew — exports fire asynchronously on each completed event
crew.kickoff()

Public API​

MethodReturnsDescription
get_records()List[Dict[str, Any]]All captured decision records
clear()NoneReset all records and inflight state
decision_countintNumber of captured records (property)
setup_listeners(bus)NoneRegister handlers on custom event bus

Advanced: Per-Instance Exporter​

Override the global exporter for this listener instance:

from briefcase_ai.exporters import SplunkHECExporter

listener = CrewAIEventListener(
    engagement_id="acme",
    exporter=SplunkHECExporter(url="https://splunk.example.com:8088", token="your-hec-token"),
)

Troubleshooting​

ImportError: crewai is required: Install with pip install crewai>=0.80.0. The listener raises at instantiation time if the package is absent.

Events not captured: Ensure you instantiate CrewAIEventListener before creating or running your crew. The auto-registration happens in __init__.

Missing optional event types: Some event classes (KnowledgeRetrievalStartedEvent, flow events, etc.) were introduced in specific crewai versions. Briefcase silently skips missing event types — no error is raised.

See Also​