Merge branch 'main' of github.com:crewAIInc/crewAI into lorenze/experimental-environment-tools

Modified the default value of allowed_paths in BaseEnvironmentTool and EnvironmentTools to default to the current directory (".") instead of an empty list. This change enhances security by ensuring that operations are restricted to the current directory unless explicitly specified otherwise. Additionally, updated related tests to reflect this new default behavior and improved type hints for better clarity.

docs/docs.json

@@ -291,7 +291,6 @@
                   "en/observability/arize-phoenix",
                   "en/observability/braintrust",
                   "en/observability/datadog",
-                  "en/observability/galileo",
                   "en/observability/langdb",
                   "en/observability/langfuse",
                   "en/observability/langtrace",
@@ -743,7 +742,6 @@
                   "pt-BR/observability/arize-phoenix",
                   "pt-BR/observability/braintrust",
                   "pt-BR/observability/datadog",
-                  "pt-BR/observability/galileo",
                   "pt-BR/observability/langdb",
                   "pt-BR/observability/langfuse",
                   "pt-BR/observability/langtrace",
@@ -1205,7 +1203,6 @@
                   "ko/observability/arize-phoenix",
                   "ko/observability/braintrust",
                   "ko/observability/datadog",
-                  "ko/observability/galileo",
                   "ko/observability/langdb",
                   "ko/observability/langfuse",
                   "ko/observability/langtrace",

docs/en/observability/galileo.mdx

