From 7c1950b547b74cceaff6290d9097365083dccda9 Mon Sep 17 00:00:00 2001 From: Pierre-Yves Le Borgne Date: Wed, 28 Oct 2026 11:10:28 +0000 Subject: [PATCH 1/2] fix(crewai): trace tools run by native function calling When the LLM supports function calling, as crewAI's native providers do, crewAI sends the tools to the model and runs the returned tool calls through BaseTool.run (AgentExecutor's _execute_single_native_tool_call, for one), not through ToolUsage._use. The instrumentation only wrapped ToolUsage._use, so those tools ran without a TOOL span. Also wrap BaseTool.run, and Tool.run, which overrides it for @tool functions, with the same execute_tool span: same name, attributes and parent as the ReAct path. ToolUsage._use calls the tool's _run through its CrewStructuredTool, not run, but a run of the same tool inside a traced _use is skipped anyway, so a call never gets two spans. crewAI runs parallel tool calls in copied contexts, so their spans keep the agent span as parent. Co-Authored-By: Claude Sonnet 5.5 Claude-Session: https://claude.ai/code/session_01GwUvhmzgfkiZ6vo4uGyesN --- .../CHANGELOG.md | 4 + .../instrumentation/crewai/__init__.py | 97 +++++- .../tests/test_native_tool_calls.py | 311 ++++++++++++++++++ 3 files changed, 410 insertions(+), 2 deletions(-) create mode 100644 instrumentation-loongsuite/loongsuite-instrumentation-crewai/tests/test_native_tool_calls.py diff --git a/instrumentation-loongsuite/loongsuite-instrumentation-crewai/CHANGELOG.md b/instrumentation-loongsuite/loongsuite-instrumentation-crewai/CHANGELOG.md index 69b06d97a..6d4f6e8dd 100644 --- a/instrumentation-loongsuite/loongsuite-instrumentation-crewai/CHANGELOG.md +++ b/instrumentation-loongsuite/loongsuite-instrumentation-crewai/CHANGELOG.md @@ -9,6 +9,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- Trace the tools that crewAI calls by native function calling: it runs them + through `BaseTool.run` instead of `ToolUsage._use`, so they had no TOOL + span. + ([#292](https://github.com/alibaba/loongsuite-python/pull/292)) - Keep incidental provider requests out of the CrewAI VCR cassettes so a first-use model-metadata fetch cannot fail the recorded tests. ([#270](https://github.com/alibaba/loongsuite-python/pull/270)) diff --git a/instrumentation-loongsuite/loongsuite-instrumentation-crewai/src/opentelemetry/instrumentation/crewai/__init__.py b/instrumentation-loongsuite/loongsuite-instrumentation-crewai/src/opentelemetry/instrumentation/crewai/__init__.py index 266919452..076a2facd 100644 --- a/instrumentation-loongsuite/loongsuite-instrumentation-crewai/src/opentelemetry/instrumentation/crewai/__init__.py +++ b/instrumentation-loongsuite/loongsuite-instrumentation-crewai/src/opentelemetry/instrumentation/crewai/__init__.py @@ -46,6 +46,8 @@ logger = logging.getLogger(__name__) _ENTRY_DEPTH: ContextVar[int] = ContextVar("crewai_entry_depth", default=0) _TASK_DEPTH: ContextVar[int] = ContextVar("crewai_task_depth", default=0) +# The tool whose execute_tool span is open in this context, if any. +_ACTIVE_TOOL: ContextVar[Any] = ContextVar("crewai_active_tool", default=None) _CREWAI_UNINSTRUMENT_TARGETS = ( ("crewai.crew", "Crew", "kickoff"), ("crewai.crew", "Crew", "kickoff_async"), @@ -54,6 +56,10 @@ ("crewai.agent", "Agent", "execute_task"), ("crewai.task", "Task", "execute_sync"), ("crewai.tools.tool_usage", "ToolUsage", "_use"), + # BaseTool first: unwrapping Tool.run, when Tool inherits it, is then a + # no-op. + ("crewai.tools.base_tool", "BaseTool", "run"), + ("crewai.tools.base_tool", "Tool", "run"), ) @@ -406,6 +412,31 @@ def _fail_tool( ) +def _is_active_tool(tool: Any) -> bool: + """Whether an execute_tool span for ``tool`` is already open here. + + ``ToolUsage._use`` gets the ``CrewStructuredTool`` built from a + ``BaseTool``, so the structured tool's ``_original_tool`` counts too. + """ + active = _ACTIVE_TOOL.get() + return active is not None and ( + active is tool or getattr(active, "_original_tool", None) is tool + ) + + +def _tool_class_overrides_run() -> bool: + """Whether crewAI's ``Tool`` (behind ``@tool``) has its own ``run``. + + Older releases inherit ``BaseTool.run``; wrapping it again there would + leave a stale wrapper behind on uninstrumentation. + """ + try: + tool_class = import_module("crewai.tools.base_tool").Tool + except (ImportError, AttributeError): + return False + return "run" in vars(tool_class) + + def _wrap_stream_result( result: Any, on_success: Any, @@ -595,7 +626,7 @@ def _instrument(self, **kwargs: Any) -> None: logger_provider=logger_provider, ) - for module, name, wrapper in ( + wrappers = [ ( "crewai.crew", "Crew.kickoff", @@ -631,7 +662,22 @@ def _instrument(self, **kwargs: Any) -> None: "ToolUsage._use", _ToolUseWrapper(self._handler), ), - ): + # Native function calls run tools through ``run``, not ``_use``. + ( + "crewai.tools.base_tool", + "BaseTool.run", + _ToolRunWrapper(self._handler), + ), + ] + if _tool_class_overrides_run(): + wrappers.append( + ( + "crewai.tools.base_tool", + "Tool.run", + _ToolRunWrapper(self._handler), + ) + ) + for module, name, wrapper in wrappers: try: wrap_function_wrapper( module=module, name=name, wrapper=wrapper @@ -824,11 +870,14 @@ def __call__(self, wrapped: Any, instance: Any, args: Any, kwargs: Any): started = _safe_handler_call( "start_execute_tool", self._handler.start_execute_tool, invocation ) + token = _ACTIVE_TOOL.set(tool) try: result = wrapped(*args, **kwargs) except Exception as exc: _fail_tool(self._handler, invocation, started, exc) raise + finally: + _ACTIVE_TOOL.reset(token) _finish_tool_success( self._handler, @@ -841,6 +890,49 @@ def __call__(self, wrapped: Any, instance: Any, args: Any, kwargs: Any): return result +class _ToolRunWrapper: + """Wrap ``BaseTool.run`` as a util-genai execute_tool span. + + With native function calling, crewAI runs tools through ``run`` instead + of ``ToolUsage._use``. A ``run`` inside an already traced execution of + the same tool gets no second span. + """ + + def __init__(self, handler: ExtendedTelemetryHandler): + self._handler = handler + + def __call__(self, wrapped: Any, instance: Any, args: Any, kwargs: Any): + if _is_active_tool(instance): + return wrapped(*args, **kwargs) + + invocation = _safe_build_invocation( + "tool", + create_tool_invocation, + instance, + None, + tool_call_arguments=kwargs or (list(args) if args else None), + ) + if invocation is None: + return wrapped(*args, **kwargs) + + started = _safe_handler_call( + "start_execute_tool", self._handler.start_execute_tool, invocation + ) + token = _ACTIVE_TOOL.set(instance) + try: + result = wrapped(*args, **kwargs) + except Exception as exc: + _fail_tool(self._handler, invocation, started, exc) + raise + finally: + _ACTIVE_TOOL.reset(token) + + _finish_tool_success( + self._handler, invocation, started, result, None, instance + ) + return result + + __all__ = [ "__version__", "CrewAIInstrumentor", @@ -851,5 +943,6 @@ def __call__(self, wrapped: Any, instance: Any, args: Any, kwargs: Any): "_FlowKickoffAsyncWrapper", "_AgentExecuteTaskWrapper", "_TaskExecuteSyncWrapper", + "_ToolRunWrapper", "_ToolUseWrapper", ] diff --git a/instrumentation-loongsuite/loongsuite-instrumentation-crewai/tests/test_native_tool_calls.py b/instrumentation-loongsuite/loongsuite-instrumentation-crewai/tests/test_native_tool_calls.py new file mode 100644 index 000000000..157e5ffe5 --- /dev/null +++ b/instrumentation-loongsuite/loongsuite-instrumentation-crewai/tests/test_native_tool_calls.py @@ -0,0 +1,311 @@ +# Copyright The OpenTelemetry Authors +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +""" +Test cases for TOOL spans of tools that CrewAI runs by native function calling. + +When the LLM supports function calling, as CrewAI's native OpenAI provider +does, CrewAI sends the tools to the model and runs the returned tool calls +through ``BaseTool.run`` instead of ``ToolUsage._use``. The OpenAI-compatible +chat completions are mocked with respx, so these tests need no cassette. +""" + +import json +import os +from types import SimpleNamespace + +import httpx +import respx +from crewai import LLM, Agent, Crew, Task +from crewai.tools.base_tool import BaseTool, Tool + +from opentelemetry.instrumentation.crewai import ( + CrewAIInstrumentor, + _ToolUseWrapper, +) +from opentelemetry.test.test_base import TestBase +from opentelemetry.trace import StatusCode +from opentelemetry.util.genai.extended_handler import ExtendedTelemetryHandler + +try: + from opentelemetry.instrumentation._semconv import ( + _OpenTelemetrySemanticConventionStability, + _OpenTelemetryStabilitySignalType, + _StabilityMode, + ) +except ImportError: + _OpenTelemetrySemanticConventionStability = None + _OpenTelemetryStabilitySignalType = None + _StabilityMode = None + +BASE_URL = "https://dashscope.aliyuncs.com/compatible-mode/v1" + + +class WeatherTool(BaseTool): + """Sample tool for weather lookup.""" + + name: str = "get_weather" + description: str = "Get current weather for a location" + + def _run(self, location: str) -> str: + """Execute the tool.""" + return f"Weather in {location}: Sunny, 72°F" + + +class FailingTool(BaseTool): + """Sample tool that always fails.""" + + name: str = "failing_tool" + description: str = "A tool that always fails" + + def _run(self, input_str: str) -> str: + """Execute the tool.""" + raise RuntimeError("Tool execution failed") + + +def chat_completion(content=None, tool_calls=None): + """Return an OpenAI-compatible chat completion body.""" + message = {"role": "assistant", "content": content} + if tool_calls: + message["tool_calls"] = tool_calls + return { + "id": "chatcmpl-native-tools", + "object": "chat.completion", + "created": 0, + "model": "qwen-turbo", + "choices": [ + { + "index": 0, + "finish_reason": "tool_calls" if tool_calls else "stop", + "message": message, + } + ], + "usage": { + "prompt_tokens": 10, + "completion_tokens": 5, + "total_tokens": 15, + }, + } + + +def tool_call(call_id, name, arguments): + """Return an OpenAI-compatible tool call.""" + return { + "id": call_id, + "type": "function", + "function": {"name": name, "arguments": json.dumps(arguments)}, + } + + +class TestNativeToolCalls(TestBase): + """Test TOOL spans of natively called tools.""" + + def setUp(self): + """Setup test resources.""" + super().setUp() + os.environ["CREWAI_TRACING_ENABLED"] = "false" + # Enable experimental mode and content capture for testing + os.environ["OTEL_SEMCONV_STABILITY_OPT_IN"] = ( + "gen_ai_latest_experimental" + ) + os.environ["OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT"] = ( + "SPAN_ONLY" + ) + + if _OpenTelemetrySemanticConventionStability: + try: + _OpenTelemetrySemanticConventionStability._OTEL_SEMCONV_STABILITY_SIGNAL_MAPPING[ + _OpenTelemetryStabilitySignalType.GEN_AI + ] = _StabilityMode.GEN_AI_LATEST_EXPERIMENTAL + except (AttributeError, KeyError): + pass + + self.instrumentor = CrewAIInstrumentor() + self.instrumentor.instrument(tracer_provider=self.tracer_provider) + + # A model that is not a known OpenAI model, on an explicit endpoint, + # gets CrewAI's native OpenAI provider, which calls tools natively. + self.llm = LLM( + model="openai/qwen-turbo", base_url=BASE_URL, api_key="fake-key" + ) + + def tearDown(self): + """Cleanup test resources.""" + with self.disable_logging(): + self.instrumentor.uninstrument() + super().tearDown() + + def kickoff(self, tools, *tool_call_responses): + """Run a one-task crew; the LLM answers with the given tool calls.""" + responses = iter(tool_call_responses) + + def respond(request): + body = next( + responses, chat_completion(content="All tools were used.") + ) + return httpx.Response(200, json=body) + + agent = Agent( + role="Tool User", + goal="Use the tools", + backstory="Agent that uses tools", + verbose=False, + llm=self.llm, + tools=tools, + ) + task = Task( + description="Use the tools", + expected_output="The tool results", + agent=agent, + ) + crew = Crew(agents=[agent], tasks=[task], verbose=False) + with respx.mock(assert_all_called=False) as router: + router.post(f"{BASE_URL}/chat/completions").mock( + side_effect=respond + ) + crew.kickoff() + + return self.memory_exporter.get_finished_spans() + + @staticmethod + def spans_of_kind(spans, kind): + return [ + s for s in spans if s.attributes.get("gen_ai.span.kind") == kind + ] + + def test_native_tool_call_creates_tool_span(self): + """A native tool call gets one TOOL span under the AGENT span.""" + self.assertTrue(self.llm.supports_function_calling()) + + spans = self.kickoff( + [WeatherTool()], + chat_completion( + tool_calls=[ + tool_call( + "call_1", "get_weather", {"location": "San Francisco"} + ) + ] + ), + ) + + tool_spans = self.spans_of_kind(spans, "TOOL") + self.assertEqual(len(tool_spans), 1) + tool_span = tool_spans[0] + self.assertEqual(tool_span.name, "execute_tool get_weather") + self.assertEqual( + tool_span.attributes.get("gen_ai.operation.name"), "execute_tool" + ) + self.assertEqual( + tool_span.attributes.get("gen_ai.crewai.operation"), + "tool.execute", + ) + self.assertEqual( + tool_span.attributes.get("gen_ai.tool.name"), "get_weather" + ) + self.assertIn( + "San Francisco", + tool_span.attributes.get("gen_ai.tool.call.arguments"), + ) + self.assertIn( + "Sunny", tool_span.attributes.get("gen_ai.tool.call.result") + ) + self.assertEqual(tool_span.status.status_code, StatusCode.OK) + + (agent_span,) = self.spans_of_kind(spans, "AGENT") + self.assertEqual(tool_span.parent.span_id, agent_span.context.span_id) + + def test_failing_native_tool_call_creates_error_span(self): + """A native tool call that raises gets a TOOL span with ERROR status.""" + spans = self.kickoff( + [FailingTool()], + chat_completion( + tool_calls=[ + tool_call("call_1", "failing_tool", {"input_str": "x"}) + ] + ), + ) + + tool_spans = self.spans_of_kind(spans, "TOOL") + self.assertEqual(len(tool_spans), 1) + tool_span = tool_spans[0] + self.assertEqual(tool_span.name, "execute_tool failing_tool") + self.assertEqual(tool_span.status.status_code, StatusCode.ERROR) + self.assertIn( + "Tool execution failed", + tool_span.events[0].attributes["exception.message"], + ) + + (agent_span,) = self.spans_of_kind(spans, "AGENT") + self.assertEqual(tool_span.parent.span_id, agent_span.context.span_id) + + def test_parallel_native_tool_calls_create_one_span_each(self): + """Tool calls that CrewAI runs in parallel threads keep the parent.""" + spans = self.kickoff( + [WeatherTool()], + chat_completion( + tool_calls=[ + tool_call("call_1", "get_weather", {"location": "Paris"}), + tool_call("call_2", "get_weather", {"location": "Tokyo"}), + ] + ), + ) + + tool_spans = self.spans_of_kind(spans, "TOOL") + self.assertEqual(len(tool_spans), 2) + results = sorted( + s.attributes.get("gen_ai.tool.call.result") for s in tool_spans + ) + self.assertIn("Paris", results[0]) + self.assertIn("Tokyo", results[1]) + + (agent_span,) = self.spans_of_kind(spans, "AGENT") + for tool_span in tool_spans: + self.assertEqual( + tool_span.parent.span_id, agent_span.context.span_id + ) + + def test_tool_run_inside_traced_tool_use_is_not_traced_again(self): + """``run`` inside a traced ``ToolUsage._use`` of that tool: one span.""" + tool = WeatherTool() + wrapper = _ToolUseWrapper( + ExtendedTelemetryHandler(tracer_provider=self.tracer_provider) + ) + + result = wrapper( + lambda *args, **kwargs: tool.run(location="Paris"), + SimpleNamespace(), + ( + "get_weather", + tool.to_structured_tool(), + SimpleNamespace( + tool_name="get_weather", arguments={"location": "Paris"} + ), + ), + {}, + ) + + self.assertIn("Paris", result) + spans = self.memory_exporter.get_finished_spans() + self.assertEqual(len(self.spans_of_kind(spans, "TOOL")), 1) + + def test_uninstrument_unwraps_tool_run(self): + """Uninstrumenting restores ``BaseTool.run`` and ``Tool.run``.""" + self.assertTrue(hasattr(BaseTool.run, "__wrapped__")) + + self.instrumentor.uninstrument() + + self.assertFalse(hasattr(BaseTool.run, "__wrapped__")) + self.assertFalse(hasattr(Tool.run, "__wrapped__")) + WeatherTool().run(location="Paris") + self.assertEqual(len(self.memory_exporter.get_finished_spans()), 0) From 0a4adedcc1eaaa99c37f26ad9554ca09477e4fca Mon Sep 17 00:00:00 2001 From: Pierre-Yves Le Borgne Date: Thu, 29 Oct 2026 13:00:27 +0000 Subject: [PATCH 2/2] fix(crewai): keep native tool call ids and mark returned ToolFailures Address the review of the native tool spans: - Give each native tool span the model's tool call id. CrewAI runs one native call in AgentExecutor._execute_single_native_tool_call or, for planned steps, agent_utils.execute_single_native_tool_call (as imported by step_executor). Wrap both to keep the id that CrewAI's extract_tool_call_info reads from the call, and use it in the BaseTool.run span. Two parallel calls to one tool no longer share the tool name as their id. Where neither runs (other CrewAI versions, or a direct run), the span keeps the tool name. - Mark a span ERROR when the tool returns a ToolFailure instead of raising: BaseTool.run returns it, ToolUsage._use keeps it in last_failure. Both paths now report it the same way: status description from ToolFailure.as_agent_message(), error.type RuntimeError, no exception event. - Record the result as the text the agent gets (the tool's format_output_for_agent), as ToolUsage._use returns it. A raw result that JSON can't encode, such as a pydantic model or a ToolFailure, made the handler raise before ending the span, so the span was lost and its context stayed attached. - Record both positional and keyword run() arguments. Co-Authored-By: Claude Sonnet 5.5 Claude-Session: https://claude.ai/code/session_01GwUvhmzgfkiZ6vo4uGyesN --- .../CHANGELOG.md | 4 +- .../instrumentation/crewai/__init__.py | 129 +++++++++- .../tests/test_native_tool_calls.py | 228 ++++++++++++++++-- 3 files changed, 331 insertions(+), 30 deletions(-) diff --git a/instrumentation-loongsuite/loongsuite-instrumentation-crewai/CHANGELOG.md b/instrumentation-loongsuite/loongsuite-instrumentation-crewai/CHANGELOG.md index a417f4177..97f8e6eb9 100644 --- a/instrumentation-loongsuite/loongsuite-instrumentation-crewai/CHANGELOG.md +++ b/instrumentation-loongsuite/loongsuite-instrumentation-crewai/CHANGELOG.md @@ -11,7 +11,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Trace the tools that crewAI calls by native function calling: it runs them through `BaseTool.run` instead of `ToolUsage._use`, so they had no TOOL - span. + span. Their spans carry the model's tool call id. A tool that returns a + `ToolFailure` instead of raising now gets an ERROR span, on the ReAct path + too. ([#292](https://github.com/alibaba/loongsuite-python/pull/292)) - Keep incidental provider requests out of the CrewAI VCR cassettes so a first-use model-metadata fetch cannot fail the recorded tests. diff --git a/instrumentation-loongsuite/loongsuite-instrumentation-crewai/src/opentelemetry/instrumentation/crewai/__init__.py b/instrumentation-loongsuite/loongsuite-instrumentation-crewai/src/opentelemetry/instrumentation/crewai/__init__.py index 076a2facd..453386845 100644 --- a/instrumentation-loongsuite/loongsuite-instrumentation-crewai/src/opentelemetry/instrumentation/crewai/__init__.py +++ b/instrumentation-loongsuite/loongsuite-instrumentation-crewai/src/opentelemetry/instrumentation/crewai/__init__.py @@ -48,6 +48,21 @@ _TASK_DEPTH: ContextVar[int] = ContextVar("crewai_task_depth", default=0) # The tool whose execute_tool span is open in this context, if any. _ACTIVE_TOOL: ContextVar[Any] = ContextVar("crewai_active_tool", default=None) +# The model's id of the native tool call that CrewAI runs in this context. +_NATIVE_TOOL_CALL_ID: ContextVar[str | None] = ContextVar( + "crewai_native_tool_call_id", default=None +) +# Where CrewAI runs one native tool call; each takes the model's tool call +# first. Missing ones (other CrewAI versions) are skipped, and the spans of +# their tool calls fall back to the tool name as the call id. +_NATIVE_TOOL_CALL_TARGETS = ( + ( + "crewai.agents.agent_executor", + "AgentExecutor", + "_execute_single_native_tool_call", + ), + ("crewai.agents.step_executor", None, "execute_single_native_tool_call"), +) _CREWAI_UNINSTRUMENT_TARGETS = ( ("crewai.crew", "Crew", "kickoff"), ("crewai.crew", "Crew", "kickoff_async"), @@ -60,6 +75,7 @@ # no-op. ("crewai.tools.base_tool", "BaseTool", "run"), ("crewai.tools.base_tool", "Tool", "run"), + *_NATIVE_TOOL_CALL_TARGETS, ) @@ -125,7 +141,9 @@ def _uninstrument_targets() -> list[tuple[Any, str]]: for module_name, class_name, method_name in _CREWAI_UNINSTRUMENT_TARGETS: try: module = import_module(module_name) - target = getattr(module, class_name) + target = ( + module if class_name is None else getattr(module, class_name) + ) except ImportError: logger.debug( "CrewAI module %s was not available for uninstrumentation.", @@ -365,7 +383,7 @@ def _finish_tool_success( def post_process() -> None: nonlocal error - invocation.tool_call_result = result + invocation.tool_call_result = _tool_result_for_agent(tool, result) if _is_tool_parsing_error(instance): error = RuntimeError( "CrewAI tool parsing attempts exceeded for " @@ -374,8 +392,13 @@ def post_process() -> None: f"{getattr(instance, '_max_parsing_attempts', None)}" ) _record_exception(invocation, error) - else: + return + failure = _returned_tool_failure(result, instance) + if failure is None: _set_ok(invocation) + else: + # Returned, not raised: an error status but no exception event. + error = RuntimeError(failure.as_agent_message()) _safe_post_process("tool success", post_process) if not started: @@ -412,6 +435,76 @@ def _fail_tool( ) +def _tool_result_for_agent(tool: Any, result: Any) -> Any: + """``result`` as the text the agent gets, as ``ToolUsage._use`` returns. + + ``BaseTool.run`` returns the raw result, which need not be JSON + serializable: a pydantic model or a ``ToolFailure``, for instance. + """ + if result is None or isinstance(result, str): + return result + format_output = getattr(tool, "format_output_for_agent", None) + if callable(format_output): + try: + return format_output(result) + except Exception: # pylint: disable=broad-exception-caught + logger.debug( + "Could not format a CrewAI tool result", exc_info=True + ) + return str(result) + + +def _returned_tool_failure(result: Any, instance: Any) -> Any: + """The ``ToolFailure`` that the tool returned instead of raising, if any. + + ``ToolUsage._use`` keeps it in ``last_failure`` and returns the text the + agent gets; ``BaseTool.run`` returns it as is. + """ + try: + tool_failure_type = import_module( + "crewai.tools.tool_failure" + ).ToolFailure + except (ImportError, AttributeError): + return None + for candidate in (getattr(instance, "last_failure", None), result): + if isinstance(candidate, tool_failure_type): + return candidate + return None + + +def _native_tool_call_id(tool_call: Any) -> str | None: + """The model's id of a native tool call, as CrewAI itself reads it.""" + try: + info = import_module( + "crewai.utilities.agent_utils" + ).extract_tool_call_info(tool_call) + except Exception: # pylint: disable=broad-exception-caught + return None + return info[0] if info and isinstance(info[0], str) else None + + +def _wrap_native_tool_call( + wrapped: Any, instance: Any, args: Any, kwargs: Any +) -> Any: + """Keep the model's call id while CrewAI runs one native tool call.""" + del instance + tool_call = args[0] if args else kwargs.get("tool_call") + token = _NATIVE_TOOL_CALL_ID.set(_native_tool_call_id(tool_call)) + try: + return wrapped(*args, **kwargs) + finally: + _NATIVE_TOOL_CALL_ID.reset(token) + + +def _run_arguments(args: tuple[Any, ...], kwargs: dict[str, Any]) -> Any: + """The arguments of a ``BaseTool.run`` call, as recorded on its span.""" + if args and kwargs: + return {"args": list(args), "kwargs": kwargs} + if args: + return list(args) + return kwargs or None + + def _is_active_tool(tool: Any) -> bool: """Whether an execute_tool span for ``tool`` is already open here. @@ -685,6 +778,19 @@ def _instrument(self, **kwargs: Any) -> None: except Exception as exc: logger.warning("Could not wrap %s: %s", name, exc) + for module, class_name, method_name in _NATIVE_TOOL_CALL_TARGETS: + name = ( + method_name + if class_name is None + else f"{class_name}.{method_name}" + ) + try: + wrap_function_wrapper( + module=module, name=name, wrapper=_wrap_native_tool_call + ) + except Exception as exc: # pylint: disable=broad-exception-caught + logger.debug("Could not wrap %s.%s: %s", module, name, exc) + def _uninstrument(self, **kwargs: Any) -> None: del kwargs for target, method_name in _uninstrument_targets(): @@ -894,8 +1000,9 @@ class _ToolRunWrapper: """Wrap ``BaseTool.run`` as a util-genai execute_tool span. With native function calling, crewAI runs tools through ``run`` instead - of ``ToolUsage._use``. A ``run`` inside an already traced execution of - the same tool gets no second span. + of ``ToolUsage._use``. The span takes the model's call id from the native + tool call being run, if any. A ``run`` inside an already traced + execution of the same tool gets no second span. """ def __init__(self, handler: ExtendedTelemetryHandler): @@ -905,12 +1012,13 @@ def __call__(self, wrapped: Any, instance: Any, args: Any, kwargs: Any): if _is_active_tool(instance): return wrapped(*args, **kwargs) + call_id = _NATIVE_TOOL_CALL_ID.get() invocation = _safe_build_invocation( "tool", create_tool_invocation, instance, - None, - tool_call_arguments=kwargs or (list(args) if args else None), + {"id": call_id} if call_id else None, + tool_call_arguments=_run_arguments(args, kwargs), ) if invocation is None: return wrapped(*args, **kwargs) @@ -918,14 +1026,17 @@ def __call__(self, wrapped: Any, instance: Any, args: Any, kwargs: Any): started = _safe_handler_call( "start_execute_tool", self._handler.start_execute_tool, invocation ) - token = _ACTIVE_TOOL.set(instance) + tool_token = _ACTIVE_TOOL.set(instance) + # A tool run from inside this one is not the model's tool call. + call_id_token = _NATIVE_TOOL_CALL_ID.set(None) try: result = wrapped(*args, **kwargs) except Exception as exc: _fail_tool(self._handler, invocation, started, exc) raise finally: - _ACTIVE_TOOL.reset(token) + _NATIVE_TOOL_CALL_ID.reset(call_id_token) + _ACTIVE_TOOL.reset(tool_token) _finish_tool_success( self._handler, invocation, started, result, None, instance diff --git a/instrumentation-loongsuite/loongsuite-instrumentation-crewai/tests/test_native_tool_calls.py b/instrumentation-loongsuite/loongsuite-instrumentation-crewai/tests/test_native_tool_calls.py index 157e5ffe5..74c139f88 100644 --- a/instrumentation-loongsuite/loongsuite-instrumentation-crewai/tests/test_native_tool_calls.py +++ b/instrumentation-loongsuite/loongsuite-instrumentation-crewai/tests/test_native_tool_calls.py @@ -28,10 +28,16 @@ import httpx import respx from crewai import LLM, Agent, Crew, Task +from crewai.agents import step_executor +from crewai.agents.agent_executor import AgentExecutor +from crewai.tools import tool from crewai.tools.base_tool import BaseTool, Tool +from crewai.tools.tool_failure import ToolFailure +from pydantic import BaseModel from opentelemetry.instrumentation.crewai import ( CrewAIInstrumentor, + _run_arguments, _ToolUseWrapper, ) from opentelemetry.test.test_base import TestBase @@ -74,6 +80,43 @@ def _run(self, input_str: str) -> str: raise RuntimeError("Tool execution failed") +class ForecastTool(BaseTool): + """Sample tool that reports a failure instead of raising.""" + + name: str = "get_forecast" + description: str = "Get the weather forecast for a location" + + def _run(self, location: str) -> ToolFailure: + """Execute the tool.""" + return ToolFailure( + message=f"No forecast for {location}", code="unknown_location" + ) + + +class Forecast(BaseModel): + """Structured tool result.""" + + location: str + summary: str + + +class StructuredForecastTool(BaseTool): + """Sample tool that returns a pydantic model.""" + + name: str = "get_structured_forecast" + description: str = "Get a structured weather forecast for a location" + + def _run(self, location: str) -> Forecast: + """Execute the tool.""" + return Forecast(location=location, summary="Sunny") + + +@tool("lookup_capital") +def lookup_capital(country: str) -> str: + """Return the capital city of a country.""" + return {"France": "Paris"}.get(country, "unknown") + + def chat_completion(content=None, tool_calls=None): """Return an OpenAI-compatible chat completion body.""" message = {"role": "assistant", "content": content} @@ -146,14 +189,12 @@ def tearDown(self): self.instrumentor.uninstrument() super().tearDown() - def kickoff(self, tools, *tool_call_responses): - """Run a one-task crew; the LLM answers with the given tool calls.""" - responses = iter(tool_call_responses) + def kickoff(self, tools, *responses, llm=None): + """Run a one-task crew; the LLM answers with ``responses``.""" + replies = iter(responses) def respond(request): - body = next( - responses, chat_completion(content="All tools were used.") - ) + body = next(replies, chat_completion(content="All tools used.")) return httpx.Response(200, json=body) agent = Agent( @@ -161,7 +202,7 @@ def respond(request): goal="Use the tools", backstory="Agent that uses tools", verbose=False, - llm=self.llm, + llm=llm or self.llm, tools=tools, ) task = Task( @@ -213,6 +254,9 @@ def test_native_tool_call_creates_tool_span(self): self.assertEqual( tool_span.attributes.get("gen_ai.tool.name"), "get_weather" ) + self.assertEqual( + tool_span.attributes.get("gen_ai.tool.call.id"), "call_1" + ) self.assertIn( "San Francisco", tool_span.attributes.get("gen_ai.tool.call.arguments"), @@ -250,7 +294,7 @@ def test_failing_native_tool_call_creates_error_span(self): self.assertEqual(tool_span.parent.span_id, agent_span.context.span_id) def test_parallel_native_tool_calls_create_one_span_each(self): - """Tool calls that CrewAI runs in parallel threads keep the parent.""" + """Parallel tool calls keep the parent and the model's call ids.""" spans = self.kickoff( [WeatherTool()], chat_completion( @@ -263,11 +307,15 @@ def test_parallel_native_tool_calls_create_one_span_each(self): tool_spans = self.spans_of_kind(spans, "TOOL") self.assertEqual(len(tool_spans), 2) - results = sorted( - s.attributes.get("gen_ai.tool.call.result") for s in tool_spans - ) - self.assertIn("Paris", results[0]) - self.assertIn("Tokyo", results[1]) + call_ids = { + s.attributes.get("gen_ai.tool.call.id"): s.attributes.get( + "gen_ai.tool.call.result" + ) + for s in tool_spans + } + self.assertEqual(set(call_ids), {"call_1", "call_2"}) + self.assertIn("Paris", call_ids["call_1"]) + self.assertIn("Tokyo", call_ids["call_2"]) (agent_span,) = self.spans_of_kind(spans, "AGENT") for tool_span in tool_spans: @@ -275,19 +323,146 @@ def test_parallel_native_tool_calls_create_one_span_each(self): tool_span.parent.span_id, agent_span.context.span_id ) + def test_native_call_of_decorated_tool_creates_one_span(self): + """A ``@tool`` function, whose ``Tool`` has its own ``run``.""" + spans = self.kickoff( + [lookup_capital], + chat_completion( + tool_calls=[ + tool_call( + "call_1", "lookup_capital", {"country": "France"} + ) + ] + ), + ) + + tool_spans = self.spans_of_kind(spans, "TOOL") + self.assertEqual(len(tool_spans), 1) + tool_span = tool_spans[0] + self.assertEqual(tool_span.name, "execute_tool lookup_capital") + self.assertEqual( + tool_span.attributes.get("gen_ai.tool.call.id"), "call_1" + ) + self.assertEqual( + tool_span.attributes.get("gen_ai.tool.call.result"), "Paris" + ) + + def test_native_tool_returning_tool_failure_creates_error_span(self): + """A ``ToolFailure`` returned instead of raised marks the span ERROR.""" + spans = self.kickoff( + [ForecastTool()], + chat_completion( + tool_calls=[ + tool_call( + "call_1", "get_forecast", {"location": "Atlantis"} + ) + ] + ), + ) + + (tool_span,) = self.spans_of_kind(spans, "TOOL") + self.assert_tool_failure_span(tool_span) + + def test_react_tool_returning_tool_failure_creates_same_error_span(self): + """The ReAct path (``ToolUsage._use``) reports it the same way.""" + react_llm = LLM( + model="openai/qwen-turbo", + base_url=BASE_URL, + api_key="fake-key", + is_litellm=True, + ) + self.assertFalse(react_llm.supports_function_calling()) + + spans = self.kickoff( + [ForecastTool()], + chat_completion( + content=( + "Thought: I need the forecast.\n" + "Action: get_forecast\n" + 'Action Input: {"location": "Atlantis"}' + ) + ), + chat_completion( + content=( + "Thought: I now know the final answer\n" + "Final Answer: There is no forecast for Atlantis." + ) + ), + llm=react_llm, + ) + + (tool_span,) = self.spans_of_kind(spans, "TOOL") + self.assert_tool_failure_span(tool_span) + + def assert_tool_failure_span(self, tool_span): + message = "No forecast for Atlantis (code: unknown_location)" + self.assertEqual(tool_span.name, "execute_tool get_forecast") + self.assertEqual(tool_span.status.status_code, StatusCode.ERROR) + self.assertEqual(tool_span.status.description, message) + self.assertEqual( + tool_span.attributes.get("error.type"), "RuntimeError" + ) + self.assertEqual( + tool_span.attributes.get("gen_ai.tool.call.result"), message + ) + self.assertEqual(len(tool_span.events), 0) + + def test_native_tool_returning_model_records_text_result(self): + """A non-JSON result (a pydantic model) still ends the span.""" + spans = self.kickoff( + [StructuredForecastTool()], + chat_completion( + tool_calls=[ + tool_call( + "call_1", + "get_structured_forecast", + {"location": "Paris"}, + ) + ] + ), + ) + + (tool_span,) = self.spans_of_kind(spans, "TOOL") + self.assertEqual(tool_span.status.status_code, StatusCode.OK) + # The text the agent gets, as CrewAI formats it. + self.assertIn( + "Sunny", tool_span.attributes.get("gen_ai.tool.call.result") + ) + + def test_direct_tool_run_uses_tool_name_as_call_id(self): + """Outside a native tool call there is no model call id.""" + WeatherTool().run(location="Paris") + + (tool_span,) = self.spans_of_kind( + self.memory_exporter.get_finished_spans(), "TOOL" + ) + self.assertEqual( + tool_span.attributes.get("gen_ai.tool.call.id"), "get_weather" + ) + + def test_run_arguments_keep_positional_and_keyword_arguments(self): + """``run`` arguments are recorded without dropping any of them.""" + self.assertEqual( + _run_arguments(("Paris",), {"unit": "C"}), + {"args": ["Paris"], "kwargs": {"unit": "C"}}, + ) + self.assertEqual(_run_arguments(("Paris",), {}), ["Paris"]) + self.assertEqual(_run_arguments((), {"unit": "C"}), {"unit": "C"}) + self.assertIsNone(_run_arguments((), {})) + def test_tool_run_inside_traced_tool_use_is_not_traced_again(self): """``run`` inside a traced ``ToolUsage._use`` of that tool: one span.""" - tool = WeatherTool() + weather_tool = WeatherTool() wrapper = _ToolUseWrapper( ExtendedTelemetryHandler(tracer_provider=self.tracer_provider) ) result = wrapper( - lambda *args, **kwargs: tool.run(location="Paris"), + lambda *args, **kwargs: weather_tool.run(location="Paris"), SimpleNamespace(), ( "get_weather", - tool.to_structured_tool(), + weather_tool.to_structured_tool(), SimpleNamespace( tool_name="get_weather", arguments={"location": "Paris"} ), @@ -300,12 +475,25 @@ def test_tool_run_inside_traced_tool_use_is_not_traced_again(self): self.assertEqual(len(self.spans_of_kind(spans, "TOOL")), 1) def test_uninstrument_unwraps_tool_run(self): - """Uninstrumenting restores ``BaseTool.run`` and ``Tool.run``.""" - self.assertTrue(hasattr(BaseTool.run, "__wrapped__")) + """Uninstrumenting restores the tools' ``run`` and native dispatch.""" + wrapped = ( + BaseTool.run, + Tool.run, + AgentExecutor._execute_single_native_tool_call, + step_executor.execute_single_native_tool_call, + ) + for function in wrapped: + self.assertTrue(hasattr(function, "__wrapped__")) self.instrumentor.uninstrument() - self.assertFalse(hasattr(BaseTool.run, "__wrapped__")) - self.assertFalse(hasattr(Tool.run, "__wrapped__")) + unwrapped = ( + BaseTool.run, + Tool.run, + AgentExecutor._execute_single_native_tool_call, + step_executor.execute_single_native_tool_call, + ) + for function in unwrapped: + self.assertFalse(hasattr(function, "__wrapped__")) WeatherTool().run(location="Paris") self.assertEqual(len(self.memory_exporter.get_finished_spans()), 0)