Tracing and event listeners

CrewAI · Expert · 12 min

You can't fix or trust a multi-agent system whose individual steps you can't see.

Multi-agent runs make many hidden model and tool calls. Observability means recording them so you can answer "why did it say that?" and "where did the money go?".

CrewAI tracing sends a structured trace of each run to CrewAI AMP, the company's hosted platform: agent reasoning, task timeline, tool calls and results, LLM prompts and responses, tokens, timings and errors. Enable it with:

  • Crew(..., tracing=True), or tracing=True when creating a flow, or
  • CREWAI_TRACING_ENABLED=true in the environment for all runs, or crewai traces enable.

Run crewai login first. Traces may contain user data and prompts, so check your data policies before turning them on.

Event listeners let your own code react to what happens inside a run. CrewAI emits typed events on an event bus, such as CrewKickoffCompletedEvent, TaskCompletedEvent, ToolUsageFinishedEvent and LLMCallCompletedEvent. Subclass BaseEventListener, register handlers in setup_listeners, and create one instance before kickoff. Use this for custom logging, metrics or alerts.

Third-party tools. CrewAI integrates with many observability platforms, including Langfuse, Arize Phoenix, MLflow, Datadog, Opik and Weave, mostly through OpenTelemetry (an open standard for traces and metrics).

Separately, CrewAI collects anonymous telemetry about library usage. It is independent of tracing and documented on the Telemetry page, including how to opt out.

Whatever you use, record inputs, outputs, tool calls, tokens and latency per step. Most "the agent is dumb" bugs turn out to be a bad tool result or missing context, visible only in traces.

Example

from crewai.events import (BaseEventListener, CrewKickoffCompletedEvent,
                           TaskCompletedEvent, ToolUsageFinishedEvent)

class CostLogger(BaseEventListener):
    def setup_listeners(self, crewai_event_bus):
        @crewai_event_bus.on(TaskCompletedEvent)
        def on_task(source, event):
            print(f"[task] {event.task_name} done: {str(event.output)[:60]}")

        @crewai_event_bus.on(ToolUsageFinishedEvent)
        def on_tool(source, event):
            print(f"[tool] {event.tool_name}")

        @crewai_event_bus.on(CrewKickoffCompletedEvent)
        def on_done(source, event):
            print(f"[crew] {event.crew_name} finished, tokens: {event.total_tokens}")

cost_logger = CostLogger()   # create once, before any kickoff
# crew = Crew(..., tracing=True)   # also send traces to CrewAI AMP (after `crewai login`)

Remember

  • tracing=True or CREWAI_TRACING_ENABLED=true sends traces to CrewAI AMP
  • Subclass BaseEventListener and use @crewai_event_bus.on(Event)
  • Integrations exist for Langfuse, Phoenix, MLflow, Datadog and more

Watch out for

  • Debugging a wrong final answer by editing prompts without first checking the trace for a failing tool or missing context.

Check yourself

  1. You want to send a metric to your own monitoring system every time any task finishes. What fits best?

Sign in to answer, see the explanations, and get this lesson back as a quick review later.

Official docs: docs.crewai.com/en/observability/tracing