Files
crewAI/lib/crewai/tests/utilities/test_agent_utils.py
João Moura 453676c61a
Some checks failed
CodeQL Advanced / Analyze (actions) (push) Has been cancelled
CodeQL Advanced / Analyze (python) (push) Has been cancelled
Check Documentation Broken Links / Check broken links (push) Has been cancelled
Vulnerability Scan / pip-audit (push) Has been cancelled
Nightly Canary Release / Check for new commits (push) Has been cancelled
Nightly Canary Release / Build nightly packages (push) Has been cancelled
Nightly Canary Release / Publish nightly to PyPI (push) Has been cancelled
feat(tools): surface tool failures instead of reporting them as success (#6712)
* feat(tools): surface tool failures instead of reporting them as success

A tool can finish without raising and still fail to do what it was asked.
Slack answers HTTP 200 with `{"ok": false, "error": "channel_not_found"}`;
an MCP server sets `isError`; a CrewAI AMP action returns
`API request failed: ...`. In every case the call "worked", so the error
text reached the agent as an ordinary result, the agent narrated the
problem in prose, and the run was recorded as a success.

Concretely: five failed `slackbot_send_message` calls each rendered as
"Tool Execution Completed", the task passed, and the crew passed -- with
the only evidence being a sentence in the final answer. Nothing
downstream could tell the difference, and an agent that keeps going on a
step that silently did nothing builds the rest of its work on it.

Give that outcome a type and a reaction:

- `ToolFailure` -- what a tool returns instead of an error string. The
  agent still reads prose via `as_agent_message()`, so model behavior is
  unchanged; the framework now knows the call failed.
- `ToolFailurePolicy` -- `ignore` (previous behavior), `warn` (default:
  record + emit, keep going), `raise` (abort with
  `ToolExecutionFailedError`). Resolved most-specific-first: tool, task,
  agent, crew.
- `ToolFailureDetectedEvent` -- emitted before a `raise` aborts, so
  subscribers always observe the failure. `ToolUsageFinishedEvent` also
  carries a `failure` field so a trace UI can mark the call failed
  without correlating two events.
- `tool_failures` on `TaskOutput`, `CrewOutput` and `LiteAgentOutput`,
  plus `has_tool_failures`, so consumers never parse a string.

Detection is strictly declarative -- no string sniffing, so a tool that
legitimately returns text about an error is never misread as failing.
Failures come from a returned `ToolFailure`, a raised exception, MCP
`isError`, a spent `max_usage_count`, or an unknown tool.

Wired into all four tool-execution paths (the ReAct path and the three
native function-calling implementations). Sources updated to report
structurally: `MCPClient.call_tool_result()` preserves `isError` that
`call_tool()` dropped, and `CrewAIPlatformActionTool` returns a
`ToolFailure` for non-2xx and for caught exceptions.

Two latent bugs fixed along the way: `ToolUsage` assumed every agent has
a `fingerprint` (LiteAgent does not), and policy resolution now tolerates
malformed values rather than letting telemetry take down a tool call.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01ETacm2dMASfpMAYUiDu5YG

* fix(tools): address review round 1 on tool-failure signalling

Five real defects from Bugbot, none of them cosmetic.

Tool-scoped policy never applied (high). `resolve_tool_failure_policy`
read `tool_failure_policy` off the object handed to it, but every
execution path passes the `CrewStructuredTool` wrapper, which never
carried the attribute -- and `BaseTool` never declared it in the first
place. A tool-scoped `raise`/`ignore` was silently ignored while the
docs and a unit test claimed otherwise; the test passed only because it
called the resolver directly with an authored tool. Declared the field on
`BaseTool`, propagated it through `to_structured_tool()` and
`CrewStructuredTool`, and made resolution fall back through
`_original_tool` so either shape works.

A failed call still printed the green "Completed" panel, then the red
one. That is the terminal version of the exact bug this PR is about.
Suppressed the success panel when the call reported failure.

A raised tool printed twice: `ToolUsageErrorEvent` already renders a red
panel, and the new failure panel repeated it. The event is still emitted
-- policy and traces need it -- but the duplicate console output is gone.
Both decisions now live in named predicates on `ConsoleFormatter` rather
than inline in the listener closure, so they are directly testable.

Unknown tools were reported on the ReAct path but silently ignored on all
three native paths, so the same miss was loud or silent depending on
executor style. Native paths now record `UNKNOWN_TOOL` too. This also
surfaced a live `NameError`: ruff had pruned `ToolFailureReason` from
`agent_utils` as unused, so the new branch would have crashed at runtime.

`LiteAgentOutput` had `tool_failures` but not `has_tool_failures`, which
the PR promised on all three output types -- an `AttributeError` for any
caller sharing one check across result types.

Testing: 16 further tests, 45 total. Two console tests were passing
vacuously because `emit()` dispatches sync handlers on a thread pool, so
the assertions raced the handler; they now assert on the predicates
directly, and the native-path test drains the bus with `flush()` and
checks the synchronously-written record. Full suite still matches
baseline exactly at 377 pre-existing failures.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01ETacm2dMASfpMAYUiDu5YG

* chore: update tool specifications

* fix(tools): address review round 2 and fix CI type failure

CI caught a type error I should have: widening `agent` to accept a
`LiteAgent` (so a standalone LiteAgent resolves its own policy) left the
declared signatures behind. Widened `execute_tool_and_check_finality`,
its async twin, and `ToolCallHookContext` to `Agent | BaseAgent |
LiteAgent | None`, which is what those actually receive now.

Seven CodeRabbit findings, all verified against the code first:

`raise` was being downgraded by three enclosing handlers. With
`max_execution_time` set, `_execute_with_timeout` wrapped every exception
in `RuntimeError`, so `_check_execution_error` no longer recognized the
passthrough and sent the task through the retry loop instead of aborting.
`StepExecutor.execute` turned it into `StepResult(success=False)` and let
the plan continue. `LiteAgent.kickoff` ran it through
`handle_unknown_error` and printed "This is likely a bug - please report
it" for what is a deliberate, configured stop.

Failure records were dropped on two paths. `reset_tool_failures()` only
ran in `_prepare_task_execution`, so `Agent.kickoff()` / `kickoff_async()`
— which enter through `_prepare_kickoff` — accumulated records across
runs. And a guardrail retry calls `execute_task` again, which resets the
agent, so a tool that failed on a blocked attempt vanished from the final
output entirely: a run could report zero failures having demonstrably
failed one. Failures now accumulate across guardrail attempts.

Writing the tests for that surfaced a further miss of my own:
`Agent.kickoff()` builds its `LiteAgentOutput` in `agent/core.py` via
`AgentExecutor`, not through `LiteAgent`, so `tool_failures` was always
empty there regardless of the recording fix. Wired up, and the LiteAgent
path now reads from whichever agent the executor was handed
(`original_agent` under kickoff, `self` standalone) rather than assuming.

`last_tool_failures` returns a copy, so a caller cannot mutate the
agent's record or watch it shift mid-run.

Testing: 7 further tests, 52 total, covering the timeout wrapper, the
retry limit, kickoff reset, the kickoff output path, copy semantics and
guardrail accumulation. Full suite matches baseline exactly at 377
pre-existing failures; mypy clean on every changed file.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01ETacm2dMASfpMAYUiDu5YG

* fix(tools): make crew-scoped policy real and close the last raise leak

Two findings, and the first was a documented feature that never worked.

`resolve_tool_failure_policy` consulted a crew, and the docs advertised
crew as a scope, but `Crew` had no `tool_failure_policy` field at all --
and even with one it was unreachable, because `BaseAgent` defaulted the
policy to `WARN` rather than `None`, so resolution always stopped at the
agent. Crew-level configuration was silently ignored.

Fixed by making "inherit" the default everywhere instead of baking `warn`
into one layer: `Crew` gains the field, and `BaseAgent`/`LiteAgent`
default to `None` like `Task` and `BaseTool` already did. The resolver
owns the single fallback, so the chain is genuinely
tool > task > agent > crew > warn and the effective default with nothing
configured is still `warn`. Reading `agent.tool_failure_policy` now
returns `None` (meaning "inherit") rather than `WARN`.

The other: `StepExecutor` re-raised `ToolExecutionFailedError` from its
outer handler, but the nested handler around the native-to-text tooling
fallback still caught it and returned `StepResult(success=False)`. An
agent whose LLM lacked native tool calling would therefore not abort
under `raise`. That is the third distinct place this exception was being
downgraded; it now re-raises there too.

Testing: 8 further tests, 60 total, including the full precedence chain
walked one level at a time and crew-scoped `raise`/`ignore` driven
end-to-end through `kickoff()` rather than only through the resolver --
the gap that let the original crew bug pass review. Full suite matches
baseline exactly at 377 pre-existing failures; mypy clean on every
changed file.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01ETacm2dMASfpMAYUiDu5YG

* docs: trim comments and docstrings on tool-failure signalling

Prose only -- no behavior change. Cut the module docstring, the longer
class and method docstrings, the multi-line inline comments, and the
verbose Field descriptions down to what actually earns its place. Net 87
lines lighter.

Kept the "why" in every case where the reason is non-obvious (why the
event fires before a raise, why the policy reads through the tool wrapper,
why the bus needs draining in tests) and dropped the restatements of what
the code already says.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01ETacm2dMASfpMAYUiDu5YG

* fix(tools): make ignore truly silent, stop caching failures, close 4 gaps

Six findings from the latest review round, all verified against the code
before touching it.

`ignore` was not silent. `ToolUsageFinishedEvent.failure` was set before
the policy ran, so traces still saw a failed call under a policy documented
as surfacing nothing. Worse, the console then showed *no* panel at all:
green was suppressed because `failure` was present, red was skipped because
`ignore` never emits `ToolFailureDetectedEvent`. New `reportable_failure()`
resolves the policy before the finished event and drops the flag under
`ignore`; wired into all four execution paths.

Failures were being cached. `CacheHandler.add` stored a `ToolFailure` like
any other result, so a transient error became permanent for the rest of the
run and every later hit re-reported a call that never re-ran. The cache now
refuses to store declared failures -- fixed at the single choke point rather
than at each of the four call sites.

A spent `max_usage_count` was invisible on the shared native path.
`BaseTool._claim_usage` returned a bare string that only the executors
recognising that exact message treated as a failure. It now returns a
`ToolFailure` with `USAGE_LIMIT`, so every path records it.

A guardrail returning a whole `TaskOutput` replaced the output without
carrying accumulated failures over, so earlier attempts vanished from
`CrewOutput.tool_failures`. New `merge_tool_failures()` combines and
deduplicates, and the retry-rebuild path uses it too.

A hook-blocked call inherited a cached failure and attributed it to a call
that never ran. Now cleared. Not reachable through the built-in cache once
failures stop being cached, so the test injects a custom cache handler that
does retain them -- verified to fail without the guard.

Also removed a `datetime` import left unused by the earlier console-test
rewrite.

Testing: 13 further tests, 73 total. Full suite matches baseline exactly at
377 pre-existing failures; the usage-limit suites that `_claim_usage`
touches pass unchanged; mypy clean on every changed file.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01ETacm2dMASfpMAYUiDu5YG

* fix(tools): let raise through the parallel native path, guard all handlers

Chasing down CodeRabbit's note about callers of
execute_single_native_tool_call turned up a fifth place this exception was
being downgraded: the experimental executor's parallel branch wrapped
future.result() in a broad except and folded the abort into a fake tool
result, so the remaining parallel calls carried on. The sequential path and
crew_agent_executor's parallel branch were already fine.

Five separate handlers have swallowed this during review, so added a guard
test asserting the passthrough at every site rather than trusting the next
one gets spotted.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01ETacm2dMASfpMAYUiDu5YG

* fix(tools): keep a failed tool out of the final answer, finish crew scope

Three more findings, all confirmed against the code.

A failed `result_as_answer` tool still became the task's output. The native
paths already excluded raised errors and hook blocks from short-circuiting,
but not declared failures -- so an error message silently became the answer,
which is the exact shape of bug this PR exists to prevent. Fixed on all
paths, and there were three independent override points, not one:
`ToolResult.result_as_answer` in tool_utils, the `execution_result`
finality checks in both executors, and `process_tool_results()`, which
reads `agent.tools_results` back separately. The first two fixes alone left
the behavior unchanged; only the third made the test pass.

`ToolUsage` never received a crew, so a crew-level `ignore` half-applied:
recording and `ToolFailureDetectedEvent` stayed quiet, but the flag was
still attached to `ToolUsageFinishedEvent`. It now takes and stores `crew`.

`CrewAgentExecutor.invoke`/`ainvoke` routed a deliberate stop through
`handle_unknown_error`, printing "An unknown error occurred" on verbose
runs. LiteAgent already special-cased this; both now do.

Testing: 5 further tests, 79 total, including that a *successful*
`result_as_answer` tool still short-circuits. Full suite matches baseline
exactly at 377 pre-existing failures; mypy clean on every changed file.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01ETacm2dMASfpMAYUiDu5YG

* fix(tools): report malformed tool args, correlate the failure event

Two findings from the latest round.

Malformed native tool arguments returned early with a plain error dict and
never reported a failure, so `ToolFailureReason.INVALID_INPUT` was declared
but unreferenced -- a bad tool call was absent from records, events and
`raise` aborts. `parse_tool_call_args` now carries an INVALID_INPUT failure
on the error dict and both executors report it before returning.

`ToolFailureDetectedEvent` never set `agent_id`, so a trace could not tie it
to a specific agent instance. Fixing that exposed the same gap running the
other way: `ToolUsage`'s own started/finished/error events never set
`agent_id` either, so on the ReAct path the paired finished event had
nothing to correlate against. Both now set it.

Set explicitly rather than via `from_agent`, which would also overwrite
`agent_role` and lose the `_original_role` preference those events already
apply -- a behavior change that has nothing to do with correlation.

Testing: 5 further tests, 84 total, asserting the ids match between the
failure event and its paired finished event. One existing test pinned the
exact key set of the parse-error dict and was updated for the new key. Full
suite matches baseline at 377 pre-existing failures; the one apparent
addition was the known `test_trace_enable_disable` order-flake, confirmed by
re-running rather than assumed.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01ETacm2dMASfpMAYUiDu5YG

* fix(tools): scope failure accumulation per execution, drop deprecated executor

Two review requests from @lorenzejay.

Accumulation no longer lives as mutable state on the shared agent. A
ContextVar collector is opened around each execution -- task, kickoff, and
each guardrail retry -- and the output reads that collector directly instead
of copying the agent's list. ContextVars are copied per asyncio task and per
thread, so concurrent executions cannot see each other's records, and
nesting is safe for retries. `last_tool_failures` prefers the active
collector and falls back to the last completed execution, so the accessor is
correct during a run too. The per-execution reset that caused the erasure is
gone.

Reproducing this took some digging and the finding is worth recording: crew
tasks *cannot* hit it, because `AgentExecutor` refuses concurrent reuse of
one instance and raises. `agent.kickoff()` has no such guard, and there the
bug reproduces exactly as reported -- two concurrent kickoffs each returned
two records. The regression test forces the overlap with a barrier so it is
deterministic rather than timing-dependent, and I verified it reports [2, 2]
against the old behavior and [1, 1] now.

Removed the tool-failure integration from `CrewAgentExecutor` entirely; that
file is back to its state on main. Note the shared ReAct helper it calls
still records failures, since that is common code rather than new behavior in
the deprecated file -- so a `raise` policy will be swallowed by that
executor's generic handler. Flagged on the PR rather than papered over.

Testing: 89 total. Two tests I wrote for this were vacuous on the first
attempt -- they passed against the simulated pre-fix code -- so each
concurrency test was checked against the old behavior before being kept.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01ETacm2dMASfpMAYUiDu5YG

* fix(tools): report malformed calls everywhere, drop the unused block reason

Four findings.

`execute_single_native_tool_call` swallowed a JSON decode error into an empty
args dict and ran the tool with no input at all -- worse than not reporting
it. It now routes through `parse_tool_call_args` like the executors do, so
the StepExecutor/planning path reports INVALID_INPUT and returns instead of
executing. That also removes a duplicated inline parse.

The ReAct path returned a `ToolUsageError` message as an ordinary result
without reporting it, so a malformed call there was invisible while the
equivalent native failure was recorded. Now reported as INVALID_INPUT too.

`Agent.kickoff` opened a collector but no longer reset the agent-level list,
so `last_tool_failures` grew across kickoffs. Reset restored, matching task
execution.

`ToolFailureReason.BLOCKED_BY_HOOK` was declared and never produced. Rather
than start reporting hook blocks as failures, the member is removed: a block
is a deliberate decision by the hook author, and treating it as a failure
would make `raise` abort on an intentional veto. Added a guard test that every
remaining reason is actually produced somewhere, so a dead member cannot
reappear -- the same smell that flagged INVALID_INPUT last round.

Also switched the deprecation guard test to a single import style.

Testing: 6 further tests, 95 total, including that the tool does not run when
its args fail to parse. Full suite matches baseline at 377 pre-existing
failures; mypy clean.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01ETacm2dMASfpMAYUiDu5YG

* fix(tools): merge failures across kickoff guardrail retries, cancel siblings

Kickoff guardrail retries discarded the blocked attempt's failures. Each
retry calls `_execute_and_build_output`, which opens a fresh collector and
builds a new output, so a run could report zero failures having demonstrably
failed one -- the same bug already fixed on the task guardrail path, which
merges. Now merged there too. Verified the test fails without the fix.

Under `raise`, one parallel native tool aborting left its siblings running:
the pool waited for them and pending ones still started. It now shuts the
pool down with `cancel_futures=True` so a not-yet-started sibling never runs.
Threads already in flight cannot be interrupted in Python, so a concurrent
tool may still complete before the abort surfaces; that is noted at the call
site rather than left implied.

Also satisfied CodeQL by materialising the enum in the guard test's loop.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01ETacm2dMASfpMAYUiDu5YG

---------

Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
Co-authored-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com>
Co-authored-by: Lorenze Jay <63378463+lorenzejay@users.noreply.github.com>
2026-07-29 19:30:14 -07:00

1344 lines
48 KiB
Python

"""Tests for agent utility functions."""
from __future__ import annotations
import asyncio
import json
from typing import Any, Literal, Optional
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from pydantic import BaseModel, Field
from crewai.hooks.tool_hooks import (
ToolCallHookContext,
clear_after_tool_call_hooks,
clear_before_tool_call_hooks,
register_after_tool_call_hook,
)
from crewai.tools.base_tool import BaseTool
from crewai.utilities.agent_utils import (
_asummarize_chunks,
_estimate_token_count,
_extract_summary_tags,
_format_messages_for_summary,
_split_messages_into_chunks,
convert_tools_to_openai_schema,
execute_single_native_tool_call,
NativeToolCallResult,
parse_tool_call_args,
summarize_messages,
)
class CalculatorInput(BaseModel):
"""Input schema for calculator tool."""
expression: str = Field(description="Mathematical expression to evaluate")
class CalculatorTool(BaseTool):
"""A simple calculator tool for testing."""
name: str = "calculator"
description: str = "Perform mathematical calculations"
args_schema: type[BaseModel] = CalculatorInput
def _run(self, expression: str) -> str:
"""Execute the calculation."""
try:
result = eval(expression) # noqa: S307
return str(result)
except Exception as e:
return f"Error: {e}"
class SearchInput(BaseModel):
"""Input schema for search tool."""
query: str = Field(description="Search query")
max_results: int = Field(default=10, description="Maximum number of results")
class SearchTool(BaseTool):
"""A search tool for testing."""
name: str = "web_search"
description: str = "Search the web for information"
args_schema: type[BaseModel] = SearchInput
def _run(self, query: str, max_results: int = 10) -> str:
"""Execute the search."""
return f"Search results for '{query}' (max {max_results})"
class NoSchemaTool(BaseTool):
"""A tool without an args schema for testing edge cases."""
name: str = "simple_tool"
description: str = "A simple tool with no schema"
def _run(self, **kwargs: Any) -> str:
"""Execute the tool."""
return "Simple tool executed"
class TestConvertToolsToOpenaiSchema:
"""Tests for convert_tools_to_openai_schema function."""
def test_converts_single_tool(self) -> None:
"""Test converting a single tool to OpenAI schema."""
tools = [CalculatorTool()]
schemas, functions, _ = convert_tools_to_openai_schema(tools)
assert len(schemas) == 1
assert len(functions) == 1
schema = schemas[0]
assert schema["type"] == "function"
assert schema["function"]["name"] == "calculator"
assert schema["function"]["description"] == "Perform mathematical calculations"
assert "properties" in schema["function"]["parameters"]
assert "expression" in schema["function"]["parameters"]["properties"]
def test_converts_multiple_tools(self) -> None:
"""Test converting multiple tools to OpenAI schema."""
tools = [CalculatorTool(), SearchTool()]
schemas, functions, _ = convert_tools_to_openai_schema(tools)
assert len(schemas) == 2
assert len(functions) == 2
calc_schema = next(s for s in schemas if s["function"]["name"] == "calculator")
assert calc_schema["function"]["description"] == "Perform mathematical calculations"
search_schema = next(s for s in schemas if s["function"]["name"] == "web_search")
assert search_schema["function"]["description"] == "Search the web for information"
assert "query" in search_schema["function"]["parameters"]["properties"]
assert "max_results" in search_schema["function"]["parameters"]["properties"]
def test_functions_dict_contains_callables(self) -> None:
"""Test that the functions dict maps names to callable run methods."""
tools = [CalculatorTool(), SearchTool()]
schemas, functions, _ = convert_tools_to_openai_schema(tools)
assert "calculator" in functions
assert "web_search" in functions
assert callable(functions["calculator"])
assert callable(functions["web_search"])
def test_function_can_be_called(self) -> None:
"""Test that the returned function can be called."""
tools = [CalculatorTool()]
schemas, functions, _ = convert_tools_to_openai_schema(tools)
result = functions["calculator"](expression="2 + 2")
assert result == "4"
def test_empty_tools_list(self) -> None:
"""Test with an empty tools list."""
schemas, functions, _ = convert_tools_to_openai_schema([])
assert schemas == []
assert functions == {}
def test_schema_has_required_fields(self) -> None:
"""Test that the schema includes required fields information."""
tools = [SearchTool()]
schemas, functions, _ = convert_tools_to_openai_schema(tools)
schema = schemas[0]
params = schema["function"]["parameters"]
assert "required" in params
assert "query" in params["required"]
def test_tool_without_args_schema(self) -> None:
"""Test converting a tool that doesn't have an args_schema."""
class MinimalTool(BaseTool):
name: str = "minimal"
description: str = "A minimal tool"
def _run(self) -> str:
return "done"
tools = [MinimalTool()]
schemas, functions, _ = convert_tools_to_openai_schema(tools)
assert len(schemas) == 1
schema = schemas[0]
assert schema["function"]["name"] == "minimal"
# Parameters should be empty dict or have minimal schema
assert isinstance(schema["function"]["parameters"], dict)
def test_schema_structure_matches_openai_format(self) -> None:
"""Test that the schema structure matches OpenAI's expected format."""
tools = [CalculatorTool()]
schemas, functions, _ = convert_tools_to_openai_schema(tools)
schema = schemas[0]
# Top level must have "type": "function"
assert schema["type"] == "function"
# Must have "function" key with nested structure
assert "function" in schema
func = schema["function"]
# Function must have name and description
assert "name" in func
assert "description" in func
assert isinstance(func["name"], str)
assert isinstance(func["description"], str)
# Parameters should be a valid JSON schema
assert "parameters" in func
params = func["parameters"]
assert isinstance(params, dict)
def test_removes_redundant_schema_fields(self) -> None:
"""Test that redundant title and description are removed from parameters."""
tools = [CalculatorTool()]
schemas, functions, _ = convert_tools_to_openai_schema(tools)
params = schemas[0]["function"]["parameters"]
# Title should be removed as it's redundant with function name
assert "title" not in params
def test_preserves_field_descriptions(self) -> None:
"""Test that field descriptions are preserved in the schema."""
tools = [SearchTool()]
schemas, functions, _ = convert_tools_to_openai_schema(tools)
params = schemas[0]["function"]["parameters"]
query_prop = params["properties"]["query"]
# Field description should be preserved
assert "description" in query_prop
assert query_prop["description"] == "Search query"
def test_preserves_default_values(self) -> None:
"""Test that default values are preserved in the schema."""
tools = [SearchTool()]
schemas, functions, _ = convert_tools_to_openai_schema(tools)
params = schemas[0]["function"]["parameters"]
max_results_prop = params["properties"]["max_results"]
# Default value should be preserved
assert "default" in max_results_prop
assert max_results_prop["default"] == 10
class MCPStyleInput(BaseModel):
"""Input schema mimicking an MCP tool with optional fields."""
query: str = Field(description="Search query")
filter_type: Optional[Literal["internal", "user"]] = Field(
default=None, description="Filter type"
)
page_id: Optional[str] = Field(
default=None, description="Page UUID"
)
class MCPStyleTool(BaseTool):
"""A tool mimicking MCP tool schemas with optional fields."""
name: str = "mcp_search"
description: str = "Search with optional filters"
args_schema: type[BaseModel] = MCPStyleInput
def _run(self, **kwargs: Any) -> str:
return "result"
class TestOptionalFieldsPreserveNull:
"""Tests that optional tool fields preserve null in the schema."""
def test_optional_string_allows_null(self) -> None:
"""Optional[str] fields should include null in the schema so the LLM
can send null instead of being forced to guess a value."""
tools = [MCPStyleTool()]
schemas, _, _ = convert_tools_to_openai_schema(tools)
params = schemas[0]["function"]["parameters"]
page_id_prop = params["properties"]["page_id"]
assert "anyOf" in page_id_prop
type_options = [opt.get("type") for opt in page_id_prop["anyOf"]]
assert "string" in type_options
assert "null" in type_options
def test_optional_literal_allows_null(self) -> None:
"""Optional[Literal[...]] fields should include null."""
tools = [MCPStyleTool()]
schemas, _, _ = convert_tools_to_openai_schema(tools)
params = schemas[0]["function"]["parameters"]
filter_prop = params["properties"]["filter_type"]
assert "anyOf" in filter_prop
has_null = any(opt.get("type") == "null" for opt in filter_prop["anyOf"])
assert has_null
def test_required_field_stays_non_null(self) -> None:
"""Required fields without Optional should NOT have null."""
tools = [MCPStyleTool()]
schemas, _, _ = convert_tools_to_openai_schema(tools)
params = schemas[0]["function"]["parameters"]
query_prop = params["properties"]["query"]
assert query_prop.get("type") == "string"
assert "anyOf" not in query_prop
def test_all_fields_in_required_for_strict_mode(self) -> None:
"""All fields (including optional) must be in required for strict mode."""
tools = [MCPStyleTool()]
schemas, _, _ = convert_tools_to_openai_schema(tools)
params = schemas[0]["function"]["parameters"]
assert "query" in params["required"]
assert "filter_type" in params["required"]
assert "page_id" in params["required"]
class TestSummarizeMessages:
"""Tests for summarize_messages function."""
def test_preserves_files_from_user_messages(self) -> None:
"""Test that files attached to user messages are preserved after summarization."""
mock_files = {"image.png": MagicMock(), "doc.pdf": MagicMock()}
messages: list[dict[str, Any]] = [
{"role": "system", "content": "You are a helpful assistant."},
{"role": "user", "content": "Analyze this image", "files": mock_files},
{"role": "assistant", "content": "I can see the image shows..."},
{"role": "user", "content": "What about the colors?"},
]
mock_llm = MagicMock()
mock_llm.get_context_window_size.return_value = 1000
mock_llm.call.return_value = "<summary>Summarized conversation about image analysis.</summary>"
summarize_messages(
messages=messages,
llm=mock_llm,
callbacks=[],
)
# System message preserved + summary message = 2
assert len(messages) == 2
assert messages[0]["role"] == "system"
summary_msg = messages[1]
assert summary_msg["role"] == "user"
assert "files" in summary_msg
assert summary_msg["files"] == mock_files
def test_merges_files_from_multiple_user_messages(self) -> None:
"""Test that files from multiple user messages are merged."""
file1 = MagicMock()
file2 = MagicMock()
file3 = MagicMock()
messages: list[dict[str, Any]] = [
{"role": "user", "content": "First image", "files": {"img1.png": file1}},
{"role": "assistant", "content": "I see the first image."},
{"role": "user", "content": "Second image", "files": {"img2.png": file2, "doc.pdf": file3}},
{"role": "assistant", "content": "I see the second image and document."},
]
mock_llm = MagicMock()
mock_llm.get_context_window_size.return_value = 1000
mock_llm.call.return_value = "<summary>Summarized conversation.</summary>"
summarize_messages(
messages=messages,
llm=mock_llm,
callbacks=[],
)
assert len(messages) == 1
assert "files" in messages[0]
assert messages[0]["files"] == {
"img1.png": file1,
"img2.png": file2,
"doc.pdf": file3,
}
def test_works_without_files(self) -> None:
"""Test that summarization works when no files are attached."""
messages: list[dict[str, Any]] = [
{"role": "user", "content": "Hello"},
{"role": "assistant", "content": "Hi there!"},
]
mock_llm = MagicMock()
mock_llm.get_context_window_size.return_value = 1000
mock_llm.call.return_value = "<summary>A greeting exchange.</summary>"
summarize_messages(
messages=messages,
llm=mock_llm,
callbacks=[],
)
assert len(messages) == 1
assert "files" not in messages[0]
def test_modifies_original_messages_list(self) -> None:
"""Test that the original messages list is modified in-place."""
messages: list[dict[str, Any]] = [
{"role": "user", "content": "First message"},
{"role": "assistant", "content": "Response"},
{"role": "user", "content": "Second message"},
]
original_list_id = id(messages)
mock_llm = MagicMock()
mock_llm.get_context_window_size.return_value = 1000
mock_llm.call.return_value = "<summary>Summary</summary>"
summarize_messages(
messages=messages,
llm=mock_llm,
callbacks=[],
)
assert id(messages) == original_list_id
assert len(messages) == 1
def test_preserves_system_messages(self) -> None:
"""Test that system messages are preserved and not summarized."""
messages: list[dict[str, Any]] = [
{"role": "system", "content": "You are a research assistant."},
{"role": "user", "content": "Find information about AI."},
{"role": "assistant", "content": "I found several resources on AI."},
]
mock_llm = MagicMock()
mock_llm.get_context_window_size.return_value = 1000
mock_llm.call.return_value = "<summary>User asked about AI, assistant found resources.</summary>"
summarize_messages(
messages=messages,
llm=mock_llm,
callbacks=[],
)
assert len(messages) == 2
assert messages[0]["role"] == "system"
assert messages[0]["content"] == "You are a research assistant."
assert messages[1]["role"] == "user"
def test_formats_conversation_with_role_labels(self) -> None:
"""Test that the LLM receives role-labeled conversation text."""
messages: list[dict[str, Any]] = [
{"role": "system", "content": "System prompt."},
{"role": "user", "content": "Hello there"},
{"role": "assistant", "content": "Hi! How can I help?"},
]
mock_llm = MagicMock()
mock_llm.get_context_window_size.return_value = 1000
mock_llm.call.return_value = "<summary>Greeting exchange.</summary>"
summarize_messages(
messages=messages,
llm=mock_llm,
callbacks=[],
)
call_args = mock_llm.call.call_args[0][0]
user_msg_content = call_args[1]["content"]
assert "[USER]:" in user_msg_content
assert "[ASSISTANT]:" in user_msg_content
# System content should NOT appear in summarization input
assert "System prompt." not in user_msg_content
def test_extracts_summary_from_tags(self) -> None:
"""Test that <summary> tags are extracted from LLM response."""
messages: list[dict[str, Any]] = [
{"role": "user", "content": "Do something."},
{"role": "assistant", "content": "Done."},
]
mock_llm = MagicMock()
mock_llm.get_context_window_size.return_value = 1000
mock_llm.call.return_value = "Here is the summary:\n<summary>The extracted summary content.</summary>\nExtra text."
summarize_messages(
messages=messages,
llm=mock_llm,
callbacks=[],
)
assert "The extracted summary content." in messages[0]["content"]
def test_handles_tool_messages(self) -> None:
"""Test that tool messages are properly formatted in summarization."""
messages: list[dict[str, Any]] = [
{"role": "user", "content": "Search for Python."},
{"role": "assistant", "content": None, "tool_calls": [
{"function": {"name": "web_search", "arguments": '{"query": "Python"}'}}
]},
{"role": "tool", "content": "Python is a programming language.", "name": "web_search"},
{"role": "assistant", "content": "Python is a programming language."},
]
mock_llm = MagicMock()
mock_llm.get_context_window_size.return_value = 1000
mock_llm.call.return_value = "<summary>User searched for Python info.</summary>"
summarize_messages(
messages=messages,
llm=mock_llm,
callbacks=[],
)
call_args = mock_llm.call.call_args[0][0]
user_msg_content = call_args[1]["content"]
assert "[TOOL_RESULT (web_search)]:" in user_msg_content
def test_only_system_messages_no_op(self) -> None:
"""Test that only system messages results in no-op (no summarization)."""
messages: list[dict[str, Any]] = [
{"role": "system", "content": "You are a helpful assistant."},
{"role": "system", "content": "Additional system instructions."},
]
mock_llm = MagicMock()
mock_llm.get_context_window_size.return_value = 1000
summarize_messages(
messages=messages,
llm=mock_llm,
callbacks=[],
)
# No LLM call should have been made
mock_llm.call.assert_not_called()
# System messages should remain untouched
assert len(messages) == 2
assert messages[0]["content"] == "You are a helpful assistant."
assert messages[1]["content"] == "Additional system instructions."
class TestFormatMessagesForSummary:
"""Tests for _format_messages_for_summary helper."""
def test_skips_system_messages(self) -> None:
messages: list[dict[str, Any]] = [
{"role": "system", "content": "System prompt"},
{"role": "user", "content": "Hello"},
]
result = _format_messages_for_summary(messages)
assert "System prompt" not in result
assert "[USER]: Hello" in result
def test_formats_user_and_assistant(self) -> None:
messages: list[dict[str, Any]] = [
{"role": "user", "content": "Question"},
{"role": "assistant", "content": "Answer"},
]
result = _format_messages_for_summary(messages)
assert "[USER]: Question" in result
assert "[ASSISTANT]: Answer" in result
def test_formats_tool_messages(self) -> None:
messages: list[dict[str, Any]] = [
{"role": "tool", "content": "Result data", "name": "search_tool"},
]
result = _format_messages_for_summary(messages)
assert "[TOOL_RESULT (search_tool)]:" in result
assert "Result data" in result
def test_handles_none_content_with_tool_calls(self) -> None:
messages: list[dict[str, Any]] = [
{"role": "assistant", "content": None, "tool_calls": [
{"function": {"name": "calculator", "arguments": "{}"}}
]},
]
result = _format_messages_for_summary(messages)
assert "[Called tools: calculator]" in result
def test_handles_none_content_without_tool_calls(self) -> None:
messages: list[dict[str, Any]] = [
{"role": "assistant", "content": None},
]
result = _format_messages_for_summary(messages)
assert "[ASSISTANT]:" in result
def test_handles_multimodal_content(self) -> None:
messages: list[dict[str, Any]] = [
{"role": "user", "content": [
{"type": "text", "text": "Describe this image"},
{"type": "image_url", "image_url": {"url": "data:image/png;base64,..."}}
]},
]
result = _format_messages_for_summary(messages)
assert "[USER]: Describe this image" in result
def test_empty_messages(self) -> None:
result = _format_messages_for_summary([])
assert result == ""
class TestExtractSummaryTags:
"""Tests for _extract_summary_tags helper."""
def test_extracts_content_from_tags(self) -> None:
text = "Preamble\n<summary>The actual summary.</summary>\nPostamble"
assert _extract_summary_tags(text) == "The actual summary."
def test_handles_multiline_content(self) -> None:
text = "<summary>\nLine 1\nLine 2\nLine 3\n</summary>"
result = _extract_summary_tags(text)
assert "Line 1" in result
assert "Line 2" in result
assert "Line 3" in result
def test_falls_back_when_no_tags(self) -> None:
text = "Just a plain summary without tags."
assert _extract_summary_tags(text) == text
def test_handles_empty_string(self) -> None:
assert _extract_summary_tags("") == ""
def test_extracts_first_match(self) -> None:
text = "<summary>First</summary> text <summary>Second</summary>"
assert _extract_summary_tags(text) == "First"
class TestSplitMessagesIntoChunks:
"""Tests for _split_messages_into_chunks helper."""
def test_single_chunk_when_under_limit(self) -> None:
messages: list[dict[str, Any]] = [
{"role": "user", "content": "Hello"},
{"role": "assistant", "content": "Hi"},
]
chunks = _split_messages_into_chunks(messages, max_tokens=1000)
assert len(chunks) == 1
assert len(chunks[0]) == 2
def test_splits_at_message_boundaries(self) -> None:
messages: list[dict[str, Any]] = [
{"role": "user", "content": "A" * 100},
{"role": "assistant", "content": "B" * 100},
{"role": "user", "content": "C" * 100},
]
# max_tokens=30 should cause splits
chunks = _split_messages_into_chunks(messages, max_tokens=30)
assert len(chunks) == 3
def test_excludes_system_messages(self) -> None:
messages: list[dict[str, Any]] = [
{"role": "system", "content": "System prompt"},
{"role": "user", "content": "Hello"},
]
chunks = _split_messages_into_chunks(messages, max_tokens=1000)
assert len(chunks) == 1
for chunk in chunks:
for msg in chunk:
assert msg.get("role") != "system"
def test_empty_messages(self) -> None:
chunks = _split_messages_into_chunks([], max_tokens=1000)
assert chunks == []
def test_only_system_messages(self) -> None:
messages: list[dict[str, Any]] = [
{"role": "system", "content": "System prompt"},
]
chunks = _split_messages_into_chunks(messages, max_tokens=1000)
assert chunks == []
def test_handles_none_content(self) -> None:
messages: list[dict[str, Any]] = [
{"role": "assistant", "content": None},
{"role": "user", "content": "Follow up"},
]
chunks = _split_messages_into_chunks(messages, max_tokens=1000)
assert len(chunks) == 1
assert len(chunks[0]) == 2
class TestEstimateTokenCount:
"""Tests for _estimate_token_count helper."""
def test_empty_string(self) -> None:
assert _estimate_token_count("") == 0
def test_short_string(self) -> None:
assert _estimate_token_count("hello") == 1 # 5 // 4 = 1
def test_longer_string(self) -> None:
assert _estimate_token_count("a" * 100) == 25 # 100 // 4 = 25
def test_approximation_is_conservative(self) -> None:
# For English text, actual token count is typically lower than char/4
text = "The quick brown fox jumps over the lazy dog."
estimated = _estimate_token_count(text)
assert estimated > 0
assert estimated == len(text) // 4
class TestParallelSummarization:
"""Tests for parallel chunk summarization via asyncio."""
def _make_messages_for_n_chunks(self, n: int) -> list[dict[str, Any]]:
"""Build a message list that will produce exactly *n* chunks.
Each message has 400 chars (~100 tokens). With max_tokens=100 returned
by the mock LLM, each message lands in its own chunk.
"""
msgs: list[dict[str, Any]] = []
for i in range(n):
msgs.append({"role": "user", "content": f"msg-{i} " + "x" * 400})
return msgs
def test_multiple_chunks_use_acall(self) -> None:
"""When there are multiple chunks, summarize_messages should use
llm.acall (parallel) instead of llm.call (sequential)."""
messages = self._make_messages_for_n_chunks(3)
mock_llm = MagicMock()
mock_llm.get_context_window_size.return_value = 100
mock_llm.acall = AsyncMock(
side_effect=[
"<summary>Summary chunk 1</summary>",
"<summary>Summary chunk 2</summary>",
"<summary>Summary chunk 3</summary>",
]
)
summarize_messages(
messages=messages,
llm=mock_llm,
callbacks=[],
)
# acall should have been awaited once per chunk
assert mock_llm.acall.await_count == 3
# sync call should NOT have been used for chunk summarization
mock_llm.call.assert_not_called()
def test_single_chunk_uses_sync_call(self) -> None:
"""When there is only one chunk, summarize_messages should use
the sync llm.call path (no async overhead)."""
messages: list[dict[str, Any]] = [
{"role": "user", "content": "Short message"},
{"role": "assistant", "content": "Short reply"},
]
mock_llm = MagicMock()
mock_llm.get_context_window_size.return_value = 100_000
mock_llm.call.return_value = "<summary>Short summary</summary>"
summarize_messages(
messages=messages,
llm=mock_llm,
callbacks=[],
)
mock_llm.call.assert_called_once()
def test_parallel_results_preserve_order(self) -> None:
"""Summaries must appear in the same order as the original chunks,
regardless of which async call finishes first."""
messages = self._make_messages_for_n_chunks(3)
mock_llm = MagicMock()
mock_llm.get_context_window_size.return_value = 100
# Simulate varying latencies — chunk 2 finishes before chunk 0
async def _delayed_acall(msgs: Any, **kwargs: Any) -> str:
user_content = msgs[1]["content"]
if "msg-0" in user_content:
await asyncio.sleep(0.05)
return "<summary>Summary-A</summary>"
elif "msg-1" in user_content:
return "<summary>Summary-B</summary>"
else:
await asyncio.sleep(0.02)
return "<summary>Summary-C</summary>"
mock_llm.acall = _delayed_acall
summarize_messages(
messages=messages,
llm=mock_llm,
callbacks=[],
)
summary_content = messages[-1]["content"]
pos_a = summary_content.index("Summary-A")
pos_b = summary_content.index("Summary-B")
pos_c = summary_content.index("Summary-C")
assert pos_a < pos_b < pos_c
def test_asummarize_chunks_returns_ordered_results(self) -> None:
"""Direct test of the async helper _asummarize_chunks."""
chunk_a: list[dict[str, Any]] = [{"role": "user", "content": "Chunk A"}]
chunk_b: list[dict[str, Any]] = [{"role": "user", "content": "Chunk B"}]
mock_llm = MagicMock()
mock_llm.acall = AsyncMock(
side_effect=[
"<summary>Result A</summary>",
"<summary>Result B</summary>",
]
)
results = asyncio.run(
_asummarize_chunks(
chunks=[chunk_a, chunk_b],
llm=mock_llm,
callbacks=[],
)
)
assert len(results) == 2
assert results[0]["content"] == "Result A"
assert results[1]["content"] == "Result B"
@patch("crewai.utilities.agent_utils.is_inside_event_loop", return_value=True)
def test_works_inside_existing_event_loop(self, _mock_loop: Any) -> None:
"""When called from inside a running event loop (e.g. a Flow),
the ThreadPoolExecutor fallback should still work."""
messages = self._make_messages_for_n_chunks(2)
mock_llm = MagicMock()
mock_llm.get_context_window_size.return_value = 100
mock_llm.acall = AsyncMock(
side_effect=[
"<summary>Flow summary 1</summary>",
"<summary>Flow summary 2</summary>",
]
)
summarize_messages(
messages=messages,
llm=mock_llm,
callbacks=[],
)
assert mock_llm.acall.await_count == 2
assert "Flow summary 1" in messages[-1]["content"]
assert "Flow summary 2" in messages[-1]["content"]
def _build_long_conversation() -> list[dict[str, Any]]:
"""Build a multi-turn conversation that produces multiple chunks at max_tokens=200.
Each non-system message is ~100-140 estimated tokens (400-560 chars),
so a max_tokens of 200 yields roughly 3 chunks from 6 messages.
"""
return [
{
"role": "system",
"content": "You are a helpful research assistant.",
},
{
"role": "user",
"content": (
"Tell me about the history of the Python programming language. "
"Who created it, when was it first released, and what were the "
"main design goals? Please provide a detailed overview covering "
"the major milestones from its inception through Python 3."
),
},
{
"role": "assistant",
"content": (
"Python was created by Guido van Rossum and first released in 1991. "
"The main design goals were code readability and simplicity. Key milestones: "
"Python 1.0 (1994) introduced functional programming tools like lambda and map. "
"Python 2.0 (2000) added list comprehensions and garbage collection. "
"Python 3.0 (2008) was a major backward-incompatible release that fixed "
"fundamental design flaws. Python 2 reached end-of-life in January 2020."
),
},
{
"role": "user",
"content": (
"What about the async/await features? When were they introduced "
"and how do they compare to similar features in JavaScript and C#? "
"Also explain the Global Interpreter Lock and its implications."
),
},
{
"role": "assistant",
"content": (
"Async/await was introduced in Python 3.5 (PEP 492, 2015). "
"Unlike JavaScript which is single-threaded by design, Python's asyncio "
"is an opt-in framework. C# introduced async/await in 2012 (C# 5.0) and "
"was a major inspiration for Python's implementation. "
"The GIL (Global Interpreter Lock) is a mutex that protects access to "
"Python objects, preventing multiple threads from executing Python bytecodes "
"simultaneously. This means CPU-bound multithreaded programs don't benefit "
"from multiple cores. PEP 703 proposes making the GIL optional in CPython."
),
},
{
"role": "user",
"content": (
"Explain the Python package ecosystem. How does pip work, what is PyPI, "
"and what are virtual environments? Compare pip with conda and uv."
),
},
{
"role": "assistant",
"content": (
"PyPI (Python Package Index) is the official repository hosting 400k+ packages. "
"pip is the standard package installer that downloads from PyPI. "
"Virtual environments (venv) create isolated Python installations to avoid "
"dependency conflicts between projects. conda is a cross-language package manager "
"popular in data science that can manage non-Python dependencies. "
"uv is a new Rust-based tool that is 10-100x faster than pip and aims to replace "
"pip, pip-tools, and virtualenv with a single unified tool."
),
},
]
class TestParallelSummarizationVCR:
"""VCR-backed integration tests for parallel summarization.
These tests use a real LLM but patch get_context_window_size to force
multiple chunks, exercising the asyncio.gather + acall parallel path.
To record cassettes:
PYTEST_VCR_RECORD_MODE=all uv run pytest lib/crewai/tests/utilities/test_agent_utils.py::TestParallelSummarizationVCR -v
"""
@pytest.mark.vcr()
def test_parallel_summarize_openai(self) -> None:
"""Test that parallel summarization with gpt-4o-mini produces a valid summary."""
from crewai.llm import LLM
llm = LLM(model="gpt-4o-mini", temperature=0)
messages = _build_long_conversation()
original_system = messages[0]["content"]
# Patch get_context_window_size to return 200 — forces multiple chunks
with patch.object(type(llm), "get_context_window_size", return_value=200):
non_system = [m for m in messages if m.get("role") != "system"]
chunks = _split_messages_into_chunks(non_system, max_tokens=200)
assert len(chunks) > 1, f"Expected multiple chunks, got {len(chunks)}"
summarize_messages(
messages=messages,
llm=llm,
callbacks=[],
)
# System message preserved
assert messages[0]["role"] == "system"
assert messages[0]["content"] == original_system
# Summary produced as a user message
summary_msg = messages[-1]
assert summary_msg["role"] == "user"
assert len(summary_msg["content"]) > 0
@pytest.mark.vcr()
def test_parallel_summarize_preserves_files(self) -> None:
"""Test that file references survive parallel summarization."""
from crewai.llm import LLM
llm = LLM(model="gpt-4o-mini", temperature=0)
messages = _build_long_conversation()
mock_file = MagicMock()
messages[1]["files"] = {"report.pdf": mock_file}
with patch.object(type(llm), "get_context_window_size", return_value=200):
summarize_messages(
messages=messages,
llm=llm,
callbacks=[],
)
summary_msg = messages[-1]
assert summary_msg["role"] == "user"
assert "files" in summary_msg
assert "report.pdf" in summary_msg["files"]
class TestParseToolCallArgs:
"""Unit tests for parse_tool_call_args."""
def test_valid_json_string_returns_dict(self) -> None:
args_dict, error = parse_tool_call_args('{"code": "print(1)"}', "run_code", "call_1")
assert error is None
assert args_dict == {"code": "print(1)"}
def test_malformed_json_returns_error_dict(self) -> None:
args_dict, error = parse_tool_call_args('{"code": "print("hi")"}', "run_code", "call_1")
assert args_dict is None
assert error is not None
assert error["call_id"] == "call_1"
assert error["func_name"] == "run_code"
assert error["from_cache"] is False
assert "Failed to parse tool arguments as JSON" in error["result"]
assert "run_code" in error["result"]
def test_malformed_json_preserves_original_tool(self) -> None:
mock_tool = object()
_, error = parse_tool_call_args("{bad}", "my_tool", "call_2", original_tool=mock_tool)
assert error is not None
assert error["original_tool"] is mock_tool
def test_malformed_json_original_tool_defaults_to_none(self) -> None:
_, error = parse_tool_call_args("{bad}", "my_tool", "call_3")
assert error is not None
assert error["original_tool"] is None
def test_dict_input_returned_directly(self) -> None:
func_args = {"code": "x = 42"}
args_dict, error = parse_tool_call_args(func_args, "run_code", "call_4")
assert error is None
assert args_dict == {"code": "x = 42"}
def test_empty_dict_input_returned_directly(self) -> None:
args_dict, error = parse_tool_call_args({}, "run_code", "call_5")
assert error is None
assert args_dict == {}
def test_valid_json_with_nested_values(self) -> None:
args_dict, error = parse_tool_call_args(
'{"query": "hello", "options": {"limit": 10}}', "search", "call_6"
)
assert error is None
assert args_dict == {"query": "hello", "options": {"limit": 10}}
def test_error_result_has_correct_keys(self) -> None:
_, error = parse_tool_call_args("{bad json}", "tool", "call_7")
assert error is not None
assert set(error.keys()) == {
"call_id",
"func_name",
"result",
"from_cache",
"original_tool",
"tool_failure",
}
class TestExecuteSingleNativeToolCall:
"""Tests for execute_single_native_tool_call."""
def test_typed_tool_output_is_json_agent_text(self) -> None:
clear_before_tool_call_hooks()
clear_after_tool_call_hooks()
class SearchOutput(BaseModel):
query: str
score: float
class TypedSearchTool(BaseTool):
name: str = "typed_search"
description: str = "Search for a query"
result_schema: type[BaseModel] = SearchOutput
def _run(self, query: str) -> SearchOutput:
return SearchOutput(query=query, score=0.9)
tool = TypedSearchTool()
tool_call = MagicMock()
tool_call.id = "call_1"
tool_call.function.name = "typed_search"
tool_call.function.arguments = '{"query": "crew"}'
result = execute_single_native_tool_call(
tool_call,
available_functions={"typed_search": tool._run},
original_tools=[tool],
structured_tools=[tool.to_structured_tool()],
tools_handler=None,
agent=None,
task=None,
crew=None,
event_source=MagicMock(),
printer=None,
verbose=False,
)
assert json.loads(result.result) == {"query": "crew", "score": 0.9}
assert json.loads(result.tool_message["content"]) == {
"query": "crew",
"score": 0.9,
}
def test_custom_agent_output_formatter_is_used_from_structured_tool(
self,
) -> None:
clear_before_tool_call_hooks()
clear_after_tool_call_hooks()
class SearchOutput(BaseModel):
query: str
score: float
class MarkdownSearchTool(BaseTool):
name: str = "markdown_search"
description: str = "Search for a query"
result_schema: type[BaseModel] = SearchOutput
def _run(self, query: str) -> SearchOutput:
return SearchOutput(query=query, score=0.9)
def format_output_for_agent(self, raw_result: Any) -> str:
result = self.result_schema.model_validate(raw_result)
return f"### {result.query}\n\nScore: **{result.score}**"
tool = MarkdownSearchTool()
tool_call = MagicMock()
tool_call.id = "call_1"
tool_call.function.name = "markdown_search"
tool_call.function.arguments = '{"query": "crew"}'
result = execute_single_native_tool_call(
tool_call,
available_functions={"markdown_search": tool._run},
original_tools=[],
structured_tools=[tool.to_structured_tool()],
tools_handler=None,
agent=None,
task=None,
crew=None,
event_source=MagicMock(),
printer=None,
verbose=False,
)
assert result.result == "### crew\n\nScore: **0.9**"
assert result.tool_message["content"] == "### crew\n\nScore: **0.9**"
def test_after_hook_includes_raw_tool_result_for_typed_output(self) -> None:
clear_after_tool_call_hooks()
class SearchOutput(BaseModel):
query: str
score: float
class TypedSearchTool(BaseTool):
name: str = "typed_search"
description: str = "Search for a query"
result_schema: type[BaseModel] = SearchOutput
def _run(self, query: str) -> SearchOutput:
return SearchOutput(query=query, score=0.9)
seen_results: list[tuple[str | None, object]] = []
def after_hook(context: ToolCallHookContext) -> None:
seen_results.append((context.tool_result, context.raw_tool_result))
tool = TypedSearchTool()
tool_call = MagicMock()
tool_call.id = "call_1"
tool_call.function.name = "typed_search"
tool_call.function.arguments = '{"query": "crew"}'
register_after_tool_call_hook(after_hook)
try:
result = execute_single_native_tool_call(
tool_call,
available_functions={"typed_search": tool._run},
original_tools=[tool],
structured_tools=[tool.to_structured_tool()],
tools_handler=None,
agent=None,
task=None,
crew=None,
event_source=MagicMock(),
printer=None,
verbose=False,
)
finally:
clear_after_tool_call_hooks()
assert json.loads(result.result) == {"query": "crew", "score": 0.9}
assert seen_results == [
('{"query":"crew","score":0.9}', SearchOutput(query="crew", score=0.9))
]
def test_result_as_answer_false_on_tool_error(self) -> None:
"""When a tool with result_as_answer=True raises, result_as_answer must be False.
Regression test for https://github.com/crewAIInc/crewAI/issues/5156
"""
from unittest.mock import MagicMock
class FailingTool(BaseTool):
name: str = "failing_tool"
description: str = "A tool that always fails"
result_as_answer: bool = True
def _run(self, **kwargs: Any) -> str:
raise RuntimeError("intentional failure")
tool = FailingTool()
tool_call = MagicMock()
tool_call.id = "call_1"
tool_call.function.name = "failing_tool"
tool_call.function.arguments = "{}"
result = execute_single_native_tool_call(
tool_call,
available_functions={"failing_tool": tool._run},
original_tools=[tool],
structured_tools=None,
tools_handler=None,
agent=None,
task=None,
crew=None,
event_source=MagicMock(),
printer=None,
verbose=False,
)
assert isinstance(result, NativeToolCallResult)
assert result.result_as_answer is False
assert "Error executing tool" in result.result
def test_result_as_answer_false_when_hook_blocks(self) -> None:
"""When a before-hook blocks a tool with result_as_answer=True, result_as_answer must be False."""
from unittest.mock import MagicMock
from crewai.hooks.tool_hooks import (
clear_before_tool_call_hooks,
register_before_tool_call_hook,
)
class BlockedTool(BaseTool):
name: str = "blocked_tool"
description: str = "A tool whose execution will be blocked by a hook"
result_as_answer: bool = True
def _run(self, **kwargs: Any) -> str:
return "should not run"
tool = BlockedTool()
tool_call = MagicMock()
tool_call.id = "call_1"
tool_call.function.name = "blocked_tool"
tool_call.function.arguments = "{}"
register_before_tool_call_hook(lambda _ctx: False)
try:
result = execute_single_native_tool_call(
tool_call,
available_functions={"blocked_tool": tool._run},
original_tools=[tool],
structured_tools=None,
tools_handler=None,
agent=None,
task=None,
crew=None,
event_source=MagicMock(),
printer=None,
verbose=False,
)
finally:
clear_before_tool_call_hooks()
assert isinstance(result, NativeToolCallResult)
assert result.result_as_answer is False
assert "blocked by hook" in result.result
class TestResolvePlusClient:
def test_builds_the_default_when_no_client_is_installed(self) -> None:
from crewai.utilities.agent_utils import resolve_plus_client
default = MagicMock()
assert resolve_plus_client(lambda: default) is default
def test_prefers_an_installed_client(
self, monkeypatch: pytest.MonkeyPatch
) -> None:
"""A hosted runtime installs a client; the default must not be built,
since looking up a user credential raises when there isn't one."""
from crewai.utilities import agent_utils
installed = MagicMock()
monkeypatch.setattr(agent_utils, "_create_plus_client_hook", lambda: installed)
default = MagicMock(side_effect=AssertionError("must not be called"))
assert agent_utils.resolve_plus_client(default) is installed
default.assert_not_called()
class TestResolvePlusResponse:
def test_passes_through_a_sync_response(self) -> None:
from crewai.utilities.agent_utils import resolve_plus_response
response = MagicMock()
assert resolve_plus_response(response) is response
@pytest.mark.parametrize("inside_loop", [False, True])
def test_awaits_an_async_response(self, inside_loop: bool) -> None:
from crewai.utilities.agent_utils import resolve_plus_response
response = MagicMock()
async def call() -> Any:
return response
if not inside_loop:
assert resolve_plus_response(call()) is response
return
async def main() -> Any:
return resolve_plus_response(call())
assert asyncio.run(main()) is response
def test_carries_context_vars_into_the_worker_thread(self) -> None:
"""Inside a running loop the coroutine runs on another thread; a client
reading runtime state (the platform token, flow context) must still see
the caller's values rather than defaults."""
from crewai.context import get_platform_integration_token, platform_context
from crewai.utilities.agent_utils import resolve_plus_response
async def call() -> Any:
return get_platform_integration_token()
async def main() -> Any:
with platform_context("token-from-caller"):
return resolve_plus_response(call())
assert asyncio.run(main()) == "token-from-caller"
def test_rejects_an_awaitable_bound_to_a_loop(self) -> None:
from crewai.utilities.agent_utils import resolve_plus_response
async def main() -> None:
future: asyncio.Future[Any] = asyncio.get_running_loop().create_future()
future.set_result(MagicMock())
with pytest.raises(TypeError, match="must return a coroutine"):
resolve_plus_response(future)
asyncio.run(main())