fix(crewai): CrewAI Flow Traces Not Generated With Async (#2308)

Author: ialisaleh

python/instrumentation/openinference-instrumentation-crewai/examples/advanced_flow.py

@@ -11,6 +11,7 @@
 and display structured and human-readable results.
 """
 
+import asyncio
 import os
 from typing import Any, Dict, List, Optional
 
@@ -186,7 +187,7 @@ def create_advanced_flow(flow_name: Optional[str] = None) -> Flow:
     return flow
 
 
-def run_advanced_flow():
+async def run_advanced_flow():
     """
     Executes the advanced flow and handles any runtime exceptions.
     """
@@ -197,7 +198,7 @@ def run_advanced_flow():
         # Inputs are passed here and injected into the @start() method
         inputs = {"product": "AI-powered Chatbots"}
 
-        flow.kickoff(inputs=inputs)
+        await flow.kickoff_async(inputs=inputs)
         print("✅ Flow execution completed successfully.")
     except Exception as e:
         print(f"⚠️ Flow execution failed: {type(e).__name__}")
@@ -207,7 +208,7 @@ def main():
     """Run the CrewAI instrumentation demonstration."""
     print("CrewAI Instrumentation Demo - Advanced Flow")
     print("Check Phoenix UI at http://localhost:6006 for trace visualization\n")
-    run_advanced_flow()
+    asyncio.run(run_advanced_flow())
 
 
 if __name__ == "__main__":

python/instrumentation/openinference-instrumentation-crewai/src/openinference/instrumentation/crewai/__init__.py

@@ -13,7 +13,7 @@
 from openinference.instrumentation.crewai._wrappers import (
     _CrewKickoffWrapper,
     _ExecuteCoreWrapper,
-    _FlowKickoffWrapper,
+    _FlowKickoffAsyncWrapper,
     _ToolUseWrapper,
 )
 from openinference.instrumentation.crewai.version import __version__
@@ -27,7 +27,7 @@ class CrewAIInstrumentor(BaseInstrumentor):  # type: ignore
     __slots__ = (
         "_original_execute_core",
         "_original_crew_kickoff",
-        "_original_flow_kickoff",
+        "_original_flow_kickoff_async",
         "_original_tool_use",
         "_tracer",
     )
@@ -63,12 +63,14 @@ def _instrument(self, **kwargs: Any) -> None:
             wrapper=crew_kickoff_wrapper,
         )
 
-        flow_kickoff_wrapper = _FlowKickoffWrapper(tracer=self._tracer)
-        self._original_flow_kickoff = getattr(import_module("crewai").Flow, "kickoff", None)
+        flow_kickoff_async_wrapper = _FlowKickoffAsyncWrapper(tracer=self._tracer)
+        self._original_flow_kickoff_async = getattr(
+            import_module("crewai").Flow, "kickoff_async", None
+        )
         wrap_function_wrapper(
             module="crewai",
-            name="Flow.kickoff",
-            wrapper=flow_kickoff_wrapper,
+            name="Flow.kickoff_async",
+            wrapper=flow_kickoff_async_wrapper,
         )
 
         use_wrapper = _ToolUseWrapper(tracer=self._tracer)
@@ -92,10 +94,10 @@ def _uninstrument(self, **kwargs: Any) -> None:
             crew_module.Crew.kickoff = self._original_crew_kickoff
             self._original_crew_kickoff = None
 
-        if self._original_flow_kickoff is not None:
+        if self._original_flow_kickoff_async is not None:
             crew_module = import_module("crewai")
-            crew_module.Flow.kickoff = self._original_flow_kickoff
-            self._original_flow_kickoff = None
+            crew_module.Flow.kickoff_async = self._original_flow_kickoff_async
+            self._original_flow_kickoff_async = None
 
         if self._original_tool_use is not None:
             tool_usage_module = import_module("crewai.tools.tool_usage")

python/instrumentation/openinference-instrumentation-crewai/src/openinference/instrumentation/crewai/_wrappers.py

@@ -327,19 +327,19 @@ def __call__(
         return crew_output
 
 
-class _FlowKickoffWrapper:
+class _FlowKickoffAsyncWrapper:
     def __init__(self, tracer: trace_api.Tracer) -> None:
         self._tracer = tracer
 
-    def __call__(
+    async def __call__(
         self,
         wrapped: Callable[..., Any],
         instance: Any,
         args: Tuple[Any, ...],
         kwargs: Mapping[str, Any],
     ) -> Any:
         if context_api.get_value(context_api._SUPPRESS_INSTRUMENTATION_KEY):
-            return wrapped(*args, **kwargs)
+            return await wrapped(*args, **kwargs)
         # Enhanced flow naming - use meaningful flow name instead of generic "Flow.kickoff"
         flow_name = _get_flow_name(instance)
         span_name = f"{flow_name}.kickoff"
@@ -365,7 +365,7 @@ def __call__(
             span.set_attribute("flow_inputs", json.dumps(inputs) if inputs else "")
 
             try:
-                flow_output = wrapped(*args, **kwargs)
+                flow_output = await wrapped(*args, **kwargs)
             except Exception as exception:
                 span.set_status(trace_api.Status(trace_api.StatusCode.ERROR, str(exception)))
                 span.record_exception(exception)