andre/crewAI
1
0
Fork 0
mirror of https://github.com/crewAIInc/crewAI.git synced 2026-07-06 23:49:24 +00:00
Code Issues Actions 5 Packages Projects Releases Wiki Activity

refactor: remove CodeInterpreterTool and deprecate code execution params (#5309)
Some checks failed
CodeQL Advanced / Analyze (actions) (push) Has been cancelled
Details
CodeQL Advanced / Analyze (python) (push) Has been cancelled
Details
Vulnerability Scan / pip-audit (push) Has been cancelled
Details
Mark stale issues and pull requests / stale (push) Has been cancelled
Details

Browse Source 
* refactor: remove CodeInterpreterTool and deprecate code execution params

CodeInterpreterTool has been removed. The allow_code_execution and
code_execution_mode parameters on Agent are deprecated and will be
removed in v2.0. Use dedicated sandbox services (E2B, Modal, etc.)
for code execution needs.

Changes:
- Remove CodeInterpreterTool from crewai-tools (tool, Dockerfile, tests, imports)
- Remove docker dependency from crewai-tools
- Deprecate allow_code_execution and code_execution_mode on Agent
- get_code_execution_tools() returns empty list with deprecation warning
- _validate_docker_installation() is a no-op with deprecation warning
- Bedrock CodeInterpreter (AWS hosted) and OpenAI code_interpreter are NOT affected

* fix: remove empty code_interpreter imports and unused stdlib imports

- Remove empty `from code_interpreter_tool import ()` blocks in both
  crewai_tools/__init__.py and tools/__init__.py that caused SyntaxError
  after CodeInterpreterTool was removed
- Remove unused `shutil` and `subprocess` imports from agent/core.py
  left over from the code execution params deprecation

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* fix: remove redundant _validate_docker_installation call and fix list type annotation

- Drop the _validate_docker_installation() call inside the allow_code_execution
  block — it fired a second DeprecationWarning identical to the one emitted
  just above it, making the warning fire twice.
- Annotate get_code_execution_tools() return type as list[Any] to satisfy mypy
  (bare `list` fails the type-arg check introduced by this branch).

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* ci: retrigger

* fix: update test_crew.py to remove CodeInterpreterTool references

CodeInterpreterTool was removed from crewai_tools. Update tests to
reflect that get_code_execution_tools() now returns an empty list.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* chore: update tool specifications

---------

Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>
Co-authored-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com>
This commit is contained in:
alex-clawd
2026-04-06 23:59:40 -07:00
committed by GitHub
parent c132d57a36
commit e64b37c5fc
11 changed files with 727 additions and 989 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
lib/crewai-tools/pyproject.toml
View File

@@ -10,7 +10,6 @@ requires-python = ">=3.10, <3.14"
dependencies = [
"pytube~=15.0.0",
"requests~=2.32.5",
"docker~=7.1.0",
"crewai==1.14.0a3",
"tiktoken~=0.8.0",
"beautifulsoup4~=4.13.4",

4
lib/crewai-tools/src/crewai_tools/__init__.py
View File

@@ -35,9 +35,6 @@ from crewai_tools.tools.browserbase_load_tool.browserbase_load_tool import (
from crewai_tools.tools.code_docs_search_tool.code_docs_search_tool import (
CodeDocsSearchTool,
)
from crewai_tools.tools.code_interpreter_tool.code_interpreter_tool import (
CodeInterpreterTool,
)
from crewai_tools.tools.composio_tool.composio_tool import ComposioTool
from crewai_tools.tools.contextualai_create_agent_tool.contextual_create_agent_tool import (
ContextualAICreateAgentTool,
@@ -225,7 +222,6 @@ __all__ = [
"BrowserbaseLoadTool",
"CSVSearchTool",
"CodeDocsSearchTool",
"CodeInterpreterTool",
"ComposioTool",
"ContextualAICreateAgentTool",
"ContextualAIParseTool",

4
lib/crewai-tools/src/crewai_tools/tools/__init__.py
View File

@@ -24,9 +24,6 @@ from crewai_tools.tools.browserbase_load_tool.browserbase_load_tool import (
from crewai_tools.tools.code_docs_search_tool.code_docs_search_tool import (
CodeDocsSearchTool,
)
from crewai_tools.tools.code_interpreter_tool.code_interpreter_tool import (
CodeInterpreterTool,
)
from crewai_tools.tools.composio_tool.composio_tool import ComposioTool
from crewai_tools.tools.contextualai_create_agent_tool.contextual_create_agent_tool import (
ContextualAICreateAgentTool,
@@ -210,7 +207,6 @@ __all__ = [
"BrowserbaseLoadTool",
"CSVSearchTool",
"CodeDocsSearchTool",
"CodeInterpreterTool",
"ComposioTool",
"ContextualAICreateAgentTool",
"ContextualAIParseTool",

6
lib/crewai-tools/src/crewai_tools/tools/code_interpreter_tool/Dockerfile
View File

@@ -1,6 +0,0 @@
FROM python:3.12-alpine
RUN pip install requests beautifulsoup4
# Set the working directory
WORKDIR /workspace

95
lib/crewai-tools/src/crewai_tools/tools/code_interpreter_tool/README.md
View File

@@ -1,95 +0,0 @@
# CodeInterpreterTool
## Description
This tool is used to give the Agent the ability to run code (Python3) from the code generated by the Agent itself. The code is executed in a Docker container for secure isolation.
It is incredibly useful since it allows the Agent to generate code, run it in an isolated environment, get the result and use it to make decisions.
## ⚠️ Security Requirements
**Docker is REQUIRED** for safe code execution. The tool will refuse to execute code without Docker to prevent security vulnerabilities.
### Why Docker is Required
Previous versions included a "restricted sandbox" fallback when Docker was unavailable. This has been **removed** due to critical security vulnerabilities:
- The Python-based sandbox could be escaped via object introspection
- Attackers could recover the original `__import__` function and access any module
- This allowed arbitrary command execution on the host system
**Docker provides real process isolation** and is the only secure way to execute untrusted code.
## Requirements
- **Docker (REQUIRED)** - Install from [docker.com](https://docs.docker.com/get-docker/)
## Installation
Install the crewai_tools package
```shell
pip install 'crewai[tools]'
```
## Example
Remember that when using this tool, the code must be generated by the Agent itself. The code must be Python3 code. It will take some time the first time to run because it needs to build the Docker image.
### Basic Usage (Docker Container - Recommended)
```python
from crewai_tools import CodeInterpreterTool
Agent(
...
tools=[CodeInterpreterTool()],
)
```
### Custom Dockerfile
If you need to pass your own Dockerfile:
```python
from crewai_tools import CodeInterpreterTool
Agent(
...
tools=[CodeInterpreterTool(user_dockerfile_path="<Dockerfile_path>")],
)
```
### Manual Docker Host Configuration
If it is difficult to connect to the Docker daemon automatically (especially for macOS users), you can set up the Docker host manually:
```python
from crewai_tools import CodeInterpreterTool
Agent(
...
tools=[CodeInterpreterTool(
user_docker_base_url="<Docker Host Base Url>",
user_dockerfile_path="<Dockerfile_path>"
)],
)
```
### Unsafe Mode (NOT RECOMMENDED)
If you absolutely cannot use Docker and **fully trust the code source**, you can use unsafe mode:
```python
from crewai_tools import CodeInterpreterTool
# WARNING: Only use with fully trusted code!
Agent(
...
tools=[CodeInterpreterTool(unsafe_mode=True)],
)
```
**⚠️ SECURITY WARNING:** `unsafe_mode=True` executes code directly on the host without any isolation. Only use this if:
- You completely trust the code being executed
- You understand the security risks
- You cannot install Docker in your environment
For production use, **always use Docker** (the default mode).

0
lib/crewai-tools/src/crewai_tools/tools/code_interpreter_tool/__init__.py
View File

424
lib/crewai-tools/src/crewai_tools/tools/code_interpreter_tool/code_interpreter_tool.py
View File

@@ -1,424 +0,0 @@
"""Code Interpreter Tool for executing Python code in isolated environments.
This module provides a tool for executing Python code either in a Docker container for
safe isolation or directly in a restricted sandbox. It includes mechanisms for blocking
potentially unsafe operations and importing restricted modules.
"""
import importlib.util
import os
import subprocess
import sys
from types import ModuleType
from typing import Any, ClassVar, TypedDict
from crewai.tools import BaseTool
from docker import ( # type: ignore[import-untyped]
DockerClient,
from_env as docker_from_env,
)
from docker.errors import ImageNotFound, NotFound # type: ignore[import-untyped]
from pydantic import BaseModel, Field
from typing_extensions import Unpack
from crewai_tools.printer import Printer
class RunKwargs(TypedDict, total=False):
"""Keyword arguments for the _run method."""
code: str
libraries_used: list[str]
class CodeInterpreterSchema(BaseModel):
"""Schema for defining inputs to the CodeInterpreterTool.
This schema defines the required parameters for code execution,
including the code to run and any libraries that need to be installed.
"""
code: str = Field(
...,
description="Python3 code used to be interpreted in the Docker container. ALWAYS PRINT the final result and the output of the code",
)
libraries_used: list[str] = Field(
...,
description="List of libraries used in the code with proper installing names separated by commas. Example: numpy,pandas,beautifulsoup4",
)
class SandboxPython:
"""INSECURE: A restricted Python execution environment with known vulnerabilities.
WARNING: This class does NOT provide real security isolation and is vulnerable to
sandbox escape attacks via Python object introspection. Attackers can recover the
original __import__ function and bypass all restrictions.
DO NOT USE for untrusted code execution. Use Docker containers instead.
This class attempts to restrict access to dangerous modules and built-in functions
but provides no real security boundary against a motivated attacker.
"""
BLOCKED_MODULES: ClassVar[set[str]] = {
"os",
"sys",
"subprocess",
"shutil",
"importlib",
"inspect",
"tempfile",
"sysconfig",
"builtins",
}
UNSAFE_BUILTINS: ClassVar[set[str]] = {
"exec",
"eval",
"open",
"compile",
"input",
"globals",
"locals",
"vars",
"help",
"dir",
}
@staticmethod
def restricted_import(
name: str,
custom_globals: dict[str, Any] | None = None,
custom_locals: dict[str, Any] | None = None,
fromlist: list[str] | None = None,
level: int = 0,
) -> ModuleType:
"""A restricted import function that blocks importing of unsafe modules.
Args:
name: The name of the module to import.
custom_globals: Global namespace to use.
custom_locals: Local namespace to use.
fromlist: List of items to import from the module.
level: The level value passed to __import__.
Returns:
The imported module if allowed.
Raises:
ImportError: If the module is in the blocked modules list.
"""
if name in SandboxPython.BLOCKED_MODULES:
raise ImportError(f"Importing '{name}' is not allowed.")
return __import__(name, custom_globals, custom_locals, fromlist or (), level)
@staticmethod
def safe_builtins() -> dict[str, Any]:
"""Creates a dictionary of built-in functions with unsafe ones removed.
Returns:
A dictionary of safe built-in functions and objects.
"""
import builtins
safe_builtins = {
k: v
for k, v in builtins.__dict__.items()
if k not in SandboxPython.UNSAFE_BUILTINS
}
safe_builtins["__import__"] = SandboxPython.restricted_import
return safe_builtins
@staticmethod
def exec(code: str, locals_: dict[str, Any]) -> None:
"""Executes Python code in a restricted environment.
Args:
code: The Python code to execute as a string.
locals_: A dictionary that will be used for local variable storage.
"""
exec(code, {"__builtins__": SandboxPython.safe_builtins()}, locals_) # noqa: S102
class CodeInterpreterTool(BaseTool):
"""A tool for executing Python code in isolated environments.
This tool provides functionality to run Python code either in a Docker container
for safe isolation or directly in a restricted sandbox. It can handle installing
Python packages and executing arbitrary Python code.
"""
name: str = "Code Interpreter"
description: str = "Interprets Python3 code strings with a final print statement."
args_schema: type[BaseModel] = CodeInterpreterSchema
default_image_tag: str = "code-interpreter:latest"
code: str | None = None
user_dockerfile_path: str | None = None
user_docker_base_url: str | None = None
unsafe_mode: bool = False
@staticmethod
def _get_installed_package_path() -> str:
"""Gets the installation path of the crewai_tools package.
Returns:
The directory path where the package is installed.
Raises:
RuntimeError: If the package cannot be found.
"""
spec = importlib.util.find_spec("crewai_tools")
if spec is None or spec.origin is None:
raise RuntimeError("Cannot find crewai_tools package installation path")
return os.path.dirname(spec.origin)
def _verify_docker_image(self) -> None:
"""Verifies if the Docker image is available or builds it if necessary.
Checks if the required Docker image exists. If not, builds it using either a
user-provided Dockerfile or the default one included with the package.
Raises:
FileNotFoundError: If the Dockerfile cannot be found.
"""
client = (
docker_from_env()
if self.user_docker_base_url is None
else DockerClient(base_url=self.user_docker_base_url)
)
try:
client.images.get(self.default_image_tag)
except ImageNotFound:
if self.user_dockerfile_path and os.path.exists(self.user_dockerfile_path):
dockerfile_path = self.user_dockerfile_path
else:
package_path = self._get_installed_package_path()
dockerfile_path = os.path.join(
package_path, "tools/code_interpreter_tool"
)
if not os.path.exists(dockerfile_path):
raise FileNotFoundError(
f"Dockerfile not found in {dockerfile_path}"
) from None
client.images.build(
path=dockerfile_path,
tag=self.default_image_tag,
rm=True,
)
def _run(self, **kwargs: Unpack[RunKwargs]) -> str:
"""Runs the code interpreter tool with the provided arguments.
Args:
**kwargs: Keyword arguments that should include 'code' and 'libraries_used'.
Returns:
The output of the executed code as a string.
"""
code: str | None = kwargs.get("code", self.code)
libraries_used: list[str] = kwargs.get("libraries_used", [])
if not code:
return "No code provided to execute."
if self.unsafe_mode:
return self.run_code_unsafe(code, libraries_used)
return self.run_code_safety(code, libraries_used)
@staticmethod
def _install_libraries(container: Any, libraries: list[str]) -> None:
"""Installs required Python libraries in the Docker container.
Args:
container: The Docker container where libraries will be installed.
libraries: A list of library names to install using pip.
"""
for library in libraries:
container.exec_run(["pip", "install", library])
def _init_docker_container(self) -> Any:
"""Initializes and returns a Docker container for code execution.
Stops and removes any existing container with the same name before creating
a new one. Maps the current working directory to /workspace in the container.
Returns:
A Docker container object ready for code execution.
"""
container_name = "code-interpreter"
client = docker_from_env()
current_path = os.getcwd()
# Check if the container is already running
try:
existing_container = client.containers.get(container_name)
existing_container.stop()
existing_container.remove()
except NotFound:
pass # Container does not exist, no need to remove
return client.containers.run(
self.default_image_tag,
detach=True,
tty=True,
working_dir="/workspace",
name=container_name,
volumes={current_path: {"bind": "/workspace", "mode": "rw"}},
)
@staticmethod
def _check_docker_available() -> bool:
"""Checks if Docker is available and running on the system.
Attempts to run the 'docker info' command to verify Docker availability.
Prints appropriate messages if Docker is not installed or not running.
Returns:
True if Docker is available and running, False otherwise.
"""
try:
subprocess.run(
["docker", "info"], # noqa: S607
check=True,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
timeout=1,
)
return True
except (subprocess.CalledProcessError, subprocess.TimeoutExpired):
Printer.print(
"Docker is installed but not running or inaccessible.",
color="bold_purple",
)
return False
except FileNotFoundError:
Printer.print("Docker is not installed", color="bold_purple")
return False
def run_code_safety(self, code: str, libraries_used: list[str]) -> str:
"""Runs code in the safest available environment.
Requires Docker to be available for secure code execution. Fails closed
if Docker is not available to prevent sandbox escape vulnerabilities.
Args:
code: The Python code to execute as a string.
libraries_used: A list of Python library names to install before execution.
Returns:
The output of the executed code as a string.
Raises:
RuntimeError: If Docker is not available, as the restricted sandbox
is vulnerable to escape attacks and should not be used
for untrusted code execution.
"""
if self._check_docker_available():
return self.run_code_in_docker(code, libraries_used)
error_msg = (
"Docker is required for safe code execution but is not available. "
"The restricted sandbox fallback has been removed due to security vulnerabilities "
"that allow sandbox escape via Python object introspection. "
"Please install Docker (https://docs.docker.com/get-docker/) or use unsafe_mode=True "
"if you trust the code source and understand the security risks."
)
Printer.print(error_msg, color="bold_red")
raise RuntimeError(error_msg)
def run_code_in_docker(self, code: str, libraries_used: list[str]) -> str:
"""Runs Python code in a Docker container for safe isolation.
Creates a Docker container, installs the required libraries, executes the code,
and then cleans up by stopping and removing the container.
Args:
code: The Python code to execute as a string.
libraries_used: A list of Python library names to install before execution.
Returns:
The output of the executed code as a string, or an error message if execution failed.
"""
Printer.print("Running code in Docker environment", color="bold_blue")
self._verify_docker_image()
container = self._init_docker_container()
self._install_libraries(container, libraries_used)
exec_result: Any = container.exec_run(["python3", "-c", code])
container.stop()
container.remove()
if exec_result.exit_code != 0:
return f"Something went wrong while running the code: \n{exec_result.output.decode('utf-8')}"
return str(exec_result.output.decode("utf-8"))
@staticmethod
def run_code_in_restricted_sandbox(code: str) -> str:
"""DEPRECATED AND INSECURE: Runs Python code in a restricted sandbox environment.
WARNING: This method is vulnerable to sandbox escape attacks via Python object
introspection and should NOT be used for untrusted code execution. It has been
deprecated and is only kept for backward compatibility with trusted code.
The "restricted" environment can be bypassed by attackers who can:
- Use object graph introspection to recover the original __import__ function
- Access any Python module including os, subprocess, sys, etc.
- Execute arbitrary commands on the host system
Use run_code_in_docker() for secure code execution, or run_code_unsafe()
if you explicitly acknowledge the security risks.
Args:
code: The Python code to execute as a string.
Returns:
The value of the 'result' variable from the executed code,
or an error message if execution failed.
"""
Printer.print(
"WARNING: Running code in INSECURE restricted sandbox (vulnerable to escape attacks)",
color="bold_red",
)
exec_locals: dict[str, Any] = {}
try:
SandboxPython.exec(code=code, locals_=exec_locals)
return exec_locals.get("result", "No result variable found.") # type: ignore[no-any-return]
except Exception as e:
return f"An error occurred: {e!s}"
@staticmethod
def run_code_unsafe(code: str, libraries_used: list[str]) -> str:
"""Runs code directly on the host machine without any safety restrictions.
WARNING: This mode is unsafe and should only be used in trusted environments
with code from trusted sources.
Args:
code: The Python code to execute as a string.
libraries_used: A list of Python library names to install before execution.
Returns:
The value of the 'result' variable from the executed code,
or an error message if execution failed.
"""
Printer.print("WARNING: Running code in unsafe mode", color="bold_magenta")
# Install libraries on the host machine
for library in libraries_used:
subprocess.run( # noqa: S603
[sys.executable, "-m", "pip", "install", library], check=False
)
# Execute the code
try:
exec_locals: dict[str, Any] = {}
exec(code, {}, exec_locals) # noqa: S102
return exec_locals.get("result", "No result variable found.") # type: ignore[no-any-return]
except Exception as e:
return f"An error occurred: {e!s}"

253
lib/crewai-tools/tests/tools/test_code_interpreter_tool.py
View File

@@ -1,253 +0,0 @@
import sys
from unittest.mock import patch
from crewai_tools.tools.code_interpreter_tool.code_interpreter_tool import (
CodeInterpreterTool,
SandboxPython,
)
import pytest
@pytest.fixture
def printer_mock():
with patch("crewai_tools.printer.Printer.print") as mock:
yield mock
@pytest.fixture
def docker_unavailable_mock():
with patch(
"crewai_tools.tools.code_interpreter_tool.code_interpreter_tool.CodeInterpreterTool._check_docker_available",
return_value=False,
) as mock:
yield mock
@patch("crewai_tools.tools.code_interpreter_tool.code_interpreter_tool.docker_from_env")
def test_run_code_in_docker(docker_mock, printer_mock):
tool = CodeInterpreterTool()
code = "print('Hello, World!')"
libraries_used = ["numpy", "pandas"]
expected_output = "Hello, World!\n"
docker_mock().containers.run().exec_run().exit_code = 0
docker_mock().containers.run().exec_run().output = expected_output.encode()
result = tool.run_code_in_docker(code, libraries_used)
assert result == expected_output
printer_mock.assert_called_with(
"Running code in Docker environment", color="bold_blue"
)
@patch("crewai_tools.tools.code_interpreter_tool.code_interpreter_tool.docker_from_env")
def test_run_code_in_docker_with_error(docker_mock, printer_mock):
tool = CodeInterpreterTool()
code = "print(1/0)"
libraries_used = ["numpy", "pandas"]
expected_output = "Something went wrong while running the code: \nZeroDivisionError: division by zero\n"
docker_mock().containers.run().exec_run().exit_code = 1
docker_mock().containers.run().exec_run().output = (
b"ZeroDivisionError: division by zero\n"
)
result = tool.run_code_in_docker(code, libraries_used)
assert result == expected_output
printer_mock.assert_called_with(
"Running code in Docker environment", color="bold_blue"
)
@patch("crewai_tools.tools.code_interpreter_tool.code_interpreter_tool.docker_from_env")
def test_run_code_in_docker_with_script(docker_mock, printer_mock):
tool = CodeInterpreterTool()
code = """print("This is line 1")
print("This is line 2")"""
libraries_used = []
expected_output = "This is line 1\nThis is line 2\n"
docker_mock().containers.run().exec_run().exit_code = 0
docker_mock().containers.run().exec_run().output = expected_output.encode()
result = tool.run_code_in_docker(code, libraries_used)
assert result == expected_output
printer_mock.assert_called_with(
"Running code in Docker environment", color="bold_blue"
)
def test_docker_unavailable_raises_error(printer_mock, docker_unavailable_mock):
"""Test that execution fails when Docker is unavailable in safe mode."""
tool = CodeInterpreterTool()
code = """
result = 2 + 2
print(result)
"""
with pytest.raises(RuntimeError) as exc_info:
tool.run(code=code, libraries_used=[])
assert "Docker is required for safe code execution" in str(exc_info.value)
assert "sandbox escape" in str(exc_info.value)
def test_restricted_sandbox_running_with_blocked_modules():
"""Test that restricted modules cannot be imported when using the deprecated sandbox directly."""
tool = CodeInterpreterTool()
restricted_modules = SandboxPython.BLOCKED_MODULES
for module in restricted_modules:
code = f"""
import {module}
result = "Import succeeded"
"""
# Note: run_code_in_restricted_sandbox is deprecated and insecure
# This test verifies the old behavior but should not be used in production
result = tool.run_code_in_restricted_sandbox(code)
assert f"An error occurred: Importing '{module}' is not allowed" in result
def test_restricted_sandbox_running_with_blocked_builtins():
"""Test that restricted builtins are not available when using the deprecated sandbox directly."""
tool = CodeInterpreterTool()
restricted_builtins = SandboxPython.UNSAFE_BUILTINS
for builtin in restricted_builtins:
code = f"""
{builtin}("test")
result = "Builtin available"
"""
# Note: run_code_in_restricted_sandbox is deprecated and insecure
# This test verifies the old behavior but should not be used in production
result = tool.run_code_in_restricted_sandbox(code)
assert f"An error occurred: name '{builtin}' is not defined" in result
def test_restricted_sandbox_running_with_no_result_variable(
printer_mock, docker_unavailable_mock
):
"""Test behavior when no result variable is set in deprecated sandbox."""
tool = CodeInterpreterTool()
code = """
x = 10
"""
# Note: run_code_in_restricted_sandbox is deprecated and insecure
# This test verifies the old behavior but should not be used in production
result = tool.run_code_in_restricted_sandbox(code)
assert result == "No result variable found."
def test_unsafe_mode_running_with_no_result_variable(
printer_mock, docker_unavailable_mock
):
"""Test behavior when no result variable is set."""
tool = CodeInterpreterTool(unsafe_mode=True)
code = """
x = 10
"""
result = tool.run(code=code, libraries_used=[])
printer_mock.assert_called_with(
"WARNING: Running code in unsafe mode", color="bold_magenta"
)
assert result == "No result variable found."
@patch("crewai_tools.tools.code_interpreter_tool.code_interpreter_tool.subprocess.run")
def test_unsafe_mode_installs_libraries_without_shell(
subprocess_run_mock, printer_mock, docker_unavailable_mock
):
"""Test that library installation uses subprocess.run with shell=False, not os.system."""
tool = CodeInterpreterTool(unsafe_mode=True)
code = "result = 1"
libraries_used = ["numpy", "pandas"]
tool.run(code=code, libraries_used=libraries_used)
assert subprocess_run_mock.call_count == 2
for call, library in zip(subprocess_run_mock.call_args_list, libraries_used):
args, kwargs = call
# Must be list form (no shell expansion possible)
assert args[0] == [sys.executable, "-m", "pip", "install", library]
# shell= must not be True (defaults to False)
assert kwargs.get("shell", False) is False
@patch("crewai_tools.tools.code_interpreter_tool.code_interpreter_tool.subprocess.run")
def test_unsafe_mode_library_name_with_shell_metacharacters_does_not_invoke_shell(
subprocess_run_mock, printer_mock, docker_unavailable_mock
):
"""Test that a malicious library name cannot inject shell commands."""
tool = CodeInterpreterTool(unsafe_mode=True)
code = "result = 1"
malicious_library = "numpy; rm -rf /"
tool.run(code=code, libraries_used=[malicious_library])
subprocess_run_mock.assert_called_once()
args, kwargs = subprocess_run_mock.call_args
# The entire malicious string is passed as a single argument — no shell parsing
assert args[0] == [sys.executable, "-m", "pip", "install", malicious_library]
assert kwargs.get("shell", False) is False
def test_unsafe_mode_running_unsafe_code(printer_mock, docker_unavailable_mock):
"""Test behavior when no result variable is set."""
tool = CodeInterpreterTool(unsafe_mode=True)
code = """
import os
os.system("ls -la")
result = eval("5/1")
"""
result = tool.run(code=code, libraries_used=[])
printer_mock.assert_called_with(
"WARNING: Running code in unsafe mode", color="bold_magenta"
)
assert 5.0 == result
@pytest.mark.xfail(
reason=(
"run_code_in_restricted_sandbox is known to be vulnerable to sandbox "
"escape via object introspection. This test encodes the desired secure "
"behavior (no escape possible) and will start passing once the "
"vulnerability is fixed or the function is removed."
)
)
def test_sandbox_escape_vulnerability_demonstration(printer_mock):
"""Demonstrate that the restricted sandbox is vulnerable to escape attacks.
This test shows that an attacker can use Python object introspection to bypass
the restricted sandbox and access blocked modules like 'os'. This is why the
sandbox should never be used for untrusted code execution.
NOTE: This test uses the deprecated run_code_in_restricted_sandbox directly
to demonstrate the vulnerability. In production, Docker is now required.
"""
tool = CodeInterpreterTool()
# Classic Python sandbox escape via object introspection
escape_code = """
# Recover the real __import__ function via object introspection
for cls in ().__class__.__bases__[0].__subclasses__():
if cls.__name__ == 'catch_warnings':
# Get the real builtins module
real_builtins = cls()._module.__builtins__
real_import = real_builtins['__import__']
# Now we can import os and execute commands
os = real_import('os')
# Demonstrate we have escaped the sandbox
result = "SANDBOX_ESCAPED" if hasattr(os, 'system') else "FAILED"
break
"""
# The deprecated sandbox is vulnerable to this attack
result = tool.run_code_in_restricted_sandbox(escape_code)
# Desired behavior: the restricted sandbox should prevent this escape.
# If this assertion fails, run_code_in_restricted_sandbox remains vulnerable.
assert result != "SANDBOX_ESCAPED", (
"The restricted sandbox was bypassed via object introspection. "
"This indicates run_code_in_restricted_sandbox is still vulnerable and "
"is why Docker is now required for safe code execution."
)

841
lib/crewai-tools/tool.specs.json
View File

File diff suppressed because it is too large Load Diff

71
lib/crewai/src/crewai/agent/core.py
View File

@@ -9,8 +9,6 @@ import contextvars
from datetime import datetime
import json
from pathlib import Path
import shutil
import subprocess
import time
from typing import (
TYPE_CHECKING,
@@ -116,7 +114,6 @@ except ImportError:
if TYPE_CHECKING:
from crewai_files import FileInput
from crewai_tools import CodeInterpreterTool
from crewai.a2a.config import A2AClientConfig, A2AConfig, A2AServerConfig
from crewai.agents.agent_builder.base_agent import PlatformAppOrAction
@@ -211,7 +208,9 @@ class Agent(BaseAgent):
default=None, description="Response format for the agent."
)
allow_code_execution: bool | None = Field(
default=False, description="Enable code execution for the agent."
default=False,
deprecated=True,
description="Deprecated. CodeInterpreterTool is no longer available. Use dedicated sandbox services instead.",
)
respect_context_window: bool = Field(
default=True,
@@ -236,7 +235,8 @@ class Agent(BaseAgent):
)
code_execution_mode: Literal["safe", "unsafe"] = Field(
default="safe",
description="Mode for code execution: 'safe' (using Docker) or 'unsafe' (direct execution).",
deprecated=True,
description="Deprecated. CodeInterpreterTool is no longer available. Use dedicated sandbox services instead.",
)
planning_config: PlanningConfig | None = Field(
default=None,
@@ -329,7 +329,13 @@ class Agent(BaseAgent):
self._setup_agent_executor()
if self.allow_code_execution:
self._validate_docker_installation()
warnings.warn(
"allow_code_execution is deprecated and will be removed in v2.0. "
"CodeInterpreterTool is no longer available. "
"Use dedicated sandbox services like E2B or Modal.",
DeprecationWarning,
stacklevel=2,
)
self.set_skills()
@@ -1123,20 +1129,15 @@ class Agent(BaseAgent):
return [AddImageTool()]
def get_code_execution_tools(self) -> list[CodeInterpreterTool]:
"""Return code interpreter tools based on the agent's execution mode."""
try:
from crewai_tools import (
CodeInterpreterTool,
)
unsafe_mode = self.code_execution_mode == "unsafe"
return [CodeInterpreterTool(unsafe_mode=unsafe_mode)]
except ModuleNotFoundError:
self._logger.log(
"info", "Coding tools not available. Install crewai_tools. "
)
return []
def get_code_execution_tools(self) -> list[Any]:
"""Deprecated: CodeInterpreterTool is no longer available."""
warnings.warn(
"CodeInterpreterTool is no longer available. "
"Use dedicated sandbox services like E2B or Modal.",
DeprecationWarning,
stacklevel=2,
)
return []
@staticmethod
def get_output_converter(
@@ -1216,28 +1217,14 @@ class Agent(BaseAgent):
self._logger.log("warning", f"Failed to inject date: {e!s}")
def _validate_docker_installation(self) -> None:
"""Check if Docker is installed and running."""
docker_path = shutil.which("docker")
if not docker_path:
raise RuntimeError(
f"Docker is not installed. Please install Docker to use code execution with agent: {self.role}"
)
try:
subprocess.run( # noqa: S603
[str(docker_path), "info"],
check=True,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
)
except subprocess.CalledProcessError as e:
raise RuntimeError(
f"Docker is not running. Please start Docker to use code execution with agent: {self.role}"
) from e
except subprocess.TimeoutExpired as e:
raise RuntimeError(
f"Docker command timed out. Please check your Docker installation for agent: {self.role}"
) from e
"""Deprecated: No-op. CodeInterpreterTool is no longer available."""
warnings.warn(
"CodeInterpreterTool is no longer available. "
"Use dedicated sandbox services like E2B or Modal.",
DeprecationWarning,
stacklevel=2,
)
return
def __repr__(self) -> str:
return f"Agent(role={self.role}, goal={self.goal}, backstory={self.backstory})"

17
lib/crewai/tests/test_crew.py
View File

@@ -48,7 +48,6 @@ from crewai.tools.agent_tools.add_image_tool import AddImageTool
from crewai.types.usage_metrics import UsageMetrics
from crewai.utilities.rpm_controller import RPMController
from crewai.utilities.task_output_storage_handler import TaskOutputStorageHandler
from crewai_tools import CodeInterpreterTool
from pydantic import BaseModel, Field
import pydantic_core
import pytest
@@ -1648,11 +1647,8 @@ def test_code_execution_flag_adds_code_tool_upon_kickoff():
_, kwargs = mock_execute_sync.call_args
used_tools = kwargs["tools"]
# Verify that exactly one tool was used and it was a CodeInterpreterTool
assert len(used_tools) == 1, "Should have exactly one tool"
assert isinstance(used_tools[0], CodeInterpreterTool), (
"Tool should be CodeInterpreterTool"
)
# CodeInterpreterTool was removed; get_code_execution_tools() now returns []
assert len(used_tools) == 0, "Should have no tools (code execution tools are deprecated)"
@pytest.mark.vcr()
@@ -3918,16 +3914,13 @@ def test_task_tools_preserve_code_execution_tools():
assert any(isinstance(tool, TestTool) for tool in used_tools), (
"Task's TestTool should be present"
)
assert any(isinstance(tool, CodeInterpreterTool) for tool in used_tools), (
"CodeInterpreterTool should be present"
)
assert any("delegate" in tool.name.lower() for tool in used_tools), (
"Delegation tool should be present"
)
# Verify the total number of tools (TestTool + CodeInterpreter + 2 delegation tools)
assert len(used_tools) == 4, (
"Should have TestTool, CodeInterpreter, and 2 delegation tools"
# Verify the total number of tools (TestTool + 2 delegation tools; CodeInterpreterTool removed)
assert len(used_tools) == 3, (
"Should have TestTool and 2 delegation tools"
)