add all unit/contract tests

Author: liustve

.coveragerc-py39

@@ -0,0 +1,8 @@
+[run]
+omit =
+    */instrumentation/crewai/*
+
+[report]
+include_namespace_packages = true
+fail_under = 95.00
+precision = 2

CHANGELOG.md

@@ -12,6 +12,8 @@ If your change does not need a CHANGELOG entry, add the "skip changelog" label t
 
 ## Unreleased
 
+- Add native CrewAI instrumentation support
+  ([#586](https://github.com/aws-observability/aws-otel-python-instrumentation/pull/586))
 - Fix: Support new fields in X-Ray API responses
   ([#577](https://github.com/aws-observability/aws-otel-python-instrumentation/pull/577))
 - Sign Lambda layer by AWS Signer

aws-opentelemetry-distro/src/amazon/opentelemetry/distro/aws_opentelemetry_configurator.py

@@ -897,4 +897,4 @@ def _create_aws_otlp_exporter(endpoint: str, service: str, region: str):
     # pylint: disable=broad-exception-caught
     except Exception as errors:
         _logger.error("Failed to create AWS OTLP exporter: %s", errors)
-        return None
+        return None

aws-opentelemetry-distro/src/amazon/opentelemetry/distro/instrumentation/crewai/__init__.py

@@ -24,18 +24,20 @@ class CrewAIInstrumentor(BaseInstrumentor):
     Note: Semantic conventions may change in future versions.
     """
 
-    def instrumentation_dependencies(self) -> Collection[str]:
-        return ("crewai >= 0.119.0",)
+    def instrumentation_dependencies(self) -> Collection[str]:  # pylint: disable=no-self-use
+        return ("crewai >= 0.41.0",)
 
-    def _instrument(self, **kwargs: Any) -> None:
+    # disabling these linters rules as these are instance methods from BaseInstrumentor
+    def _instrument(self, **kwargs: Any) -> None:  # pylint: disable=no-self-use
         tracer_provider = kwargs.get("tracer_provider") or trace.get_tracer_provider()
         tracer = trace.get_tracer(__name__, __version__, tracer_provider=tracer_provider)
 
         wrap_function_wrapper("crewai", "Crew.kickoff", _CrewKickoffWrapper(tracer))
         wrap_function_wrapper("crewai", "Task._execute_core", _TaskExecuteCoreWrapper(tracer))
         wrap_function_wrapper("crewai.tools.tool_usage", "ToolUsage._use", _ToolUseWrapper(tracer))
 
-    def _uninstrument(self, **kwargs: Any) -> None:
+    def _uninstrument(self, **kwargs: Any) -> None:  # pylint: disable=no-self-use
+        # pylint: disable=import-outside-toplevel
         import crewai
         from crewai.tools import tool_usage
 

aws-opentelemetry-distro/src/amazon/opentelemetry/distro/instrumentation/crewai/_wrappers.py

@@ -5,18 +5,6 @@
 from abc import ABC, abstractmethod
 from typing import TYPE_CHECKING, Any, Callable, Dict, Mapping, Optional, Tuple
 
-from opentelemetry import context
-
-if TYPE_CHECKING:
-    from crewai.agent import Agent
-    from crewai.crew import Crew
-    from crewai.llm import LLM
-    from crewai.task import Task
-    from crewai.tools.structured_tool import CrewStructuredTool
-    from crewai.tools.tool_calling import ToolCalling
-    from crewai.tools.tool_usage import ToolUsage
-    from pydantic import BaseModel
-
 from amazon.opentelemetry.distro.semconv._incubating.attributes.gen_ai_attributes import (
     GEN_AI_AGENT_DESCRIPTION,
     GEN_AI_AGENT_ID,
@@ -28,21 +16,33 @@
     GEN_AI_REQUEST_TEMPERATURE,
     GEN_AI_SYSTEM_INSTRUCTIONS,
     GEN_AI_TOOL_CALL_ARGUMENTS,
-    GEN_AI_TOOL_CALL_ID,
+    GEN_AI_TOOL_CALL_RESULT,
     GEN_AI_TOOL_DEFINITIONS,
     GEN_AI_TOOL_DESCRIPTION,
     GEN_AI_TOOL_NAME,
+    GEN_AI_TOOL_TYPE,
 )
-from opentelemetry import trace
+from opentelemetry import context, trace
+from opentelemetry.semconv._incubating.attributes.error_attributes import ERROR_TYPE
 from opentelemetry.trace import SpanKind, Status, StatusCode
 
+if TYPE_CHECKING:
+    from crewai.agent import Agent
+    from crewai.crew import Crew
+    from crewai.llm import LLM
+    from crewai.task import Task
+    from crewai.tools.structured_tool import CrewStructuredTool
+    from crewai.tools.tool_calling import ToolCalling
+    from crewai.tools.tool_usage import ToolUsage
+    from pydantic import BaseModel
+
 _OPERATION_INVOKE_AGENT = "invoke_agent"
 _OPERATION_EXECUTE_TOOL = "execute_tool"
-# default value for gen_ai.provider.name, a required attribute per OpenTelemetry semantic conventions.
+# default value for gen_ai.provider.name, a required attribute per OpenTelemetry
+# semantic conventions.
 # "crewai" is not a standard provider name in semconv v1.39, but serves as a fallback when the
 # underlying LLM provider cannot be determined.
 _PROVIDER_CREWAI = "crewai"
-_ERROR_TYPE = "error.type"
 
 
 class _BaseWrapper(ABC):
@@ -70,6 +70,32 @@ class _BaseWrapper(ABC):
     def __init__(self, tracer: Optional[trace.Tracer] = None) -> None:
         self._tracer = tracer or trace.get_tracer(__name__)
 
+    def __call__(
+        self,
+        wrapped: Callable[..., Any],
+        instance: Any,
+        args: Tuple[Any, ...],
+        kwargs: Mapping[str, Any],
+    ) -> Any:
+        if context.get_value(context._SUPPRESS_INSTRUMENTATION_KEY):
+            return wrapped(*args, **kwargs)
+
+        with self._tracer.start_as_current_span(
+            self._get_span_name(instance, args, kwargs),
+            kind=SpanKind.INTERNAL,
+            attributes=self._get_attributes(instance, args, kwargs),
+        ) as span:
+            try:
+                result = wrapped(*args, **kwargs)
+                self._on_success(span, result)
+                span.set_status(Status(StatusCode.OK))
+                return result
+            except Exception as exc:  # pylint: disable=broad-exception-caught
+                span.set_status(Status(StatusCode.ERROR, str(exc)))
+                span.set_attribute(ERROR_TYPE, type(exc).__name__)
+                span.record_exception(exc)
+                raise
+
     def _extract_provider_and_model(self, llm: Optional["LLM"]) -> Tuple[Optional[str], Optional[str]]:
         # extracts provider name and model from CrewAI LLM object
         if not llm:
@@ -93,7 +119,8 @@ def _extract_provider_and_model(self, llm: Optional["LLM"]) -> Tuple[Optional[st
 
         return None, model
 
-    def _serialize_to_json(self, value: Any, max_depth: int = 10) -> str:
+    @staticmethod
+    def _serialize_to_json(value: Any, max_depth: int = 10) -> str:
         def _truncate(obj: Any, depth: int) -> Any:
             if depth <= 0:
                 return "..."
@@ -116,49 +143,24 @@ def _get_span_name(self, instance: Any, args: Tuple[Any, ...], kwargs: Mapping[s
     def _get_attributes(self, instance: Any, args: Tuple[Any, ...], kwargs: Mapping[str, Any]) -> Dict[str, Any]:
         pass
 
-    def __call__(
-        self,
-        wrapped: Callable[..., Any],
-        instance: Any,
-        args: Tuple[Any, ...],
-        kwargs: Mapping[str, Any],
-    ) -> Any:
-        if context.get_value(context._SUPPRESS_INSTRUMENTATION_KEY):
-            return wrapped(*args, **kwargs)
-
-        with self._tracer.start_as_current_span(
-            self._get_span_name(instance, args, kwargs),
-            kind=SpanKind.INTERNAL,
-            attributes=self._get_attributes(instance, args, kwargs),
-        ) as span:
-            try:
-                result = wrapped(*args, **kwargs)
-                span.set_status(Status(StatusCode.OK))
-                return result
-            except Exception as e:
-                span.set_status(Status(StatusCode.ERROR, str(e)))
-                span.set_attribute(_ERROR_TYPE, type(e).__name__)
-                span.record_exception(e)
-                raise
+    def _on_success(self, span: trace.Span, result: Any) -> None:
+        """Hook called on successful execution."""
 
 
 class _CrewKickoffWrapper(_BaseWrapper):
-    # wraps Crew.kickoff which is responsible for starting the entire agentic workflow.
-    # see: https://github.com/crewAIInc/crewAI/blob/main/src/crewai/crew.py
+    # wraps Crew.kickoff which is responsible for starting the agentic workflow.
+    # see:
+    # https://github.com/crewAIInc/crewAI/blob/06d953bf46c636ff9f2d64f45574493d05fb7771/lib/crewai/src/crewai/crew.py#L676-L679
     # Note: The span name "crew_kickoff {crew_name}" does not conform to any current OTel semantic
     # conventions. This is because CrewAI's orchestration workflow where a Crew can contain multiple
     # agents but there currently does not exist any semantic convention naming schema to capture
     # this architecture.
 
-    def _get_span_name(
-        self, instance: "Crew", args: Tuple[Any, ...], kwargs: Mapping[str, Any]
-    ) -> str:
+    def _get_span_name(self, instance: "Crew", args: Tuple[Any, ...], kwargs: Mapping[str, Any]) -> str:
         crew_name = getattr(instance, "name", None)
         return f"crew_kickoff {crew_name}" if crew_name else "crew_kickoff"
 
-    def _get_attributes(
-        self, instance: "Crew", args: Tuple[Any, ...], kwargs: Mapping[str, Any]
-    ) -> Dict[str, Any]:
+    def _get_attributes(self, instance: "Crew", args: Tuple[Any, ...], kwargs: Mapping[str, Any]) -> Dict[str, Any]:
         attributes: Dict[str, Any] = {
             GEN_AI_OPERATION_NAME: _OPERATION_INVOKE_AGENT,
             GEN_AI_PROVIDER_NAME: _PROVIDER_CREWAI,
@@ -190,7 +192,8 @@ def _get_attributes(
 
         return attributes
 
-    def _extract_tool_definitions(self, tools: Any) -> list:
+    @staticmethod
+    def _extract_tool_definitions(tools: Any) -> list:
         defs = []
         for tool in tools:
             tool_def: Dict[str, Any] = {"type": "function"}
@@ -202,26 +205,24 @@ def _extract_tool_definitions(self, tools: Any) -> list:
             if args_schema is not None:
                 try:
                     tool_def["parameters"] = args_schema.model_json_schema()
-                except Exception:
+                except Exception:  # pylint: disable=broad-exception-caught
                     pass
             defs.append(tool_def)
         return defs
 
 
 class _TaskExecuteCoreWrapper(_BaseWrapper):
-    # wraps Task._execute_core which is responsible for running a single task with its assigned agent.
-    # see: https://github.com/crewAIInc/crewAI/blob/main/src/crewai/task.py
+    # wraps Task._execute_core which is responsible for running a single task
+    # with its assigned agent.
+    # see:
+    # https://github.com/crewAIInc/crewAI/blob/06d953bf46c636ff9f2d64f45574493d05fb7771/lib/crewai/src/crewai/task.py#L604-L608
 
-    def _get_span_name(
-        self, instance: "Task", args: Tuple[Any, ...], kwargs: Mapping[str, Any]
-    ) -> str:
+    def _get_span_name(self, instance: "Task", args: Tuple[Any, ...], kwargs: Mapping[str, Any]) -> str:
         agent: Optional[Agent] = args[0] if args else kwargs.get("agent")
         agent_role = getattr(agent, "role", None) if agent else None
         return f"{_OPERATION_INVOKE_AGENT} {agent_role}" if agent_role else _OPERATION_INVOKE_AGENT
 
-    def _get_attributes(
-        self, instance: "Task", args: Tuple[Any, ...], kwargs: Mapping[str, Any]
-    ) -> Dict[str, Any]:
+    def _get_attributes(self, instance: "Task", args: Tuple[Any, ...], kwargs: Mapping[str, Any]) -> Dict[str, Any]:
         agent: Optional[Agent] = args[0] if args else kwargs.get("agent")
         attributes: Dict[str, Any] = {
             GEN_AI_OPERATION_NAME: _OPERATION_INVOKE_AGENT,
@@ -263,11 +264,10 @@ def _get_attributes(
 
 class _ToolUseWrapper(_BaseWrapper):
     # Wraps ToolUsage._use which executes a tool call during agent task execution.
-    # see: https://github.com/crewAIInc/crewAI/blob/main/src/crewai/tools/tool_usage.py
+    # see:
+    # https://github.com/crewAIInc/crewAI/blob/06d953bf46c636ff9f2d64f45574493d05fb7771/lib/crewai/src/crewai/tools/tool_usage.py#L423-L427
 
-    def _get_span_name(
-        self, instance: "ToolUsage", args: Tuple[Any, ...], kwargs: Mapping[str, Any]
-    ) -> str:
+    def _get_span_name(self, instance: "ToolUsage", args: Tuple[Any, ...], kwargs: Mapping[str, Any]) -> str:
         tool: Optional[CrewStructuredTool] = args[1] if len(args) > 1 else kwargs.get("tool")
         tool_name = getattr(tool, "name", None) if tool else None
         return f"{_OPERATION_EXECUTE_TOOL} {tool_name}" if tool_name else _OPERATION_EXECUTE_TOOL
@@ -280,6 +280,7 @@ def _get_attributes(
         attributes: Dict[str, Any] = {
             GEN_AI_OPERATION_NAME: _OPERATION_EXECUTE_TOOL,
             GEN_AI_PROVIDER_NAME: _PROVIDER_CREWAI,
+            GEN_AI_TOOL_TYPE: "function",
         }
 
         if tool:
@@ -291,9 +292,6 @@ def _get_attributes(
                 attributes[GEN_AI_TOOL_DESCRIPTION] = tool_desc
 
         if calling:
-            call_id = getattr(calling, "id", None)
-            if call_id:
-                attributes[GEN_AI_TOOL_CALL_ID] = str(call_id)
             call_args = getattr(calling, "arguments", None)
             if call_args:
                 attributes[GEN_AI_TOOL_CALL_ARGUMENTS] = self._serialize_to_json(call_args)
@@ -308,3 +306,7 @@ def _get_attributes(
                 attributes[GEN_AI_REQUEST_MODEL] = model
 
         return attributes
+
+    def _on_success(self, span: trace.Span, result: Any) -> None:
+        if result is not None:
+            span.set_attribute(GEN_AI_TOOL_CALL_RESULT, self._serialize_to_json(result))

aws-opentelemetry-distro/src/amazon/opentelemetry/distro/semconv/_incubating/attributes/gen_ai_attributes.py

@@ -25,8 +25,7 @@
 GEN_AI_TOOL_DESCRIPTION = "gen_ai.tool.description"
 GEN_AI_TOOL_CALL_ID = "gen_ai.tool.call.id"
 GEN_AI_TOOL_CALL_ARGUMENTS = "gen_ai.tool.call.arguments"
+GEN_AI_TOOL_CALL_RESULT = "gen_ai.tool.call.result"
 GEN_AI_TOOL_DEFINITIONS = "gen_ai.tool.definitions"
+GEN_AI_TOOL_TYPE = "gen_ai.tool.type"
 GEN_AI_SYSTEM_INSTRUCTIONS = "gen_ai.system_instructions"
-
-# Deprecated - use GEN_AI_PROVIDER_NAME instead
-GEN_AI_SYSTEM = "gen_ai.system"