CrewAI

instrument_crewai() patches Crew.kickoff and Crew.kickoff_async. Every kickoff inside an active run emits one llm_call event with the crew's aggregated usage_metrics (input tokens, cached prompt tokens, output tokens, request count) and latency.

The event reports provider: "crewai", because a single crew can span multiple LLM providers and CrewAI aggregates usage at the crew level. See the advanced note for per-provider Spend.

Install

pip install "agentping-sdk[crewai]"

Supports crewai >= 1.0.0.

Quickstart

import agentping
from crewai import Agent, Crew, Task

agentping.init()
agentping.instrument_crewai()

researcher = Agent(role="Researcher", goal="Find recent industry reports", llm="gpt-4o-mini")
writer = Agent(role="Writer", goal="Write a one-paragraph summary", llm="gpt-4o-mini")

research_task = Task(description="Find 5 recent reports on EV adoption in EU", agent=researcher)
writeup_task = Task(description="Summarise findings in one paragraph", agent=writer)

with agentping.run("market-brief"):
    crew = Crew(agents=[researcher, writer], tasks=[research_task, writeup_task])
    result = crew.kickoff()

instrument_crewai() is idempotent; call it once at module load. If a task raises, the run's context manager records status="failed" and stamps output.error automatically, no extra wiring needed.

What's captured

One llm_call event per Crew.kickoff():

Field Source
provider always crewai
input_tokens / cached_input_tokens usage_metrics.prompt_tokens (cached split out)
output_tokens usage_metrics.completion_tokens
request_count usage_metrics.successful_requests
latency_ms wall-clock kickoff duration

The auto-instrumented event is crew-level only. Per-agent or per-task events are not captured automatically.

Advanced (optional)

For per-provider Spend (one agent on Claude, another on GPT-4o), emit your own llm_call per task from CrewAI's task callback. The crew-level event still fires; this adds the per-provider breakdown:

def on_task(task_output):
    provider = "anthropic" if "claude" in task_output.agent.llm else "openai"
    usage = task_output.token_usage
    r.event("llm_call", {
        "provider": provider,
        "model": task_output.agent.llm.split("/")[-1],
        "input_tokens": usage.prompt_tokens,
        "output_tokens": usage.completion_tokens,
    })

Wire on_task into your Task callbacks and capture the active run r from the surrounding with agentping.run(...) as r: block.

Source