datarobot_genai.crewai.telemetry

telemetry

Idempotent CrewAI auto-instrumentation for agent telemetry.

The released opentelemetry-instrumentation-crewai only instruments CrewAI's synchronous execution path (kickoff -> Agent.execute_task -> Task.execute_sync -> LLM.call). CrewAI's native async path (akickoff -> Agent.aexecute_task -> Task.aexecute_sync -> LLM.acall) is uninstrumented, so agents driven via akickoff emit no framework spans.

Until async support lands upstream, :class:DataRobotCrewAIInstrumentor adds the async wrappers on top of the released synchronous instrumentor, reusing the released module's span helpers so span shape stays identical across both paths.

TODO (BUZZOK-31424): remove :class:DataRobotCrewAIInstrumentor and the async wrappers once opentelemetry-instrumentation-crewai ships native akickoff instrumentation, and revert to instantiating CrewAIInstrumentor directly.

https://github.com/traceloop/openllmetry/pull/4342

DataRobotCrewAIInstrumentor

Bases: CrewAIInstrumentor

CrewAIInstrumentor extended with async execution-path (akickoff) wrappers.

The synchronous wrapping is delegated to the released base class; this subclass adds wrappers for CrewAI's native async methods. Each async method is wrapped defensively so instrumentation still succeeds on CrewAI releases that predate a given method.

Source code in datarobot_genai/crewai/telemetry.py

class DataRobotCrewAIInstrumentor(CrewAIInstrumentor):
    """CrewAIInstrumentor extended with async execution-path (akickoff) wrappers.

    The synchronous wrapping is delegated to the released base class; this
    subclass adds wrappers for CrewAI's native async methods. Each async method
    is wrapped defensively so instrumentation still succeeds on CrewAI releases
    that predate a given method.
    """

    def _instrument(self, **kwargs: Any) -> None:
        super()._instrument(**kwargs)

        tracer = get_tracer(_INSTRUMENTATION_NAME, __version__, kwargs.get("tracer_provider"))
        meter = get_meter(_INSTRUMENTATION_NAME, __version__, kwargs.get("meter_provider"))
        token_histogram = None
        duration_histogram = None
        if is_metrics_enabled():
            token_histogram, duration_histogram = _create_metrics(meter)

        for module, method, factory in _ASYNC_WRAP_TARGETS:
            try:
                wrap_function_wrapper(
                    module, method, factory(tracer, duration_histogram, token_histogram)
                )
            except (AttributeError, ModuleNotFoundError):
                logger.debug("CrewAI async method %s.%s not found; skipping", module, method)

    def _uninstrument(self, **kwargs: Any) -> None:
        super()._uninstrument(**kwargs)
        for module, method in _ASYNC_UNWRAP_TARGETS:
            try:
                unwrap(module, method)
            except (AttributeError, ModuleNotFoundError):
                pass

instrument

instrument() -> None

Idempotently enable CrewAI instrumentation, including the async path.

Source code in datarobot_genai/crewai/telemetry.py

def instrument() -> None:
    """Idempotently enable CrewAI instrumentation, including the async path."""
    if _INSTRUMENTED["crewai"]:
        logger.info("CrewAI instrumentation already enabled")
        return
    try:
        DataRobotCrewAIInstrumentor().instrument()
        os.environ.setdefault("CREWAI_TESTING", "true")
        _INSTRUMENTED["crewai"] = True
    except Exception as e:
        logger.info(f"CrewAI instrumentation failed: {e}")