Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
122 changes: 115 additions & 7 deletions libs/deepagents/deepagents/middleware/filesystem.py
Original file line number Diff line number Diff line change
Expand Up @@ -1052,7 +1052,13 @@ class ExecuteSchema(BaseModel):
GREP_TOOL_DESCRIPTION = _GREP_TOOL_DESCRIPTION_TEMPLATE.format(execute_fallback=_GREP_REGEX_EXECUTE_FALLBACK)
_GREP_TOOL_DESCRIPTION_WITHOUT_EXECUTE = _GREP_TOOL_DESCRIPTION_TEMPLATE.format(execute_fallback="")

EXECUTE_TOOL_DESCRIPTION = """Executes a shell command in an isolated sandbox environment.
_EXECUTE_SEARCH_GUIDANCE = "You MUST avoid using search commands like find and grep. Instead use the grep, glob tools to search. "
_EXECUTE_GREP_SEARCH_GUIDANCE = "You MUST avoid using shell grep for searches. Instead use the grep tool to search text. "
_EXECUTE_GLOB_SEARCH_GUIDANCE = "You MUST avoid using shell find for searches. Instead use the glob tool to find files. "
_EXECUTE_GLOB_BAD_EXAMPLE = "\n - execute(command=\"find . -name '*.py'\") # Use glob tool instead"
_EXECUTE_GREP_BAD_EXAMPLE = "\n - execute(command=\"grep -r 'pattern' .\") # Use grep tool instead"

_EXECUTE_TOOL_DESCRIPTION_TEMPLATE = """Executes a shell command in an isolated sandbox environment.

Usage:
Executes a given command in the sandbox environment with proper handling and security measures.
Expand All @@ -1075,7 +1081,7 @@ class ExecuteSchema(BaseModel):
- If the output is very large, it may be truncated
- For long-running commands, use the optional timeout parameter to override the default timeout (e.g., execute(command="make build", timeout=300))
- A timeout of 0 may disable timeouts on backends that support no-timeout execution
- VERY IMPORTANT: You MUST avoid using search commands like find and grep. Instead use the grep, glob tools to search. You MUST avoid read tools like cat, head, tail, and use read_file to read files.
- VERY IMPORTANT: {search_guidance}You MUST avoid read tools like cat, head, tail, and use read_file to read files.
- When issuing multiple commands, use the ';' or '&&' operator to separate them. DO NOT use newlines (newlines are ok in quoted strings)
- Use '&&' when commands depend on each other (e.g., "mkdir dir && cd dir")
- Use ';' only when you need to run commands sequentially but don't care if earlier commands fail
Expand All @@ -1090,13 +1096,32 @@ class ExecuteSchema(BaseModel):

Bad examples (avoid these):
- execute(command="cd /foo/bar && pytest tests") # Use absolute path instead
- execute(command="cat file.txt") # Use read_file tool instead
- execute(command="find . -name '*.py'") # Use glob tool instead
- execute(command="grep -r 'pattern' .") # Use grep tool instead
- execute(command="cat file.txt") # Use read_file tool instead{glob_bad_example}{grep_bad_example}

Note: This tool is only available if the backend supports execution (SandboxBackendProtocol).
If execution is not supported, the tool will return an error message."""

EXECUTE_TOOL_DESCRIPTION = _EXECUTE_TOOL_DESCRIPTION_TEMPLATE.format(
search_guidance=_EXECUTE_SEARCH_GUIDANCE,
glob_bad_example=_EXECUTE_GLOB_BAD_EXAMPLE,
grep_bad_example=_EXECUTE_GREP_BAD_EXAMPLE,
)
_EXECUTE_TOOL_DESCRIPTION_WITH_GREP_ONLY = _EXECUTE_TOOL_DESCRIPTION_TEMPLATE.format(
search_guidance=_EXECUTE_GREP_SEARCH_GUIDANCE,
glob_bad_example="",
grep_bad_example=_EXECUTE_GREP_BAD_EXAMPLE,
)
_EXECUTE_TOOL_DESCRIPTION_WITH_GLOB_ONLY = _EXECUTE_TOOL_DESCRIPTION_TEMPLATE.format(
search_guidance=_EXECUTE_GLOB_SEARCH_GUIDANCE,
glob_bad_example=_EXECUTE_GLOB_BAD_EXAMPLE,
grep_bad_example="",
)
_EXECUTE_TOOL_DESCRIPTION_WITHOUT_SEARCH = _EXECUTE_TOOL_DESCRIPTION_TEMPLATE.format(
search_guidance="",
glob_bad_example="",
grep_bad_example="",
)

FsToolName = Literal["ls", "read_file", "write_file", "edit_file", "delete", "glob", "grep", "execute"]
"""Names of the built-in filesystem tools that can be passed to `FilesystemMiddleware(tools=...)`."""

Expand Down Expand Up @@ -2473,6 +2498,82 @@ def _with_filtered_grep_description(

return rewritten if changed else tools

def _execute_tool_description(self, *, visible_search_tools: set[str]) -> str:
"""Return the execute description for the visible search tools.

