Chapter 14 · Callbacks, Logging, and Event Listening

Chapter goals: Distinguish the trigger timing and signatures of step_callback vs task_callback, master CrewAI's event bus (Event Bus) and BaseEventListener体系, implement a TaskCompletedEvent auditor, and persist structured logs as JSONL.

14.1 Two-Level Callbacks: Process vs Product

CrewAI offers two callback tiers. Clarify the division of labor before discussing the event system:

Callback	Mount location	Trigger timing	Typical use
step_callback	Agent or Crew	After each step of each agent	Process logs, real-time UI refresh
callback / task_callback	Task (or Crew's task_callback)	When a task completes	Product auditing, result persistence, metrics

# callbacks_demo.py —— minimal two-level callback对照 example
import os
from crewai import Agent, Task, Crew, LLM

llm = LLM(
    model="openai/deepseek-chat",
    base_url="https://api.deepseek.com/v1",
    api_key=os.getenv("DEEPSEEK_API_KEY"),
)

def on_step(step):
    """Step-level: fires once per step; arg is the step's output/action object."""
    print(f"[STEP] {str(step)[:120]}")

def on_task_done(task_output):
    """Task-level: fires once on task completion; arg is the TaskOutput object."""
    print(f"[TASK] {task_output.description[:40]} -> "
          f"{len(task_output.raw)} chars, from {task_output.agent}")

analyst = Agent(role="Analyst", goal="Produce a conclusion", backstory="Senior analyst", llm=llm)

t1 = Task(
    description="Analyze the impact of remote work on team efficiency.",
    expected_output="A conclusion paragraph under 150 words.",
    agent=analyst,
    callback=on_task_done,      # task-level callback
)

crew = Crew(
    agents=[analyst],
    tasks=[t1],
    step_callback=on_step,      # crew-level step callback
)
crew.kickoff()

Three pitfalls to watch:

• step_callback defined on an Agent overrides the同名 one on the Crew — don't expect both to fire;
• TaskOutput.raw is the string body; pydantic / json_dict attributes carry structured results (Chapter 12);
• Exceptions thrown inside callbacks disrupt the execution flow — audit code must have its own try/except; don't let logging bugs crash the业务.

14.2 Event Bus: A More Complete Observation Surface

Callbacks only cover "step end" and "task complete" — two points. Internally, CrewAI runs on an event bus architecture:

1. CrewAIEventsBus — singleton event bus responsible for registration and dispatch;
2. BaseEvent — base class for all events, carrying timestamp and type;
3. BaseEventListener — abstract base class for custom listeners.

Crew start, agent complete, tool invocation, memory read/write, LLM call — the entire lifecycle fires events on the bus. Listening to events is far more powerful than callbacks: no need to侵入 Crew definition code to achieve global monitoring, auditing, and third-party integration.

14.3 Writing a Custom Event Listener

Standard four steps: inherit BaseEventListener → implement setup_listeners → register handlers with @crewai_event_bus.on(EventType) → create a listener instance in the module where the Crew/Flow lives.

# my_listeners.py —— listen to Crew start/completion and agent completion
from crewai.events import (
    BaseEventListener,
    CrewKickoffStartedEvent,
    CrewKickoffCompletedEvent,
    AgentExecutionCompletedEvent,
)

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}' execution complete!")
            print(f"Output: {event.output}")

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

# Critical: must create an instance, otherwise handlers are not registered
my_listener = MyCustomListener()

Instance must be alive and imported