@@ -1,115 +0,0 @@
----
-title: Galileo
-description: Galileo integration for CrewAI tracing and evaluation
-icon: telescope
-mode: "wide"
----
-
-## Overview
-
-This guide demonstrates how to integrate **Galileo** with **CrewAI**
-for comprehensive tracing and Evaluation Engineering.
-By the end of this guide, you will be able to trace your CrewAI agents,
-monitor their performance, and evaluate their behaviour with
-Galileo's powerful observability platform.
-
-> **What is Galileo?** [Galileo](https://galileo.ai) is AI evaluation and observability
-platform that delivers end-to-end tracing, evaluation,
-and monitoring for AI applications. It enables teams to capture ground truth,
-create robust guardrails, and run systematic experiments with
-built-in experiment tracking and performance analytics—ensuring reliability,
-transparency, and continuous improvement across the AI lifecycle.
-
-## Getting started
-
-This tutorial follows the [CrewAI quickstart](/en/quickstart) and shows how to add
-Galileo's [CrewAIEventListener](https://v2docs.galileo.ai/sdk-api/python/reference/handlers/crewai/handler),
-an event handler.
-For more information, see Galileo’s
-[Add Galileo to a CrewAI Application](https://v2docs.galileo.ai/how-to-guides/third-party-integrations/add-galileo-to-crewai/add-galileo-to-crewai)
-how-to guide.
-
-> **Note** This tutorial assumes you have completed the [CrewAI quickstart](/en/quickstart).
-If you want a completed comprehensive example, see the Galileo
-[CrewAI sdk-example repo](https://github.com/rungalileo/sdk-examples/tree/main/python/agent/crew-ai).
-
-### Step 1: Install dependencies
-
-Install the required dependencies for your app.
-Create a virtual environment using your preferred method,
-then install dependencies inside that environment using your
-preferred tool:
-
-```bash
-uv add galileo 
-```
-
-### Step 2: Add to the .env file from the [CrewAI quickstart](/en/quickstart)
-
-```bash
-# Your Galileo API key
-GALILEO_API_KEY="your-galileo-api-key"
-
-# Your Galileo project name
-GALILEO_PROJECT="your-galileo-project-name"
-
-# The name of the Log stream you want to use for logging
-GALILEO_LOG_STREAM="your-galileo-log-stream "
-```
-
-### Step 3: Add the Galileo event listener
-
-To enable logging with Galileo, you need to create an instance of the `CrewAIEventListener`.
-Import the Galileo CrewAI handler package by
-adding the following code at the top of your main.py file:
-
-```python
-from galileo.handlers.crewai.handler import CrewAIEventListener
-```
-
-At the start of your run function, create the event listener:
-
-```python
-def run():
-    # Create the event listener
-    CrewAIEventListener()
-    # The rest of your existing code goes here
-```
-
-When you create the listener instance, it is automatically
-registered with CrewAI.
-
-### Step 4: Run your crew
-
-Run your crew with the CrewAI CLI:
-
-```bash
-crewai run
-```
-
-### Step 5: View the traces in Galileo
-
-Once your crew has finished, the traces will be flushed and appear in Galileo.
-
-![Galileo trace view](/images/galileo-trace-veiw.png)
-  
-## Understanding the Galileo Integration
-
-Galileo integrates with CrewAI by registering an event listener
-that captures Crew execution events (e.g., agent actions, tool calls, model responses)
-and forwards them to Galileo for observability and evaluation.
-
-### Understanding the event listener
-
-Creating a `CrewAIEventListener()` instance is all that’s
-required to enable Galileo for a CrewAI run. When instantiated, the listener:
-
-- Automatically registers itself with CrewAI
-- Reads Galileo configuration from environment variables
-- Logs all run data to the Galileo project and log stream specified by
-  `GALILEO_PROJECT` and `GALILEO_LOG_STREAM`
-
-No additional configuration or code changes are required.
-All data from this run is logged to the Galileo project and
-log stream specified by your environment configuration
-(for example, GALILEO_PROJECT and GALILEO_LOG_STREAM).

docs/ko/observability/galileo.mdx

@@ -1,115 +0,0 @@
----
-title: Galileo 갈릴레오
-description: CrewAI 추적 및 평가를 위한 Galileo 통합
-icon: telescope
-mode: "wide"
----
-
-## 개요
-
-이 가이드는 **Galileo**를 **CrewAI**와 통합하는 방법을 보여줍니다.
-포괄적인 추적 및 평가 엔지니어링을 위한 것입니다.
-이 가이드가 끝나면 CrewAI 에이전트를 추적할 수 있게 됩니다.
-성과를 모니터링하고 행동을 평가합니다.
-Galileo의 강력한 관측 플랫폼.
-
-> **갈릴레오(Galileo)란 무엇인가요?**[Galileo](https://galileo.ai/)는 AI 평가 및 관찰 가능성입니다.
-엔드투엔드 추적, 평가,
-AI 애플리케이션 모니터링. 이를 통해 팀은 실제 사실을 포착할 수 있습니다.
-견고한 가드레일을 만들고 체계적인 실험을 실행하세요.
-내장된 실험 추적 및 성능 분석으로 신뢰성 보장
-AI 수명주기 전반에 걸쳐 투명성과 지속적인 개선을 제공합니다.
-
-## 시작하기
-
-이 튜토리얼은 [CrewAI 빠른 시작](/ko/quickstart.mdx)을 따르며 추가하는 방법을 보여줍니다.
-갈릴레오의 [CrewAIEventListener](https://v2docs.galileo.ai/sdk-api/python/reference/handlers/crewai/handler),
-이벤트 핸들러.
-자세한 내용은 갈릴레오 문서를 참고하세요.
-[CrewAI 애플리케이션에 Galileo 추가](https://v2docs.galileo.ai/how-to-guides/third-party-integrations/add-galileo-to-crewai/add-galileo-to-crewai)
-방법 안내.
-
-> **참고**이 튜토리얼에서는 [CrewAI 빠른 시작](/ko/quickstart.mdx)을 완료했다고 가정합니다.
-완전한 포괄적인 예제를 원한다면 Galileo
-[CrewAI SDK 예제 저장소](https://github.com/rungalileo/sdk-examples/tree/main/python/agent/crew-ai).
-
-### 1단계: 종속성 설치
-
-앱에 필요한 종속성을 설치합니다.
-원하는 방법으로 가상 환경을 생성하고,
-그런 다음 다음을 사용하여 해당 환경 내에 종속성을 설치하십시오.
-선호하는 도구:
-
-```bash
-uv add galileo 
-```
-
-### 2단계: [CrewAI 빠른 시작](/ko/quickstart.mdx)에서 .env 파일에 추가
-
-```bash
-# Your Galileo API key
-GALILEO_API_KEY="your-galileo-api-key"
-
-# Your Galileo project name
-GALILEO_PROJECT="your-galileo-project-name"
-
-# The name of the Log stream you want to use for logging
-GALILEO_LOG_STREAM="your-galileo-log-stream "
-```
-
-### 3단계: Galileo 이벤트 리스너 추가
-
-Galileo로 로깅을 활성화하려면 `CrewAIEventListener`의 인스턴스를 생성해야 합니다.
-다음을 통해 Galileo CrewAI 핸들러 패키지를 가져옵니다.
-main.py 파일 상단에 다음 코드를 추가하세요.
-
-```python
-from galileo.handlers.crewai.handler import CrewAIEventListener
-```
-
-실행 함수 시작 시 이벤트 리스너를 생성합니다.
-
-```python
-def run():
-    # Create the event listener
-    CrewAIEventListener()
-    # The rest of your existing code goes here
-```
-
-리스너 인스턴스를 생성하면 자동으로
-CrewAI에 등록되었습니다.
-
-### 4단계: Crew Agent 실행
-
-CrewAI CLI를 사용하여 Crew Agent를 실행하세요.
-
-```bash
-crewai run
-```
-
-### 5단계: Galileo에서 추적 보기
-
-승무원 에이전트가 완료되면 흔적이 플러시되어 Galileo에 나타납니다.
-
-![Galileo trace view](/images/galileo-trace-veiw.png)
-  
-## 갈릴레오 통합 이해
-
-Galileo는 이벤트 리스너를 등록하여 CrewAI와 통합됩니다.
-승무원 실행 이벤트(예: 에이전트 작업, 도구 호출, 모델 응답)를 캡처합니다.
-관찰 가능성과 평가를 위해 이를 갈릴레오에 전달합니다.
-
-### 이벤트 리스너 이해
-
-`CrewAIEventListener()` 인스턴스를 생성하는 것이 전부입니다.
-CrewAI 실행을 위해 Galileo를 활성화하는 데 필요합니다. 인스턴스화되면 리스너는 다음을 수행합니다.
-
--CrewAI에 자동으로 등록됩니다.
--환경 변수에서 Galileo 구성을 읽습니다.
--모든 실행 데이터를 Galileo 프로젝트 및 다음에서 지정한 로그 스트림에 기록합니다.
-  `GALILEO_PROJECT` 및 `GALILEO_LOG_STREAM`
-
-추가 구성이나 코드 변경이 필요하지 않습니다.
-이 실행의 모든 데이터는 Galileo 프로젝트에 기록되며
-환경 구성에 따라 지정된 로그 스트림
-(예: GALILEO_PROJECT 및 GALILEO_LOG_STREAM)

docs/pt-BR/observability/galileo.mdx

@@ -1,115 +0,0 @@
----
-title: Galileo Galileu
-description: Integração Galileo para rastreamento e avaliação CrewAI
-icon: telescope
-mode: "wide"
----
-
-## Visão geral
-
-Este guia demonstra como integrar o **Galileo**com o **CrewAI**
-para rastreamento abrangente e engenharia de avaliação.
-Ao final deste guia, você será capaz de rastrear seus agentes CrewAI,
-monitorar seu desempenho e avaliar seu comportamento com
-A poderosa plataforma de observabilidade do Galileo.
-
-> **O que é Galileo?**[Galileo](https://galileo.ai/) é avaliação e observabilidade de IA
-plataforma que oferece rastreamento, avaliação e
-e monitoramento de aplicações de IA. Ele permite que as equipes capturem a verdade,
-criar grades de proteção robustas e realizar experimentos sistemáticos com
-rastreamento de experimentos integrado e análise de desempenho -garantindo confiabilidade,
-transparência e melhoria contínua em todo o ciclo de vida da IA.
-
-## Primeiros passos
-
-Este tutorial segue o [CrewAI Quickstart](pt-BR/quickstart) e mostra como adicionar
-[CrewAIEventListener] do Galileo(https://v2docs.galileo.ai/sdk-api/python/reference/handlers/crewai/handler),
-um manipulador de eventos.
-Para mais informações, consulte Galileu
-[Adicionar Galileo a um aplicativo CrewAI](https://v2docs.galileo.ai/how-to-guides/third-party-integrations/add-galileo-to-crewai/add-galileo-to-crewai)
-guia prático.
-
-> **Observação**Este tutorial pressupõe que você concluiu o [CrewAI Quickstart](pt-BR/quickstart).
-Se você quiser um exemplo completo e abrangente, consulte o Galileo
-[Repositório de exemplo SDK da CrewAI](https://github.com/rungalileo/sdk-examples/tree/main/python/agent/crew-ai).
-
-### Etapa 1: instalar dependências
-
-Instale as dependências necessárias para seu aplicativo.
-Crie um ambiente virtual usando seu método preferido,
-em seguida, instale dependências dentro desse ambiente usando seu
-ferramenta preferida:
-
-```bash
-uv add galileo 
-```
-
-### Etapa 2: adicione ao arquivo .env do [CrewAI Quickstart](/pt-BR/quickstart)
-
-```bash
-# Your Galileo API key
-GALILEO_API_KEY="your-galileo-api-key"
-
-# Your Galileo project name
-GALILEO_PROJECT="your-galileo-project-name"
-
-# The name of the Log stream you want to use for logging
-GALILEO_LOG_STREAM="your-galileo-log-stream "
-```
-
-### Etapa 3: adicionar o ouvinte de eventos Galileo
-
-Para habilitar o registro com Galileo, você precisa criar uma instância do `CrewAIEventListener`.
-Importe o pacote manipulador Galileo CrewAI por
-adicionando o seguinte código no topo do seu arquivo main.py:
-
-```python
-from galileo.handlers.crewai.handler import CrewAIEventListener
-```
-
-No início da sua função run, crie o ouvinte de evento:
-
-```python
-def run():
-    # Create the event listener
-    CrewAIEventListener()
-    # The rest of your existing code goes here
-```
-
-Quando você cria a instância do listener, ela é automaticamente
-registrado na CrewAI.
-
-### Etapa 4: administre sua Crew
-
-Administre sua Crew com o CrewAI CLI:
-
-```bash
-crewai run
-```
-
-### Passo 5: Visualize os traços no Galileo
-
-Assim que sua tripulação terminar, os rastros serão eliminados e aparecerão no Galileo.
-
-![Galileo trace view](/images/galileo-trace-veiw.png)
-  
-## Compreendendo a integração do Galileo
-
-Galileo se integra ao CrewAI registrando um ouvinte de evento
-que captura eventos de execução da tripulação (por exemplo, ações do agente, chamadas de ferramentas, respostas do modelo)
-e os encaminha ao Galileo para observabilidade e avaliação.
-
-### Compreendendo o ouvinte de eventos
-
-Criar uma instância `CrewAIEventListener()` é tudo o que você precisa
-necessário para habilitar o Galileo para uma execução do CrewAI. Quando instanciado, o ouvinte:
-
--Registra-se automaticamente no CrewAI
--Lê a configuração do Galileo a partir de variáveis de ambiente
--Registra todos os dados de execução no projeto Galileo e fluxo de log especificado por
-  `GALILEO_PROJECT` e `GALILEO_LOG_STREAM`
-
-Nenhuma configuração adicional ou alterações de código são necessárias.
-Todos os dados desta execução são registados no projecto Galileo e
-fluxo de log especificado pela configuração do seu ambiente
-(por exemplo, GALILEO_PROJECT e GALILEO_LOG_STREAM).

lib/crewai/src/crewai/experimental/__init__.py

@@ -1,4 +1,12 @@
 from crewai.experimental.crew_agent_executor_flow import CrewAgentExecutorFlow
+from crewai.experimental.environment_tools import (
+    BaseEnvironmentTool,
+    EnvironmentTools,
+    FileReadTool,
+    FileSearchTool,
+    GrepTool,
+    ListDirTool,
+)
 from crewai.experimental.evaluation import (
     AgentEvaluationResult,
     AgentEvaluator,
@@ -23,14 +31,20 @@ from crewai.experimental.evaluation import (
 __all__ = [
     "AgentEvaluationResult",
     "AgentEvaluator",
+    "BaseEnvironmentTool",
     "BaseEvaluator",
     "CrewAgentExecutorFlow",
+    "EnvironmentTools",
     "EvaluationScore",
     "EvaluationTraceCallback",
     "ExperimentResult",
     "ExperimentResults",
     "ExperimentRunner",
+    "FileReadTool",
+    "FileSearchTool",
     "GoalAlignmentEvaluator",
+    "GrepTool",
+    "ListDirTool",
     "MetricCategory",
     "ParameterExtractionEvaluator",
     "ReasoningEfficiencyEvaluator",

lib/crewai/src/crewai/experimental/environment_tools/__init__.py

@@ -0,0 +1,24 @@
+"""Environment tools for file system operations.
+
+These tools provide agents with the ability to explore and read from
+the filesystem for context engineering purposes.
+"""
+
+from crewai.experimental.environment_tools.base_environment_tool import (
+    BaseEnvironmentTool,
+)
+from crewai.experimental.environment_tools.environment_tools import EnvironmentTools
+from crewai.experimental.environment_tools.file_read_tool import FileReadTool
+from crewai.experimental.environment_tools.file_search_tool import FileSearchTool
+from crewai.experimental.environment_tools.grep_tool import GrepTool
+from crewai.experimental.environment_tools.list_dir_tool import ListDirTool
+
+
+__all__ = [
+    "BaseEnvironmentTool",
+    "EnvironmentTools",
+    "FileReadTool",
+    "FileSearchTool",
+    "GrepTool",
+    "ListDirTool",
+]

lib/crewai/src/crewai/experimental/environment_tools/base_environment_tool.py

@@ -0,0 +1,84 @@
+"""Base class for environment tools with path security."""
+
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Any
+
+from pydantic import Field
+
+from crewai.tools.base_tool import BaseTool
+
+
+class BaseEnvironmentTool(BaseTool):
+    """Base class for environment/file system tools with path security.
+
+    Provides path validation to restrict file operations to allowed directories.
+    This prevents path traversal attacks and enforces security sandboxing.
+
+    Attributes:
+        allowed_paths: List of paths that operations are restricted to.
+            Empty list means allow all paths (no restrictions).
+    """
+
+    allowed_paths: list[str] = Field(
+        default_factory=lambda: ["."],
+        description="Restrict operations to these paths. Defaults to current directory.",
+    )
+
+    def _validate_path(self, path: str) -> tuple[bool, Path | str]:
+        """Validate and resolve a path against allowed_paths whitelist.
+
+        Args:
+            path: The path to validate.
+
+        Returns:
+            A tuple of (is_valid, result) where:
+            - If valid: (True, resolved_path as Path)
+            - If invalid: (False, error_message as str)
+        """
+        try:
+            resolved = Path(path).resolve()
+
+            # If no restrictions, allow all paths
+            if not self.allowed_paths:
+                return True, resolved
+
+            # Check if path is within any allowed path
+            for allowed in self.allowed_paths:
+                allowed_resolved = Path(allowed).resolve()
+                try:
+                    # This will raise ValueError if resolved is not relative to allowed_resolved
+                    resolved.relative_to(allowed_resolved)
+                    return True, resolved
+                except ValueError:
+                    continue
+
+            return (
+                False,
+                f"Path '{path}' is outside allowed paths: {self.allowed_paths}",
+            )
+
+        except Exception as e:
+            return False, f"Invalid path '{path}': {e}"
+
+    def _format_size(self, size: int) -> str:
+        """Format file size in human-readable format.
+
+        Args:
+            size: Size in bytes.
+
+        Returns:
+            Human-readable size string (e.g., "1.5KB", "2.3MB").
+        """
+        if size < 1024:
+            return f"{size}B"
+        if size < 1024 * 1024:
+            return f"{size / 1024:.1f}KB"
+        if size < 1024 * 1024 * 1024:
+            return f"{size / (1024 * 1024):.1f}MB"
+        return f"{size / (1024 * 1024 * 1024):.1f}GB"
+
+    def _run(self, *args: Any, **kwargs: Any) -> Any:
+        """Subclasses must implement this method."""
+        raise NotImplementedError("Subclasses must implement _run method")

lib/crewai/src/crewai/experimental/environment_tools/environment_tools.py

@@ -0,0 +1,77 @@
+"""Manager class for environment tools."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from crewai.experimental.environment_tools.file_read_tool import FileReadTool
+from crewai.experimental.environment_tools.file_search_tool import FileSearchTool
+from crewai.experimental.environment_tools.grep_tool import GrepTool
+from crewai.experimental.environment_tools.list_dir_tool import ListDirTool
+
+
+if TYPE_CHECKING:
+    from crewai.tools.base_tool import BaseTool
+
+
+class EnvironmentTools:
+    """Manager class for file system/environment tools.
+
+    Provides a convenient way to create a set of file system tools
+    with shared security configuration (allowed_paths).
+
+    Similar to AgentTools but for file system operations. Use this to
+    give agents the ability to explore and read files for context engineering.
+
+    Example:
+        from crewai.experimental import EnvironmentTools
+
+        # Create tools with security sandbox
+        env_tools = EnvironmentTools(
+            allowed_paths=["./src", "./docs"],
+        )
+
+        # Use with an agent
+        agent = Agent(
+            role="Code Analyst",
+            tools=env_tools.tools(),
+        )
+    """
+
+    def __init__(
+        self,
+        allowed_paths: list[str] | None = None,
+        include_grep: bool = True,
+        include_search: bool = True,
+    ) -> None:
+        """Initialize EnvironmentTools.
+
+        Args:
+            allowed_paths: List of paths to restrict operations to.
+                Defaults to current directory ["."] if None.
+                Pass empty list [] to allow all paths (not recommended).
+            include_grep: Whether to include GrepTool (requires grep installed).
+            include_search: Whether to include FileSearchTool.
+        """
+        self.allowed_paths = allowed_paths if allowed_paths is not None else ["."]
+        self.include_grep = include_grep
+        self.include_search = include_search
+
+    def tools(self) -> list[BaseTool]:
+        """Get all configured environment tools.
+
+        Returns:
+            List of BaseTool instances with shared allowed_paths configuration.
+        """
+        tool_list: list[BaseTool] = [
+            FileReadTool(allowed_paths=self.allowed_paths),
+            ListDirTool(allowed_paths=self.allowed_paths),
+        ]
+
+        if self.include_grep:
+            tool_list.append(GrepTool(allowed_paths=self.allowed_paths))
+
+        if self.include_search:
+            tool_list.append(FileSearchTool(allowed_paths=self.allowed_paths))
+
+        return tool_list

lib/crewai/src/crewai/experimental/environment_tools/file_read_tool.py

@@ -0,0 +1,124 @@
+"""Tool for reading file contents."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+from pydantic import BaseModel, Field
+
+from crewai.experimental.environment_tools.base_environment_tool import (
+    BaseEnvironmentTool,
+)
+
+
+class FileReadInput(BaseModel):
+    """Input schema for reading files."""
+
+    path: str = Field(..., description="Path to the file to read")
+    start_line: int | None = Field(
+        default=None,
+        description="Line to start reading from (1-indexed). If None, starts from beginning.",
+    )
+    line_count: int | None = Field(
+        default=None,
+        description="Number of lines to read. If None, reads to end of file.",
+    )
+
+
+class FileReadTool(BaseEnvironmentTool):
+    """Read contents of text files with optional line ranges.
+
+    Use this tool to:
+    - Read configuration files, source code, logs
+    - Inspect file contents before making decisions
+    - Load reference documentation or data files
+
+    Supports reading entire files or specific line ranges for efficiency.
+    """
+
+    name: str = "read_file"
+    description: str = """Read the contents of a text file.
+
+Use this to read configuration files, source code, logs, or any text file.
+You can optionally specify start_line and line_count to read specific portions.
+
+Examples:
+- Read entire file: path="config.yaml"
+- Read lines 100-149: path="large.log", start_line=100, line_count=50
+"""
+    args_schema: type[BaseModel] = FileReadInput
+
+    def _run(
+        self,
+        path: str,
+        start_line: int | None = None,
+        line_count: int | None = None,
+    ) -> str:
+        """Read file contents with optional line range.
+
+        Args:
+            path: Path to the file to read.
+            start_line: Line to start reading from (1-indexed).
+            line_count: Number of lines to read.
+
+        Returns:
+            File contents with metadata header, or error message.
+        """
+        # Validate path against allowed_paths
+        valid, result = self._validate_path(path)
+        if not valid:
+            return f"Error: {result}"
+
+        assert isinstance(result, Path)  # noqa: S101
+        file_path = result
+
+        # Check file exists and is a file
+        if not file_path.exists():
+            return f"Error: File not found: {path}"
+
+        if not file_path.is_file():
+            return f"Error: Not a file: {path}"
+
+        try:
+            with open(file_path, "r", encoding="utf-8") as f:
+                if start_line is None and line_count is None:
+                    # Read entire file
+                    content = f.read()
+                else:
+                    # Read specific line range
+                    lines = f.readlines()
+                    start_idx = (start_line or 1) - 1  # Convert to 0-indexed
+                    start_idx = max(0, start_idx)  # Ensure non-negative
+
+                    if line_count is not None:
+                        end_idx = start_idx + line_count
+                    else:
+                        end_idx = len(lines)
+
+                    content = "".join(lines[start_idx:end_idx])
+
+            # Get file metadata
+            stat = file_path.stat()
+            total_lines = content.count("\n") + (
+                1 if content and not content.endswith("\n") else 0
+            )
+
+            # Format output with metadata header
+            header = f"File: {path}\n"
+            header += f"Size: {self._format_size(stat.st_size)} | Lines: {total_lines}"
+
+            if start_line is not None or line_count is not None:
+                header += (
+                    f" | Range: {start_line or 1}-{(start_line or 1) + total_lines - 1}"
+                )
+
+            header += "\n" + "=" * 60 + "\n"
+
+            return header + content
+
+        except UnicodeDecodeError:
+            return f"Error: File is not a text file or has encoding issues: {path}"
+        except PermissionError:
+            return f"Error: Permission denied: {path}"
+        except Exception as e:
+            return f"Error reading file: {e}"

lib/crewai/src/crewai/experimental/environment_tools/file_search_tool.py

@@ -0,0 +1,127 @@
+"""Tool for finding files by name pattern."""
+
+from __future__ import annotations
+
+from typing import Literal
+
+from pydantic import BaseModel, Field
+
+from crewai.experimental.environment_tools.base_environment_tool import (
+    BaseEnvironmentTool,
+)
+
+
+class FileSearchInput(BaseModel):
+    """Input schema for file search."""
+
+    pattern: str = Field(
+        ...,
+        description="Filename pattern to search for (glob syntax, e.g., '*.py', 'test_*.py')",
+    )
+    path: str = Field(
+        default=".",
+        description="Directory to search in",
+    )
+    file_type: Literal["file", "dir", "all"] | None = Field(
+        default="all",
+        description="Filter by type: 'file' for files only, 'dir' for directories only, 'all' for both",
+    )
+
+
+class FileSearchTool(BaseEnvironmentTool):
+    """Find files by name pattern.
+
+    Use this tool to:
+    - Find specific files in a codebase
+    - Locate configuration files
+    - Search for files matching a pattern
+    """
+
+    name: str = "find_files"
+    description: str = """Find files by name pattern using glob syntax.
+
+Searches recursively through directories to find matching files.
+
+Examples:
+- Find Python files: pattern="*.py", path="src/"
+- Find test files: pattern="test_*.py"
+- Find configs: pattern="*.yaml", path="."
+- Find directories only: pattern="*", file_type="dir"
+"""
+    args_schema: type[BaseModel] = FileSearchInput
+
+    def _run(
+        self,
+        pattern: str,
+        path: str = ".",
+        file_type: Literal["file", "dir", "all"] | None = "all",
+    ) -> str:
+        """Find files matching a pattern.
+
+        Args:
+            pattern: Glob pattern for filenames.
+            path: Directory to search in.
+            file_type: Filter by type ('file', 'dir', or 'all').
+
+        Returns:
+            List of matching files or error message.
+        """
+        # Validate path against allowed_paths
+        valid, result = self._validate_path(path)
+        if not valid:
+            return f"Error: {result}"
+
+        search_path = result
+
+        # Check directory exists
+        if not search_path.exists():
+            return f"Error: Directory not found: {path}"
+
+        if not search_path.is_dir():
+            return f"Error: Not a directory: {path}"
+
+        try:
+            # Find matching entries recursively
+            matches = list(search_path.rglob(pattern))
+
+            # Filter by type
+            if file_type == "file":
+                matches = [m for m in matches if m.is_file()]
+            elif file_type == "dir":
+                matches = [m for m in matches if m.is_dir()]
+
+            # Filter out hidden files
+            matches = [
+                m for m in matches if not any(part.startswith(".") for part in m.parts)
+            ]
+
+            # Sort alphabetically
+            matches.sort(key=lambda x: str(x).lower())
+
+            if not matches:
+                return f"No {file_type if file_type != 'all' else 'files'} matching '{pattern}' found in {path}"
+
+            # Format output
+            result_lines = [f"Found {len(matches)} matches for '{pattern}' in {path}:"]
+            result_lines.append("=" * 60)
+
+            for match in matches:
+                # Get relative path from search directory
+                rel_path = match.relative_to(search_path)
+
+                if match.is_dir():
+                    result_lines.append(f"📁 {rel_path}/")
+                else:
+                    try:
+                        size = match.stat().st_size
+                    except (OSError, PermissionError):
+                        continue  # Skip files we can't stat
+                    size_str = self._format_size(size)
+                    result_lines.append(f"📄 {rel_path} ({size_str})")
+
+            return "\n".join(result_lines)
+
+        except PermissionError:
+            return f"Error: Permission denied: {path}"
+        except Exception as e:
+            return f"Error searching files: {e}"

lib/crewai/src/crewai/experimental/environment_tools/grep_tool.py

@@ -0,0 +1,149 @@
+"""Tool for searching patterns in files using grep."""
+
+from __future__ import annotations
+
+import subprocess
+
+from pydantic import BaseModel, Field
+
+from crewai.experimental.environment_tools.base_environment_tool import (
+    BaseEnvironmentTool,
+)
+
+
+class GrepInput(BaseModel):
+    """Input schema for grep search."""
+
+    pattern: str = Field(..., description="Search pattern (supports regex)")
+    path: str = Field(..., description="File or directory to search in")
+    recursive: bool = Field(
+        default=True,
+        description="Search recursively in directories",
+    )
+    ignore_case: bool = Field(
+        default=False,
+        description="Case-insensitive search",
+    )
+    context_lines: int = Field(
+        default=2,
+        description="Number of context lines to show before/after matches",
+    )
+
+
+class GrepTool(BaseEnvironmentTool):
+    """Search for text patterns in files using grep.
+
+    Use this tool to:
+    - Find where a function or class is defined
+    - Search for error messages in logs
+    - Locate configuration values
+    - Find TODO comments or specific patterns
+    """
+
+    name: str = "grep_search"
+    description: str = """Search for text patterns in files using grep.
+
+Supports regex patterns. Returns matching lines with context.
+
+Examples:
+- Find function: pattern="def process_data", path="src/"
+- Search logs: pattern="ERROR", path="logs/app.log"
+- Case-insensitive: pattern="todo", path=".", ignore_case=True
+"""
+    args_schema: type[BaseModel] = GrepInput
+
+    def _run(
+        self,
+        pattern: str,
+        path: str,
+        recursive: bool = True,
+        ignore_case: bool = False,
+        context_lines: int = 2,
+    ) -> str:
+        """Search for patterns in files.
+
+        Args:
+            pattern: Search pattern (regex supported).
+            path: File or directory to search in.
+            recursive: Whether to search recursively.
+            ignore_case: Whether to ignore case.
+            context_lines: Lines of context around matches.
+
+        Returns:
+            Search results or error message.
+        """
+        # Validate path against allowed_paths
+        valid, result = self._validate_path(path)
+        if not valid:
+            return f"Error: {result}"
+
+        search_path = result
+
+        # Check path exists
+        if not search_path.exists():
+            return f"Error: Path not found: {path}"
+
+        try:
+            # Build grep command safely
+            cmd = ["grep", "--color=never"]
+
+            # Add recursive flag if searching directory
+            if recursive and search_path.is_dir():
+                cmd.append("-r")
+
+            # Case insensitive
+            if ignore_case:
+                cmd.append("-i")
+
+            # Context lines
+            if context_lines > 0:
+                cmd.extend(["-C", str(context_lines)])
+
+            # Show line numbers
+            cmd.append("-n")
+
+            # Use -- to prevent pattern from being interpreted as option
+            cmd.append("--")
+            cmd.append(pattern)
+            cmd.append(str(search_path))
+
+            # Execute with timeout
+            # Security: cmd is a list (no shell injection), path is validated above
+            result = subprocess.run(  # noqa: S603
+                cmd,
+                capture_output=True,
+                text=True,
+                timeout=30,
+            )
+
+            if result.returncode == 0:
+                # Found matches
+                output = result.stdout
+                # Count actual match lines (not context lines)
+                match_lines = [
+                    line
+                    for line in output.split("\n")
+                    if line and not line.startswith("--")
+                ]
+                match_count = len(match_lines)
+
+                header = f"Found {match_count} matches for '{pattern}' in {path}\n"
+                header += "=" * 60 + "\n"
+                return header + output
+
+            if result.returncode == 1:
+                # No matches found (grep returns 1 for no matches)
+                return f"No matches found for '{pattern}' in {path}"
+
+            # Error occurred
+            error_msg = result.stderr.strip() if result.stderr else "Unknown error"
+            return f"Error: {error_msg}"
+
+        except subprocess.TimeoutExpired:
+            return "Error: Search timed out (>30s). Try narrowing the search path."
+        except FileNotFoundError:
+            return (
+                "Error: grep command not found. Ensure grep is installed on the system."
+            )
+        except Exception as e:
+            return f"Error during search: {e}"

lib/crewai/src/crewai/experimental/environment_tools/list_dir_tool.py

@@ -0,0 +1,147 @@
+"""Tool for listing directory contents."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+from pydantic import BaseModel, Field
+
+from crewai.experimental.environment_tools.base_environment_tool import (
+    BaseEnvironmentTool,
+)
+
+
+class ListDirInput(BaseModel):
+    """Input schema for listing directories."""
+
+    path: str = Field(default=".", description="Directory path to list")
+    pattern: str | None = Field(
+        default=None,
+        description="Glob pattern to filter entries (e.g., '*.py', '*.md')",
+    )
+    recursive: bool = Field(
+        default=False,
+        description="If True, list contents recursively including subdirectories",
+    )
+
+
+class ListDirTool(BaseEnvironmentTool):
+    """List contents of a directory with optional filtering.
+
+    Use this tool to:
+    - Explore project structure
+    - Find specific file types
+    - Check what files exist in a directory
+    - Navigate the file system
+    """
+
+    name: str = "list_directory"
+    description: str = """List contents of a directory.
+
+Use this to explore directories and find files. You can filter by pattern
+and optionally list recursively.
+
+Examples:
+- List current dir: path="."
+- List src folder: path="src/"
+- Find Python files: path=".", pattern="*.py"
+- Recursive listing: path="src/", recursive=True
+"""
+    args_schema: type[BaseModel] = ListDirInput
+
+    def _run(
+        self,
+        path: str = ".",
+        pattern: str | None = None,
+        recursive: bool = False,
+    ) -> str:
+        """List directory contents.
+
+        Args:
+            path: Directory path to list.
+            pattern: Glob pattern to filter entries.
+            recursive: Whether to list recursively.
+
+        Returns:
+            Formatted directory listing or error message.
+        """
+        # Validate path against allowed_paths
+        valid, result = self._validate_path(path)
+        if not valid:
+            return f"Error: {result}"
+
+        assert isinstance(result, Path)  # noqa: S101
+        dir_path = result
+
+        # Check directory exists
+        if not dir_path.exists():
+            return f"Error: Directory not found: {path}"
+
+        if not dir_path.is_dir():
+            return f"Error: Not a directory: {path}"
+
+        try:
+            # Get entries based on pattern and recursive flag
+            if pattern:
+                if recursive:
+                    entries = list(dir_path.rglob(pattern))
+                else:
+                    entries = list(dir_path.glob(pattern))
+            else:
+                if recursive:
+                    entries = list(dir_path.rglob("*"))
+                else:
+                    entries = list(dir_path.iterdir())
+
+            # Filter out hidden files (starting with .)
+            entries = [e for e in entries if not e.name.startswith(".")]
+
+            # Sort: directories first, then files, alphabetically
+            entries.sort(key=lambda x: (not x.is_dir(), x.name.lower()))
+
+            if not entries:
+                if pattern:
+                    return f"No entries matching '{pattern}' in {path}"
+                return f"Directory is empty: {path}"
+
+            # Format output
+            result_lines = [f"Contents of {path}:"]
+            result_lines.append("=" * 60)
+
+            dirs = []
+            files = []
+
+            for entry in entries:
+                # Get relative path for recursive listings
+                if recursive:
+                    display_name = str(entry.relative_to(dir_path))
+                else:
+                    display_name = entry.name
+
+                if entry.is_dir():
+                    dirs.append(f"📁 {display_name}/")
+                else:
+                    try:
+                        size = entry.stat().st_size
+                    except (OSError, PermissionError):
+                        continue  # Skip files we can't stat
+                    size_str = self._format_size(size)
+                    files.append(f"📄 {display_name} ({size_str})")
+
+            # Output directories first, then files
+            if dirs:
+                result_lines.extend(dirs)
+            if files:
+                if dirs:
+                    result_lines.append("")  # Blank line between dirs and files
+                result_lines.extend(files)
+
+            result_lines.append("")
+            result_lines.append(f"Total: {len(dirs)} directories, {len(files)} files")
+
+            return "\n".join(result_lines)
+
+        except PermissionError:
+            return f"Error: Permission denied: {path}"
+        except Exception as e:
+            return f"Error listing directory: {e}"

lib/crewai/src/crewai/flow/flow_trackable.py

@@ -1,5 +1,4 @@
 import inspect
-from typing import Any
 
 from pydantic import BaseModel, Field, InstanceOf, model_validator
 from typing_extensions import Self
@@ -15,14 +14,14 @@ class FlowTrackable(BaseModel):
     inspecting the call stack.
     """
 
-    parent_flow: InstanceOf[Flow[Any]] | None = Field(
+    parent_flow: InstanceOf[Flow] | None = Field(
         default=None,
         description="The parent flow of the instance, if it was created inside a flow.",
     )
 
     @model_validator(mode="after")
     def _set_parent_flow(self) -> Self:
-        max_depth = 8
+        max_depth = 5
         frame = inspect.currentframe()
 
         try:

lib/crewai/src/crewai/llms/providers/azure/completion.py

@@ -443,7 +443,7 @@ class AzureCompletion(BaseLLM):
             params["presence_penalty"] = self.presence_penalty
         if self.max_tokens is not None:
             params["max_tokens"] = self.max_tokens
-        if self.stop and self.supports_stop_words():
+        if self.stop:
             params["stop"] = self.stop
 
         # Handle tools/functions for Azure OpenAI models
@@ -931,28 +931,8 @@ class AzureCompletion(BaseLLM):
         return self.is_openai_model
 
     def supports_stop_words(self) -> bool:
-        """Check if the model supports stop words.
-
-        Models using the Responses API (GPT-5 family, o-series reasoning models,
-        computer-use-preview) do not support stop sequences.
-        See: https://learn.microsoft.com/en-us/azure/ai-foundry/foundry-models/concepts/models-sold-directly-by-azure
-        """
-        model_lower = self.model.lower() if self.model else ""
-
-        if "gpt-5" in model_lower:
-            return False
-
-        o_series_models = ["o1", "o3", "o4", "o1-mini", "o3-mini", "o4-mini"]
-
-        responses_api_models = ["computer-use-preview"]
-
-        unsupported_stop_models = o_series_models + responses_api_models
-
-        for unsupported in unsupported_stop_models:
-            if unsupported in model_lower:
-                return False
-
-        return True
+        """Check if the model supports stop words."""
+        return True  # Most Azure models support stop sequences
 
     def get_context_window_size(self) -> int:
         """Get the context window size for the model."""

lib/crewai/src/crewai/utilities/agent_utils.py

@@ -229,48 +229,6 @@ def enforce_rpm_limit(
         request_within_rpm_limit()
 
 
-def _extract_tools_from_context(
-    executor_context: CrewAgentExecutor | LiteAgent | None,
-) -> list[dict[str, Any]] | None:
-    """Extract tools from executor context and convert to LLM-compatible format.
-
-    Args:
-        executor_context: The executor context containing tools.
-
-    Returns:
-        List of tool dictionaries in LLM-compatible format, or None if no tools.
-    """
-    if executor_context is None:
-        return None
-
-    # Get tools from executor context
-    # CrewAgentExecutor has 'tools' attribute, LiteAgent has '_parsed_tools'
-    tools: list[CrewStructuredTool] | None = None
-    if hasattr(executor_context, "tools"):
-        context_tools = executor_context.tools
-        if isinstance(context_tools, list) and len(context_tools) > 0:
-            tools = context_tools
-    if tools is None and hasattr(executor_context, "_parsed_tools"):
-        parsed_tools = executor_context._parsed_tools
-        if isinstance(parsed_tools, list) and len(parsed_tools) > 0:
-            tools = parsed_tools
-
-    if not tools:
-        return None
-
-    # Convert CrewStructuredTool to dict format expected by LLM
-    tool_dicts: list[dict[str, Any]] = []
-    for tool in tools:
-        tool_dict: dict[str, Any] = {
-            "name": tool.name,
-            "description": tool.description,
-            "args_schema": tool.args_schema,
-        }
-        tool_dicts.append(tool_dict)
-
-    return tool_dicts if tool_dicts else None
-
-
 def get_llm_response(
     llm: LLM | BaseLLM,
     messages: list[LLMMessage],
@@ -306,29 +264,14 @@ def get_llm_response(
             raise ValueError("LLM call blocked by before_llm_call hook")
         messages = executor_context.messages
 
-    # Extract tools from executor context for native function calling support
-    tools = _extract_tools_from_context(executor_context)
-
     try:
-        # Only pass tools parameter if tools are available to maintain backward compatibility
-        # with code that checks "tools" in kwargs
-        if tools is not None:
-            answer = llm.call(
-                messages,
-                tools=tools,
-                callbacks=callbacks,
-                from_task=from_task,
-                from_agent=from_agent,  # type: ignore[arg-type]
-                response_model=response_model,
-            )
-        else:
-            answer = llm.call(
-                messages,
-                callbacks=callbacks,
-                from_task=from_task,
-                from_agent=from_agent,  # type: ignore[arg-type]
-                response_model=response_model,
-            )
+        answer = llm.call(
+            messages,
+            callbacks=callbacks,
+            from_task=from_task,
+            from_agent=from_agent,  # type: ignore[arg-type]
+            response_model=response_model,
+        )
     except Exception as e:
         raise e
     if not answer:
@@ -349,7 +292,7 @@ async def aget_llm_response(
     from_task: Task | None = None,
     from_agent: Agent | LiteAgent | None = None,
     response_model: type[BaseModel] | None = None,
-    executor_context: CrewAgentExecutor | LiteAgent | None = None,
+    executor_context: CrewAgentExecutor | None = None,
 ) -> str:
     """Call the LLM asynchronously and return the response.
 
@@ -375,29 +318,14 @@ async def aget_llm_response(
             raise ValueError("LLM call blocked by before_llm_call hook")
         messages = executor_context.messages
 
-    # Extract tools from executor context for native function calling support
-    tools = _extract_tools_from_context(executor_context)
-
     try:
-        # Only pass tools parameter if tools are available to maintain backward compatibility
-        # with code that checks "tools" in kwargs
-        if tools is not None:
-            answer = await llm.acall(
-                messages,
-                tools=tools,
-                callbacks=callbacks,
-                from_task=from_task,
-                from_agent=from_agent,  # type: ignore[arg-type]
-                response_model=response_model,
-            )
-        else:
-            answer = await llm.acall(
-                messages,
-                callbacks=callbacks,
-                from_task=from_task,
-                from_agent=from_agent,  # type: ignore[arg-type]
-                response_model=response_model,
-            )
+        answer = await llm.acall(
+            messages,
+            callbacks=callbacks,
+            from_task=from_task,
+            from_agent=from_agent,  # type: ignore[arg-type]
+            response_model=response_model,
+        )
     except Exception as e:
         raise e
     if not answer:

lib/crewai/tests/experimental/environment_tools/test_environment_tools.py

@@ -0,0 +1,408 @@
+"""Tests for experimental environment tools."""
+
+from __future__ import annotations
+
+import os
+import tempfile
+from collections.abc import Generator
+from pathlib import Path
+
+import pytest
+
+from crewai.experimental.environment_tools import (
+    BaseEnvironmentTool,
+    EnvironmentTools,
+    FileReadTool,
+    FileSearchTool,
+    GrepTool,
+    ListDirTool,
+)
+
+
+# ============================================================================
+# Fixtures
+# ============================================================================
+
+
+@pytest.fixture
+def temp_dir() -> Generator[str, None, None]:
+    """Create a temporary directory with test files."""
+    with tempfile.TemporaryDirectory() as tmpdir:
+        # Create test files
+        test_file = Path(tmpdir) / "test.txt"
+        test_file.write_text("Line 1\nLine 2\nLine 3\nLine 4\nLine 5\n")
+
+        python_file = Path(tmpdir) / "example.py"
+        python_file.write_text("def hello():\n    print('Hello World')\n")
+
+        # Create subdirectory with files
+        subdir = Path(tmpdir) / "subdir"
+        subdir.mkdir()
+        (subdir / "nested.txt").write_text("Nested content\n")
+        (subdir / "another.py").write_text("# Another Python file\n")
+
+        yield tmpdir
+
+
+@pytest.fixture
+def restricted_temp_dir() -> Generator[tuple[str, str], None, None]:
+    """Create two directories - one allowed, one not."""
+    with tempfile.TemporaryDirectory() as allowed_dir:
+        with tempfile.TemporaryDirectory() as forbidden_dir:
+            # Create files in both
+            (Path(allowed_dir) / "allowed.txt").write_text("Allowed content\n")
+            (Path(forbidden_dir) / "forbidden.txt").write_text("Forbidden content\n")
+
+            yield allowed_dir, forbidden_dir
+
+
+# ============================================================================
+# BaseEnvironmentTool Tests
+# ============================================================================
+
+
+class TestBaseEnvironmentTool:
+    """Tests for BaseEnvironmentTool path validation."""
+
+    def test_default_allowed_paths_is_current_directory(self) -> None:
+        """Default allowed_paths should be current directory for security."""
+        tool = FileReadTool()
+
+        assert tool.allowed_paths == ["."]
+
+    def test_validate_path_explicit_no_restrictions(self, temp_dir: str) -> None:
+        """With explicit empty allowed_paths, all paths should be allowed."""
+        tool = FileReadTool(allowed_paths=[])
+        valid, result = tool._validate_path(temp_dir)
+
+        assert valid is True
+        assert isinstance(result, Path)
+
+    def test_validate_path_within_allowed(self, temp_dir: str) -> None:
+        """Paths within allowed_paths should be valid."""
+        tool = FileReadTool(allowed_paths=[temp_dir])
+        test_file = os.path.join(temp_dir, "test.txt")
+
+        valid, result = tool._validate_path(test_file)
+
+        assert valid is True
+        assert isinstance(result, Path)
+
+    def test_validate_path_outside_allowed(self, restricted_temp_dir: tuple[str, str]) -> None:
+        """Paths outside allowed_paths should be rejected."""
+        allowed_dir, forbidden_dir = restricted_temp_dir
+        tool = FileReadTool(allowed_paths=[allowed_dir])
+
+        forbidden_file = os.path.join(forbidden_dir, "forbidden.txt")
+        valid, result = tool._validate_path(forbidden_file)
+
+        assert valid is False
+        assert isinstance(result, str)
+        assert "outside allowed paths" in result
+
+    def test_format_size(self) -> None:
+        """Test human-readable size formatting."""
+        tool = FileReadTool()
+
+        assert tool._format_size(500) == "500B"
+        assert tool._format_size(1024) == "1.0KB"
+        assert tool._format_size(1536) == "1.5KB"
+        assert tool._format_size(1024 * 1024) == "1.0MB"
+        assert tool._format_size(1024 * 1024 * 1024) == "1.0GB"
+
+
+# ============================================================================
+# FileReadTool Tests
+# ============================================================================
+
+
+class TestFileReadTool:
+    """Tests for FileReadTool."""
+
+    def test_read_entire_file(self, temp_dir: str) -> None:
+        """Should read entire file contents."""
+        tool = FileReadTool(allowed_paths=[temp_dir])
+        test_file = os.path.join(temp_dir, "test.txt")
+
+        result = tool._run(path=test_file)
+
+        assert "Line 1" in result
+        assert "Line 2" in result
+        assert "Line 5" in result
+        assert "File:" in result  # Metadata header
+
+    def test_read_with_line_range(self, temp_dir: str) -> None:
+        """Should read specific line range."""
+        tool = FileReadTool(allowed_paths=[temp_dir])
+        test_file = os.path.join(temp_dir, "test.txt")
+
+        result = tool._run(path=test_file, start_line=2, line_count=2)
+
+        assert "Line 2" in result
+        assert "Line 3" in result
+        # Should not include lines outside range
+        assert "Line 1" not in result.split("=" * 60)[-1]  # Check content after header
+
+    def test_read_file_not_found(self, temp_dir: str) -> None:
+        """Should return error for missing file."""
+        tool = FileReadTool(allowed_paths=[temp_dir])
+        missing_file = os.path.join(temp_dir, "nonexistent.txt")
+
+        result = tool._run(path=missing_file)
+
+        assert "Error: File not found" in result
+
+    def test_read_file_path_restricted(self, restricted_temp_dir: tuple[str, str]) -> None:
+        """Should reject paths outside allowed_paths."""
+        allowed_dir, forbidden_dir = restricted_temp_dir
+        tool = FileReadTool(allowed_paths=[allowed_dir])
+
+        forbidden_file = os.path.join(forbidden_dir, "forbidden.txt")
+        result = tool._run(path=forbidden_file)
+
+        assert "Error:" in result
+        assert "outside allowed paths" in result
+
+
+# ============================================================================
+# ListDirTool Tests
+# ============================================================================
+
+
+class TestListDirTool:
+    """Tests for ListDirTool."""
+
+    def test_list_directory(self, temp_dir: str) -> None:
+        """Should list directory contents."""
+        tool = ListDirTool(allowed_paths=[temp_dir])
+
+        result = tool._run(path=temp_dir)
+
+        assert "test.txt" in result
+        assert "example.py" in result
+        assert "subdir" in result
+        assert "Total:" in result
+
+    def test_list_with_pattern(self, temp_dir: str) -> None:
+        """Should filter by pattern."""
+        tool = ListDirTool(allowed_paths=[temp_dir])
+
+        result = tool._run(path=temp_dir, pattern="*.py")
+
+        assert "example.py" in result
+        assert "test.txt" not in result
+
+    def test_list_recursive(self, temp_dir: str) -> None:
+        """Should list recursively when enabled."""
+        tool = ListDirTool(allowed_paths=[temp_dir])
+
+        result = tool._run(path=temp_dir, recursive=True)
+
+        assert "nested.txt" in result
+        assert "another.py" in result
+
+    def test_list_nonexistent_directory(self, temp_dir: str) -> None:
+        """Should return error for missing directory."""
+        tool = ListDirTool(allowed_paths=[temp_dir])
+
+        result = tool._run(path=os.path.join(temp_dir, "nonexistent"))
+
+        assert "Error: Directory not found" in result
+
+    def test_list_path_restricted(self, restricted_temp_dir: tuple[str, str]) -> None:
+        """Should reject paths outside allowed_paths."""
+        allowed_dir, forbidden_dir = restricted_temp_dir
+        tool = ListDirTool(allowed_paths=[allowed_dir])
+
+        result = tool._run(path=forbidden_dir)
+
+        assert "Error:" in result
+        assert "outside allowed paths" in result
+
+
+# ============================================================================
+# GrepTool Tests
+# ============================================================================
+
+
+class TestGrepTool:
+    """Tests for GrepTool."""
+
+    def test_grep_finds_pattern(self, temp_dir: str) -> None:
+        """Should find matching patterns."""
+        tool = GrepTool(allowed_paths=[temp_dir])
+        test_file = os.path.join(temp_dir, "test.txt")
+
+        result = tool._run(pattern="Line 2", path=test_file)
+
+        assert "Line 2" in result
+        assert "matches" in result.lower() or "found" in result.lower()
+
+    def test_grep_no_matches(self, temp_dir: str) -> None:
+        """Should report when no matches found."""
+        tool = GrepTool(allowed_paths=[temp_dir])
+        test_file = os.path.join(temp_dir, "test.txt")
+
+        result = tool._run(pattern="nonexistent pattern xyz", path=test_file)
+
+        assert "No matches found" in result
+
+    def test_grep_recursive(self, temp_dir: str) -> None:
+        """Should search recursively in directories."""
+        tool = GrepTool(allowed_paths=[temp_dir])
+
+        result = tool._run(pattern="Nested", path=temp_dir, recursive=True)
+
+        assert "Nested" in result
+
+    def test_grep_case_insensitive(self, temp_dir: str) -> None:
+        """Should support case-insensitive search."""
+        tool = GrepTool(allowed_paths=[temp_dir])
+        test_file = os.path.join(temp_dir, "test.txt")
+
+        result = tool._run(pattern="LINE", path=test_file, ignore_case=True)
+
+        assert "Line" in result or "matches" in result.lower()
+
+    def test_grep_path_restricted(self, restricted_temp_dir: tuple[str, str]) -> None:
+        """Should reject paths outside allowed_paths."""
+        allowed_dir, forbidden_dir = restricted_temp_dir
+        tool = GrepTool(allowed_paths=[allowed_dir])
+
+        result = tool._run(pattern="test", path=forbidden_dir)
+
+        assert "Error:" in result
+        assert "outside allowed paths" in result
+
+
+# ============================================================================
+# FileSearchTool Tests
+# ============================================================================
+
+
+class TestFileSearchTool:
+    """Tests for FileSearchTool."""
+
+    def test_find_files_by_pattern(self, temp_dir: str) -> None:
+        """Should find files matching pattern."""
+        tool = FileSearchTool(allowed_paths=[temp_dir])
+
+        result = tool._run(pattern="*.py", path=temp_dir)
+
+        assert "example.py" in result
+        assert "another.py" in result
+
+    def test_find_no_matches(self, temp_dir: str) -> None:
+        """Should report when no files match."""
+        tool = FileSearchTool(allowed_paths=[temp_dir])
+
+        result = tool._run(pattern="*.xyz", path=temp_dir)
+
+        assert "No" in result and "found" in result
+
+    def test_find_files_only(self, temp_dir: str) -> None:
+        """Should filter to files only."""
+        tool = FileSearchTool(allowed_paths=[temp_dir])
+
+        result = tool._run(pattern="*", path=temp_dir, file_type="file")
+
+        # Should include files
+        assert "test.txt" in result or "example.py" in result
+        # Directories should have trailing slash in output
+        # Check that subdir is not listed as a file
+
+    def test_find_dirs_only(self, temp_dir: str) -> None:
+        """Should filter to directories only."""
+        tool = FileSearchTool(allowed_paths=[temp_dir])
+
+        result = tool._run(pattern="*", path=temp_dir, file_type="dir")
+
+        assert "subdir" in result
+
+    def test_find_path_restricted(self, restricted_temp_dir: tuple[str, str]) -> None:
+        """Should reject paths outside allowed_paths."""
+        allowed_dir, forbidden_dir = restricted_temp_dir
+        tool = FileSearchTool(allowed_paths=[allowed_dir])
+
+        result = tool._run(pattern="*", path=forbidden_dir)
+
+        assert "Error:" in result
+        assert "outside allowed paths" in result
+
+
+# ============================================================================
+# EnvironmentTools Manager Tests
+# ============================================================================
+
+
+class TestEnvironmentTools:
+    """Tests for EnvironmentTools manager class."""
+
+    def test_default_allowed_paths_is_current_directory(self) -> None:
+        """Default should restrict to current directory for security."""
+        env_tools = EnvironmentTools()
+        tools = env_tools.tools()
+
+        # All tools should default to current directory
+        for tool in tools:
+            assert isinstance(tool, BaseEnvironmentTool)
+            assert tool.allowed_paths == ["."]
+
+    def test_explicit_empty_allowed_paths_allows_all(self) -> None:
+        """Passing empty list should allow all paths."""
+        env_tools = EnvironmentTools(allowed_paths=[])
+        tools = env_tools.tools()
+
+        for tool in tools:
+            assert isinstance(tool, BaseEnvironmentTool)
+            assert tool.allowed_paths == []
+
+    def test_returns_all_tools_by_default(self) -> None:
+        """Should return all four tools by default."""
+        env_tools = EnvironmentTools()
+        tools = env_tools.tools()
+
+        assert len(tools) == 4
+
+        tool_names = [t.name for t in tools]
+        assert "read_file" in tool_names
+        assert "list_directory" in tool_names
+        assert "grep_search" in tool_names
+        assert "find_files" in tool_names
+
+    def test_exclude_grep(self) -> None:
+        """Should exclude grep tool when disabled."""
+        env_tools = EnvironmentTools(include_grep=False)
+        tools = env_tools.tools()
+
+        assert len(tools) == 3
+        tool_names = [t.name for t in tools]
+        assert "grep_search" not in tool_names
+
+    def test_exclude_search(self) -> None:
+        """Should exclude search tool when disabled."""
+        env_tools = EnvironmentTools(include_search=False)
+        tools = env_tools.tools()
+
+        assert len(tools) == 3
+        tool_names = [t.name for t in tools]
+        assert "find_files" not in tool_names
+
+    def test_allowed_paths_propagated(self, temp_dir: str) -> None:
+        """Should propagate allowed_paths to all tools."""
+        env_tools = EnvironmentTools(allowed_paths=[temp_dir])
+        tools = env_tools.tools()
+
+        for tool in tools:
+            assert isinstance(tool, BaseEnvironmentTool)
+            assert tool.allowed_paths == [temp_dir]
+
+    def test_tools_are_base_tool_instances(self) -> None:
+        """All returned tools should be BaseTool instances."""
+        from crewai.tools.base_tool import BaseTool
+
+        env_tools = EnvironmentTools()
+        tools = env_tools.tools()
+
+        for tool in tools:
+            assert isinstance(tool, BaseTool)

lib/crewai/tests/llms/azure/test_azure.py

@@ -515,94 +515,6 @@ def test_azure_supports_stop_words():
     assert llm.supports_stop_words() == True
 
 
-def test_azure_gpt5_models_do_not_support_stop_words():
-    """
-    Test that GPT-5 family models do not support stop words.
-    GPT-5 models use the Responses API which doesn't support stop sequences.
-    See: https://learn.microsoft.com/en-us/azure/ai-foundry/foundry-models/concepts/models-sold-directly-by-azure
-    """
-    # GPT-5 base models
-    gpt5_models = [
-        "azure/gpt-5",
-        "azure/gpt-5-mini",
-        "azure/gpt-5-nano",
-        "azure/gpt-5-chat",
-        # GPT-5.1 series
-        "azure/gpt-5.1",
-        "azure/gpt-5.1-chat",
-        "azure/gpt-5.1-codex",
-        "azure/gpt-5.1-codex-mini",
-        # GPT-5.2 series
-        "azure/gpt-5.2",
-        "azure/gpt-5.2-chat",
-    ]
-
-    for model_name in gpt5_models:
-        llm = LLM(model=model_name)
-        assert llm.supports_stop_words() == False, f"Expected {model_name} to NOT support stop words"
-
-
-def test_azure_o_series_models_do_not_support_stop_words():
-    """
-    Test that o-series reasoning models do not support stop words.
-    """
-    o_series_models = [
-        "azure/o1",
-        "azure/o1-mini",
-        "azure/o3",
-        "azure/o3-mini",
-        "azure/o4",
-        "azure/o4-mini",
-    ]
-
-    for model_name in o_series_models:
-        llm = LLM(model=model_name)
-        assert llm.supports_stop_words() == False, f"Expected {model_name} to NOT support stop words"
-
-
-def test_azure_responses_api_models_do_not_support_stop_words():
-    """
-    Test that models using the Responses API do not support stop words.
-    """
-    responses_api_models = [
-        "azure/computer-use-preview",
-    ]
-
-    for model_name in responses_api_models:
-        llm = LLM(model=model_name)
-        assert llm.supports_stop_words() == False, f"Expected {model_name} to NOT support stop words"
-
-
-def test_azure_stop_words_not_included_for_unsupported_models():
-    """
-    Test that stop words are not included in completion params for models that don't support them.
-    """
-    with patch.dict(os.environ, {
-        "AZURE_API_KEY": "test-key",
-        "AZURE_ENDPOINT": "https://models.inference.ai.azure.com"
-    }):
-        # Test GPT-5 model - stop should NOT be included even if set
-        llm_gpt5 = LLM(
-            model="azure/gpt-5-nano",
-            stop=["STOP", "END"]
-        )
-        params = llm_gpt5._prepare_completion_params(
-            messages=[{"role": "user", "content": "test"}]
-        )
-        assert "stop" not in params, "stop should not be included for GPT-5 models"
-
-        # Test regular model - stop SHOULD be included
-        llm_gpt4 = LLM(
-            model="azure/gpt-4",
-            stop=["STOP", "END"]
-        )
-        params = llm_gpt4._prepare_completion_params(
-            messages=[{"role": "user", "content": "test"}]
-        )
-        assert "stop" in params, "stop should be included for GPT-4 models"
-        assert params["stop"] == ["STOP", "END"]
-
-
 def test_azure_context_window_size():
     """
     Test that Azure models return correct context window sizes

lib/crewai/tests/test_crew.py

@@ -4500,71 +4500,6 @@ def test_crew_copy_with_memory():
         pytest.fail(f"Copying crew raised an unexpected exception: {e}")
 
 
-def test_sets_parent_flow_when_using_crewbase_pattern_inside_flow():
-    @CrewBase
-    class TestCrew:
-        agents_config = None
-        tasks_config = None
-
-        agents: list[BaseAgent]
-        tasks: list[Task]
-
-        @agent
-        def researcher(self) -> Agent:
-            return Agent(
-                role="Researcher",
-                goal="Research things",
-                backstory="Expert researcher",
-            )
-
-        @agent
-        def writer_agent(self) -> Agent:
-            return Agent(
-                role="Writer",
-                goal="Write things",
-                backstory="Expert writer",
-            )
-
-        @task
-        def research_task(self) -> Task:
-            return Task(
-                description="Test task for researcher",
-                expected_output="output",
-                agent=self.researcher(),
-            )
-
-        @task
-        def write_task(self) -> Task:
-            return Task(
-                description="Test task for writer",
-                expected_output="output",
-                agent=self.writer_agent(),
-            )
-
-        @crew
-        def crew(self) -> Crew:
-            return Crew(
-                agents=self.agents,
-                tasks=self.tasks,
-                process=Process.sequential,
-            )
-
-    captured_crew = None
-
-    class MyFlow(Flow):
-        @start()
-        def start_method(self):
-            nonlocal captured_crew
-            captured_crew = TestCrew().crew()
-            return captured_crew
-
-    flow = MyFlow()
-    flow.kickoff()
-
-    assert captured_crew is not None
-    assert captured_crew.parent_flow is flow
-
-
 def test_sets_parent_flow_when_outside_flow(researcher, writer):
     crew = Crew(
         agents=[researcher, writer],

lib/crewai/tests/utilities/test_agent_utils.py

@@ -1,457 +0,0 @@
-"""Unit tests for agent_utils module.
-
-Tests the utility functions for agent execution including tool extraction
-and LLM response handling.
-"""
-
-from unittest.mock import AsyncMock, MagicMock, Mock, patch
-
-import pytest
-from pydantic import BaseModel, Field
-
-from crewai.tools.structured_tool import CrewStructuredTool
-from crewai.utilities.agent_utils import (
-    _extract_tools_from_context,
-    aget_llm_response,
-    get_llm_response,
-)
-from crewai.utilities.printer import Printer
-
-
-class MockArgsSchema(BaseModel):
-    """Mock args schema for testing."""
-
-    query: str = Field(description="The search query")
-
-
-class TestExtractToolsFromContext:
-    """Test _extract_tools_from_context function."""
-
-    def test_returns_none_when_context_is_none(self):
-        """Test that None is returned when executor_context is None."""
-        result = _extract_tools_from_context(None)
-        assert result is None
-
-    def test_returns_none_when_no_tools_attribute(self):
-        """Test that None is returned when context has no tools."""
-        mock_context = Mock(spec=[])
-        result = _extract_tools_from_context(mock_context)
-        assert result is None
-
-    def test_returns_none_when_tools_is_empty(self):
-        """Test that None is returned when tools list is empty."""
-        mock_context = Mock()
-        mock_context.tools = []
-        result = _extract_tools_from_context(mock_context)
-        assert result is None
-
-    def test_extracts_tools_from_crew_agent_executor(self):
-        """Test tool extraction from CrewAgentExecutor (has 'tools' attribute)."""
-        mock_tool = CrewStructuredTool(
-            name="search_tool",
-            description="A tool for searching",
-            args_schema=MockArgsSchema,
-            func=lambda query: f"Results for {query}",
-        )
-
-        mock_context = Mock()
-        mock_context.tools = [mock_tool]
-
-        result = _extract_tools_from_context(mock_context)
-
-        assert result is not None
-        assert len(result) == 1
-        assert result[0]["name"] == "search_tool"
-        assert result[0]["description"] == "A tool for searching"
-        assert result[0]["args_schema"] == MockArgsSchema
-
-    def test_extracts_tools_from_lite_agent(self):
-        """Test tool extraction from LiteAgent (has '_parsed_tools' attribute)."""
-        mock_tool = CrewStructuredTool(
-            name="calculator_tool",
-            description="A tool for calculations",
-            args_schema=MockArgsSchema,
-            func=lambda query: f"Calculated {query}",
-        )
-
-        mock_context = Mock(spec=["_parsed_tools"])
-        mock_context._parsed_tools = [mock_tool]
-
-        result = _extract_tools_from_context(mock_context)
-
-        assert result is not None
-        assert len(result) == 1
-        assert result[0]["name"] == "calculator_tool"
-        assert result[0]["description"] == "A tool for calculations"
-        assert result[0]["args_schema"] == MockArgsSchema
-
-    def test_extracts_multiple_tools(self):
-        """Test extraction of multiple tools."""
-        tool1 = CrewStructuredTool(
-            name="tool1",
-            description="First tool",
-            args_schema=MockArgsSchema,
-            func=lambda query: "result1",
-        )
-        tool2 = CrewStructuredTool(
-            name="tool2",
-            description="Second tool",
-            args_schema=MockArgsSchema,
-            func=lambda query: "result2",
-        )
-
-        mock_context = Mock()
-        mock_context.tools = [tool1, tool2]
-
-        result = _extract_tools_from_context(mock_context)
-
-        assert result is not None
-        assert len(result) == 2
-        assert result[0]["name"] == "tool1"
-        assert result[1]["name"] == "tool2"
-
-    def test_prefers_tools_over_parsed_tools(self):
-        """Test that 'tools' attribute is preferred over '_parsed_tools'."""
-        tool_from_tools = CrewStructuredTool(
-            name="from_tools",
-            description="Tool from tools attribute",
-            args_schema=MockArgsSchema,
-            func=lambda query: "from_tools",
-        )
-        tool_from_parsed = CrewStructuredTool(
-            name="from_parsed",
-            description="Tool from _parsed_tools attribute",
-            args_schema=MockArgsSchema,
-            func=lambda query: "from_parsed",
-        )
-
-        mock_context = Mock()
-        mock_context.tools = [tool_from_tools]
-        mock_context._parsed_tools = [tool_from_parsed]
-
-        result = _extract_tools_from_context(mock_context)
-
-        assert result is not None
-        assert len(result) == 1
-        assert result[0]["name"] == "from_tools"
-
-
-class TestGetLlmResponse:
-    """Test get_llm_response function."""
-
-    @pytest.fixture
-    def mock_llm(self):
-        """Create a mock LLM."""
-        llm = Mock()
-        llm.call = Mock(return_value="LLM response")
-        return llm
-
-    @pytest.fixture
-    def mock_printer(self):
-        """Create a mock printer."""
-        return Mock(spec=Printer)
-
-    def test_passes_tools_to_llm_call(self, mock_llm, mock_printer):
-        """Test that tools are extracted and passed to llm.call()."""
-        mock_tool = CrewStructuredTool(
-            name="test_tool",
-            description="A test tool",
-            args_schema=MockArgsSchema,
-            func=lambda query: "result",
-        )
-
-        mock_context = Mock()
-        mock_context.tools = [mock_tool]
-        mock_context.messages = [{"role": "user", "content": "test"}]
-        mock_context.before_llm_call_hooks = []
-        mock_context.after_llm_call_hooks = []
-
-        with patch(
-            "crewai.utilities.agent_utils._setup_before_llm_call_hooks",
-            return_value=True,
-        ):
-            with patch(
-                "crewai.utilities.agent_utils._setup_after_llm_call_hooks",
-                return_value="LLM response",
-            ):
-                result = get_llm_response(
-                    llm=mock_llm,
-                    messages=[{"role": "user", "content": "test"}],
-                    callbacks=[],
-                    printer=mock_printer,
-                    executor_context=mock_context,
-                )
-
-        # Verify llm.call was called with tools parameter
-        mock_llm.call.assert_called_once()
-        call_kwargs = mock_llm.call.call_args[1]
-        assert "tools" in call_kwargs
-        assert call_kwargs["tools"] is not None
-        assert len(call_kwargs["tools"]) == 1
-        assert call_kwargs["tools"][0]["name"] == "test_tool"
-
-    def test_does_not_pass_tools_when_no_context(self, mock_llm, mock_printer):
-        """Test that tools parameter is not passed when no executor_context."""
-        result = get_llm_response(
-            llm=mock_llm,
-            messages=[{"role": "user", "content": "test"}],
-            callbacks=[],
-            printer=mock_printer,
-            executor_context=None,
-        )
-
-        mock_llm.call.assert_called_once()
-        call_kwargs = mock_llm.call.call_args[1]
-        # tools should NOT be in kwargs when there are no tools
-        # This maintains backward compatibility with code that checks "tools" in kwargs
-        assert "tools" not in call_kwargs
-
-    def test_does_not_pass_tools_when_context_has_no_tools(
-        self, mock_llm, mock_printer
-    ):
-        """Test that tools parameter is not passed when context has no tools."""
-        mock_context = Mock()
-        mock_context.tools = []
-        mock_context.messages = [{"role": "user", "content": "test"}]
-        mock_context.before_llm_call_hooks = []
-        mock_context.after_llm_call_hooks = []
-
-        with patch(
-            "crewai.utilities.agent_utils._setup_before_llm_call_hooks",
-            return_value=True,
-        ):
-            with patch(
-                "crewai.utilities.agent_utils._setup_after_llm_call_hooks",
-                return_value="LLM response",
-            ):
-                result = get_llm_response(
-                    llm=mock_llm,
-                    messages=[{"role": "user", "content": "test"}],
-                    callbacks=[],
-                    printer=mock_printer,
-                    executor_context=mock_context,
-                )
-
-        mock_llm.call.assert_called_once()
-        call_kwargs = mock_llm.call.call_args[1]
-        # tools should NOT be in kwargs when there are no tools
-        # This maintains backward compatibility with code that checks "tools" in kwargs
-        assert "tools" not in call_kwargs
-
-
-class TestAgetLlmResponse:
-    """Test aget_llm_response async function."""
-
-    @pytest.fixture
-    def mock_llm(self):
-        """Create a mock LLM with async call."""
-        llm = Mock()
-        llm.acall = AsyncMock(return_value="Async LLM response")
-        return llm
-
-    @pytest.fixture
-    def mock_printer(self):
-        """Create a mock printer."""
-        return Mock(spec=Printer)
-
-    @pytest.mark.asyncio
-    async def test_passes_tools_to_llm_acall(self, mock_llm, mock_printer):
-        """Test that tools are extracted and passed to llm.acall()."""
-        mock_tool = CrewStructuredTool(
-            name="async_test_tool",
-            description="An async test tool",
-            args_schema=MockArgsSchema,
-            func=lambda query: "async result",
-        )
-
-        mock_context = Mock()
-        mock_context.tools = [mock_tool]
-        mock_context.messages = [{"role": "user", "content": "async test"}]
-        mock_context.before_llm_call_hooks = []
-        mock_context.after_llm_call_hooks = []
-
-        with patch(
-            "crewai.utilities.agent_utils._setup_before_llm_call_hooks",
-            return_value=True,
-        ):
-            with patch(
-                "crewai.utilities.agent_utils._setup_after_llm_call_hooks",
-                return_value="Async LLM response",
-            ):
-                result = await aget_llm_response(
-                    llm=mock_llm,
-                    messages=[{"role": "user", "content": "async test"}],
-                    callbacks=[],
-                    printer=mock_printer,
-                    executor_context=mock_context,
-                )
-
-        # Verify llm.acall was called with tools parameter
-        mock_llm.acall.assert_called_once()
-        call_kwargs = mock_llm.acall.call_args[1]
-        assert "tools" in call_kwargs
-        assert call_kwargs["tools"] is not None
-        assert len(call_kwargs["tools"]) == 1
-        assert call_kwargs["tools"][0]["name"] == "async_test_tool"
-
-    @pytest.mark.asyncio
-    async def test_does_not_pass_tools_when_no_context(self, mock_llm, mock_printer):
-        """Test that tools parameter is not passed when no executor_context."""
-        result = await aget_llm_response(
-            llm=mock_llm,
-            messages=[{"role": "user", "content": "test"}],
-            callbacks=[],
-            printer=mock_printer,
-            executor_context=None,
-        )
-
-        mock_llm.acall.assert_called_once()
-        call_kwargs = mock_llm.acall.call_args[1]
-        # tools should NOT be in kwargs when there are no tools
-        # This maintains backward compatibility with code that checks "tools" in kwargs
-        assert "tools" not in call_kwargs
-
-
-class TestToolsPassedToGeminiModels:
-    """Test that tools are properly passed for Gemini models.
-
-    This test class specifically addresses GitHub issue #4238 where
-    Gemini models fail with UNEXPECTED_TOOL_CALL errors because tools
-    were not being passed to llm.call().
-    """
-
-    @pytest.fixture
-    def mock_gemini_llm(self):
-        """Create a mock Gemini LLM."""
-        llm = Mock()
-        llm.model = "gemini/gemini-2.0-flash-exp"
-        llm.call = Mock(return_value="Gemini response with tool call")
-        return llm
-
-    @pytest.fixture
-    def mock_printer(self):
-        """Create a mock printer."""
-        return Mock(spec=Printer)
-
-    @pytest.fixture
-    def delegation_tools(self):
-        """Create mock delegation tools similar to hierarchical crew setup."""
-
-        class DelegateWorkArgsSchema(BaseModel):
-            task: str = Field(description="The task to delegate")
-            context: str = Field(description="Context for the task")
-            coworker: str = Field(description="The coworker to delegate to")
-
-        class AskQuestionArgsSchema(BaseModel):
-            question: str = Field(description="The question to ask")
-            context: str = Field(description="Context for the question")
-            coworker: str = Field(description="The coworker to ask")
-
-        delegate_tool = CrewStructuredTool(
-            name="Delegate work to coworker",
-            description="Delegate a specific task to one of your coworkers",
-            args_schema=DelegateWorkArgsSchema,
-            func=lambda task, context, coworker: f"Delegated {task} to {coworker}",
-        )
-
-        ask_question_tool = CrewStructuredTool(
-            name="Ask question to coworker",
-            description="Ask a specific question to one of your coworkers",
-            args_schema=AskQuestionArgsSchema,
-            func=lambda question, context, coworker: f"Asked {question} to {coworker}",
-        )
-
-        return [delegate_tool, ask_question_tool]
-
-    def test_gemini_receives_tools_for_hierarchical_crew(
-        self, mock_gemini_llm, mock_printer, delegation_tools
-    ):
-        """Test that Gemini models receive tools when used in hierarchical crew.
-
-        This test verifies the fix for issue #4238 where the manager agent
-        in a hierarchical crew would fail because tools weren't passed to
-        the Gemini model, causing UNEXPECTED_TOOL_CALL errors.
-        """
-        mock_context = Mock()
-        mock_context.tools = delegation_tools
-        mock_context.messages = [
-            {"role": "system", "content": "You are a manager agent"},
-            {"role": "user", "content": "Coordinate the team to answer this question"},
-        ]
-        mock_context.before_llm_call_hooks = []
-        mock_context.after_llm_call_hooks = []
-
-        with patch(
-            "crewai.utilities.agent_utils._setup_before_llm_call_hooks",
-            return_value=True,
-        ):
-            with patch(
-                "crewai.utilities.agent_utils._setup_after_llm_call_hooks",
-                return_value="Gemini response with tool call",
-            ):
-                result = get_llm_response(
-                    llm=mock_gemini_llm,
-                    messages=mock_context.messages,
-                    callbacks=[],
-                    printer=mock_printer,
-                    executor_context=mock_context,
-                )
-
-        # Verify that tools were passed to the Gemini model
-        mock_gemini_llm.call.assert_called_once()
-        call_kwargs = mock_gemini_llm.call.call_args[1]
-
-        assert "tools" in call_kwargs
-        assert call_kwargs["tools"] is not None
-        assert len(call_kwargs["tools"]) == 2
-
-        # Verify the delegation tools are properly formatted
-        tool_names = [t["name"] for t in call_kwargs["tools"]]
-        assert "Delegate work to coworker" in tool_names
-        assert "Ask question to coworker" in tool_names
-
-        # Verify each tool has the required fields
-        for tool_dict in call_kwargs["tools"]:
-            assert "name" in tool_dict
-            assert "description" in tool_dict
-            assert "args_schema" in tool_dict
-
-    def test_tool_dict_format_compatible_with_llm_providers(
-        self, mock_gemini_llm, mock_printer, delegation_tools
-    ):
-        """Test that extracted tools are in a format compatible with LLM providers.
-
-        The tool dictionaries should have 'name', 'description', and 'args_schema'
-        fields that can be processed by the LLM's _prepare_completion_params method.
-        """
-        mock_context = Mock()
-        mock_context.tools = delegation_tools
-        mock_context.messages = [{"role": "user", "content": "test"}]
-        mock_context.before_llm_call_hooks = []
-        mock_context.after_llm_call_hooks = []
-
-        with patch(
-            "crewai.utilities.agent_utils._setup_before_llm_call_hooks",
-            return_value=True,
-        ):
-            with patch(
-                "crewai.utilities.agent_utils._setup_after_llm_call_hooks",
-                return_value="response",
-            ):
-                get_llm_response(
-                    llm=mock_gemini_llm,
-                    messages=mock_context.messages,
-                    callbacks=[],
-                    printer=mock_printer,
-                    executor_context=mock_context,
-                )
-
-        call_kwargs = mock_gemini_llm.call.call_args[1]
-        tools = call_kwargs["tools"]
-
-        for tool_dict in tools:
-            # Verify the format matches what extract_tool_info() in common.py expects
-            assert isinstance(tool_dict["name"], str)
-            assert isinstance(tool_dict["description"], str)
-            # args_schema should be a Pydantic model class
-            assert hasattr(tool_dict["args_schema"], "model_json_schema")