Merge branch 'Arize-ai:main' into alisaleh/ticket-2037

Author: ialisaleh

.release-please-manifest.json

@@ -10,7 +10,7 @@
     "python/openinference-instrumentation": "0.1.41",
     "python/instrumentation/openinference-instrumentation-guardrails": "0.1.10",
     "python/instrumentation/openinference-instrumentation-vertexai": "0.1.11",
-    "python/instrumentation/openinference-instrumentation-crewai": "0.1.14",
+    "python/instrumentation/openinference-instrumentation-crewai": "0.1.15",
     "python/instrumentation/openinference-instrumentation-haystack": "0.1.26",
     "python/instrumentation/openinference-instrumentation-litellm": "0.1.27",
     "python/instrumentation/openinference-instrumentation-groq": "0.1.12",

python/instrumentation/openinference-instrumentation-crewai/CHANGELOG.md

@@ -1,5 +1,12 @@
 # Changelog
 
+## [0.1.15](https://github.com/Arize-ai/openinference/compare/python-openinference-instrumentation-crewai-v0.1.14...python-openinference-instrumentation-crewai-v0.1.15) (2025-10-21)
+
+
+### Bug Fixes
+
+* **crewai:** CrewAI Flow Traces Not Generated With Async ([#2308](https://github.com/Arize-ai/openinference/issues/2308)) ([7e7f532](https://github.com/Arize-ai/openinference/commit/7e7f53283df61a0d78df01ab808bacdddd14620a))
+
 ## [0.1.14](https://github.com/Arize-ai/openinference/compare/python-openinference-instrumentation-crewai-v0.1.13...python-openinference-instrumentation-crewai-v0.1.14) (2025-10-14)
 
 

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

@@ -14,7 +14,7 @@
     _AgentActionWrapper,
     _CrewKickoffWrapper,
     _ExecuteCoreWrapper,
-    _FlowKickoffWrapper,
+    _FlowKickoffAsyncWrapper,
     _ToolUseWrapper,
 )
 from openinference.instrumentation.crewai.version import __version__
@@ -29,7 +29,7 @@ class CrewAIInstrumentor(BaseInstrumentor):  # type: ignore
         "_original_agent_action",
         "_original_execute_core",
         "_original_crew_kickoff",
-        "_original_flow_kickoff",
+        "_original_flow_kickoff_async",
         "_original_tool_use",
         "_tracer",
     )
@@ -77,12 +77,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)
@@ -113,10 +115,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

@@ -412,19 +412,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"
@@ -450,7 +450,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)

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

@@ -1 +1 @@
-__version__ = "0.1.14"
+__version__ = "0.1.15"