Defining the class alone is not enough. The official docs stress three things: handlers must be registered on the event bus; the listener instance must stay in memory (not GC'd); the module containing the instance must be imported by your application. The simplest approach: from my_listeners import MyCustomListener at the top of the file where you define the Crew, and instantiate it.

Handler functions follow a uniform signature (source, event): source is the object firing the event, event is the event instance — beyond the base timestamp/type, each event has its own fields (e.g., CrewKickoffCompletedEvent.output, event.agent.role).

14.4 Listening to TaskCompletedEvent for Audit Logs

Event types cover a wide range. Common families:

• Crew events: CrewKickoffStartedEvent / CrewKickoffCompletedEvent / CrewKickoffFailedEvent
• Agent events: AgentExecutionStartedEvent / AgentExecutionCompletedEvent / AgentExecutionErrorEvent
• Task events: TaskStartedEvent / TaskCompletedEvent / TaskFailedEvent
• Tool events: ToolUsageStartedEvent / ToolUsageFinishedEvent / ToolUsageErrorEvent
• LLM events: LLMCallStartedEvent / LLMCallCompletedEvent / LLMCallFailedEvent
• Flow events: FlowStartedEvent / MethodExecutionStartedEvent / MethodExecutionFinishedEvent, etc.

Below is a production-ready audit listener: appending each task's completion status to a JSONL file.

# audit_listener.py —— TaskCompletedEvent audit persistence
import json
import os
from datetime import datetime
from crewai.events import BaseEventListener, TaskCompletedEvent

class TaskAuditListener(BaseEventListener):
    def __init__(self, path="audit_log.jsonl"):
        super().__init__()
        self.path = path

    def setup_listeners(self, crewai_event_bus):
        @crewai_event_bus.on(TaskCompletedEvent)
        def on_task_completed(source, event):
            try:                       # audit code must never throw to disrupt the main flow
                record = {
                    "ts": datetime.now().isoformat(timespec="seconds"),
                    "task": getattr(event, "task", None) is not None
                            and str(event.task.description)[:80],
                    "summary": str(getattr(event, "output", ""))[:2000],
                }
                with open(self.path, "a", encoding="utf-8") as f:
                    f.write(json.dumps(record, ensure_ascii=False) + "\n")
            except Exception as e:     # noqa: BLE001
                print(f"[audit] log write failed: {e}")

task_audit = TaskAuditListener()   # module-level instantiation; effective on import

# main.py —— import the listener in business code to activate
from crewai import Agent, Task, Crew, LLM
import audit_listener   # noqa: F401  import registers; no explicit usage needed

llm = LLM(
    model="openai/deepseek-chat",
    base_url="https://api.deepseek.com/v1",
    api_key=os.getenv("DEEPSEEK_API_KEY"),
)
writer = Agent(role="Writer", goal="Write summaries", backstory="Professional writer", llm=llm)
crew = Crew(
    agents=[writer],
    tasks=[Task(description="Write a 50-word summary of the following product copy.",
                expected_output="50-word summary.", agent=writer)],
)
crew.kickoff()   # after execution, check audit_log.jsonl for audit records

With multiple listeners, the official recommendation is a listeners/ package: each listener module creates an instance at the bottom; __init__.py re-exports them; the business file只需 import my_project.listeners one line.

14.5 Structured Logs and Scoped Temporary Listeners

Production logs should be structured (JSON lines) rather than print: fixed fields, machine-parseable,可 directly feed to ELK/Loki. The JSONL pattern above is the minimal implementation. One step further: include event.type and event.timestamp in records, and split by event family into separate files (llm.jsonl / tool.jsonl / task.jsonl).

When debugging a specific flow, use the official scoped_handlers context manager to register temporary listeners, auto-removed on exit:

from crewai.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 exists only within this with block")

    crew.kickoff()   # events during this window are captured by the temporary handler
# after exiting the with block, temp_handler is removed

14.6 Callbacks vs Event Listeners: How to Choose

• One-off,局部 observation (persist a specific Task's result) → use Task callback, code is近 and直观;
• Global, cross-Crew cross-cutting concern (auditing, monitoring, billing, notifications) → use event listeners, zero侵入,可 uniformly switched;
• They are not mutually exclusive: mature projects通常 let callbacks handle "business actions" and events handle "platform observability."

Chapter Summary

• step_callback fires per-step (agent-level overrides crew-level); Task(callback=...) fires on task completion;
• Event system three-piece: CrewAIEventsBus singleton bus, BaseEvent base class, BaseEventListener listener base class;
• Listeners must be instantiated and imported to take effect; handler signature is uniformly (source, event);
• TaskCompletedEvent + JSONL persistence is the minimal可用 audit方案; scoped_handlers适合临时调试;
• Audit/logging code must have its own exception protection — never let the observability system crash the business execution.

🧪 随堂测验

点击你认为正确的选项。答错时会展示正确答案与原因解析。

1. Regarding the division of labor between step_callback and Task callback (callback/task_callback), which is correct?

2. You defined a BaseEventListener subclass but the listener has no effect at runtime. Most likely cause?

3. What is the unified signature for event handler functions?

4. You want to temporarily listen to events while debugging a specific flow. Official recommended approach?

🛠️ Hands-On Exercises

1. Add a ToolUsageStartedEvent / ToolUsageFinishedEvent listener to the Chapter 13 crew; count each tool's invocation次数 and total latency; output as tool_stats.json.
2. Extend the TaskAuditListener from this chapter to also listen for TaskFailedEvent and CrewKickoffFailedEvent; write failure records to errors.jsonl with an event type field.
3. Write a调试 script using scoped_handlers: temporarily listen for LLMCallStartedEvent, run a two-task crew, and count the LLM calls发起 during this execution.

With observability mastered, next enters CrewAI's other pillar: Chapter 15 · Flow入门: Event-Driven Workflows.