Building observability dashboards for CrewAI crews

A CrewAI observability dashboard turns a black-box multi-agent run into a debuggable timeline of agent steps, token spend, and failures. This tutorial builds one from scratch using CrewAI’s native step callbacks, a SQLite event log, and a Streamlit front end. You’ll see exactly where a crew stalls or burns tokens before your users do.

Prerequisites

• Python 3.11 or newer
• pip install crewai streamlit pandas plotly prometheus_client
• An OpenAI-compatible API key. If you route CrewAI through n4n.ai, its OpenAI-compatible endpoint delivers per-token usage metering and automatic fallback, so your client-side token counter can be replaced by provider-side totals.
• Basic familiarity with CrewAI Agent, Task, and Crew primitives.

Step 1: Persist every crew step

CrewAI invokes step_callback after each agent finishes a task. The passed StepOutput exposes agent_role, task_id, and output. We write these to SQLite. SQLite handles single-writer concurrency well enough for one worker process and avoids standing up Postgres for a first pass.

import sqlite3, time, threading

class CrewTelemetry:
    def __init__(self, db_path="crew_telemetry.db"):
        self.conn = sqlite3.connect(db_path, check_same_thread=False)
        self.lock = threading.Lock()
        with self.lock:
            self.conn.execute("""
                CREATE TABLE IF NOT EXISTS steps (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    run_id TEXT,
                    ts REAL,
                    agent_role TEXT,
                    task_id TEXT,
                    output TEXT
                )
            """)
            self.conn.commit()

    def step_callback(self, step_output, run_id="default"):
        with self.lock:
            self.conn.execute(
                "INSERT INTO steps (run_id, ts, agent_role, task_id, output) VALUES (?,?,?,?,?)",
                (run_id, time.time(),
                 getattr(step_output, "agent_role", "unknown"),
                 getattr(step_output, "task_id", "n/a"),
                 str(step_output.output)[:500])
            )
            self.conn.commit()

Wire it into a crew:

from crewai import Agent, Task, Crew

telemetry = CrewTelemetry()

researcher = Agent(role="Researcher", goal="Summarize LLM gateways",
                   backstory="Infra analyst", allow_delegation=False)
writer = Agent(role="Writer", goal="Draft a short post",
               backstory="Technical writer")

task1 = Task(description="List three inference gateways", agent=researcher)
task2 = Task(description="Write one paragraph comparing them", agent=writer)

crew = Crew(
    agents=[researcher, writer],
    tasks=[task1, task2],
    step_callback=lambda s: telemetry.step_callback(s, "run-1")
)
crew.kickoff()

Checkpoint: after kickoff() returns, run sqlite3 crew_telemetry.db "SELECT agent_role, count(*) FROM steps GROUP BY 1;". You should see two rows—one for Researcher, one for Writer. If you see zero rows, your CrewAI version passes a different attribute name; print dir(step_output) to inspect.

Step 2: Capture token usage without guesswork

StepOutput does not reliably carry token counts across CrewAI releases. Attach a LangChain callback to the underlying LLM instead. This works because CrewAI wraps LangChain chat models.

from langchain_core.callbacks import BaseCallbackHandler

class TokenCounter(BaseCallbackHandler):
    def __init__(self):
        self.tokens = 0
    def on_llm_end(self, response, **kwargs):
        usage = (response.llm_output or {}).get("token_usage", {})
        self.tokens += usage.get("total_tokens", 0)

counter = TokenCounter()
researcher.llm.callbacks = [counter]
writer.llm.callbacks = [counter]

crew.kickoff()
print(f"Total tokens: {counter.tokens}")

Expected output: a printed integer, e.g. Total tokens: 1842. Store this in a runs table alongside run_id and duration. Opinion: never bill from client-side counts. If your gateway already meters per-token usage, pipe those numbers into the same dashboard to avoid drift between what you show and what you pay.

Step 3: Render the CrewAI observability dashboard

Streamlit gives a zero-JS path to a live view. The script below reads the SQLite log, plots a step timeline, and shows a per-agent count bar.

import sqlite3, pandas as pd, streamlit as st, plotly.express as px

st.set_page_config(page_title="CrewAI observability dashboard", layout="wide")
conn = sqlite3.connect("crew_telemetry.db")
df = pd.read_sql("SELECT ts, agent_role, task_id, output FROM steps ORDER BY ts", conn)

st.title("CrewAI observability dashboard")
st.metric("Recorded steps", len(df))

if not df.empty:
    df["ts"] = pd.to_datetime(df["ts"], unit="s")
    col1, col2 = st.columns(2)
    with col1:
        fig = px.scatter(df, x="ts", y="agent_role", color="agent_role",
                         hover_data=["task_id", "output"])
        st.plotly_chart(fig, use_container_width=True)
    with col2:
        counts = df["agent_role"].value_counts().reset_index()
        st.plotly_chart(px.bar(counts, x="agent_role", y="count"),
                        use_container_width=True)
    st.dataframe(df[["ts", "agent_role", "task_id", "output"]])

Run streamlit run dashboard.py. Expected UI: a scatter plot placing Researcher and Writer points on a time axis, a bar chart showing equal step counts, and a table of truncated outputs. This is a minimal but functional CrewAI observability dashboard—enough to spot a crew that never reached the Writer.

Step 4: Expose Prometheus metrics for alerting

Dashboards are for humans; alerts need a pull endpoint. Use prometheus_client to count steps per agent role and expose /metrics.

from prometheus_client import Counter, start_http_server

STEP_COUNTER = Counter("crew_steps_total", "Agent steps", ["agent_role", "run_id"])

def metric_callback(step_output, run_id="default"):
    STEP_COUNTER.labels(getattr(step_output, "agent_role", "unknown"), run_id).inc()
    telemetry.step_callback(step_output, run_id)

start_http_server(8000)
crew = Crew(agents=[researcher, writer], tasks=[task1, task2],
            step_callback=lambda s: metric_callback(s, "run-1"))
crew.kickoff()

Scrape http://localhost:8000/metrics. You will see lines like crew_steps_total{agent_role="Researcher",run_id="run-1"} 1.0. A Grafana alert on increase(crew_steps_total[5m]) == 0 catches a silent crew—the most common production failure mode for multi-agent systems.

Step 5: Harden for production

• Concurrency: SQLite locks on write. Move to Postgres when you run multiple crew processes, or batch inserts behind a queue.
• Run isolation: Always tag rows with run_id (or trace id). The dashboard should filter by run, not show a global soup.
• Errors: Capture getattr(step_output, "error", None) in the callback. Plot error rate per agent; a Writer that throws on every call is a prompt bug, not a model outage.
• Correlation: If you use n4n.ai, forward the x-request-id from response headers into the steps table. That lets you join crew events with gateway-side latency and cache-hit logs.
• Privacy: Truncate output to 500 chars as shown. Full agent outputs belong in object storage, not in a dashboard query.

A CrewAI observability dashboard is only as good as the events you emit. Start with step callbacks, add token and error signals, then graduate to Prometheus when the crew becomes mission-critical. The code above is runnable today; extend the schema as your debugging needs grow.