CrewAI workflow telemetry

Test the integration

Send and verify events with CrewAI workflow telemetry

Use CrewAI workflow telemetry where your app knows the final result. Collect only the fields you need, then verify a test event before building charts.

1
Choose the outcome
Multi-agent workflow observability
2
Define the contract
run_id, crew_name, and invocation status
3
Log the final result
Pass your existing Crew and inputs to run_research_crew. This wrapper supports synchronous kickoff only; async methods, streams and resumed checkpoints need their own lifecycle handling.
4
Check the stored event
Exercise a known fixture, then inspect crew_invocation for one correctly typed terminal row.

Before you start

An existing synchronous Crew configured with stream=False and application-owned inputs
A server-side TELEMETRY_API_KEY; this example uses Python standard-library HTTP rather than the Telemetry SDK
A content policy covering task inputs, agent outputs, memory, knowledge, and tool payloads

Delivery setup

Install and initialize server-side

This synchronous Crew wrapper sends events with Python standard-library HTTP; no Telemetry SDK installation or initialization is needed. Keep ingestion credentials out of browser bundles, client-visible environment variables, source control, logs, and exception messages.

1Wrap the existing synchronous Crew with run_research_crew and keep stream=False.
2Log an event when the operation succeeds, fails, retries, or times out.
3Send test events with known results and inspect the stored rows before enabling an alert.

Snippet

Start with one structured event

Add this shape where the workflow completes, fails, or retries. Then build the dashboard from real fields.

crewai

CrewAI workflow telemetry event

python

import json
import os
import sys
from time import perf_counter
from urllib.request import Request, urlopen
from uuid import uuid4


def emit_crew_invocation(event):
    try:
        key = os.environ.get("TELEMETRY_API_KEY")
        if not key:
            return
        request = Request(
            "https://api.telemetry.sh/log",
            data=json.dumps({"table": "crew_invocation", "data": event}).encode("utf-8"),
            headers={"Authorization": key, "Content-Type": "application/json",
                     "User-Agent": "telemetry-crew-example/1"},
            method="POST",
        )
        with urlopen(request, timeout=1) as response:
            if not 200 <= response.status < 300:
                raise RuntimeError("telemetry_delivery_unconfirmed")
    except Exception:
        try:
            print("Telemetry delivery unconfirmed", file=sys.stderr)
        except Exception:
            pass


def run_research_crew(research_crew, inputs):
    # Use an already configured, synchronous, non-streaming Crew.
    if getattr(research_crew, "stream", False):
        raise ValueError("This wrapper requires stream=False")
    run_id = str(uuid4())
    started_at = perf_counter()
    status, error_type = "interrupted", None
    try:
        result = research_crew.kickoff(inputs=inputs)
        status = "returned"
        return result
    except Exception as error:
        status = "failed"
        error_type = "timeout" if isinstance(error, TimeoutError) else "crew_error"
        raise
    finally:
        # Neither metadata creation nor export may replace the crew outcome.
        try:
            emit_crew_invocation({
                "run_id": run_id,
                "crew_name": "research_crew",
                "status": status,
                "error_type": error_type,
                "duration_ms": round((perf_counter() - started_at) * 1000),
                "release": os.environ.get("APP_RELEASE"),
            })
        except Exception:
            pass


# Pass your application's existing Crew and inputs:
# result = run_research_crew(research_crew, inputs)

Event schema

run_id, crew_name, and invocation status

duration_ms, bounded error_type, and release

Check your setup

Checkpoint 1

Pass your existing Crew and inputs to run_research_crew. This wrapper supports synchronous kickoff only; async methods, streams and resumed checkpoints need their own lifecycle handling.

Checkpoint 2

A returned kickoff is not proof of accepted work or downstream delivery. Record reviewer decisions separately; this example does not measure task counts, delegation, retries or cost.

Checkpoint 3

Keep inputs, outputs, memory, knowledge, tool payloads and exception text out of these events.

Checkpoint 4

HTTP export has a one-second socket timeout, not a hard total deadline. Export and diagnostic failures preserve the original result or exception; synchronous delivery can still add latency. No durable queue, retry or shutdown drain is provided.

Checkpoint 5

Use CrewAI's own tracing or callbacks when step-level diagnosis is required, and correlate with an approved run ID.

Verification

Prove the event arrived

Run this after exercising known success and failure cases. Replace the fallback table name if your final event contract differs from the snippet.

crewai-verification

CrewAI workflow telemetry verification query

sql

SELECT *
FROM crew_invocation
ORDER BY timestamp_utc DESC
LIMIT 20;

Confirm one terminal row per logical outcome, with the expected status, identifiers, units, and UTC time.

Inspect the inferred schema and verify that retries do not change field types or generate a new logical event ID.

Search the stored fields for credentials, raw payloads, prompts, private content, and unbounded error messages.

Exercise a provider timeout, ingestion rejection, and process shutdown before treating the dashboard as complete.

Implementation references

Review the event contract, data-safety guidance, and upstream primary documentation before enabling a new production path.

AI agent observability template Nested tool-call SQL Event sampling CrewAI crews CrewAI documentation

Where to log

Keep the outcome event small and recoverable

This pattern provides

Record the outcome as an event you can query with SQL.
Stable fields for dashboards, alerts, and cross-event correlation.
Test events for checking success, failure, retries, and timeouts.

This pattern does not provide

An OTLP exporter, automatic collection pipeline, or replacement for detailed traces and diagnostic logs.
Exactly-once delivery merely because the payload contains an event ID.
Permission to collect raw provider payloads, user content, credentials, or regulated data.

Delivery and idempotency Batching and shutdown Troubleshoot ingestion

Example event schemas

Check what each event records, when to send it, and which field types it needs. Review the example payload and privacy checklist before using it in production.

AI agent operations10 fields