diff --git a/libs/code/deepagents_code/_constants.py b/libs/code/deepagents_code/_constants.py index 10e7537e42..b97acefc67 100644 --- a/libs/code/deepagents_code/_constants.py +++ b/libs/code/deepagents_code/_constants.py @@ -13,6 +13,20 @@ DEFAULT_AGENT_NAME: Final[str] = "agent" """Default agent / assistant identifier when no `-a` flag is given.""" +FS_TOOL_NAMES: Final[frozenset[str]] = frozenset( + {"ls", "read_file", "write_file", "edit_file", "delete", "glob", "grep", "execute"} +) +"""Mirror of the SDK's `FsToolName` literal members. + +Hardcoded here rather than derived from `deepagents.FsToolName` because +`deepagents` must not be imported on the arg-parsing hot path (see AGENTS.md +"Startup performance"); this module is dependency-free and safe for `main.py` to +import. Consumers (`main._parse_allow_fs_tools_flag`, +`tool_catalog.collect_built_in_tools`) alias this set, and `get_args(FsToolName)` +drift guards in `test_main_args` and `test_tool_catalog` pin it so a new or +renamed SDK filesystem tool fails a test instead of silently diverging. +""" + FIREWORKS_PROVIDER_ID_PREFIX: Final[str] = "accounts/fireworks/" """Prefix used to infer Fireworks from fully-qualified IDs.""" diff --git a/libs/code/deepagents_code/_server_config.py b/libs/code/deepagents_code/_server_config.py index 2c4afca7cb..0e07dd40ce 100644 --- a/libs/code/deepagents_code/_server_config.py +++ b/libs/code/deepagents_code/_server_config.py @@ -15,12 +15,14 @@ import os from dataclasses import dataclass from pathlib import Path -from typing import TYPE_CHECKING, Any +from typing import TYPE_CHECKING, Any, cast from deepagents_code._constants import DEFAULT_AGENT_NAME as DEFAULT_ASSISTANT_ID from deepagents_code._env_vars import SERVER_ENV_PREFIX if TYPE_CHECKING: + from deepagents import FsToolName + from deepagents_code.project_utils import ProjectContext @@ -68,6 +70,63 @@ def _read_env_json(suffix: str) -> Any: # noqa: ANN401 raise ValueError(msg) from exc +def _read_env_allow_fs_tools() -> list[FsToolName] | None: + """Read and shape-validate the `ALLOW_FS_TOOLS` filesystem allowlist. + + The parent writes only an absent variable (unrestricted — `None`, which is + also what `--allow-fs-tools all` collapses to) or a non-empty JSON list of + tool names (`main._parse_allow_fs_tools_flag`). This runs in the server + subprocess, where the variable could be tampered with, so — because the + value is a security control — any unrecognized shape must fail closed + (raise) rather than fall through to an unrestricted filesystem. + (`_read_env_json` already fails closed on malformed JSON.) + + `[]` and unknown tool names are rejected here, not deferred downstream, so + the returned list genuinely satisfies `list[FsToolName]` and the `cast` + asserts membership that was actually checked. Importing `deepagents` here is + fine: the subprocess already imports the SDK to build the agent (this is not + the arg-parsing hot path guarded in `main`). The `"read_file"` requirement + is not checked here: `ServerConfig.__post_init__` enforces it when the + returned value is placed on the config (with `FilesystemMiddleware` as a + final backstop), so a tampered list without `read_file` still fails closed + at construction. + + Returns: + `None` when the variable is absent, or a non-empty list of filesystem + tool-name strings, each a valid `FsToolName`. + + Raises: + ValueError: If the present variable parses to anything other than a + non-empty list of strings, or if any list element is not a + recognized filesystem tool name. + """ + env_name = f"{SERVER_ENV_PREFIX}ALLOW_FS_TOOLS" + if env_name not in os.environ: + return None + + raw = _read_env_json("ALLOW_FS_TOOLS") + if isinstance(raw, list) and raw and all(isinstance(name, str) for name in raw): + from typing import get_args + + from deepagents import FsToolName + + valid_names = frozenset(get_args(FsToolName)) + unknown = [name for name in raw if name not in valid_names] + if unknown: + msg = ( + f"Invalid {SERVER_ENV_PREFIX}ALLOW_FS_TOOLS value: unknown " + f"filesystem tool name(s) {unknown!r}; valid names are " + f"{sorted(valid_names)}." + ) + raise ValueError(msg) + return cast("list[FsToolName]", raw) + msg = ( + f"Invalid {SERVER_ENV_PREFIX}ALLOW_FS_TOOLS value: {raw!r}; expected " + "a non-empty list of filesystem tool names." + ) + raise ValueError(msg) + + def _read_env_str(suffix: str) -> str | None: """Read an optional `DEEPAGENTS_CODE_SERVER_*` string variable. @@ -261,6 +320,17 @@ class ServerConfig: `interpreter_ptc="all"` is paired with non-`auto_approve` mode. """ + allow_fs_tools: list[FsToolName] | None = None + """Allowlist for `FilesystemMiddleware`'s `tools` param, from + `--allow-fs-tools`. + + `None` means "all filesystem tools" and is also what `--allow-fs-tools all` + parses to: it leaves the SDK's own default `FilesystemMiddleware` in place + (no replacement). A list is an explicit allowlist of filesystem tool names, + must include `"read_file"`, and installs a restricted replacement (see + `create_cli_agent`). + """ + rubric_model: str | None = None """Grader model spec for `RubricMiddleware` (e.g. `'anthropic:...'`). @@ -307,7 +377,8 @@ def __post_init__(self) -> None: Raises: TypeError: If `rubric_max_iterations` is a boolean. - ValueError: If `shell_allow_list` is an empty list or + ValueError: If `shell_allow_list` is an empty list, + `allow_fs_tools` is an empty list or omits `"read_file"`, or `rubric_max_iterations` is non-positive. """ if self.sandbox_type == "none": @@ -315,6 +386,20 @@ def __post_init__(self) -> None: if self.shell_allow_list is not None and len(self.shell_allow_list) == 0: msg = "shell_allow_list must be None or non-empty" raise ValueError(msg) + # `allow_fs_tools` is a security control: `None` means unrestricted, but + # an explicit list must be a usable allowlist. Own the non-empty + + # `read_file`-required invariant here (the single authoritative point + # for both the env round-trip via `from_env` and direct construction) + # rather than deferring to `FilesystemMiddleware`, which would only + # surface the violation a process boundary away. `_parse_allow_fs_tools_flag` + # still enforces the same rule at the CLI for a friendlier error. + if self.allow_fs_tools is not None: + if len(self.allow_fs_tools) == 0: + msg = "allow_fs_tools must be None or a non-empty list" + raise ValueError(msg) + if "read_file" not in self.allow_fs_tools: + msg = "allow_fs_tools must include 'read_file'" + raise ValueError(msg) if isinstance(self.rubric_max_iterations, bool): msg = "rubric_max_iterations must be None or a positive integer" raise TypeError(msg) @@ -370,6 +455,11 @@ def to_env(self) -> dict[str, str | None]: "INTERPRETER_PTC_ACKNOWLEDGE_UNSAFE": str( self.interpreter_ptc_acknowledge_unsafe ).lower(), + "ALLOW_FS_TOOLS": ( + json.dumps(self.allow_fs_tools) + if self.allow_fs_tools is not None + else None + ), "RUBRIC_MODEL": self.rubric_model, "RUBRIC_MAX_ITERATIONS": ( str(self.rubric_max_iterations) @@ -425,6 +515,7 @@ def from_env(cls) -> ServerConfig: interpreter_ptc_acknowledge_unsafe=_read_env_bool( "INTERPRETER_PTC_ACKNOWLEDGE_UNSAFE" ), + allow_fs_tools=_read_env_allow_fs_tools(), rubric_model=_read_env_str("RUBRIC_MODEL") or None, rubric_max_iterations=_read_env_int("RUBRIC_MAX_ITERATIONS", default=None), sandbox_type=_read_env_str("SANDBOX_TYPE"), @@ -463,6 +554,7 @@ def from_cli_args( enable_interpreter: bool | None = None, interpreter_ptc: str | list[str] | None = None, interpreter_ptc_acknowledge_unsafe: bool = False, + allow_fs_tools: list[FsToolName] | None = None, rubric_model: str | None = None, rubric_max_iterations: int | None = None, mcp_config_path: str | None, @@ -499,6 +591,9 @@ def from_cli_args( interpreter_ptc: Override for `settings.interpreter_ptc`. interpreter_ptc_acknowledge_unsafe: Mirror of `settings.interpreter_ptc_acknowledge_unsafe`. + allow_fs_tools: Allowlist for `FilesystemMiddleware`'s `tools` + param to forward to the server subprocess. `None` leaves the + SDK default (all tools). rubric_model: Grader model spec; `None` reuses the main model. rubric_max_iterations: Explicit grader iterations per rubric attempt; `None` uses the SDK default. @@ -530,6 +625,7 @@ def from_cli_args( enable_interpreter=resolved_enable_interpreter, interpreter_ptc=interpreter_ptc, interpreter_ptc_acknowledge_unsafe=interpreter_ptc_acknowledge_unsafe, + allow_fs_tools=allow_fs_tools, rubric_model=rubric_model, rubric_max_iterations=rubric_max_iterations, sandbox_type=sandbox_type, diff --git a/libs/code/deepagents_code/agent.py b/libs/code/deepagents_code/agent.py index 72c764581a..9f8f090927 100644 --- a/libs/code/deepagents_code/agent.py +++ b/libs/code/deepagents_code/agent.py @@ -13,7 +13,7 @@ from pathlib import Path, PurePosixPath from typing import TYPE_CHECKING, Any, cast -from deepagents import create_deep_agent +from deepagents import FsToolName, create_deep_agent from deepagents.backends import CompositeBackend, LocalShellBackend from deepagents.backends.filesystem import FilesystemBackend from deepagents.middleware import ( @@ -174,6 +174,95 @@ class _NoTodoListMiddleware(AgentMiddleware): """ +def _get_harness_tool_descriptions( + model: str | BaseChatModel, +) -> dict[str, str]: + """Return the SDK harness's tool-description overrides for `model`. + + The CLI supplies its own `FilesystemMiddleware` when filesystem tools are + allowlisted. Because that middleware replaces the SDK-created instance, + it must carry forward the same model-specific descriptions. + + Args: + model: Model spec or resolved chat model used by the agent. + + Returns: + Copy of the matching harness profile's tool-description overrides. + """ + # deepagents-code exactly pins the SDK, and these are the same resolution + # helpers used by `create_deep_agent` for its filesystem middleware. + from deepagents.profiles.harness.harness_profiles import ( + _get_harness_profile, # noqa: PLC2701 # Mirrors SDK profile lookup. + _harness_profile_for_model, # noqa: PLC2701 # Mirrors SDK profile lookup. + ) + + if isinstance(model, str): + profile = _get_harness_profile(model) + return dict(profile.tool_description_overrides) if profile is not None else {} + return dict(_harness_profile_for_model(model, None).tool_description_overrides) + + +def _inject_fs_tools_into_subagents( + custom_subagents: list[SubAgent | CompiledSubAgent], + *, + fs_tools: list[FsToolName], + backend: CompositeBackend, + main_tool_descriptions: dict[str, str], +) -> None: + """Inject a filesystem-restricted `FilesystemMiddleware` into each subagent. + + Mutates each sync subagent spec in place, appending a `FilesystemMiddleware` + bound to `fs_tools` so delegating via `task` cannot bypass the allowlist. + Each subagent keeps its own harness tool descriptions (by its `model`, or + `main_tool_descriptions` when it inherits the runtime model). + + Args: + custom_subagents: Sync subagent specs to mutate. Must be raw `SubAgent` + dicts; see the `CompiledSubAgent` guard below. + fs_tools: The explicit allowlist to pass through to each subagent's + `FilesystemMiddleware`. + backend: Composite backend shared with the main agent's middleware. + main_tool_descriptions: Harness tool descriptions to use for a subagent + that inherits the runtime model (no explicit `model` key). + + Raises: + ValueError: If a `CompiledSubAgent` (identified by a `"runnable"` key, + matching the SDK's own `"runnable" in spec` discriminator in + `deepagents.middleware.subagents`) is present. Such a spec is used + as-is by the SDK and its `middleware` + key is never read, so we cannot enforce the restriction on it. dcode + adds only raw `SubAgent` dicts today, but the declared type admits + compiled specs: fail loud rather than silently exposing an + unrestricted filesystem via `task` delegation. + """ + for subagent in custom_subagents: + if "runnable" in subagent: + msg = ( + "Cannot enforce --allow-fs-tools on compiled subagent " + f"{subagent.get('name', '')!r}: its middleware is " + "not configurable, so the filesystem restriction would be " + "silently bypassed." + ) + raise ValueError(msg) + # `"runnable" in subagent` above narrows the union to `SubAgent`. + subagent_tool_descriptions = ( + _get_harness_tool_descriptions(subagent["model"]) + if "model" in subagent + else main_tool_descriptions + ) + subagent["middleware"] = cast( + "list[AgentMiddleware]", + [ + *subagent.get("middleware", []), + FilesystemMiddleware( + backend=backend, + tools=fs_tools, + custom_tool_descriptions=subagent_tool_descriptions, + ), + ], + ) + + def _todo_list_middleware_override() -> list[AgentMiddleware]: """Return the middleware needed to strip `TodoListMiddleware`, if enabled. @@ -882,6 +971,36 @@ def reset_agent( """Matches the `### Model Identity` section in the system prompt, up to the next heading or end of string.""" +_FS_TOOL_USAGE_INSTRUCTIONS: tuple[tuple[FsToolName, str], ...] = ( + ("edit_file", "- `edit_file` over `sed`/`awk`"), + ("write_file", "- `write_file` over `echo`/heredoc"), +) +"""dcode filesystem-tool preferences included in the generated prompt.""" + + +def _build_fs_tool_prompt_guidance(fs_tools: list[FsToolName] | None) -> str: + """Build dcode prompt guidance for the enabled filesystem tools. + + Args: + fs_tools: Filesystem tool allowlist, or `None` for all tools. + + Returns: + Filesystem preference guidance, or an empty string when neither + applicable tool is enabled. + """ + enabled = None if fs_tools is None else frozenset(fs_tools) + instructions = [ + instruction + for name, instruction in _FS_TOOL_USAGE_INSTRUCTIONS + if enabled is None or name in enabled + ] + if not instructions: + return "" + return ( + "IMPORTANT: Use specialized tools instead of shell commands:\n\n" + + "\n".join(instructions) + ) + def build_model_identity_section( name: str | None, @@ -932,6 +1051,7 @@ def get_system_prompt( *, interactive: bool = True, cwd: str | Path | None = None, + fs_tools: list[FsToolName] | None = None, ) -> str: """Get the base system prompt for the agent. @@ -949,6 +1069,8 @@ def get_system_prompt( interactive: When `False`, the prompt is tailored for headless non-interactive execution (no human in the loop). cwd: Override the working directory shown in the prompt. + fs_tools: Filesystem tool allowlist. Restricted prompts omit guidance + for unavailable tools; `None` retains all guidance. Returns: The system prompt string @@ -1029,6 +1151,7 @@ def get_system_prompt( context_limit=settings.model_context_limit, unsupported_modalities=settings.model_unsupported_modalities, ) + filesystem_tool_guidance = _build_fs_tool_prompt_guidance(fs_tools) # Build working directory section (local vs sandbox) if sandbox_type: @@ -1083,6 +1206,7 @@ def get_system_prompt( .replace("{model_identity_section}", model_identity_section) .replace("{working_dir_section}", working_dir_section) .replace("{skills_path}", skills_path) + .replace("{filesystem_tool_guidance}", filesystem_tool_guidance) ) # Detect unreplaced placeholders (defense-in-depth for template typos) @@ -1685,6 +1809,7 @@ def create_cli_agent( auto_mode_enabled: bool = False, interrupt_shell_only: bool = False, shell_allow_list: list[str] | None = None, + fs_tools: list[FsToolName] | None = None, enable_ask_user: bool = True, enable_memory: bool = True, memory_auto_save: bool = True, @@ -1761,6 +1886,16 @@ def create_cli_agent( the CLI process. When provided (and `interrupt_shell_only` is `True`), used directly instead of reading `settings.shell_allow_list` (which may not be set in the server subprocess environment). + fs_tools: Allowlist of filesystem tools to expose to the agent, from + `--allow-fs-tools`. `None` (default; also what `--allow-fs-tools + all` parses to) leaves `FilesystemMiddleware` at its SDK default + (all tools). An explicit list (which must include `"read_file"`) + installs a `FilesystemMiddleware` restricted to those tool names, + replacing the SDK's default for the main agent and every synchronous + subagent (including `general-purpose`) as well as the nested + goal-criteria agent, so delegation cannot bypass the restriction. + Async subagents are unaffected (they run on their own remote + backend, not the local filesystem). enable_ask_user: Enable `AskUserMiddleware` so the agent can ask clarifying questions. @@ -2234,6 +2369,7 @@ def _subagent_cli_middleware( sandbox_type=sandbox_type, interactive=interactive, cwd=effective_cwd, + fs_tools=fs_tools, ) } else: @@ -2305,6 +2441,40 @@ def _subagent_cli_middleware( routes={}, ) + if fs_tools is not None: + # `fs_tools` is an explicit allowlist here (`--allow-fs-tools all` and an + # omitted flag both arrive as `None`, leaving the SDK default in place). + main_tool_descriptions = _get_harness_tool_descriptions(model) + # Overrides the SDK's default `FilesystemMiddleware` (matched by + # `.name` in `create_deep_agent`'s custom-middleware merge) for the + # main agent. Preserve the SDK harness's model-specific tool metadata + # on the replacement. + # + # NOTE: this replacement only carries `backend`/`tools`/descriptions. + # The SDK also builds its default with `_permissions`; dcode passes no + # filesystem `permissions` to `create_deep_agent` today, so there is + # nothing to preserve. If dcode ever adopts filesystem permissions, + # they must be threaded through here (and into + # `_inject_fs_tools_into_subagents`) or `--allow-fs-tools` would + # silently strip them. + agent_middleware.append( + FilesystemMiddleware( + backend=composite_backend, + tools=fs_tools, + custom_tool_descriptions=main_tool_descriptions, + ) + ) + # dcode always supplies its own `general-purpose` spec, so the SDK's + # auto-created-GP middleware inheritance path never fires; the + # restriction must be injected into each subagent's own `middleware` + # list, or delegating via `task` could bypass `--allow-fs-tools`. + _inject_fs_tools_into_subagents( + custom_subagents, + fs_tools=fs_tools, + backend=composite_backend, + main_tool_descriptions=main_tool_descriptions, + ) + if goal_criteria_tools is not None: from deepagents_code.goal_rubric import ( GoalCriteriaMiddleware, @@ -2334,6 +2504,7 @@ def _subagent_cli_middleware( repository_root=criteria_root, context_tools=goal_criteria_tools, auto_mode_enabled=auto_mode_enabled, + fs_tools=fs_tools, ) criteria_fallback_agent = create_goal_criteria_fallback_agent(model=model) agent_middleware.append( diff --git a/libs/code/deepagents_code/app.py b/libs/code/deepagents_code/app.py index b727871f27..88344bdf0a 100644 --- a/libs/code/deepagents_code/app.py +++ b/libs/code/deepagents_code/app.py @@ -9475,6 +9475,7 @@ async def _handle_tools_command(self, command: str) -> None: collect_built_in_tools, assistant_id=self._assistant_id or DEFAULT_AGENT_NAME, enable_interpreter=enable_interpreter, + fs_tools=self._server_kwargs.get("allow_fs_tools"), ) except Exception: logger.exception("Failed to enumerate built-in tools for /tools") diff --git a/libs/code/deepagents_code/client/commands/tools.py b/libs/code/deepagents_code/client/commands/tools.py index eb19d26448..75509df86a 100644 --- a/libs/code/deepagents_code/client/commands/tools.py +++ b/libs/code/deepagents_code/client/commands/tools.py @@ -69,8 +69,9 @@ def _run_tools_list(args: argparse.Namespace) -> int: `tool_catalog.collect_catalog`) so names and descriptions never drift from what the model sees. The same runtime options that shape the agent's tool set are honored: the resolved interpreter setting controls whether `js_eval` - is listed, and the MCP options (`--no-mcp`, `--mcp-config`, - `--trust-project-mcp`) control MCP discovery. Those are top-level flags, so + is listed, `--allow-fs-tools` restricts filesystem tools, and the MCP options + (`--no-mcp`, `--mcp-config`, `--trust-project-mcp`) control MCP discovery. + Those are top-level flags, so they must precede the subcommand (e.g. `dcode --no-mcp tools list`). MCP discovery is best-effort: the built-in tools always render. Servers that @@ -88,8 +89,8 @@ def _run_tools_list(args: argparse.Namespace) -> int: Args: args: Parsed CLI namespace. Reads `output_format`, `agent`, - `interpreter`, `sandbox`, `no_mcp`, `mcp_config`, and - `trust_project_mcp`. + `interpreter`, `sandbox`, `allow_fs_tools`, `no_mcp`, `mcp_config`, + and `trust_project_mcp`. Returns: `0` on success (including best-effort MCP degradation); `1` when an @@ -97,7 +98,7 @@ def _run_tools_list(args: argparse.Namespace) -> int: """ from deepagents_code._constants import DEFAULT_AGENT_NAME from deepagents_code._server_config import _resolve_enable_interpreter - from deepagents_code.main import _resolve_agent_arg + from deepagents_code.main import _parse_allow_fs_tools_flag, _resolve_agent_arg from deepagents_code.tool_catalog import collect_catalog output_format: OutputFormat = getattr(args, "output_format", "text") @@ -111,6 +112,7 @@ def _run_tools_list(args: argparse.Namespace) -> int: catalog = collect_catalog( assistant_id=assistant_id, enable_interpreter=enable_interpreter, + fs_tools=_parse_allow_fs_tools_flag(getattr(args, "allow_fs_tools", None)), include_mcp=not getattr(args, "no_mcp", False), mcp_config_path=mcp_config_path, trust_project_mcp=_tools_list_project_mcp_trust(args), diff --git a/libs/code/deepagents_code/client/launch/server_manager.py b/libs/code/deepagents_code/client/launch/server_manager.py index 52dbb843c4..e8dd8435d7 100644 --- a/libs/code/deepagents_code/client/launch/server_manager.py +++ b/libs/code/deepagents_code/client/launch/server_manager.py @@ -26,6 +26,8 @@ if TYPE_CHECKING: from collections.abc import AsyncIterator + from deepagents import FsToolName + from deepagents_code.client.launch.server import ServerProcess from deepagents_code.client.remote_client import RemoteAgent from deepagents_code.mcp_tools import MCPSessionManager @@ -305,6 +307,7 @@ async def start_server_and_get_agent( enable_interpreter: bool | None = None, interpreter_ptc: str | list[str] | None = None, interpreter_ptc_acknowledge_unsafe: bool = False, + allow_fs_tools: list[FsToolName] | None = None, rubric_model: str | None = None, rubric_max_iterations: int | None = None, mcp_config_path: str | None = None, @@ -335,6 +338,9 @@ async def start_server_and_get_agent( interpreter_ptc: Override for `settings.interpreter_ptc` (PTC allowlist). interpreter_ptc_acknowledge_unsafe: Explicit acknowledgement for `interpreter_ptc="all"` outside of `auto_approve`. + allow_fs_tools: Allowlist for `FilesystemMiddleware`'s `tools` param. + + `None` leaves the SDK default (all tools). rubric_model: Grader model spec; `None` reuses the main model. rubric_max_iterations: Explicit grader iterations per rubric attempt; `None` uses the SDK default. @@ -385,6 +391,7 @@ async def start_server_and_get_agent( enable_interpreter=enable_interpreter, interpreter_ptc=interpreter_ptc, interpreter_ptc_acknowledge_unsafe=interpreter_ptc_acknowledge_unsafe, + allow_fs_tools=allow_fs_tools, rubric_model=rubric_model, rubric_max_iterations=rubric_max_iterations, mcp_config_path=mcp_config_path, @@ -458,6 +465,7 @@ async def server_session( enable_interpreter: bool | None = None, interpreter_ptc: str | list[str] | None = None, interpreter_ptc_acknowledge_unsafe: bool = False, + allow_fs_tools: list[FsToolName] | None = None, rubric_model: str | None = None, rubric_max_iterations: int | None = None, mcp_config_path: str | None = None, @@ -491,6 +499,9 @@ async def server_session( interpreter_ptc: Override for `settings.interpreter_ptc` (PTC allowlist). interpreter_ptc_acknowledge_unsafe: Explicit acknowledgement for `interpreter_ptc="all"` outside of `auto_approve`. + allow_fs_tools: Allowlist for `FilesystemMiddleware`'s `tools` param. + + `None` leaves the SDK default (all tools). rubric_model: Grader model spec; `None` reuses the main model. rubric_max_iterations: Explicit grader iterations per rubric attempt; `None` uses the SDK default. @@ -526,6 +537,7 @@ async def server_session( enable_interpreter=enable_interpreter, interpreter_ptc=interpreter_ptc, interpreter_ptc_acknowledge_unsafe=interpreter_ptc_acknowledge_unsafe, + allow_fs_tools=allow_fs_tools, rubric_model=rubric_model, rubric_max_iterations=rubric_max_iterations, mcp_config_path=mcp_config_path, diff --git a/libs/code/deepagents_code/client/non_interactive.py b/libs/code/deepagents_code/client/non_interactive.py index 3e4db9fb2f..665c655a74 100644 --- a/libs/code/deepagents_code/client/non_interactive.py +++ b/libs/code/deepagents_code/client/non_interactive.py @@ -84,6 +84,7 @@ from asyncio.subprocess import Process from pathlib import Path + from deepagents import FsToolName from langchain_core.runnables import RunnableConfig logger = logging.getLogger(__name__) @@ -1359,6 +1360,7 @@ async def run_non_interactive( enable_interpreter: bool | None = None, interpreter_ptc: str | list[str] | None = None, interpreter_ptc_acknowledge_unsafe: bool = False, + allow_fs_tools: list[FsToolName] | None = None, max_turns: int | None = None, rubric: str | None = None, rubric_model: str | None = None, @@ -1424,6 +1426,10 @@ async def run_non_interactive( allowlist for `js_eval`). interpreter_ptc_acknowledge_unsafe: Explicit acknowledgement for `interpreter_ptc="all"` outside of `auto_approve`. + allow_fs_tools: Allowlist for `FilesystemMiddleware`'s `tools` param, + from `--allow-fs-tools`. + + `None` leaves the SDK default (all tools). max_turns: Optional cap on total agentic turns. When `None`, the internal safety default applies. rubric: Acceptance criteria for `RubricMiddleware`. When provided, the @@ -1642,6 +1648,7 @@ def discover_all_skills() -> tuple[list[ExtendedSkillMetadata], list[Path]]: enable_interpreter=enable_interpreter, interpreter_ptc=interpreter_ptc, interpreter_ptc_acknowledge_unsafe=interpreter_ptc_acknowledge_unsafe, + allow_fs_tools=allow_fs_tools, rubric_model=rubric_model, rubric_max_iterations=rubric_max_iterations, mcp_config_path=mcp_config_path, diff --git a/libs/code/deepagents_code/goal_rubric.py b/libs/code/deepagents_code/goal_rubric.py index 60ea6af23f..8f28eff2a2 100644 --- a/libs/code/deepagents_code/goal_rubric.py +++ b/libs/code/deepagents_code/goal_rubric.py @@ -40,6 +40,7 @@ if TYPE_CHECKING: from collections.abc import Awaitable, Callable, Sequence + from deepagents import FsToolName from deepagents.backends.protocol import FileInfo from langchain.agents.middleware.human_in_the_loop import InterruptOnConfig from langchain.agents.middleware.types import ModelRequest, ModelResponse @@ -1480,6 +1481,7 @@ def _create_goal_criteria_agent( repository_root: str, context_tools: Sequence[BaseTool | Callable[..., Any]], auto_mode_enabled: bool, + fs_tools: list[FsToolName] | None = None, ) -> Any: # noqa: ANN401 """Build a criteria agent with the parent runtime's Auto eligibility. @@ -1489,6 +1491,10 @@ def _create_goal_criteria_agent( repository_root: Absolute path that bounds repository reads. context_tools: External context tools available to the criteria agent. auto_mode_enabled: Whether Auto may bypass delegated context approval. + fs_tools: Parent filesystem-tool allowlist. + + The criteria agent exposes only the allowed subset of its read-only + repository tools. Returns: Compiled criteria agent graph. @@ -1533,11 +1539,16 @@ def _create_goal_criteria_agent( _CriteriaContextBudgetMiddleware(), ] if repository_backend is not None: + # Annotated (not `cast`) so the type checker validates each literal + # against `FsToolName` and rejects a typo at check time. + repository_tools: list[FsToolName] = ["ls", "read_file", "glob", "grep"] + if fs_tools is not None: + repository_tools = [name for name in repository_tools if name in fs_tools] middleware.extend( [ FilesystemMiddleware( backend=repository_backend, - tools=["ls", "read_file", "glob", "grep"], + tools=repository_tools, grep_max_count=_REPOSITORY_GREP_MATCH_LIMIT, tool_token_limit_before_evict=None, ), diff --git a/libs/code/deepagents_code/main.py b/libs/code/deepagents_code/main.py index 0add915957..5ad654b950 100644 --- a/libs/code/deepagents_code/main.py +++ b/libs/code/deepagents_code/main.py @@ -23,9 +23,10 @@ from collections.abc import Callable, Sequence from enum import Enum from pathlib import Path -from typing import TYPE_CHECKING, Any, Literal, NoReturn +from typing import TYPE_CHECKING, Any, Literal, NoReturn, cast if TYPE_CHECKING: + from deepagents import FsToolName from rich.console import Console from deepagents_code.app import AppResult @@ -698,6 +699,85 @@ def _parse_interpreter_tools_flag( return names +# Aliased from the dependency-free `_constants` module (see its docstring for +# why the set is hardcoded, how the drift guard pins it, and why importing it +# here keeps the arg-parsing hot path free of a `deepagents` import). +from deepagents_code._constants import FS_TOOL_NAMES as _FS_TOOL_NAMES + + +def _parse_allow_fs_tools_flag( + raw: str | None, +) -> "list[FsToolName] | None": + """Parse `--allow-fs-tools` into `FilesystemMiddleware`'s `tools` shape. + + Args: + raw: Argparse value: `None` (flag absent), `"all"`, or a + comma-separated list of filesystem tool names. + + Returns: + `None` when the flag is absent *or* the value is `"all"` (both mean + "leave the SDK default filesystem middleware in place — all tools"), + or a list of trimmed, lower-cased tool names. + + Tool names are matched case-insensitively + (like the `"all"` sentinel), so `READ_FILE` and `read_file` + are equivalent. + + Calls `sys.exit(2)` when the value is empty, contains only blank + tokens, combines the `"all"` sentinel with other tool names, + includes an unknown tool name, or is an explicit list that + omits `"read_file"` — `FilesystemMiddleware` requires it. + """ + if raw is None: + return None + text = raw.strip() + if not text: + sys.stderr.write( + "Error: --allow-fs-tools requires a value: 'all', or a " + "comma-separated list of filesystem tool names.\n" + ) + sys.exit(2) + normalized = text.lower() + if normalized == "all": + # `"all"` collapses to `None`: both mean "all filesystem tools". `None` + # leaves the SDK's own default `FilesystemMiddleware` untouched, which is + # strictly safer than reinstalling a hand-built unrestricted instance + # (that would have to re-derive descriptions/permissions and could drift + # from the SDK default). Only an explicit sub-list installs a + # replacement middleware. + return None + # Lower-case each token so tool names are case-insensitive, matching the + # `"all"` sentinel above. SDK `FsToolName` members are all lower-case. + names = [token.strip().lower() for token in text.split(",") if token.strip()] + if not names: + sys.stderr.write( + "Error: --allow-fs-tools list must contain at least one " + "non-empty tool name.\n" + ) + sys.exit(2) + if "all" in names: # `names` are already lower-cased above. + sys.stderr.write( + "Error: --allow-fs-tools 'all' cannot be combined with other tool " + "names; pass 'all' on its own.\n" + ) + sys.exit(2) + unknown = [name for name in names if name not in _FS_TOOL_NAMES] + if unknown: + sys.stderr.write( + f"Error: --allow-fs-tools has unknown tool name(s): " + f"{', '.join(unknown)}. Valid names: " + f"{', '.join(sorted(_FS_TOOL_NAMES))}.\n" + ) + sys.exit(2) + if "read_file" not in names: + sys.stderr.write( + "Error: --allow-fs-tools list must include 'read_file'; it is " + "required by FilesystemMiddleware.\n" + ) + sys.exit(2) + return cast("list[FsToolName]", names) + + def _resolve_interpreter_enabled(args: argparse.Namespace) -> bool: """Return whether the JS interpreter should run for these CLI args. @@ -1999,6 +2079,17 @@ def help_parent(help_fn: Callable[[], None]) -> list[argparse.ArgumentParser]: "list of tool names (which may include the 'safe' preset, e.g. " "'safe,task'). Default is 'safe' (read-only file tools).", ) + parser.add_argument( + "--allow-fs-tools", + dest="allow_fs_tools", + metavar="LIST", + help="Allowlist of filesystem tools to expose to the agent: 'all', or " + "a comma-separated list of tool names (ls, read_file, write_file, " + "edit_file, delete, glob, grep, execute). 'read_file' must be " + "included in an explicit list. Note 'execute' is the shell tool: " + "omitting it from the list removes shell access even if shell is " + "otherwise enabled. Default is 'all'.", + ) parser.add_argument( "--update", @@ -2174,6 +2265,7 @@ async def run_textual_cli_async( interpreter_arg: bool | None = None, interpreter_ptc: str | list[str] | None = None, interpreter_ptc_acknowledge_unsafe: bool = False, + allow_fs_tools: "list[FsToolName] | None" = None, ) -> "AppResult": """Run the Textual TUI interface (async version). @@ -2236,6 +2328,10 @@ async def run_textual_cli_async( for `js_eval`). interpreter_ptc_acknowledge_unsafe: Explicit acknowledgement for `interpreter_ptc="all"` outside of `auto_approve`. + allow_fs_tools: Allowlist for `FilesystemMiddleware`'s `tools` param, + from `--allow-fs-tools`. + + `None` leaves the SDK default (all tools). Returns: An `AppResult` with the return code and final thread ID. @@ -2315,6 +2411,7 @@ async def run_textual_cli_async( "enable_interpreter": enable_interpreter, "interpreter_ptc": interpreter_ptc, "interpreter_ptc_acknowledge_unsafe": interpreter_ptc_acknowledge_unsafe, + "allow_fs_tools": allow_fs_tools, "mcp_config_path": mcp_config_path, "no_mcp": no_mcp, "trust_project_mcp": trust_project_mcp, @@ -2375,6 +2472,7 @@ async def _run_acp_cli_async( mcp_config_path: str | None = None, no_mcp: bool = False, trust_project_mcp: bool | None = None, + allow_fs_tools: "list[FsToolName] | None" = None, ) -> int: """Run ACP server mode and return a process exit code. @@ -2389,6 +2487,10 @@ async def _run_acp_cli_async( no_mcp: Disable all MCP tool loading. trust_project_mcp: Controls project-level server trust (stdio and remote alike). + allow_fs_tools: Allowlist for `FilesystemMiddleware`'s `tools` param, + from `--allow-fs-tools`. + + `None` leaves the SDK default (all tools). Returns: Exit code for ACP mode. @@ -2484,6 +2586,7 @@ async def _run_acp_cli_async( mcp_server_info=mcp_server_info, checkpointer=InMemorySaver(), async_subagents=async_subagents, + fs_tools=allow_fs_tools, memory_auto_save=is_memory_auto_save_enabled(), ) except Exception as exc: @@ -3424,6 +3527,9 @@ def cli_main() -> None: try: args = parse_args() + allow_fs_tools = _parse_allow_fs_tools_flag( + getattr(args, "allow_fs_tools", None) + ) if _show_bare_command_group_help(args): return @@ -3570,6 +3676,7 @@ def cli_main() -> None: mcp_config_path=getattr(args, "mcp_config", None), no_mcp=getattr(args, "no_mcp", False), trust_project_mcp=getattr(args, "trust_project_mcp", False), + allow_fs_tools=allow_fs_tools, ) ) sys.exit(exit_code) @@ -4418,6 +4525,7 @@ def cli_main() -> None: trust_project_mcp=getattr(args, "trust_project_mcp", False), enable_interpreter=enable_interpreter, interpreter_ptc=interpreter_ptc, + allow_fs_tools=allow_fs_tools, max_turns=getattr(args, "max_turns", None), rubric=rubric_text, rubric_model=getattr(args, "rubric_model", None), @@ -4572,6 +4680,7 @@ def cli_main() -> None: enable_interpreter=enable_interpreter, interpreter_arg=args.interpreter, interpreter_ptc=interpreter_ptc, + allow_fs_tools=allow_fs_tools, ) ) return_code = result.return_code diff --git a/libs/code/deepagents_code/server_graph.py b/libs/code/deepagents_code/server_graph.py index 0dc7c97eb3..89c0683a48 100644 --- a/libs/code/deepagents_code/server_graph.py +++ b/libs/code/deepagents_code/server_graph.py @@ -309,6 +309,7 @@ def _create_cli_agent_sync() -> Any: # noqa: ANN401 auto_mode_enabled=auto_mode_enabled, interrupt_shell_only=config.interrupt_shell_only, shell_allow_list=config.shell_allow_list, + fs_tools=config.allow_fs_tools, enable_ask_user=config.enable_ask_user, enable_memory=config.enable_memory, memory_auto_save=is_memory_auto_save_enabled(), diff --git a/libs/code/deepagents_code/system_prompt.md b/libs/code/deepagents_code/system_prompt.md index 6a5e744999..21420a9331 100644 --- a/libs/code/deepagents_code/system_prompt.md +++ b/libs/code/deepagents_code/system_prompt.md @@ -63,13 +63,7 @@ CRITICAL: Match what the user asked for EXACTLY. ## Tool Usage -IMPORTANT: Use specialized tools instead of shell commands: - -- `read_file` over `cat`/`head`/`tail` -- `edit_file` over `sed`/`awk` -- `write_file` over `echo`/heredoc -- `grep` tool over shell `grep`/`rg` -- `glob` over shell `find`/`ls` +{filesystem_tool_guidance} When performing multiple independent operations, make all tool calls in a single response — don't make sequential calls when parallel is possible. diff --git a/libs/code/deepagents_code/tool_catalog.py b/libs/code/deepagents_code/tool_catalog.py index 0edf3d8218..abb92507bc 100644 --- a/libs/code/deepagents_code/tool_catalog.py +++ b/libs/code/deepagents_code/tool_catalog.py @@ -25,11 +25,13 @@ from pathlib import Path from typing import TYPE_CHECKING, Any, Literal, cast +from deepagents_code._constants import FS_TOOL_NAMES from deepagents_code._fake_models import _ToolBindingFakeModel if TYPE_CHECKING: from collections.abc import Sequence + from deepagents import FsToolName from langgraph.prebuilt.tool_node import ToolNode from deepagents_code.mcp_tools import MCPServerInfo, MCPServerStatus @@ -47,6 +49,14 @@ BUILT_IN_GROUP = "Built-in" """Display label for the group of tools bundled with `deepagents-code`.""" +_FILESYSTEM_TOOL_NAMES = FS_TOOL_NAMES +"""Which enumerated tools the `fs_tools` allowlist governs. + +Aliased from the shared `_constants.FS_TOOL_NAMES` (see its docstring); the +drift guard in `test_tool_catalog` pins it so a new or renamed SDK filesystem +tool fails a test instead of silently escaping the leak check below. +""" + @dataclass(frozen=True, slots=True) class ToolEntry: @@ -180,7 +190,10 @@ def _first_line(text: str | None) -> str: def collect_built_in_tools( - *, assistant_id: str = "agent", enable_interpreter: bool = False + *, + assistant_id: str = "agent", + enable_interpreter: bool = False, + fs_tools: list[FsToolName] | None = None, ) -> list[ToolEntry]: """Enumerate the built-in tools the agent binds by default. @@ -198,6 +211,12 @@ def collect_built_in_tools( appears when the default agent would bind it. Callers should pass the resolved runtime setting (see `_resolve_enable_interpreter`) so the list matches the tools the agent actually binds. + fs_tools: Filesystem tool allowlist. Forwarded to the catalog agent so + it is built exactly like the runtime session. The SDK's + `FilesystemMiddleware` omits disallowed tools from the node + entirely, so forwarding alone narrows the enumeration; a defensive + check below verifies no disallowed tool leaked through and logs + loudly if one did (see the comment on that check). Returns: Built-in tools in bind order. @@ -223,11 +242,38 @@ def collect_built_in_tools( enable_skills=False, enable_shell=True, enable_interpreter=enable_interpreter, + fs_tools=fs_tools, ) tools = collect_tools_from_agent(agent) if tools is None: msg = "Compiled agent does not expose a LangGraph tool node" raise RuntimeError(msg) + # Defensive backstop against a change in SDK behavior. The SDK's + # `FilesystemMiddleware` omits disallowed tools from the bound node + # entirely, so `collect_tools_from_agent` should already return only + # allowlisted filesystem tools. If a disallowed tool *does* leak through, + # enforcement broke on the real agent (this enumeration is built from the + # same `create_cli_agent` the runtime uses). Return the *unfiltered* list + # and log loudly rather than scrubbing: scrubbing would hide the one signal + # that enforcement failed and make `/tools` report a restricted surface over + # an unrestricted agent. (`None` — the unrestricted default — skips this.) + if isinstance(fs_tools, list): + enabled = frozenset(fs_tools) + leaked = [ + tool.name + for tool in tools + if tool.name in _FILESYSTEM_TOOL_NAMES and tool.name not in enabled + ] + if leaked: + logger.error( + "Filesystem tool allowlist backstop detected %s in the tool " + "listing: the enumerated agent exposed disallowed filesystem " + "tool(s) not in the allowlist %s. This indicates the allowlist " + "was not applied to the underlying agent; the listing reflects " + "the agent's actual (unrestricted) tools.", + leaked, + sorted(enabled), + ) return tools @@ -467,6 +513,7 @@ def collect_catalog( *, assistant_id: str = "agent", enable_interpreter: bool = False, + fs_tools: list[FsToolName] | None = None, include_mcp: bool = True, mcp_config_path: str | None = None, trust_project_mcp: bool | None = None, @@ -478,6 +525,9 @@ def collect_catalog( tools, including any agent-specific subagents. enable_interpreter: Whether the default agent binds `js_eval`; forwarded to `collect_built_in_tools`. + fs_tools: Filesystem tool allowlist; forwarded to + `collect_built_in_tools`, which filters the built-in enumeration so + it matches the configured session. include_mcp: When `True`, discover MCP servers and append their groups after the built-in group (best-effort). Pass `False` to mirror `--no-mcp`. @@ -497,6 +547,7 @@ def collect_catalog( collect_built_in_tools( assistant_id=assistant_id, enable_interpreter=enable_interpreter, + fs_tools=fs_tools, ) ), ) diff --git a/libs/code/deepagents_code/ui.py b/libs/code/deepagents_code/ui.py index 1ef8f38444..6b9ffdbdeb 100644 --- a/libs/code/deepagents_code/ui.py +++ b/libs/code/deepagents_code/ui.py @@ -184,6 +184,10 @@ def show_help() -> None: " --interpreter-tools VALUE PTC allowlist: 'safe', 'all', or comma-separated " "tool names (may include 'safe')" ) + console.print( + " --allow-fs-tools LIST Filesystem tool allowlist: 'all' or " + "comma-separated tool names (must include 'read_file')" + ) console.print(" -n, --non-interactive MSG Run a single task and exit") console.print(" -q, --quiet Clean output for piping (needs -n)") console.print( diff --git a/libs/code/tests/unit_tests/client/commands/test_tools.py b/libs/code/tests/unit_tests/client/commands/test_tools.py index 5f521745b5..77c07d30ad 100644 --- a/libs/code/tests/unit_tests/client/commands/test_tools.py +++ b/libs/code/tests/unit_tests/client/commands/test_tools.py @@ -334,12 +334,13 @@ def test_list_discovery_failure_without_explicit_config_exits_zero(self) -> None assert "showing built-in tools only" in output def test_list_forwards_runtime_options(self) -> None: - """`--no-mcp`, `--mcp-config`, and interpreter resolution reach the catalog.""" + """Agent tool options reach the catalog.""" args = argparse.Namespace( tools_command="list", output_format="json", interpreter=True, sandbox="none", + allow_fs_tools="ls,read_file", no_mcp=True, mcp_config="/tmp/mcp.json", trust_project_mcp=True, @@ -355,11 +356,40 @@ def test_list_forwards_runtime_options(self) -> None: collect.assert_called_once_with( assistant_id="agent", enable_interpreter=True, + fs_tools=["ls", "read_file"], include_mcp=False, mcp_config_path="/tmp/mcp.json", trust_project_mcp=True, ) + def test_list_invalid_allow_fs_tools_exits(self) -> None: + """A malformed `--allow-fs-tools` fails fast with exit 2, before catalog. + + `_parse_allow_fs_tools_flag` is unit-tested exhaustively in isolation; + this pins the command-level contract that the bad value aborts the + `tools list` request rather than degrading to an unrestricted listing. + """ + args = argparse.Namespace( + tools_command="list", + output_format="json", + interpreter=False, + sandbox="none", + allow_fs_tools="bogus", + no_mcp=True, + mcp_config=None, + trust_project_mcp=False, + ) + with ( + patch( + "deepagents_code.tool_catalog.collect_catalog", + return_value=ToolCatalog(groups=()), + ) as collect, + pytest.raises(SystemExit) as exc_info, + ): + run_tools_command(args) + assert exc_info.value.code == 2 + collect.assert_not_called() + def test_list_defaults_trust_project_mcp_to_none(self) -> None: """Absent `--trust-project-mcp` forwards `None`. @@ -383,6 +413,7 @@ def test_list_defaults_trust_project_mcp_to_none(self) -> None: collect.assert_called_once_with( assistant_id="agent", enable_interpreter=False, + fs_tools=None, include_mcp=True, mcp_config_path=None, trust_project_mcp=None, diff --git a/libs/code/tests/unit_tests/smoke_tests/snapshots/system_prompt_interactive_local.md b/libs/code/tests/unit_tests/smoke_tests/snapshots/system_prompt_interactive_local.md index 1e24ffe0e6..25f76eb96c 100644 --- a/libs/code/tests/unit_tests/smoke_tests/snapshots/system_prompt_interactive_local.md +++ b/libs/code/tests/unit_tests/smoke_tests/snapshots/system_prompt_interactive_local.md @@ -66,11 +66,8 @@ CRITICAL: Match what the user asked for EXACTLY. IMPORTANT: Use specialized tools instead of shell commands: -- `read_file` over `cat`/`head`/`tail` - `edit_file` over `sed`/`awk` - `write_file` over `echo`/heredoc -- `grep` tool over shell `grep`/`rg` -- `glob` over shell `find`/`ls` When performing multiple independent operations, make all tool calls in a single response — don't make sequential calls when parallel is possible. diff --git a/libs/code/tests/unit_tests/test_agent.py b/libs/code/tests/unit_tests/test_agent.py index 1a76626fab..228abf3940 100644 --- a/libs/code/tests/unit_tests/test_agent.py +++ b/libs/code/tests/unit_tests/test_agent.py @@ -292,6 +292,7 @@ def test_goal_criteria_tools_wire_fallback_and_none_backend(tmp_path: Path) -> N create_cli_agent( model=model, assistant_id="test-agent", + fs_tools=["read_file"], enable_memory=False, enable_skills=False, enable_shell=False, @@ -302,6 +303,7 @@ def test_goal_criteria_tools_wire_fallback_and_none_backend(tmp_path: Path) -> N make_criteria.assert_called_once() assert make_criteria.call_args.kwargs["repository_backend"] is None + assert make_criteria.call_args.kwargs["fs_tools"] == ["read_file"] make_fallback.assert_called_once() # Primary and fallback agents share one model, and the middleware receives # both so graph-level failures can degrade to goal-only generation. @@ -1672,6 +1674,48 @@ def test_local_mode_omits_sandbox_warnings(self) -> None: assert "remote Linux sandbox" not in prompt +class TestGetSystemPromptFilesystemTools: + """Tests for filesystem allowlist guidance in the generated prompt.""" + + def test_restricted_prompt_omits_unavailable_mutation_tools(self) -> None: + mock_settings = Mock() + mock_settings.model_name = None + + with patch("deepagents_code.agent.settings", mock_settings): + prompt = get_system_prompt( + "test-agent", + fs_tools=["read_file", "execute"], + ) + + assert "`edit_file` over" not in prompt + assert "`write_file` over" not in prompt + assert "Use specialized tools instead of shell commands" not in prompt + + def test_restricted_prompt_keeps_enabled_mutation_tool(self) -> None: + mock_settings = Mock() + mock_settings.model_name = None + + with patch("deepagents_code.agent.settings", mock_settings): + prompt = get_system_prompt( + "test-agent", + fs_tools=["read_file", "edit_file"], + ) + + assert "`edit_file` over" in prompt + assert "`write_file` over" not in prompt + assert "Use specialized tools instead of shell commands" in prompt + + def test_unrestricted_prompt_keeps_all_mutation_tool_guidance(self) -> None: + mock_settings = Mock() + mock_settings.model_name = None + + with patch("deepagents_code.agent.settings", mock_settings): + prompt = get_system_prompt("test-agent") + + assert "`edit_file` over" in prompt + assert "`write_file` over" in prompt + + class TestGetSystemPromptPlaceholderValidation: """Tests for unreplaced placeholder detection.""" @@ -1759,6 +1803,7 @@ def create_agent(**_kwargs: Any) -> Mock: create_cli_agent( model="fake-model", assistant_id="my agent", + fs_tools=["read_file", "grep"], enable_memory=False, enable_skills=False, enable_shell=False, @@ -1768,6 +1813,7 @@ def create_agent(**_kwargs: Any) -> Mock: mock_get_prompt.assert_called_once() _, kwargs = mock_get_prompt.call_args assert kwargs["interactive"] is False + assert kwargs["fs_tools"] == ["read_file", "grep"] assert mock_create_deep_agent.call_args.kwargs["name"] == "my_agent" assert ( mock_create_deep_agent.call_args.kwargs["context_schema"] @@ -1823,6 +1869,7 @@ def test_explicit_system_prompt_ignores_interactive(self, tmp_path: Path) -> Non create_cli_agent( model="fake-model", assistant_id="test", + fs_tools=["read_file", "grep"], enable_memory=False, enable_skills=False, enable_shell=False, @@ -3813,6 +3860,602 @@ def test_preserves_explicit_subagent_model_without_configurable_middleware( ) +class TestCreateCliAgentFsToolsWiring: + """Verify `create_cli_agent` wires `fs_tools` into `FilesystemMiddleware`.""" + + @staticmethod + def _build_mock_settings(tmp_path: Path) -> Mock: + """Create a settings mock suitable for `create_cli_agent` wiring tests.""" + agent_dir = tmp_path / "agent" + agent_dir.mkdir() + skills_dir = tmp_path / "skills" + skills_dir.mkdir() + + mock_settings = Mock() + mock_settings.ensure_agent_dir.return_value = agent_dir + mock_settings.ensure_user_skills_dir.return_value = skills_dir + mock_settings.get_project_skills_dir.return_value = None + mock_settings.get_built_in_skills_dir.return_value = ( + Settings.get_built_in_skills_dir() + ) + mock_settings.get_user_agent_md_path.return_value = agent_dir / "AGENTS.md" + mock_settings.get_project_agent_md_path.return_value = [] + mock_settings.get_user_agents_dir.return_value = tmp_path / "agents" + mock_settings.get_project_agents_dir.return_value = None + mock_settings.model_name = None + mock_settings.model_provider = None + mock_settings.model_unsupported_modalities = frozenset() + mock_settings.model_context_limit = None + mock_settings.project_root = None + mock_settings.shell_allow_list = None + return mock_settings + + @staticmethod + def _fs_middleware_spy() -> tuple[list[dict[str, Any]], Any]: + """Return `(recorded_calls, factory)` for spying the FS-middleware ctor. + + `factory` records each call's kwargs and returns a *real* + `FilesystemMiddleware`, so `isinstance` checks on the agent's middleware + still hold while tests assert dcode's actual contract — the `tools=` it + passes — instead of the SDK-private `_enabled_tools` attribute (which an + SDK-internal rename could silently break). + """ + from deepagents.middleware.filesystem import FilesystemMiddleware + + calls: list[dict[str, Any]] = [] + + def factory(*args: Any, **kwargs: Any) -> Any: # noqa: ANN401 + calls.append(dict(kwargs)) + return FilesystemMiddleware(*args, **kwargs) + + return calls, factory + + def test_harness_tool_descriptions_accepts_model_instance(self) -> None: + """`_get_harness_tool_descriptions` handles a resolved model, not just a spec. + + The string-spec branch is exercised throughout this class via + `model="fake-model"`; the `BaseChatModel` branch (taken when the agent is + built from an already-instantiated model) is otherwise unexercised. It + must resolve a profile and return a plain dict rather than raise. + """ + from deepagents_code.agent import _get_harness_tool_descriptions + + result = _get_harness_tool_descriptions(_make_fake_chat_model()) + assert isinstance(result, dict) + + def test_restricted_middleware_replaces_sdk_default_by_name(self) -> None: + """The security guarantee rests on the SDK's replace-by-name merge. + + The other tests in this class assert what `create_cli_agent` *passes* + to `create_deep_agent`; they trust the SDK to replace its own default + `FilesystemMiddleware` with dcode's restricted one (matched by `.name`) + rather than append a second, unrestricted instance that would win. This + exercises the real SDK merge so that contract fails loudly here if it + ever changes, instead of silently leaving the restriction inert. + """ + from deepagents.graph import _apply_custom_middleware + from deepagents.middleware.filesystem import FilesystemMiddleware + + sdk_default = FilesystemMiddleware() # unrestricted, as the SDK builds it + restricted = FilesystemMiddleware(tools=["ls", "read_file"]) + # The merge key: both instances must share a `.name` or replacement + # degrades into appending two middleware. + assert restricted.name == sdk_default.name + + merged = _apply_custom_middleware([sdk_default], [restricted]) + + fs_middleware = [m for m in merged if isinstance(m, FilesystemMiddleware)] + assert len(fs_middleware) == 1 + # Identity is the contract: the restricted instance replaced the default + # rather than a second instance being appended. (No need to read the + # SDK-private `_enabled_tools` — that the *restricted* instance survived + # is exactly what proves replace-by-name.) + assert fs_middleware[0] is restricted + + def test_none_does_not_add_filesystem_middleware(self, tmp_path: Path) -> None: + """`fs_tools=None` (default) leaves the SDK's own default in place.""" + from deepagents.middleware.filesystem import FilesystemMiddleware + + mock_settings = self._build_mock_settings(tmp_path) + + mock_agent = Mock() + mock_agent.with_config.return_value = mock_agent + + fake_model = _make_fake_chat_model() + with ( + patch("deepagents_code.agent.settings", mock_settings), + patch("deepagents_code.agent.SkillsMiddleware"), + patch("deepagents_code.agent.MemoryMiddleware"), + patch( + "deepagents_code.agent.create_deep_agent", + return_value=mock_agent, + ) as mock_create, + patch( + "deepagents._models.init_chat_model", + return_value=fake_model, + ), + ): + create_cli_agent( + model="fake-model", + assistant_id="test", + enable_memory=False, + enable_skills=False, + enable_shell=True, + ) + + _, kwargs = mock_create.call_args + middleware_types = [type(m) for m in kwargs["middleware"]] + assert FilesystemMiddleware not in middleware_types + + def test_explicit_list_adds_restricted_filesystem_middleware( + self, tmp_path: Path + ) -> None: + """`fs_tools=[...]` installs a `FilesystemMiddleware` restricted to it.""" + from deepagents.middleware.filesystem import FilesystemMiddleware + + mock_settings = self._build_mock_settings(tmp_path) + + mock_agent = Mock() + mock_agent.with_config.return_value = mock_agent + + fs_calls, fs_factory = self._fs_middleware_spy() + fake_model = _make_fake_chat_model() + with ( + patch("deepagents_code.agent.settings", mock_settings), + patch("deepagents_code.agent.SkillsMiddleware"), + patch("deepagents_code.agent.MemoryMiddleware"), + patch( + "deepagents_code.agent.FilesystemMiddleware", + side_effect=fs_factory, + ), + patch( + "deepagents_code.agent.create_deep_agent", + return_value=mock_agent, + ) as mock_create, + patch( + "deepagents._models.init_chat_model", + return_value=fake_model, + ), + ): + create_cli_agent( + model="fake-model", + assistant_id="test", + fs_tools=["ls", "read_file"], + enable_memory=False, + enable_skills=False, + enable_shell=True, + ) + + _, kwargs = mock_create.call_args + fs_middleware = [ + m for m in kwargs["middleware"] if isinstance(m, FilesystemMiddleware) + ] + assert len(fs_middleware) == 1 + # dcode's contract: it constructs each allowlist FS middleware with the + # exact tool list. Asserting the ctor `tools=` kwarg avoids coupling to + # the SDK-private `_enabled_tools`. Filter to allowlist-driven + # constructions (those passing `tools=`); unrelated FS middleware — e.g. + # the rubric grader's — is built without it. + allowlisted = [call["tools"] for call in fs_calls if "tools" in call] + assert allowlisted + assert all(tools == ["ls", "read_file"] for tools in allowlisted) + + def test_allowlist_preserves_harness_descriptions_for_main_and_subagent( + self, tmp_path: Path + ) -> None: + """Allowlisting retains model-specific filesystem tool guidance.""" + from deepagents.middleware.filesystem import FilesystemMiddleware + + mock_settings = self._build_mock_settings(tmp_path) + mock_agent = Mock() + mock_agent.with_config.return_value = mock_agent + fake_model = _make_fake_chat_model() + + with ( + patch("deepagents_code.agent.settings", mock_settings), + patch("deepagents_code.agent.SkillsMiddleware"), + patch("deepagents_code.agent.MemoryMiddleware"), + patch( + "deepagents_code.agent.create_deep_agent", + return_value=mock_agent, + ) as mock_create, + patch( + "deepagents._models.init_chat_model", + return_value=fake_model, + ), + ): + create_cli_agent( + model="nvidia:nvidia/nemotron-3-ultra-550b-a55b", + assistant_id="test", + fs_tools=["ls", "read_file"], + enable_memory=False, + enable_skills=False, + enable_shell=True, + ) + + _, kwargs = mock_create.call_args + main_filesystem = next( + middleware + for middleware in kwargs["middleware"] + if isinstance(middleware, FilesystemMiddleware) + ) + general_purpose = next( + subagent + for subagent in kwargs["subagents"] + if subagent["name"] == "general-purpose" + ) + subagent_filesystem = next( + middleware + for middleware in general_purpose["middleware"] + if isinstance(middleware, FilesystemMiddleware) + ) + + for filesystem in (main_filesystem, subagent_filesystem): + read_file = next( + tool for tool in filesystem.tools if tool.name == "read_file" + ) + assert ( + "keep reading paginated chunks until you reach EOF" + in read_file.description + ) + + def test_explicit_list_narrows_effective_tools_main_and_subagent( + self, tmp_path: Path + ) -> None: + """An explicit allowlist narrows the *effective* filesystem tool set. + + The sibling wiring tests mock `create_deep_agent` and assert only the + `tools=` kwarg dcode forwards. This one reads the `FilesystemMiddleware` + instances dcode actually constructs — on the main agent and on the + injected `general-purpose` subagent — and asserts their model-visible + `.tools` contain exactly the allowlist and none of the disallowed names. + `.tools` is public and already omits disallowed tools, so this pins the + end-to-end restriction contract rather than just the constructor input. + """ + from deepagents.middleware.filesystem import FilesystemMiddleware + + mock_settings = self._build_mock_settings(tmp_path) + mock_agent = Mock() + mock_agent.with_config.return_value = mock_agent + fake_model = _make_fake_chat_model() + + with ( + patch("deepagents_code.agent.settings", mock_settings), + patch("deepagents_code.agent.SkillsMiddleware"), + patch("deepagents_code.agent.MemoryMiddleware"), + patch( + "deepagents_code.agent.create_deep_agent", + return_value=mock_agent, + ) as mock_create, + patch( + "deepagents._models.init_chat_model", + return_value=fake_model, + ), + ): + create_cli_agent( + model="fake-model", + assistant_id="test", + fs_tools=["ls", "read_file"], + enable_memory=False, + enable_skills=False, + enable_shell=True, + ) + + _, kwargs = mock_create.call_args + main_filesystem = next( + m for m in kwargs["middleware"] if isinstance(m, FilesystemMiddleware) + ) + general_purpose = next( + s for s in kwargs["subagents"] if s["name"] == "general-purpose" + ) + subagent_filesystem = next( + m + for m in general_purpose["middleware"] + if isinstance(m, FilesystemMiddleware) + ) + + disallowed = {"write_file", "edit_file", "delete", "glob", "grep", "execute"} + for filesystem in (main_filesystem, subagent_filesystem): + names = {tool.name for tool in filesystem.tools} + assert names == {"ls", "read_file"} + assert not (disallowed & names) + + def test_explicit_list_restricts_general_purpose_subagent( + self, tmp_path: Path + ) -> None: + """The auto-added `general-purpose` subagent inherits the restriction. + + dcode always supplies its own explicit `general-purpose` spec (so the + SDK's default-subagent inheritance never fires), so the restriction + must be injected into that subagent's own `middleware` list directly, + otherwise `task` could bypass `--allow-fs-tools` entirely. + """ + from deepagents.middleware.filesystem import FilesystemMiddleware + + mock_settings = self._build_mock_settings(tmp_path) + + mock_agent = Mock() + mock_agent.with_config.return_value = mock_agent + + fs_calls, fs_factory = self._fs_middleware_spy() + fake_model = _make_fake_chat_model() + with ( + patch("deepagents_code.agent.settings", mock_settings), + patch("deepagents_code.agent.SkillsMiddleware"), + patch("deepagents_code.agent.MemoryMiddleware"), + patch( + "deepagents_code.agent.FilesystemMiddleware", + side_effect=fs_factory, + ), + patch( + "deepagents_code.agent.create_deep_agent", + return_value=mock_agent, + ) as mock_create, + patch( + "deepagents._models.init_chat_model", + return_value=fake_model, + ), + ): + create_cli_agent( + model="fake-model", + assistant_id="test", + fs_tools=["ls", "read_file"], + enable_memory=False, + enable_skills=False, + enable_shell=True, + ) + + _, kwargs = mock_create.call_args + subagents = kwargs["subagents"] + gp_subagent = next(s for s in subagents if s["name"] == "general-purpose") + gp_fs_middleware = [ + m + for m in gp_subagent.get("middleware", []) + if isinstance(m, FilesystemMiddleware) + ] + assert len(gp_fs_middleware) == 1 + # Each allowlist-driven FS middleware (main agent + every subagent) uses + # the same tool list. Filter to `tools=`-bearing constructions so an + # unrelated FS middleware (e.g. the rubric grader's) doesn't interfere. + allowlisted = [call["tools"] for call in fs_calls if "tools" in call] + assert len(allowlisted) >= 2 + assert all(tools == ["ls", "read_file"] for tools in allowlisted) + + def test_restricts_every_sync_subagent_including_user_defined( + self, tmp_path: Path + ) -> None: + """The restriction is injected into *every* sync subagent, not just GP. + + `_build_mock_settings` yields no user subagents, so the other tests + exercise only the auto-added `general-purpose` spec. Here a user-defined + subagent (with its own explicit model, exercising the per-subagent + harness-description branch) is injected via `list_subagents`, proving the + "inject into each" contract for >1 subagent. A regression narrowing + injection to general-purpose-by-name would let `task` delegate to the + user subagent with an unrestricted filesystem — exactly the bypass this + feature prevents. + """ + from deepagents.middleware.filesystem import FilesystemMiddleware + + mock_settings = self._build_mock_settings(tmp_path) + mock_agent = Mock() + mock_agent.with_config.return_value = mock_agent + + user_subagent = { + "name": "researcher", + "description": "Researches things", + "system_prompt": "You research.", + "model": "anthropic:claude-haiku-4-5-20251001", + } + + fs_calls, fs_factory = self._fs_middleware_spy() + fake_model = _make_fake_chat_model() + with ( + patch("deepagents_code.agent.settings", mock_settings), + patch("deepagents_code.agent.SkillsMiddleware"), + patch("deepagents_code.agent.MemoryMiddleware"), + patch( + "deepagents_code.agent.list_subagents", + return_value=[user_subagent], + ), + patch( + "deepagents_code.agent.FilesystemMiddleware", + side_effect=fs_factory, + ), + patch( + "deepagents_code.agent.create_deep_agent", + return_value=mock_agent, + ) as mock_create, + patch( + "deepagents._models.init_chat_model", + return_value=fake_model, + ), + ): + create_cli_agent( + model="fake-model", + assistant_id="test", + fs_tools=["ls", "read_file"], + enable_memory=False, + enable_skills=False, + enable_shell=True, + ) + + _, kwargs = mock_create.call_args + subagents = kwargs["subagents"] + names = {subagent["name"] for subagent in subagents} + assert {"researcher", "general-purpose"} <= names + # Every sync subagent must carry exactly one restricted FS middleware. + for subagent in subagents: + fs = [ + middleware + for middleware in subagent.get("middleware", []) + if isinstance(middleware, FilesystemMiddleware) + ] + assert len(fs) == 1, f"{subagent['name']} missing FS middleware" + allowlisted = [call["tools"] for call in fs_calls if "tools" in call] + assert all(tools == ["ls", "read_file"] for tools in allowlisted) + + def test_subagent_uses_its_own_model_harness_descriptions( + self, tmp_path: Path + ) -> None: + """A subagent's injected FS middleware carries *its own* model's guidance. + + `_inject_fs_tools_into_subagents` resolves harness tool descriptions per + subagent: from `subagent["model"]` when it has one, else the main + model's. Here a `researcher` subagent has an explicit model distinct from + the runtime model, while the auto-added `general-purpose` inherits the + runtime model. We stub `_get_harness_tool_descriptions` to return a + per-model sentinel and assert each subagent's `read_file` description + reflects the right model — a regression that passed the main model's + descriptions to every subagent (the pre-fix behavior all other tests + missed) would give `researcher` the main sentinel and fail here. + """ + from deepagents.middleware.filesystem import FilesystemMiddleware + + researcher_model = "anthropic:claude-haiku-4-5-20251001" + + def fake_descriptions(model: object) -> dict[str, str]: + if model == researcher_model: + return {"read_file": "RESEARCHER-MODEL-GUIDANCE"} + return {"read_file": "MAIN-MODEL-GUIDANCE"} + + mock_settings = self._build_mock_settings(tmp_path) + mock_agent = Mock() + mock_agent.with_config.return_value = mock_agent + user_subagent = { + "name": "researcher", + "description": "Researches things", + "system_prompt": "You research.", + "model": researcher_model, + } + fake_model = _make_fake_chat_model() + with ( + patch("deepagents_code.agent.settings", mock_settings), + patch("deepagents_code.agent.SkillsMiddleware"), + patch("deepagents_code.agent.MemoryMiddleware"), + patch( + "deepagents_code.agent.list_subagents", + return_value=[user_subagent], + ), + patch( + "deepagents_code.agent._get_harness_tool_descriptions", + side_effect=fake_descriptions, + ), + patch( + "deepagents_code.agent.create_deep_agent", + return_value=mock_agent, + ) as mock_create, + patch( + "deepagents._models.init_chat_model", + return_value=fake_model, + ), + ): + create_cli_agent( + model="fake-model", + assistant_id="test", + fs_tools=["ls", "read_file"], + enable_memory=False, + enable_skills=False, + enable_shell=True, + ) + + _, kwargs = mock_create.call_args + subagents = {s["name"]: s for s in kwargs["subagents"]} + + def read_file_description(subagent: dict[str, Any]) -> str: + fs = next( + m for m in subagent["middleware"] if isinstance(m, FilesystemMiddleware) + ) + return next(t for t in fs.tools if t.name == "read_file").description + + # The researcher gets its own model's guidance; general-purpose (which + # inherits the runtime model) gets the main model's. + assert "RESEARCHER-MODEL-GUIDANCE" in read_file_description( + subagents["researcher"] + ) + assert "MAIN-MODEL-GUIDANCE" in read_file_description( + subagents["general-purpose"] + ) + assert "MAIN-MODEL-GUIDANCE" not in read_file_description( + subagents["researcher"] + ) + + def test_compiled_subagent_raises_rather_than_bypassing(self) -> None: + """A compiled subagent can't carry injected middleware → fail loud. + + `_inject_fs_tools_into_subagents` cannot enforce the allowlist on a + `CompiledSubAgent` (its `middleware` key is ignored by the SDK). dcode + never adds one today, but the guard must raise rather than silently + delegate `task` to it with an unrestricted filesystem. + """ + from deepagents_code.agent import _inject_fs_tools_into_subagents + + compiled = {"name": "precompiled", "runnable": object()} + with pytest.raises(ValueError, match="compiled subagent"): + _inject_fs_tools_into_subagents( + [compiled], # ty: ignore[invalid-argument-type] + fs_tools=["ls", "read_file"], + backend=Mock(), + main_tool_descriptions={}, + ) + + def test_async_subagents_are_not_restricted(self, tmp_path: Path) -> None: + """Async subagents run on a remote backend, so they get no FS middleware. + + The injection loop mutates only `custom_subagents`; async specs are + merged in separately. This pins the documented "async subagents are + unaffected" invariant so a future refactor that widened the loop to all + subagents would fail here. + """ + from deepagents.middleware.filesystem import FilesystemMiddleware + + mock_settings = self._build_mock_settings(tmp_path) + mock_agent = Mock() + mock_agent.with_config.return_value = mock_agent + + async_subagent = { + "name": "remote-researcher", + "description": "Remote research", + "graph_id": "research-graph", + } + + fake_model = _make_fake_chat_model() + with ( + patch("deepagents_code.agent.settings", mock_settings), + patch("deepagents_code.agent.SkillsMiddleware"), + patch("deepagents_code.agent.MemoryMiddleware"), + patch( + "deepagents_code.agent.create_deep_agent", + return_value=mock_agent, + ) as mock_create, + patch( + "deepagents._models.init_chat_model", + return_value=fake_model, + ), + ): + create_cli_agent( + model="fake-model", + assistant_id="test", + fs_tools=["ls", "read_file"], + async_subagents=[async_subagent], # ty: ignore[invalid-argument-type] + enable_memory=False, + enable_skills=False, + enable_shell=True, + ) + + _, kwargs = mock_create.call_args + remote = next( + subagent + for subagent in kwargs["subagents"] + if subagent["name"] == "remote-researcher" + ) + assert not [ + middleware + for middleware in remote.get("middleware", []) + if isinstance(middleware, FilesystemMiddleware) + ] + + class TestExperimentalTodoMiddlewareWiring: """`DEEPAGENTS_CODE_EXPERIMENTAL` drops TodoListMiddleware from every stack. diff --git a/libs/code/tests/unit_tests/test_app.py b/libs/code/tests/unit_tests/test_app.py index b212173e3f..6676a70e21 100644 --- a/libs/code/tests/unit_tests/test_app.py +++ b/libs/code/tests/unit_tests/test_app.py @@ -14819,7 +14819,10 @@ async def test_mounts_user_echo_then_catalog(self) -> None: app = DeepAgentsApp(agent=MagicMock()) app._assistant_id = "agent" - app._server_kwargs = {"enable_interpreter": False} + app._server_kwargs = { + "enable_interpreter": False, + "allow_fs_tools": ["ls", "read_file"], + } app._mcp_server_info = [ MCPServerInfo( name="docs", @@ -14837,7 +14840,11 @@ async def test_mounts_user_echo_then_catalog(self) -> None: ): await app._handle_command("/tools") - collect.assert_called_once_with(assistant_id="agent", enable_interpreter=False) + collect.assert_called_once_with( + assistant_id="agent", + enable_interpreter=False, + fs_tools=["ls", "read_file"], + ) assert mount.await_count == 2 first, second = (c.args[0] for c in mount.await_args_list) assert isinstance(first, UserMessage) @@ -15087,7 +15094,9 @@ async def test_forwards_enable_interpreter_true(self) -> None: ): await app._handle_command("/tools") - collect.assert_called_once_with(assistant_id="agent", enable_interpreter=True) + collect.assert_called_once_with( + assistant_id="agent", enable_interpreter=True, fs_tools=None + ) assert mount.await_count == 2 assert "js_eval" in mount.await_args_list[-1].args[0]._content.plain diff --git a/libs/code/tests/unit_tests/test_goal_rubric.py b/libs/code/tests/unit_tests/test_goal_rubric.py index 09df5e3fe7..d0175c557d 100644 --- a/libs/code/tests/unit_tests/test_goal_rubric.py +++ b/libs/code/tests/unit_tests/test_goal_rubric.py @@ -1001,6 +1001,73 @@ def test_wires_only_read_repository_tools_plus_external_context(self) -> None: for item in kwargs["middleware"] ) + def test_parent_allowlist_restricts_repository_tools(self) -> None: + """Nested criteria generation cannot bypass the parent fs allowlist.""" + backend = MagicMock() + filesystem = MagicMock() + graph = MagicMock() + graph.with_config.return_value = graph + + with ( + patch( + "deepagents.middleware.FilesystemMiddleware", + return_value=filesystem, + ) as filesystem_type, + patch("langchain.agents.create_agent", return_value=graph), + ): + _create_goal_criteria_agent( + model=MagicMock(), + repository_backend=backend, + repository_root="/workspace", + context_tools=[], + auto_mode_enabled=True, + fs_tools=["read_file"], + ) + + filesystem_type.assert_called_once_with( + backend=backend, + tools=["read_file"], + grep_max_count=_REPOSITORY_GREP_MATCH_LIMIT, + tool_token_limit_before_evict=None, + ) + + def test_parent_allowlist_intersects_with_repository_tools(self) -> None: + """The criteria agent keeps only allowed *repository* tools. + + The repository tool set is `["ls", "read_file", "glob", "grep"]`. Given + an allowlist that overlaps it partially and also names a non-repository + tool (`write_file`), the result must be the intersection in repository + order — multiple repository names survive, the disallowed repository + names (`glob`, `grep`) drop, and the non-repository name never appears. + """ + backend = MagicMock() + filesystem = MagicMock() + graph = MagicMock() + graph.with_config.return_value = graph + + with ( + patch( + "deepagents.middleware.FilesystemMiddleware", + return_value=filesystem, + ) as filesystem_type, + patch("langchain.agents.create_agent", return_value=graph), + ): + _create_goal_criteria_agent( + model=MagicMock(), + repository_backend=backend, + repository_root="/workspace", + context_tools=[], + auto_mode_enabled=True, + fs_tools=["ls", "read_file", "write_file"], + ) + + filesystem_type.assert_called_once_with( + backend=backend, + tools=["ls", "read_file"], + grep_max_count=_REPOSITORY_GREP_MATCH_LIMIT, + tool_token_limit_before_evict=None, + ) + @staticmethod def _async_hitl(*, auto_mode_enabled: bool = True) -> AsyncApprovalHITLMiddleware: from deepagents_code.agent import AsyncApprovalHITLMiddleware diff --git a/libs/code/tests/unit_tests/test_main_acp_mode.py b/libs/code/tests/unit_tests/test_main_acp_mode.py index bcdcc91502..570f60faa5 100644 --- a/libs/code/tests/unit_tests/test_main_acp_mode.py +++ b/libs/code/tests/unit_tests/test_main_acp_mode.py @@ -203,6 +203,92 @@ def test_acp_mode_omits_web_search_without_tavily() -> None: assert call_kwargs["checkpointer"] is not None +def test_acp_mode_forwards_allow_fs_tools() -> None: + """`--acp --allow-fs-tools` forwards the parsed allowlist as `fs_tools`.""" + args = _make_acp_args(allow_fs_tools="ls,read_file") + model_obj = object() + model_result = SimpleNamespace( + model=model_obj, + provider="anthropic", + model_name="claude-sonnet-4-6", + apply_to_settings=MagicMock(), + ) + server = object() + run_agent = AsyncMock(return_value=None) + resolve_mcp_tools = AsyncMock(return_value=([], None, [])) + + with ( + patch.object(sys, "argv", ["deepagents", "--acp"]), + patch( + "deepagents_code.main.check_cli_dependencies", + side_effect=AssertionError("check_cli_dependencies should be skipped"), + ), + patch("deepagents_code.main.parse_args", return_value=args), + patch("deepagents_code.config.settings", new=SimpleNamespace(has_tavily=False)), + patch("deepagents_code.model_config.save_recent_model", return_value=True), + patch("deepagents_code.config.create_model", return_value=model_result), + patch( + "deepagents_code.mcp_tools.resolve_and_load_mcp_tools", resolve_mcp_tools + ), + patch("deepagents_code.tools.fetch_url", new=object()), + patch("deepagents_code.tools.get_current_thread_id", new=object()), + patch("deepagents_code.tools.web_search", new=object()), + patch( + "deepagents_code.agent.create_cli_agent", return_value=("graph", object()) + ) as mock_create_agent, + patch("deepagents_acp.server.AgentServerACP", return_value=server), + patch("acp.run_agent", run_agent), + pytest.raises(SystemExit) as exc_info, + ): + cli_main() + + assert exc_info.value.code == 0 + mock_create_agent.assert_called_once() + assert mock_create_agent.call_args.kwargs["fs_tools"] == ["ls", "read_file"] + + +def test_acp_mode_forwards_none_allow_fs_tools_by_default() -> None: + """`--acp` without `--allow-fs-tools` forwards `fs_tools=None` (unrestricted).""" + args = _make_acp_args() # no allow_fs_tools override + model_result = SimpleNamespace( + model=object(), + provider="anthropic", + model_name="claude-sonnet-4-6", + apply_to_settings=MagicMock(), + ) + run_agent = AsyncMock(return_value=None) + resolve_mcp_tools = AsyncMock(return_value=([], None, [])) + + with ( + patch.object(sys, "argv", ["deepagents", "--acp"]), + patch( + "deepagents_code.main.check_cli_dependencies", + side_effect=AssertionError("check_cli_dependencies should be skipped"), + ), + patch("deepagents_code.main.parse_args", return_value=args), + patch("deepagents_code.config.settings", new=SimpleNamespace(has_tavily=False)), + patch("deepagents_code.model_config.save_recent_model", return_value=True), + patch("deepagents_code.config.create_model", return_value=model_result), + patch( + "deepagents_code.mcp_tools.resolve_and_load_mcp_tools", resolve_mcp_tools + ), + patch("deepagents_code.tools.fetch_url", new=object()), + patch("deepagents_code.tools.get_current_thread_id", new=object()), + patch("deepagents_code.tools.web_search", new=object()), + patch( + "deepagents_code.agent.create_cli_agent", return_value=("graph", object()) + ) as mock_create_agent, + patch("deepagents_acp.server.AgentServerACP", return_value=object()), + patch("acp.run_agent", run_agent), + pytest.raises(SystemExit) as exc_info, + ): + cli_main() + + assert exc_info.value.code == 0 + mock_create_agent.assert_called_once() + assert mock_create_agent.call_args.kwargs["fs_tools"] is None + + def test_mcp_preload_includes_plugin_configs() -> None: """The TUI metadata preload should include enabled plugin MCP servers.""" project_root = object() diff --git a/libs/code/tests/unit_tests/test_main_args.py b/libs/code/tests/unit_tests/test_main_args.py index 17d4f0b741..d287386442 100644 --- a/libs/code/tests/unit_tests/test_main_args.py +++ b/libs/code/tests/unit_tests/test_main_args.py @@ -2564,6 +2564,277 @@ def test_empty_value_exits(self) -> None: assert exc_info.value.code == 2 +class TestParseAllowFsToolsFlag: + """Tests for `_parse_allow_fs_tools_flag`.""" + + def test_none_returns_none(self) -> None: + from deepagents_code.main import _parse_allow_fs_tools_flag + + assert _parse_allow_fs_tools_flag(None) is None + + def test_all_collapses_to_none(self) -> None: + """`all` collapses to `None` (both mean "leave the SDK default").""" + from deepagents_code.main import _parse_allow_fs_tools_flag + + assert _parse_allow_fs_tools_flag("all") is None + + def test_explicit_list(self) -> None: + from deepagents_code.main import _parse_allow_fs_tools_flag + + assert _parse_allow_fs_tools_flag("ls,read_file,grep") == [ + "ls", + "read_file", + "grep", + ] + + def test_empty_value_exits(self, capsys: pytest.CaptureFixture[str]) -> None: + from deepagents_code.main import _parse_allow_fs_tools_flag + + with pytest.raises(SystemExit) as exc_info: + _parse_allow_fs_tools_flag(" ") + assert exc_info.value.code == 2 + # Distinct message, not one of the other exit paths. + assert "requires a value" in capsys.readouterr().err + + def test_unknown_tool_name_exits(self, capsys: pytest.CaptureFixture[str]) -> None: + from deepagents_code.main import _parse_allow_fs_tools_flag + + with pytest.raises(SystemExit) as exc_info: + _parse_allow_fs_tools_flag("read_file,bogus") + assert exc_info.value.code == 2 + err = capsys.readouterr().err + assert "unknown tool name" in err + assert "bogus" in err + + def test_missing_read_file_exits(self, capsys: pytest.CaptureFixture[str]) -> None: + from deepagents_code.main import _parse_allow_fs_tools_flag + + with pytest.raises(SystemExit) as exc_info: + _parse_allow_fs_tools_flag("ls,grep") + assert exc_info.value.code == 2 + assert "must include 'read_file'" in capsys.readouterr().err + + def test_tool_names_are_case_insensitive(self) -> None: + """Tool names are matched case-insensitively (like the `all` sentinel).""" + from deepagents_code.main import _parse_allow_fs_tools_flag + + assert _parse_allow_fs_tools_flag("LS,Read_File") == ["ls", "read_file"] + + def test_all_inside_list_exits(self, capsys: pytest.CaptureFixture[str]) -> None: + from deepagents_code.main import _parse_allow_fs_tools_flag + + with pytest.raises(SystemExit) as exc_info: + _parse_allow_fs_tools_flag("all,read_file") + assert exc_info.value.code == 2 + # A dedicated message, not the generic "unknown tool name" path. + assert "cannot be combined" in capsys.readouterr().err + + def test_all_is_case_insensitive(self) -> None: + from deepagents_code.main import _parse_allow_fs_tools_flag + + assert _parse_allow_fs_tools_flag("ALL") is None + assert _parse_allow_fs_tools_flag("All") is None + + def test_list_trims_and_skips_blank_tokens(self) -> None: + from deepagents_code.main import _parse_allow_fs_tools_flag + + assert _parse_allow_fs_tools_flag(" ls , read_file , ") == ["ls", "read_file"] + assert _parse_allow_fs_tools_flag("read_file,") == ["read_file"] + + def test_blank_only_tokens_exit(self, capsys: pytest.CaptureFixture[str]) -> None: + """A non-empty value that splits to zero tokens (e.g. ',') exits.""" + from deepagents_code.main import _parse_allow_fs_tools_flag + + with pytest.raises(SystemExit) as exc_info: + _parse_allow_fs_tools_flag(", ,") + assert exc_info.value.code == 2 + assert "at least one" in capsys.readouterr().err + + def test_fs_tool_names_match_sdk(self) -> None: + """`_FS_TOOL_NAMES` must not drift from the SDK's `FsToolName`.""" + from typing import get_args + + from deepagents import FsToolName + + from deepagents_code.main import _FS_TOOL_NAMES + + assert set(get_args(FsToolName)) == _FS_TOOL_NAMES + + +class TestAllowFsToolsArgument: + """Tests for --allow-fs-tools argument parsing and forwarding.""" + + def test_not_specified_is_none(self, mock_argv: MockArgvType) -> None: + with mock_argv(): + parsed = parse_args() + assert parsed.allow_fs_tools is None + + def test_parses_raw_value(self, mock_argv: MockArgvType) -> None: + with mock_argv("-n", "task", "--allow-fs-tools", "ls,read_file"): + parsed = parse_args() + assert parsed.allow_fs_tools == "ls,read_file" + + def test_help_lists_every_fs_tool_name(self, mock_argv: MockArgvType) -> None: + """The `--allow-fs-tools` help text must name every SDK filesystem tool. + + The help string hardcodes the tool-name list (`deepagents` must not be + imported on the arg-parsing path), so — unlike `_FS_TOOL_NAMES`, which a + drift test pins — it could silently go stale when the SDK adds a tool. + Spy the argparse registration to capture that specific help string and + guard it against `FsToolName`. + """ + import argparse + from typing import Any, get_args + + from deepagents import FsToolName + + captured: dict[str, str] = {} + real_add_argument = argparse.ArgumentParser.add_argument + + def spy(*args: Any, **kwargs: Any) -> Any: # noqa: ANN401 + # args[0] is the bound ArgumentParser instance; the flag strings + # follow. Match the registration for `--allow-fs-tools`. + if "--allow-fs-tools" in args: + captured["help"] = str(kwargs.get("help", "")) + return real_add_argument(*args, **kwargs) + + with ( + patch.object(argparse.ArgumentParser, "add_argument", spy), + mock_argv("-n", "task"), + ): + parse_args() + + assert "help" in captured, "--allow-fs-tools argument was not registered" + for name in get_args(FsToolName): + assert name in captured["help"], f"--allow-fs-tools help omits {name!r}" + + def test_forwarded_to_run_non_interactive(self) -> None: + """--allow-fs-tools is parsed and forwarded as allow_fs_tools.""" + from deepagents_code.main import cli_main + + mock_stdin = MagicMock() + mock_stdin.isatty.return_value = True + with ( + patch.object( + sys, + "argv", + [ + "deepagents", + "-n", + "do the thing", + "--allow-fs-tools", + "ls,read_file", + ], + ), + patch.object(sys, "stdin", mock_stdin), + patch("deepagents_code.main.check_optional_tools", return_value=[]), + patch( + "deepagents_code.main._should_ensure_managed_ripgrep", + return_value=False, + ), + patch( + "deepagents_code.client.non_interactive.run_non_interactive", + new_callable=AsyncMock, + return_value=0, + ) as mock_run, + pytest.raises(SystemExit), + ): + cli_main() + assert mock_run.await_args.kwargs["allow_fs_tools"] == ["ls", "read_file"] # ty: ignore + + def test_invalid_value_exits_before_startup_side_effects(self) -> None: + """Malformed allowlists fail before migration, installs, or prompts.""" + from deepagents_code.main import cli_main + + mock_stdin = MagicMock() + mock_stdin.isatty.return_value = True + with ( + patch.object( + sys, + "argv", + ["deepagents", "-n", "task", "--allow-fs-tools", "bogus"], + ), + patch.object(sys, "stdin", mock_stdin), + patch("deepagents_code.state_migration.migrate_legacy_state") as migrate, + patch("deepagents_code.main.check_optional_tools") as check_tools, + patch("deepagents_code.main._run_startup_auto_update") as update, + patch("deepagents_code.main._check_mcp_project_trust") as trust, + pytest.raises(SystemExit) as exc_info, + ): + cli_main() + + assert exc_info.value.code == 2 + migrate.assert_not_called() + check_tools.assert_not_called() + update.assert_not_called() + trust.assert_not_called() + + def test_not_forwarded_as_none_when_omitted(self) -> None: + """When --allow-fs-tools is omitted, allow_fs_tools=None is forwarded.""" + from deepagents_code.main import cli_main + + mock_stdin = MagicMock() + mock_stdin.isatty.return_value = True + with ( + patch.object(sys, "argv", ["deepagents", "-n", "do the thing"]), + patch.object(sys, "stdin", mock_stdin), + patch("deepagents_code.main.check_optional_tools", return_value=[]), + patch( + "deepagents_code.main._should_ensure_managed_ripgrep", + return_value=False, + ), + patch( + "deepagents_code.client.non_interactive.run_non_interactive", + new_callable=AsyncMock, + return_value=0, + ) as mock_run, + pytest.raises(SystemExit), + ): + cli_main() + assert mock_run.await_args.kwargs["allow_fs_tools"] is None # ty: ignore + + def test_forwarded_to_run_textual_cli(self) -> None: + """--allow-fs-tools is parsed and forwarded to the TUI launch path.""" + from deepagents_code.main import cli_main + + mock_stdin = MagicMock() + mock_stdin.isatty.return_value = True + + fake_result = MagicMock() + fake_result.return_code = 0 + fake_result.thread_id = None + fake_result.update_available = (False, None) + fake_result.session_stats = MagicMock(request_count=0) + run_tui = AsyncMock(return_value=fake_result) + + with ( + patch.object( + sys, + "argv", + ["deepagents", "-m", "hello", "--allow-fs-tools", "ls,read_file"], + ), + patch.object(sys, "stdin", mock_stdin), + patch("deepagents_code.main.run_textual_cli_async", run_tui), + patch("deepagents_code.main._run_startup_auto_update"), + patch("deepagents_code.main._resolve_agent_arg", return_value="agent"), + patch("deepagents_code.main._check_mcp_project_trust", return_value=False), + patch( + "deepagents_code.main._resolve_interpreter_enabled", + return_value=False, + ), + patch("deepagents_code.main._print_session_stats"), + patch( + "deepagents_code.main._should_check_teardown_thread", + return_value=False, + ), + ): + cli_main() + + run_tui.assert_awaited_once() + assert run_tui.await_args is not None + assert run_tui.await_args.kwargs["allow_fs_tools"] == ["ls", "read_file"] + + class TestInterpreterFlagParsing: """`--interpreter` is a tri-state `BooleanOptionalAction` (default `None`).""" diff --git a/libs/code/tests/unit_tests/test_non_interactive.py b/libs/code/tests/unit_tests/test_non_interactive.py index 21a72acc17..8f24166c55 100644 --- a/libs/code/tests/unit_tests/test_non_interactive.py +++ b/libs/code/tests/unit_tests/test_non_interactive.py @@ -366,6 +366,59 @@ async def test_sandbox_snapshot_name_passed_to_server(self) -> None: assert kwargs["sandbox_snapshot_name"] == "my-snap" +class TestAllowFsToolsForwarding: + """`allow_fs_tools` must survive the run_non_interactive plumbing. + + `start_server_and_get_agent` is mocked but `server_session` is not, so this + pins the middle hops (`run_non_interactive` -> `server_session` -> + `start_server_and_get_agent`) where a dropped kwarg would silently disable + the filesystem allowlist for every `-n` server session with a green suite. + """ + + async def test_allow_fs_tools_passed_to_server(self) -> None: + mock_agent = MagicMock() + mock_agent.astream = MagicMock(return_value=_async_iter([])) + mock_server_proc = MagicMock() + + with ( + patch( + "deepagents_code.client.non_interactive.create_model", + return_value=ModelResult( + model=MagicMock(), + model_name="test-model", + provider="test", + ), + ), + patch( + "deepagents_code.client.non_interactive.generate_thread_id", + return_value="test-thread", + ), + patch( + "deepagents_code.client.non_interactive.settings", + ) as mock_settings, + patch( + "deepagents_code.client.non_interactive.build_langsmith_thread_url", + return_value=None, + ), + patch( + "deepagents_code.client.launch.server_manager.start_server_and_get_agent", + new_callable=AsyncMock, + return_value=(mock_agent, mock_server_proc, None), + ) as mock_start_server, + ): + mock_settings.shell_allow_list = None + mock_settings.has_tavily = False + mock_settings.model_name = None + + await run_non_interactive( + message="test task", + allow_fs_tools=["ls", "read_file"], + ) + + _, kwargs = mock_start_server.call_args + assert kwargs["allow_fs_tools"] == ["ls", "read_file"] + + class TestQuietMode: """Tests for --quiet flag in run_non_interactive.""" diff --git a/libs/code/tests/unit_tests/test_server_graph.py b/libs/code/tests/unit_tests/test_server_graph.py index 39753942ff..761955dcb8 100644 --- a/libs/code/tests/unit_tests/test_server_graph.py +++ b/libs/code/tests/unit_tests/test_server_graph.py @@ -201,6 +201,12 @@ async def cleanup(self) -> None: config = ServerConfig( no_mcp=False, profile_overrides={"max_input_tokens": 32000}, + # Non-default allowlist so the `fs_tools=` assertion below is + # load-bearing: it round-trips through `to_env()`/`from_env()` and + # must reach `create_cli_agent`. With the `None` default this + # assertion passed whether or not `_make_graph` read + # `config.allow_fs_tools`, so a dropped read would go unnoticed. + allow_fs_tools=["ls", "read_file"], ) env_overrides = {} for suffix, value in config.to_env().items(): @@ -267,6 +273,7 @@ async def cleanup(self) -> None: auto_mode_enabled=False, interrupt_shell_only=False, shell_allow_list=None, + fs_tools=["ls", "read_file"], enable_ask_user=False, enable_memory=True, memory_auto_save=True, diff --git a/libs/code/tests/unit_tests/test_server_manager.py b/libs/code/tests/unit_tests/test_server_manager.py index d843f58098..974e9609aa 100644 --- a/libs/code/tests/unit_tests/test_server_manager.py +++ b/libs/code/tests/unit_tests/test_server_manager.py @@ -38,6 +38,7 @@ def test_round_trip_preserves_all_fields(self) -> None: auto_approve=True, interrupt_shell_only=True, shell_allow_list=["ls", "cat", "grep"], + allow_fs_tools=["ls", "read_file"], interactive=False, enable_shell=False, enable_ask_user=True, @@ -73,6 +74,98 @@ def test_defaults_round_trip(self) -> None: assert restored == original + def test_allow_fs_tools_list_round_trips(self) -> None: + """An explicit allowlist survives the env round trip as a JSON list.""" + original = ServerConfig(allow_fs_tools=["ls", "read_file"]) + env_dict = original.to_env() + with patch.dict(os.environ, {}, clear=True): + for suffix, value in env_dict.items(): + if value is not None: + os.environ[f"{SERVER_ENV_PREFIX}{suffix}"] = value + restored = ServerConfig.from_env() + + assert restored.allow_fs_tools == ["ls", "read_file"] + + def test_rejects_allow_fs_tools_without_read_file(self) -> None: + """An explicit allowlist missing `read_file` fails at construction. + + `ServerConfig.__post_init__` owns this invariant so a tampered env value + (which `_read_env_allow_fs_tools` intentionally does not check for + `read_file`) fails closed here rather than a process boundary away in + `FilesystemMiddleware`. + """ + with pytest.raises(ValueError, match="allow_fs_tools must include"): + ServerConfig(allow_fs_tools=["ls"]) + + def test_rejects_empty_allow_fs_tools(self) -> None: + """An empty explicit allowlist is rejected at construction.""" + with pytest.raises(ValueError, match="allow_fs_tools must be None"): + ServerConfig(allow_fs_tools=[]) + + def test_from_env_absent_allow_fs_tools_is_none(self) -> None: + """An absent `ALLOW_FS_TOOLS` var deserializes to `None` (unrestricted). + + `None` is the "flag omitted" state (also what `--allow-fs-tools all` + collapses to); it leaves the SDK default in place. Guards the + absent-variable passthrough in `_read_env_allow_fs_tools`. + """ + with patch.dict(os.environ, {}, clear=True): + os.environ.pop(f"{SERVER_ENV_PREFIX}ALLOW_FS_TOOLS", None) + restored = ServerConfig.from_env() + + assert restored.allow_fs_tools is None + + def test_from_env_rejects_invalid_allow_fs_tools_shape(self) -> None: + """A tampered/skewed ALLOW_FS_TOOLS value fails closed rather than open. + + Well-formed JSON of an unexpected type must raise instead of falling + through to an unrestricted filesystem — see `_read_env_allow_fs_tools`. + Covers non-list scalars/objects, a list containing non-strings (the + `all(isinstance(...))` guard), and the empty list (rejected directly so + the fail-closed guarantee is self-contained, not SDK-dependent). + """ + bad_values = ( + "null", # explicit null is not the same as an absent variable + '"all"', # the "all" sentinel is collapsed to None before serialize + '"read_file"', # bare string, not a list + "42", # number + "true", # boolean + "{}", # object + "[1, 2]", # list of non-strings + '["ls", null]', # list with a null element + "[]", # empty list + ) + for bad in bad_values: + with ( + patch.dict( + os.environ, + {f"{SERVER_ENV_PREFIX}ALLOW_FS_TOOLS": bad}, + clear=True, + ), + pytest.raises(ValueError, match="ALLOW_FS_TOOLS"), + ): + ServerConfig.from_env() + + def test_from_env_rejects_unknown_allow_fs_tools_name(self) -> None: + """A well-shaped list with an unrecognized tool name fails closed. + + The parent CLI (`_parse_allow_fs_tools_flag`) already rejects unknown + names, but the server subprocess re-validates independently: a tampered + value like `["read_file", "evil_tool"]` is a non-empty list of strings + (so it passes the shape guard) yet must still raise here rather than be + cast to `list[FsToolName]` and have the bogus name silently dropped + downstream. This keeps the `cast` in `_read_env_allow_fs_tools` honest. + """ + with ( + patch.dict( + os.environ, + {f"{SERVER_ENV_PREFIX}ALLOW_FS_TOOLS": '["read_file", "evil_tool"]'}, + clear=True, + ), + pytest.raises(ValueError, match="unknown filesystem tool name"), + ): + ServerConfig.from_env() + def test_trust_project_mcp_none_round_trips(self) -> None: """None trust_project_mcp should survive a round trip.""" original = ServerConfig(trust_project_mcp=None) @@ -237,6 +330,60 @@ async def test_passes_scaffold_hook_to_server_process( assert mock_server_process.call_args.kwargs["scaffold"] is mock_scaffold + async def test_forwards_allow_fs_tools_into_server_config( + self, tmp_path: Path, monkeypatch + ) -> None: + """`allow_fs_tools` reaches the `ServerConfig` written to the subprocess. + + The higher-level TUI/non-interactive forwarding tests mock this function + out, so without this a dropped kwarg here would disable the feature for + every server-backed session with no failing test. + """ + project_root = tmp_path / "project" + project_root.mkdir() + monkeypatch.chdir(project_root) + + work_dir = tmp_path / "runtime" + work_dir.mkdir() + + mock_server = MagicMock() + mock_server.start = AsyncMock() + mock_server.wait_for_graph_ready = AsyncMock() + mock_server.url = "http://127.0.0.1:2024" + + captured: list[ServerConfig] = [] + + with ( + patch.dict(os.environ, {}, clear=False), + patch( + "deepagents_code.client.launch.server_manager.tempfile.mkdtemp", + return_value=str(work_dir), + ), + patch("deepagents_code.client.launch.server_manager._write_checkpointer"), + patch("deepagents_code.client.launch.server_manager._write_pyproject"), + patch( + "deepagents_code.client.launch.server_manager._apply_server_config", + side_effect=captured.append, + ), + patch("deepagents_code.client.launch.server.generate_langgraph_json"), + patch( + "deepagents_code.client.launch.server.ServerProcess", + return_value=mock_server, + ), + patch( + "deepagents_code.client.remote_client.RemoteAgent", + return_value=object(), + ), + ): + await start_server_and_get_agent( + assistant_id="agent", + mcp_config_path=None, + allow_fs_tools=["ls", "read_file"], + ) + + assert len(captured) == 1 + assert captured[0].allow_fs_tools == ["ls", "read_file"] + async def test_stops_server_when_graph_readiness_fails( self, tmp_path: Path, monkeypatch ) -> None: diff --git a/libs/code/tests/unit_tests/test_tool_catalog.py b/libs/code/tests/unit_tests/test_tool_catalog.py index e32f431b9f..9c757de596 100644 --- a/libs/code/tests/unit_tests/test_tool_catalog.py +++ b/libs/code/tests/unit_tests/test_tool_catalog.py @@ -68,6 +68,123 @@ def test_includes_core_tools(self) -> None: assert tool.description assert "\n" not in tool.description + def test_respects_filesystem_allowlist(self) -> None: + """The catalog listing is narrowed to an explicit allowlist. + + Scope: this validates the `/tools` display contract for + `collect_built_in_tools`, NOT runtime `FilesystemMiddleware` enforcement + (covered in `test_agent.py`). The narrowing is produced by the SDK + middleware, which omits disallowed tools from the node entirely; the + `collect_built_in_tools` post-filter is a defensive backstop over the + same result. Either way the listing must exclude the disallowed names. + """ + names = { + tool.name for tool in collect_built_in_tools(fs_tools=["ls", "read_file"]) + } + assert {"ls", "read_file", "task"} <= names + assert ( + not { + "write_file", + "edit_file", + "delete", + "glob", + "grep", + "execute", + } + & names + ) + + def test_none_lists_every_filesystem_tool(self) -> None: + """`fs_tools=None` (unrestricted default) lists every filesystem tool.""" + names = {tool.name for tool in collect_built_in_tools(fs_tools=None)} + assert { + "ls", + "read_file", + "write_file", + "edit_file", + "delete", + "glob", + "grep", + "execute", + } <= names + + def test_backstop_surfaces_and_logs_when_disallowed_tool_leaks_through( + self, + caplog: pytest.LogCaptureFixture, + ) -> None: + """If the SDK ever stops narrowing, the listing must not silently lie. + + Normally the SDK's `FilesystemMiddleware` omits disallowed tools from the + node, so the check is a no-op. Here we simulate that guarantee breaking + (a disallowed `write_file` reaches enumeration) and assert the backstop + (a) keeps it in the listing — because the agent really does expose it, so + hiding it would misreport a restricted surface over an unrestricted + agent — and (b) logs an error so the discrepancy is visible. + """ + leaked = [ + ToolEntry(name="read_file", description="read"), + ToolEntry(name="write_file", description="write"), + ToolEntry(name="task", description="delegate"), + ] + with ( + patch( + "deepagents_code.agent.create_cli_agent", + return_value=(SimpleNamespace(), None), + ), + patch( + "deepagents_code.tool_catalog.collect_tools_from_agent", + return_value=leaked, + ), + caplog.at_level("ERROR", logger="deepagents_code.tool_catalog"), + ): + names = { + tool.name for tool in collect_built_in_tools(fs_tools=["read_file"]) + } + + # The leaked tool is surfaced, not scrubbed: the listing reflects the + # agent's real (unrestricted) tools rather than a false restricted view. + assert "write_file" in names + assert {"read_file", "task"} <= names + assert any( + "allowlist backstop detected" in record.getMessage() + and "write_file" in record.getMessage() + for record in caplog.records + ) + + def test_backstop_silent_when_allowlist_already_applied( + self, + caplog: pytest.LogCaptureFixture, + ) -> None: + """The backstop must stay quiet when enumeration already respects it.""" + applied = [ + ToolEntry(name="read_file", description="read"), + ToolEntry(name="task", description="delegate"), + ] + with ( + patch( + "deepagents_code.agent.create_cli_agent", + return_value=(SimpleNamespace(), None), + ), + patch( + "deepagents_code.tool_catalog.collect_tools_from_agent", + return_value=applied, + ), + caplog.at_level("ERROR", logger="deepagents_code.tool_catalog"), + ): + collect_built_in_tools(fs_tools=["read_file"]) + + assert not caplog.records + + def test_filesystem_tool_names_match_sdk(self) -> None: + """`_FILESYSTEM_TOOL_NAMES` must not drift from the SDK's `FsToolName`.""" + from typing import get_args + + from deepagents import FsToolName + + from deepagents_code.tool_catalog import _FILESYSTEM_TOOL_NAMES + + assert set(get_args(FsToolName)) == _FILESYSTEM_TOOL_NAMES + def test_web_search_present_with_tavily(self) -> None: with patch.object( Settings, "has_tavily", new_callable=PropertyMock, return_value=True @@ -99,10 +216,13 @@ def test_forwards_assistant_id_to_agent_compilation(self) -> None: "deepagents_code.agent.create_cli_agent", return_value=(agent, None), ) as create: - tools = collect_built_in_tools(assistant_id="custom-agent") + tools = collect_built_in_tools( + assistant_id="custom-agent", fs_tools=["ls", "read_file"] + ) assert tools == [ToolEntry(name="task", description="Run a subagent")] create.assert_called_once() assert create.call_args.kwargs["assistant_id"] == "custom-agent" + assert create.call_args.kwargs["fs_tools"] == ["ls", "read_file"] def test_raises_when_compiled_agent_not_inspectable(self) -> None: # A compiled agent whose graph does not expose the conventional tool @@ -550,7 +670,9 @@ def test_built_in_group_first_and_mcp_optional(self) -> None: ): catalog = collect_catalog(include_mcp=False) mock_mcp.assert_not_called() - built_in.assert_called_once_with(assistant_id="agent", enable_interpreter=False) + built_in.assert_called_once_with( + assistant_id="agent", enable_interpreter=False, fs_tools=None + ) assert len(catalog.groups) == 1 assert catalog.groups[0].label == BUILT_IN_GROUP assert catalog.groups[0].source == "built-in" @@ -578,11 +700,14 @@ def test_appends_mcp_groups_and_carries_unavailable(self) -> None: ): catalog = collect_catalog( assistant_id="custom-agent", + fs_tools=["ls", "read_file"], include_mcp=True, mcp_config_path="/tmp/mcp.json", ) built_in.assert_called_once_with( - assistant_id="custom-agent", enable_interpreter=False + assistant_id="custom-agent", + enable_interpreter=False, + fs_tools=["ls", "read_file"], ) # Built-in group stays first; MCP groups follow. assert catalog.groups[0].label == BUILT_IN_GROUP