Args:
visible_search_tools: Search tool names available to the model.

Returns:
The custom description, or the default variant matching tool visibility.
"""
custom_description = self._custom_tool_descriptions.get("execute")
if custom_description:
return custom_description
if "grep" in visible_search_tools and "glob" in visible_search_tools:
return EXECUTE_TOOL_DESCRIPTION
if "grep" in visible_search_tools:
return _EXECUTE_TOOL_DESCRIPTION_WITH_GREP_ONLY
if "glob" in visible_search_tools:
return _EXECUTE_TOOL_DESCRIPTION_WITH_GLOB_ONLY
return _EXECUTE_TOOL_DESCRIPTION_WITHOUT_SEARCH

def _with_filtered_execute_description(
self,
tools: list[BaseTool | dict[str, Any]],
*,
visible_search_tools: set[str],
) -> list[BaseTool | dict[str, Any]]:
"""Copy default execute tools when their search guidance changes.

Args:
tools: Request tools after backend capability filtering.
visible_search_tools: Search tool names available to the model.

Returns:
A copied list when an execute description changes, otherwise `tools`.
"""
if self._custom_tool_descriptions.get("execute"):
return tools

target_description = self._execute_tool_description(visible_search_tools=visible_search_tools)
default_descriptions = {
EXECUTE_TOOL_DESCRIPTION,
_EXECUTE_TOOL_DESCRIPTION_WITH_GREP_ONLY,
_EXECUTE_TOOL_DESCRIPTION_WITH_GLOB_ONLY,
_EXECUTE_TOOL_DESCRIPTION_WITHOUT_SEARCH,
}
rewritten: list[BaseTool | dict[str, Any]] = []
changed = False

for tool in tools:
tool_name = self._tool_name(tool)
if tool_name != "execute":
rewritten.append(tool)
continue

if isinstance(tool, BaseTool):
if tool.description in default_descriptions and tool.description != target_description:
rewritten.append(tool.model_copy(update={"description": target_description}))
changed = True
else:
rewritten.append(tool)
continue

if not isinstance(tool, dict):
rewritten.append(cast("BaseTool | dict[str, Any]", tool))
continue

if tool.get("description") in default_descriptions and tool.get("description") != target_description:
copied_tool = tool.copy()
copied_tool["description"] = target_description
rewritten.append(copied_tool)
changed = True
else:
rewritten.append(tool)

return rewritten if changed else tools

@staticmethod
def _tool_name(tool: object) -> str | None:
"""Extract a request tool name from `BaseTool`, dict, or test doubles."""
Expand Down Expand Up @@ -2584,7 +2685,10 @@ def _interpret_capture_output(self, offload: ExecuteOffloadResult, capture_path:

def _create_execute_tool(self) -> BaseTool: # noqa: C901
"""Create the execute tool for sandbox command execution."""
tool_description = self._custom_tool_descriptions.get("execute") or EXECUTE_TOOL_DESCRIPTION
visible_search_tools = {"grep", "glob"}
if self._enabled_tools is not None:
visible_search_tools.intersection_update(self._enabled_tools)
tool_description = self._execute_tool_description(visible_search_tools=visible_search_tools)

def sync_execute( # noqa: PLR0911 - early returns for distinct error conditions
command: str,
Expand Down Expand Up @@ -2785,10 +2889,15 @@ def _filter_unsupported_tools_and_apply_prompt(self, request: ModelRequest[Conte
tool_names: set[str | None] = {self._tool_name(tool) for tool in request.tools}
unsupported, execution_active, backend = self._unsupported_tools_and_execution_state(tool_names, request.runtime)
visible_tools = [tool for tool in request.tools if self._tool_name(tool) not in unsupported]
visible_fs = {name for name in (tool_names - unsupported) if name is not None}
if unsupported:
request = request.override(tools=visible_tools)

described_tools = self._with_filtered_grep_description(visible_tools, include_execution=execution_active)
described_tools = self._with_filtered_execute_description(
described_tools,
visible_search_tools=visible_fs,
)
if described_tools is not visible_tools:
request = request.override(tools=described_tools)

Expand All @@ -2797,7 +2906,6 @@ def _filter_unsupported_tools_and_apply_prompt(self, request: ModelRequest[Conte
system_prompt = self._custom_system_prompt
else:
# Build dynamic system prompt reflecting only the tools that survived filtering
visible_fs = {n for n in (tool_names - unsupported) if n is not None}
tool_header, tool_descriptions = _build_fs_tools_section(visible_fs)
prompt_parts = [
_FILESYSTEM_SYSTEM_PROMPT_TEMPLATE.format(
Expand Down
148 changes: 148 additions & 0 deletions libs/deepagents/tests/unit_tests/test_middleware.py
Original file line number Diff line number Diff line change
Expand Up @@ -83,6 +83,13 @@ def _runtime(tool_call_id=""):
return ToolRuntime(state={}, context=None, tool_call_id=tool_call_id, store=None, stream_writer=lambda _: None, config={})


class _SandboxBackend(SandboxBackendProtocol, StateBackend):
"""State backend with shell execution enabled for tool-description tests."""

def execute(self, command: str, *, timeout: int | None = None) -> ExecuteResponse:
return ExecuteResponse(output="", exit_code=0)


class TestAddMiddleware:
def test_filesystem_middleware(self):
middleware = [FilesystemMiddleware()]
Expand Down Expand Up @@ -2731,6 +2738,147 @@ def test_grep_description_swap_copies_dict_tool_specs(self):
assert "rg '<regex>'" not in rewritten_grep["description"]
assert "LITERAL text pattern" in rewritten_grep["description"]

def test_execute_description_keeps_search_guidance_when_grep_and_glob_visible(self):
"""Default execute docs recommend both visible search tools."""
middleware = FilesystemMiddleware(backend=_SandboxBackend(), system_prompt="")
request = MagicMock()
request.tools = middleware.tools
request.runtime = _runtime()
request.override.return_value = request

middleware._filter_unsupported_tools_and_apply_prompt(request)

tools_overrides = [call for call in request.override.call_args_list if "tools" in call.kwargs]
execute_tool = next(tool for tool in middleware.tools if tool.name == "execute")
assert tools_overrides == []
assert "use the grep, glob tools to search" in execute_tool.description
assert "# Use glob tool instead" in execute_tool.description
assert "# Use grep tool instead" in execute_tool.description

def test_execute_description_omits_search_guidance_when_grep_and_glob_hidden(self):
"""Execute docs omit search guidance when neither search tool is visible."""
middleware = FilesystemMiddleware(
backend=_SandboxBackend(),
system_prompt="",
tools=["read_file", "execute"],
)
request = MagicMock()
request.tools = middleware.tools
request.runtime = _runtime()
request.override.return_value = request

middleware._filter_unsupported_tools_and_apply_prompt(request)

tools_overrides = [call for call in request.override.call_args_list if "tools" in call.kwargs]
execute_tool = next(tool for tool in middleware.tools if tool.name == "execute")
assert tools_overrides == []
assert "use the grep, glob tools" not in execute_tool.description
assert "grep, glob tools to search" not in execute_tool.description
assert "# Use glob tool instead" not in execute_tool.description
assert "# Use grep tool instead" not in execute_tool.description
assert "# Use read_file tool instead" in execute_tool.description
assert "avoid read tools like cat, head, tail" in execute_tool.description

def test_execute_description_references_only_visible_grep_tool(self):
"""Execute docs retain grep guidance when glob is hidden."""
middleware = FilesystemMiddleware(
backend=_SandboxBackend(),
system_prompt="",
tools=["read_file", "grep", "execute"],
)
request = MagicMock()
request.tools = middleware.tools
request.runtime = _runtime()
request.override.return_value = request

middleware._filter_unsupported_tools_and_apply_prompt(request)

tools_overrides = [call for call in request.override.call_args_list if "tools" in call.kwargs]
execute_tool = next(tool for tool in middleware.tools if tool.name == "execute")
assert tools_overrides == []
assert "grep tool to search text" in execute_tool.description
assert "glob" not in execute_tool.description
assert "# Use glob tool instead" not in execute_tool.description
assert "# Use grep tool instead" in execute_tool.description

def test_execute_description_references_only_visible_glob_tool(self):
"""Execute docs retain glob guidance when grep is hidden."""
middleware = FilesystemMiddleware(
backend=_SandboxBackend(),
system_prompt="",
tools=["read_file", "glob", "execute"],
)
request = MagicMock()
request.tools = middleware.tools
request.runtime = _runtime()
request.override.return_value = request

middleware._filter_unsupported_tools_and_apply_prompt(request)

tools_overrides = [call for call in request.override.call_args_list if "tools" in call.kwargs]
execute_tool = next(tool for tool in middleware.tools if tool.name == "execute")
assert tools_overrides == []
assert "glob tool to find files" in execute_tool.description
assert "grep" not in execute_tool.description
assert "# Use glob tool instead" in execute_tool.description
assert "# Use grep tool instead" not in execute_tool.description

def test_custom_execute_description_is_not_rewritten_when_search_tools_hidden(self):
"""User-provided execute docs remain authoritative."""
custom_description = "Custom."
middleware = FilesystemMiddleware(
backend=_SandboxBackend(),
system_prompt="",
tools=["read_file", "execute"],
custom_tool_descriptions={"execute": custom_description},
)
request = MagicMock()
request.tools = middleware.tools
request.runtime = _runtime()
request.override.return_value = request

middleware._filter_unsupported_tools_and_apply_prompt(request)

tools_overrides = [call for call in request.override.call_args_list if "tools" in call.kwargs]
execute_tool = next(tool for tool in middleware.tools if tool.name == "execute")
assert tools_overrides == []
assert execute_tool.description == custom_description

def test_execute_description_swap_copies_dict_tool_specs(self):
"""Dict-shaped execute specs are swapped via a copy, leaving the input untouched."""
middleware = FilesystemMiddleware(backend=_SandboxBackend(), system_prompt="")
default_description = next(tool for tool in middleware.tools if tool.name == "execute").description
original = {"name": "execute", "description": default_description}
tools = [original]

rewritten = middleware._with_filtered_execute_description(tools, visible_search_tools=set())

rewritten_execute = next(tool for tool in rewritten if tool["name"] == "execute")
assert rewritten is not tools
assert rewritten_execute is not original
assert original["description"] == default_description
assert "# Use glob tool instead" not in rewritten_execute["description"]
assert "# Use grep tool instead" not in rewritten_execute["description"]
assert "# Use read_file tool instead" in rewritten_execute["description"]

def test_execute_description_rewrites_when_search_tools_are_filtered_from_request(self):
"""Runtime filtering corrects execute docs without mutating the registered tool."""
middleware = FilesystemMiddleware(backend=_SandboxBackend(), system_prompt="")
registered_execute = next(tool for tool in middleware.tools if tool.name == "execute")
request = MagicMock()
request.tools = [tool for tool in middleware.tools if tool.name not in {"grep", "glob"}]
request.runtime = _runtime()
request.override.return_value = request

middleware._filter_unsupported_tools_and_apply_prompt(request)

tools_override = [call.kwargs["tools"] for call in request.override.call_args_list if "tools" in call.kwargs][-1]
rewritten_execute = next(tool for tool in tools_override if tool.name == "execute")
assert rewritten_execute is not registered_execute
assert "use the grep, glob tools to search" in registered_execute.description
assert "grep" not in rewritten_execute.description
assert "glob" not in rewritten_execute.description

def test_enabled_tools_system_prompt_lists_only_enabled_tools(self):
"""Dynamic system prompt only mentions tools registered on `self.tools`."""
middleware = FilesystemMiddleware(
Expand Down