feat(code): add dcode tools list command (#4461)

Add `dcode tools list` (with `--json`) to show the tools available to
the agent, grouped by source.

---

Adds a `dcode tools list` verb under the existing `dcode tools` group
that prints the tools available to the agent, grouped by source
(built-in tools first, then per-server MCP tools), in an Amp-style
layout with a total-count header and a `--json` mode. Tools are
enumerated from the *real* tool objects the agent binds — the agent is
compiled with an offline placeholder model (no credentials or network)
and its bound tool node is read — so names/descriptions never drift from
what the model sees. MCP discovery is best-effort and skipped on
failure.

Made by [Open
SWE](https://openswe.vercel.app/agents/96e689b3-3bc7-3a04-7135-c2001b8ed058)

## References
- Plan:
https://openswe.vercel.app/agents/96e689b3-3bc7-3a04-7135-c2001b8ed058/plan

---------

Co-authored-by: open-swe[bot] <open-swe@users.noreply.github.com>
This commit is contained in:
Mason Daugherty
2026-07-06 12:25:00 -04:00
committed by GitHub
parent e2a2151b79
commit 1402d0e735
10 changed files with 1430 additions and 54 deletions
+66
View File
@@ -0,0 +1,66 @@
"""Fake chat model base shared by integration tests and tool enumeration.
Holds the tool-binding base that both the local integration-test fakes
(`_testing_models`) and the `dcode tools list` tool-enumeration path
(`tool_catalog._CatalogModel`) build on. It lives in a use-neutral module — not
under a `_testing_`-prefixed name — so a production import path never depends on
something that reads as test-only and might be pruned or excluded from the wheel.
"""
from __future__ import annotations
from typing import TYPE_CHECKING, Any
from langchain_core.language_models.fake_chat_models import GenericFakeChatModel
from pydantic import Field
if TYPE_CHECKING:
from collections.abc import Callable, Sequence
from langchain_core.language_models import LanguageModelInput
from langchain_core.messages import AIMessage
from langchain_core.runnables import Runnable
from langchain_core.tools import BaseTool
_TOOL_BINDING_MODEL_PROFILE: dict[str, Any] = {
"tool_calling": True,
"max_input_tokens": 8000,
}
"""Minimal capability profile the agent runtime reads while compiling a model.
Only `tool_calling` is load-bearing — the agent negotiates tool support at
setup. `max_input_tokens` is part of the profile surface but inert here: these
models are compiled to bind tools and are never invoked, so no token budget ever
applies. Defined once so both the integration-test fakes and
`tool_catalog._CatalogModel` share a single source of truth.
"""
class _ToolBindingFakeModel(GenericFakeChatModel):
"""Base for fake chat models that must bind tools but are never invoked.
The agent runtime calls `model.bind_tools(schemas)` and reads `model.profile`
while compiling the graph, and a bare `GenericFakeChatModel` cannot be
compiled into an agent graph: it inherits `BaseChatModel.bind_tools`, which
raises `NotImplementedError`, and its `profile` is `None`, which breaks
capability negotiation. This base supplies a no-op `bind_tools` passthrough
and a minimal `profile`, leaving subclasses to add generation behavior
(tests) or nothing at all (tool enumeration).
"""
# Required by `GenericFakeChatModel`, but subclasses never consume it.
messages: object = Field(default_factory=lambda: iter(()))
profile: dict[str, Any] | None = Field(
default_factory=lambda: dict(_TOOL_BINDING_MODEL_PROFILE)
)
def bind_tools(
self,
tools: Sequence[dict[str, Any] | type | Callable | BaseTool], # noqa: ARG002
*,
tool_choice: str | None = None, # noqa: ARG002
**kwargs: Any, # noqa: ARG002
) -> Runnable[LanguageModelInput, AIMessage]:
"""Return self so the agent can bind tool schemas without a real model."""
return self
+22 -34
View File
@@ -1,21 +1,23 @@
"""Internal chat models used by local integration tests."""
"""Internal fake chat models for local integration tests.
The tool-binding base these build on (`_fake_models._ToolBindingFakeModel`) is
factored out into a use-neutral module so the `dcode tools list` enumeration
path can reuse it without importing this test-named module.
"""
from __future__ import annotations
from typing import TYPE_CHECKING, Any
from langchain_core.language_models.fake_chat_models import GenericFakeChatModel
from langchain_core.messages import AIMessage, BaseMessage
from langchain_core.outputs import ChatGeneration, ChatResult
from pydantic import Field
from deepagents_code._fake_models import _ToolBindingFakeModel
if TYPE_CHECKING:
from collections.abc import Callable, Sequence
from collections.abc import Callable
from langchain_core.callbacks import CallbackManagerForLLMRun
from langchain_core.language_models import LanguageModelInput
from langchain_core.runnables import Runnable
from langchain_core.tools import BaseTool
# Prompt markers that drive `ToolCallingIntegrationChatModel`. Each marker is the
@@ -33,13 +35,13 @@ TOP_LEVEL_WRITE_CONTENT = "auto-approved"
SUBAGENT_WRITE_CONTENT = "auto-approved-subagent"
class DeterministicIntegrationChatModel(GenericFakeChatModel):
class DeterministicIntegrationChatModel(_ToolBindingFakeModel):
"""Deterministic chat model for integration tests.
This subclasses LangChain's `GenericFakeChatModel` so the implementation
stays aligned with the core fake-chat-model test surface, while overriding
generation to remain prompt-driven and restart-safe for real CLI server
integration tests.
This subclasses `_ToolBindingFakeModel` (itself a `GenericFakeChatModel`) so
the implementation stays aligned with the core fake-chat-model test surface,
while overriding generation to remain prompt-driven and restart-safe for real
CLI server integration tests.
Why the existing `langchain_core` fakes cannot be reused here:
@@ -53,13 +55,15 @@ class DeterministicIntegrationChatModel(GenericFakeChatModel):
identical output regardless of process lifecycle.
2. The agent runtime calls `model.bind_tools(schemas)` during
initialization. None of the core fakes implement `bind_tools`, so they
raise `AttributeError` in any agent-loop context. This model provides a
initialization. A bare `GenericFakeChatModel` inherits
`BaseChatModel.bind_tools`, which raises `NotImplementedError` in any
agent-loop context. The inherited `_ToolBindingFakeModel` supplies a
no-op passthrough.
3. The app server reads `model.profile` for capability negotiation (e.g.
`tool_calling`, `max_input_tokens`). Core fakes have no such attribute,
causing `AttributeError` or silent misconfiguration at runtime.
`tool_calling`, `max_input_tokens`). A bare fake's `profile` is `None`,
causing silent misconfiguration at runtime. The inherited
`_ToolBindingFakeModel` supplies a minimal profile.
Additionally, the compact middleware issues summarization prompts mid-
conversation. A list-based model cannot distinguish these from normal user
@@ -68,24 +72,8 @@ class DeterministicIntegrationChatModel(GenericFakeChatModel):
"""
model: str = "fake"
# Required by `GenericFakeChatModel`, but our override does not consume it.
messages: object = Field(default_factory=lambda: iter(()))
profile: dict[str, Any] | None = Field(
default_factory=lambda: {
"tool_calling": True,
"max_input_tokens": 8000,
}
)
def bind_tools(
self,
tools: Sequence[dict[str, Any] | type | Callable | BaseTool], # noqa: ARG002
*,
tool_choice: str | None = None, # noqa: ARG002
**kwargs: Any, # noqa: ARG002
) -> Runnable[LanguageModelInput, AIMessage]:
"""Return self so the agent can bind tool schemas during tests."""
return self
# `messages`, `profile`, and the `bind_tools` passthrough are inherited from
# `_ToolBindingFakeModel`; this model adds only prompt-driven generation.
def _generate(
self,
+16 -9
View File
@@ -1702,6 +1702,9 @@ def create_cli_agent(
)
raise ValueError(msg)
# Lazy import keeps `dcode -v` fast — see AGENTS.md startup-perf rule.
from langchain_core._api import ( # noqa: PLC2701 # re-exported in _api.__all__
suppress_langchain_beta_warning,
)
from langchain_quickjs import CodeInterpreterMiddleware, PTCOption
ptc_names = _resolve_ptc_option(
@@ -1713,16 +1716,20 @@ def create_cli_agent(
ptc_option: PTCOption | None = (
cast("PTCOption", list(ptc_names)) if ptc_names is not None else None
)
agent_middleware.append(
CodeInterpreterMiddleware(
tool_name="js_eval",
timeout=settings.interpreter_timeout_seconds,
memory_limit=settings.interpreter_memory_limit_mb * 1024 * 1024,
max_ptc_calls=settings.interpreter_max_ptc_calls,
max_result_chars=settings.interpreter_max_result_chars,
ptc=ptc_option,
# `CodeInterpreterMiddleware` is decorated `@beta()`, which emits a
# `LangChainBetaWarning` on every instantiation. We intentionally use it
# and the warning is not actionable for users, so suppress it.
with suppress_langchain_beta_warning():
agent_middleware.append(
CodeInterpreterMiddleware(
tool_name="js_eval",
timeout=settings.interpreter_timeout_seconds,
memory_limit=settings.interpreter_memory_limit_mb * 1024 * 1024,
max_ptc_calls=settings.interpreter_max_ptc_calls,
max_result_chars=settings.interpreter_max_result_chars,
ptc=ptc_option,
)
)
)
# Local context middleware (git info, directory tree, etc.).
if isinstance(backend, (_ExecutableBackend, _AsyncExecutableBackend)):
+8
View File
@@ -1404,6 +1404,14 @@ def parse_args() -> argparse.Namespace:
)
add_json_output_arg(tools_install)
tools_list = tools_sub.add_parser(
"list",
help="List the tools available to the agent",
add_help=False,
parents=help_parent(_lazy_help("show_tools_list_help")),
)
add_json_output_arg(tools_list)
# Default interactive mode — argument order here determines the
# usage line printed by argparse; keep in sync with ui.show_help().
parser.add_argument(
+354
View File
@@ -0,0 +1,354 @@
"""Enumerate the tools available to the agent for `dcode tools list`.
The tool set is read from the *real* tool objects the agent binds rather than a
hand-maintained catalog, so names and descriptions never drift from what the
model actually sees. Built-in tools are collected by compiling the agent with a
throwaway offline chat model (no credentials, no network) and reading the bound
tool node; MCP tools are discovered via the same path the app and server use.
The collection functions here lazily import the heavy agent stack (agent
compilation, MCP discovery) inside their bodies. Only the fake-model base is
imported at module top, so importing this module is cheap relative to the agent
stack — and this module is itself imported lazily inside `_run_tools_list`,
never on the startup hot path. Those functions must only run on the
`dcode tools list` command path.
"""
from __future__ import annotations
import asyncio
import logging
from dataclasses import dataclass
from pathlib import Path
from typing import TYPE_CHECKING, Any, Literal, cast
from deepagents_code._fake_models import _ToolBindingFakeModel
if TYPE_CHECKING:
from langgraph.prebuilt.tool_node import ToolNode
from deepagents_code.mcp_tools import MCPServerStatus
logger = logging.getLogger(__name__)
ToolSource = Literal["built-in", "mcp"]
"""Stable source token identifying where a tool group comes from.
Emitted verbatim in the `--json` output, so it is a public contract; keep it a
`Literal` of stable tokens (not a bare `str`), following the same convention as
`mcp_tools.MCPServerStatus`.
"""
BUILT_IN_GROUP = "Built-in"
"""Display label for the group of tools bundled with `deepagents-code`."""
@dataclass(frozen=True, slots=True)
class ToolEntry:
"""A single tool's display metadata."""
name: str
"""Tool name as bound on the agent (e.g. `read_file`)."""
description: str
"""First non-empty line of the tool's description, whitespace-collapsed."""
@dataclass(frozen=True, slots=True)
class ToolGroup:
"""A named group of tools sharing a source."""
label: str
"""Group heading (`Built-in`, or the MCP server name)."""
source: ToolSource
"""Stable source token: `built-in` or `mcp`."""
tools: tuple[ToolEntry, ...]
"""Tools in this group, in bind order."""
@dataclass(frozen=True, slots=True)
class UnavailableServer:
"""An MCP server that was discovered but currently exposes no tools."""
name: str
"""Server name from the MCP configuration."""
status: MCPServerStatus
"""Load status token — any non-`ok` `mcp_tools.MCPServerStatus`.
Reuses `MCPServerStatus` (rather than a bare `str`) so the same closed value
set governs this field and its `--json` output, and so the token list has a
single source of truth in `mcp_tools`.
"""
detail: str
"""Human-readable reason from discovery, or `""` when none was given.
For config-load failures this is discovery's own reason string, which may
include the local config file path (e.g. `~/.deepagents/mcp.json: ...`) —
the same text the interactive `/mcp` viewer shows. See `collect_mcp_catalog`.
"""
@dataclass(frozen=True, slots=True)
class ToolCatalog:
"""Everything `dcode tools list` needs to render, in display order."""
groups: tuple[ToolGroup, ...]
"""Built-in group first, then one group per MCP server that exposes tools."""
unavailable: tuple[UnavailableServer, ...] = ()
"""MCP servers discovered with no tools (errored, needing login, or disabled).
Surfaced rather than dropped so a user debugging a missing tool can see why
it is absent.
"""
mcp_error: str | None = None
"""Generic notice set when MCP discovery itself failed; `None` on success.
Raw exception detail is logged at debug level, never embedded here, so no
file paths or stack traces leak into CLI/JSON output.
"""
class _CatalogModel(_ToolBindingFakeModel):
"""Offline placeholder model used only to compile the agent for enumeration.
Compiling the agent binds every tool but never calls the model, so this
never issues a request — enumeration only reads the bound tool node. It
exists so tool enumeration works without credentials or network access.
Inherits the `bind_tools` passthrough and minimal `profile` the agent
runtime reads during setup from `_ToolBindingFakeModel`.
"""
model: str = "catalog"
def _first_line(text: str | None) -> str:
"""Return the first non-empty line of `text`, whitespace-collapsed."""
if not text:
return ""
for line in text.splitlines():
stripped = line.strip()
if stripped:
return " ".join(stripped.split())
return ""
def collect_built_in_tools(
*, assistant_id: str = "agent", enable_interpreter: bool = False
) -> list[ToolEntry]:
"""Enumerate the built-in tools the agent binds by default.
Compiles the agent with an offline placeholder model and reads the bound
tool node. Memory and skills are disabled because they contribute no tools
(they only augment the system prompt). The selected assistant id is still
forwarded so agent-specific subagents are loaded from the same directory the
normal launch path uses. The custom CLI tools are included the same way
`server_graph._build_tools` adds them, so `web_search` appears only when
Tavily is configured.
Args:
assistant_id: Resolved dcode agent identifier to compile.
enable_interpreter: Wire the JS interpreter middleware so `js_eval`
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.
Returns:
Built-in tools in bind order.
"""
from deepagents_code.agent import create_cli_agent
from deepagents_code.config import settings
from deepagents_code.tools import fetch_url, get_current_thread_id, web_search
# Keep in sync with `server_graph._build_tools`: web_search is bound only
# when Tavily is configured, so it appears here only under the same gate.
custom_tools: list[Any] = [fetch_url, get_current_thread_id]
if settings.has_tavily:
custom_tools.append(web_search)
agent, _backend = create_cli_agent(
_CatalogModel(),
assistant_id=assistant_id,
tools=custom_tools,
enable_memory=False,
enable_skills=False,
enable_shell=True,
enable_interpreter=enable_interpreter,
)
# `agent.nodes["tools"].bound` reaches into langgraph's compiled graph; the
# "tools" node name and `ToolNode.tools_by_name` are langgraph conventions,
# not a documented contract. This real compile path is exercised by
# test_tool_catalog.py::TestCollectBuiltInTools, so a breaking langgraph
# change fails loudly there rather than silently emitting an empty list.
tool_node = cast("ToolNode", agent.nodes["tools"].bound)
tools_by_name = tool_node.tools_by_name
return [
ToolEntry(name=name, description=_first_line(tool.description))
for name, tool in tools_by_name.items()
]
def collect_mcp_catalog(
*,
mcp_config_path: str | None = None,
trust_project_mcp: bool | None = None,
) -> tuple[list[ToolGroup], list[UnavailableServer], str | None]:
"""Discover MCP servers, split into tool groups and unavailable servers.
Best-effort: if discovery itself raises (no config, offline, load error),
the technical detail is logged and a generic `mcp_error` message is
returned so `dcode tools list` still renders the built-in tools while
telling the user discovery failed. Servers that loaded but expose no tools
are reported as `UnavailableServer`s (errored, needing login, or disabled)
rather than silently dropped — surfacing exactly what a user running this
command to debug a missing tool needs to see.
Args:
mcp_config_path: Explicit MCP config path (`--mcp-config`), or `None`
to rely on auto-discovery.
trust_project_mcp: Project-level stdio trust decision
(`--trust-project-mcp`), forwarded to discovery unchanged.
Returns:
`(groups, unavailable, mcp_error)`: per-server tool groups, discovered
servers exposing no tools, and a generic discovery-failure message
(`None` when discovery succeeded).
"""
try:
server_info = asyncio.run(
_load_mcp_server_info(
mcp_config_path=mcp_config_path,
trust_project_mcp=trust_project_mcp,
)
)
except Exception:
# Log the real cause for debugging, but return a generic message so no
# file path or stack trace leaks into CLI/JSON output.
logger.warning("MCP tool discovery failed for `tools list`", exc_info=True)
return [], [], "MCP discovery failed; showing built-in tools only."
groups: list[ToolGroup] = []
unavailable: list[UnavailableServer] = []
for server in server_info or []:
if server.tools:
entries = tuple(
ToolEntry(name=tool.name, description=_first_line(tool.description))
for tool in server.tools
)
groups.append(ToolGroup(label=server.name, source="mcp", tools=entries))
elif server.status != "ok":
# A server that loaded but has no tools *and* is not "ok" is broken,
# unauthenticated, or disabled — report it so the omission is
# explained. `server.error` is discovery's own reason string (the
# same text the interactive `/mcp` viewer shows); it is not a stack
# trace, but for config-load failures it can include the local
# config file path — see `UnavailableServer.detail`.
unavailable.append(
UnavailableServer(
name=server.name,
status=server.status,
detail=server.error or "",
)
)
return groups, unavailable, None
async def _load_mcp_server_info(
*,
mcp_config_path: str | None,
trust_project_mcp: bool | None,
) -> list[Any]:
"""Load MCP server metadata, cleaning up any temporary sessions.
Args:
mcp_config_path: Explicit MCP config path, or `None` for auto-discovery.
trust_project_mcp: Project-level stdio trust decision.
Returns:
Discovered MCP server metadata, or an empty list when none load.
"""
from deepagents_code.mcp_tools import resolve_and_load_mcp_tools
from deepagents_code.project_utils import ProjectContext
try:
project_context = ProjectContext.from_user_cwd(Path.cwd())
except (OSError, RuntimeError):
# `Path.cwd()`/`.resolve()` raise OSError for a missing cwd and
# RuntimeError on a symlink loop (3.11-3.12); match the codebase's own
# convention in `project_utils` and fall back to no project context.
logger.warning("Could not determine working directory for MCP discovery")
project_context = None
session_manager = None
try:
_tools, session_manager, server_info = await resolve_and_load_mcp_tools(
explicit_config_path=mcp_config_path,
no_mcp=False,
trust_project_mcp=trust_project_mcp,
project_context=project_context,
)
return server_info or []
finally:
if session_manager is not None:
try:
await session_manager.cleanup()
except Exception:
logger.warning("MCP discovery cleanup failed", exc_info=True)
def collect_catalog(
*,
assistant_id: str = "agent",
enable_interpreter: bool = False,
include_mcp: bool = True,
mcp_config_path: str | None = None,
trust_project_mcp: bool | None = None,
) -> ToolCatalog:
"""Collect everything `dcode tools list` renders.
Args:
assistant_id: Resolved dcode agent identifier to compile for built-in
tools, including any agent-specific subagents.
enable_interpreter: Whether the default agent binds `js_eval`; forwarded
to `collect_built_in_tools`.
include_mcp: When `True`, discover MCP servers and append their groups
after the built-in group (best-effort). Pass `False` to mirror
`--no-mcp`.
mcp_config_path: Explicit MCP config path (`--mcp-config`).
trust_project_mcp: Project-level stdio trust decision
(`--trust-project-mcp`).
Returns:
A `ToolCatalog` with the built-in group first, then any MCP groups,
plus unavailable servers and any discovery-failure notice.
"""
groups: list[ToolGroup] = [
ToolGroup(
label=BUILT_IN_GROUP,
source="built-in",
tools=tuple(
collect_built_in_tools(
assistant_id=assistant_id,
enable_interpreter=enable_interpreter,
)
),
)
]
unavailable: list[UnavailableServer] = []
mcp_error: str | None = None
if include_mcp:
mcp_groups, unavailable, mcp_error = collect_mcp_catalog(
mcp_config_path=mcp_config_path,
trust_project_mcp=trust_project_mcp,
)
groups.extend(mcp_groups)
return ToolCatalog(
groups=tuple(groups),
unavailable=tuple(unavailable),
mcp_error=mcp_error,
)
+219 -8
View File
@@ -1,14 +1,18 @@
"""The `dcode tools` command group: provision managed external tools.
Currently this exposes `dcode tools install`, which fetches the pinned,
SHA-256-verified ripgrep binary into `~/.deepagents/bin` (the same managed
path used on first run) and is also handy for repairing a missing or stale
`rg`. The install script calls this verb instead of re-encoding the pinned
version + checksum table in bash.
`dcode tools install` fetches the pinned, SHA-256-verified ripgrep binary into
`~/.deepagents/bin` (the same managed path used on first run) and is also handy
for repairing a missing or stale `rg`. The install script calls this verb
instead of re-encoding the pinned version + checksum table in bash.
Help rendering for `dcode tools -h` / `dcode tools install -h` is served by
`ui.show_tools_help` / `ui.show_tools_install_help`, which do not import this
module, so the help path stays light.
`dcode tools list` prints the tools available to the agent, grouped by source
(built-in tools, then per-server MCP tools), enumerated from the real tool
objects the agent binds so the output never drifts from what the model sees.
Help rendering for `dcode tools -h` / `dcode tools install -h` /
`dcode tools list -h` is served by `ui.show_tools_help` /
`ui.show_tools_install_help` / `ui.show_tools_list_help`, which do not import
this module, so the help path stays light.
"""
from __future__ import annotations
@@ -23,6 +27,7 @@ if TYPE_CHECKING:
import argparse
from deepagents_code.output import OutputFormat
from deepagents_code.tool_catalog import ToolCatalog, UnavailableServer
logger = logging.getLogger(__name__)
@@ -46,6 +51,8 @@ def run_tools_command(args: argparse.Namespace) -> int:
subcommand = getattr(args, "tools_command", None)
if subcommand == "install":
return _run_tools_install(args)
if subcommand == "list":
return _run_tools_list(args)
# `cli_main`'s bare-group help fast path handles `dcode tools` with no
# subcommand, so this is only reached for an unexpected value.
@@ -55,6 +62,210 @@ def run_tools_command(args: argparse.Namespace) -> int:
return 0
def _run_tools_list(args: argparse.Namespace) -> int:
"""List the tools available to the agent, grouped by source.
Enumerates the real tool objects the agent binds (see
`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
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
errored, need login, or are disabled are still reported (not hidden) so a
user debugging a missing tool can see why it is absent. When discovery fails
outright while an explicit `--mcp-config` was supplied, the command exits
non-zero because the user's explicit request could not be satisfied.
The exit code is not a complete health signal: only a discovery *failure*
(missing/unparseable config) sets a non-zero code, and only for an explicit
`--mcp-config`. An explicit config that parses but whose server is merely
unreachable is surfaced as an `unavailable` entry with exit `0`. Scripts
that need per-server health must inspect `unavailable`/`mcp_error` in the
`--json` output, not the exit code alone.
Args:
args: Parsed CLI namespace. Reads `output_format`, `agent`,
`interpreter`, `sandbox`, `no_mcp`, `mcp_config`, and
`trust_project_mcp`.
Returns:
`0` on success (including best-effort MCP degradation); `1` when an
explicit `--mcp-config` was given but MCP discovery failed.
"""
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.tool_catalog import collect_catalog
output_format: OutputFormat = getattr(args, "output_format", "text")
mcp_config_path: str | None = getattr(args, "mcp_config", None)
assistant_id = (
_resolve_agent_arg(args) if hasattr(args, "agent") else DEFAULT_AGENT_NAME
)
enable_interpreter = _resolve_enable_interpreter(
getattr(args, "interpreter", None), getattr(args, "sandbox", None)
)
catalog = collect_catalog(
assistant_id=assistant_id,
enable_interpreter=enable_interpreter,
include_mcp=not getattr(args, "no_mcp", False),
mcp_config_path=mcp_config_path,
trust_project_mcp=_tools_list_project_mcp_trust(args),
)
# A failed *explicit* --mcp-config is a failed user request → non-zero exit;
# best-effort auto-discovery failures stay exit 0 (built-ins still render).
exit_code = 1 if catalog.mcp_error and mcp_config_path else 0
if output_format == "json":
tools_payload = [
{
"name": entry.name,
"description": entry.description,
"group": group.label,
"source": group.source,
}
for group in catalog.groups
for entry in group.tools
]
write_json(
"tools list",
{
"tools": tools_payload,
"count": len(tools_payload),
"unavailable": [
{
"name": server.name,
"status": server.status,
"detail": server.detail,
}
for server in catalog.unavailable
],
"mcp_error": catalog.mcp_error,
},
)
return exit_code
_print_catalog(catalog)
return exit_code
def _tools_list_project_mcp_trust(args: argparse.Namespace) -> bool | None:
"""Resolve project MCP trust behavior for `dcode tools list`.
Args:
args: Parsed CLI namespace.
Returns:
`True` when project MCP trust was explicitly requested, otherwise
`None` so MCP discovery can consult persisted trust.
"""
if getattr(args, "trust_project_mcp", False):
return True
return None
def _print_catalog(catalog: ToolCatalog) -> None:
"""Render a tool catalog to the console.
Prints the count header, the tool groups, then any unavailable MCP servers
and a discovery-failure notice.
Args:
catalog: Collected groups, unavailable servers, and discovery status.
"""
from deepagents_code.config import console, get_glyphs
ellipsis = get_glyphs().ellipsis
total = sum(len(group.tools) for group in catalog.groups)
noun = "tool" if total == 1 else "tools"
console.print()
console.print(f"{total} {noun} available", highlight=False)
for group in catalog.groups:
if not group.tools:
continue
name_width = max(len(entry.name) for entry in group.tools)
# Indent (2) + name column + gap (2) precede the description; keep each
# row on one line by truncating the description to the terminal width.
desc_width = console.width - 2 - name_width - 2
console.print()
console.print(group.label, style="bold", markup=False, highlight=False)
for entry in group.tools:
padded = entry.name.ljust(name_width)
description = _truncate(entry.description, desc_width, ellipsis)
# `markup=False`/`highlight=False`: tool names and descriptions are
# sourced from tool objects and may contain brackets or numbers.
console.print(
f" {padded} {description}".rstrip(),
markup=False,
highlight=False,
no_wrap=True,
crop=True,
)
_print_unavailable_servers(catalog.unavailable)
if catalog.mcp_error:
console.print()
console.print(f"Note: {catalog.mcp_error}", style="yellow", highlight=False)
console.print()
def _print_unavailable_servers(servers: tuple[UnavailableServer, ...]) -> None:
"""Render MCP servers that were discovered but expose no tools.
Args:
servers: Unavailable servers (errored, needing login, or disabled).
"""
if not servers:
return
from deepagents_code.config import console
name_width = max(len(server.name) for server in servers)
console.print()
console.print(
"Unavailable MCP servers", style="bold", markup=False, highlight=False
)
for server in servers:
padded = server.name.ljust(name_width)
# `status: detail` (ASCII-only, no em-dash) so legacy consoles don't hit
# an encoding error; detail is discovery's own curated reason string.
detail = f": {server.detail}" if server.detail else ""
console.print(
f" {padded} {server.status}{detail}".rstrip(),
style="dim",
markup=False,
highlight=False,
no_wrap=True,
crop=True,
)
def _truncate(text: str, width: int, ellipsis: str) -> str:
"""Truncate `text` to `width` columns, appending `ellipsis` when clipped.
Args:
text: Description text to truncate.
width: Maximum column width for the description.
ellipsis: Marker appended when `text` is clipped.
Returns:
`text` unchanged when it fits, otherwise a clipped string ending in
`ellipsis`.
"""
if width <= 0 or len(text) <= width:
return text
if width <= len(ellipsis):
return text[:width]
return text[: width - len(ellipsis)].rstrip() + ellipsis
def _run_tools_install(args: argparse.Namespace) -> int:
"""Install or repair the managed ripgrep binary.
+24
View File
@@ -545,12 +545,36 @@ def show_tools_help() -> None:
console.print()
console.print("[bold]Commands:[/bold]", style=theme.PRIMARY)
console.print(" install Install or repair the managed ripgrep binary")
console.print(" list List the tools available to the agent")
console.print()
_print_option_section()
console.print()
console.print("[bold]Examples:[/bold]", style=theme.PRIMARY)
console.print(" dcode tools install")
console.print(" dcode tools install --json")
console.print(" dcode tools list")
console.print(" dcode tools list --json")
console.print()
def show_tools_list_help() -> None:
"""Show help information for the `tools list` subcommand."""
console.print()
console.print("[bold]Usage:[/bold]", style=theme.PRIMARY)
console.print(" dcode tools list [options]")
console.print()
console.print(
"List the tools available to the agent, grouped by source: built-in",
)
console.print(
"tools first, then the tools exposed by each configured MCP server.",
)
console.print()
_print_option_section()
console.print()
console.print("[bold]Examples:[/bold]", style=theme.PRIMARY)
console.print(" dcode tools list")
console.print(" dcode tools list --json")
console.print()
@@ -168,6 +168,7 @@ def test_auth_credential_resolution_commands_run_settings_bootstrap() -> None:
["config", "show"],
["auth", "list"],
["tools", "install"],
["tools", "list"],
],
)
def test_subcommands_bypass_fast_path(argv: list[str]) -> None:
@@ -0,0 +1,334 @@
"""Tests for tool enumeration behind `dcode tools list`."""
from __future__ import annotations
import asyncio
from types import SimpleNamespace
from unittest.mock import AsyncMock, PropertyMock, patch
from deepagents_code.config import Settings
from deepagents_code.mcp_tools import MCPServerInfo, MCPToolInfo
from deepagents_code.tool_catalog import (
BUILT_IN_GROUP,
ToolEntry,
ToolGroup,
UnavailableServer,
_first_line,
_load_mcp_server_info,
collect_built_in_tools,
collect_catalog,
collect_mcp_catalog,
)
# Core tools the agent always binds, independent of optional integrations.
_CORE_BUILT_IN = {
"write_todos",
"ls",
"read_file",
"write_file",
"edit_file",
"delete",
"glob",
"grep",
"execute",
"task",
"ask_user",
"fetch_url",
"get_current_thread_id",
}
class TestFirstLine:
"""Tests for `_first_line` description normalization."""
def test_returns_first_non_empty_line_collapsed(self) -> None:
assert _first_line("\n Hello world \n\nmore") == "Hello world"
def test_empty_input(self) -> None:
assert _first_line(None) == ""
assert _first_line(" \n ") == ""
class TestCollectBuiltInTools:
"""Tests for enumerating built-in tools from the compiled agent."""
def test_includes_core_tools(self) -> None:
tools = collect_built_in_tools()
names = {tool.name for tool in tools}
assert names >= _CORE_BUILT_IN
# Every entry carries a non-empty, single-line description.
for tool in tools:
assert tool.description
assert "\n" not in tool.description
def test_web_search_present_with_tavily(self) -> None:
with patch.object(
Settings, "has_tavily", new_callable=PropertyMock, return_value=True
):
names = {tool.name for tool in collect_built_in_tools()}
assert "web_search" in names
def test_web_search_absent_without_tavily(self) -> None:
with patch.object(
Settings, "has_tavily", new_callable=PropertyMock, return_value=False
):
names = {tool.name for tool in collect_built_in_tools()}
assert "web_search" not in names
def test_interpreter_toggles_js_eval(self) -> None:
without = {tool.name for tool in collect_built_in_tools()}
assert "js_eval" not in without
with_interp = {
tool.name for tool in collect_built_in_tools(enable_interpreter=True)
}
assert "js_eval" in with_interp
def test_forwards_assistant_id_to_agent_compilation(self) -> None:
tool_node = SimpleNamespace(
tools_by_name={"task": SimpleNamespace(description="Run a subagent")}
)
agent = SimpleNamespace(nodes={"tools": SimpleNamespace(bound=tool_node)})
with patch(
"deepagents_code.agent.create_cli_agent",
return_value=(agent, None),
) as create:
tools = collect_built_in_tools(assistant_id="custom-agent")
assert tools == [ToolEntry(name="task", description="Run a subagent")]
create.assert_called_once()
assert create.call_args.kwargs["assistant_id"] == "custom-agent"
class TestCollectMcpCatalog:
"""Tests for MCP discovery: grouping tools and surfacing broken servers."""
def test_groups_ok_servers_and_surfaces_unavailable(self) -> None:
servers = [
MCPServerInfo(
name="docs",
transport="http",
tools=(
MCPToolInfo(name="search_docs", description="Search the docs\nX"),
),
status="ok",
),
MCPServerInfo(
name="broken",
transport="http",
status="error",
error="boom",
),
MCPServerInfo(
name="needslogin",
transport="http",
status="unauthenticated",
error="run login",
),
]
loader = AsyncMock(return_value=servers)
with patch("deepagents_code.tool_catalog._load_mcp_server_info", new=loader):
groups, unavailable, mcp_error = collect_mcp_catalog(
mcp_config_path="/tmp/mcp.json",
trust_project_mcp=True,
)
assert mcp_error is None
assert len(groups) == 1
group = groups[0]
assert group.label == "docs"
assert group.source == "mcp"
assert group.tools == (
ToolEntry(name="search_docs", description="Search the docs"),
)
# Non-ok servers are reported, not dropped, so the omission is explained.
assert unavailable == [
UnavailableServer(name="broken", status="error", detail="boom"),
UnavailableServer(
name="needslogin", status="unauthenticated", detail="run login"
),
]
# MCP options are forwarded to discovery unchanged.
loader.assert_awaited_once_with(
mcp_config_path="/tmp/mcp.json", trust_project_mcp=True
)
def test_ok_server_without_tools_is_neither_group_nor_unavailable(self) -> None:
servers = [MCPServerInfo(name="empty", transport="http", status="ok")]
with patch(
"deepagents_code.tool_catalog._load_mcp_server_info",
new=AsyncMock(return_value=servers),
):
groups, unavailable, mcp_error = collect_mcp_catalog()
assert groups == []
assert unavailable == []
assert mcp_error is None
def test_disabled_and_awaiting_reconnect_servers_are_surfaced(self) -> None:
# `disabled` and `awaiting_reconnect` share the `!= "ok"` branch with
# error/unauthenticated; lock the contract for the full non-ok set.
# (`MCPServerInfo` requires a reason for any non-ok status.)
servers = [
MCPServerInfo(
name="off",
transport="unknown",
status="disabled",
error="turned off via /mcp",
),
MCPServerInfo(
name="pending",
transport="http",
status="awaiting_reconnect",
error="reconnecting after login",
),
]
with patch(
"deepagents_code.tool_catalog._load_mcp_server_info",
new=AsyncMock(return_value=servers),
):
groups, unavailable, mcp_error = collect_mcp_catalog()
assert groups == []
assert mcp_error is None
assert unavailable == [
UnavailableServer(
name="off", status="disabled", detail="turned off via /mcp"
),
UnavailableServer(
name="pending",
status="awaiting_reconnect",
detail="reconnecting after login",
),
]
def test_discovery_failure_returns_generic_error_without_leaking(self) -> None:
with patch(
"deepagents_code.tool_catalog._load_mcp_server_info",
new=AsyncMock(side_effect=RuntimeError("secret /path/mcp.json boom")),
):
groups, unavailable, mcp_error = collect_mcp_catalog()
assert groups == []
assert unavailable == []
# Generic message only — raw exception text must not leak to output.
assert mcp_error == "MCP discovery failed; showing built-in tools only."
assert "secret" not in mcp_error
assert "/path/mcp.json" not in mcp_error
class TestLoadMcpServerInfo:
"""Tests for `_load_mcp_server_info` session lifecycle and cwd handling."""
def test_cleans_up_session_manager(self) -> None:
session_manager = AsyncMock()
server_info = [
MCPServerInfo(
name="docs",
transport="http",
tools=(MCPToolInfo(name="t", description="d"),),
)
]
loader = AsyncMock(return_value=([], session_manager, server_info))
with (
patch(
"deepagents_code.project_utils.ProjectContext.from_user_cwd",
return_value=None,
),
patch("deepagents_code.mcp_tools.resolve_and_load_mcp_tools", new=loader),
):
result = asyncio.run(
_load_mcp_server_info(mcp_config_path=None, trust_project_mcp=None)
)
assert result == server_info
session_manager.cleanup.assert_awaited_once()
def test_cleanup_failure_is_swallowed(self) -> None:
session_manager = AsyncMock()
session_manager.cleanup.side_effect = RuntimeError("cleanup boom")
server_info = [
MCPServerInfo(
name="docs",
transport="http",
tools=(MCPToolInfo(name="t", description="d"),),
)
]
loader = AsyncMock(return_value=([], session_manager, server_info))
with (
patch(
"deepagents_code.project_utils.ProjectContext.from_user_cwd",
return_value=None,
),
patch("deepagents_code.mcp_tools.resolve_and_load_mcp_tools", new=loader),
):
# A cleanup failure must not mask the return value or propagate.
result = asyncio.run(
_load_mcp_server_info(mcp_config_path=None, trust_project_mcp=None)
)
assert result == server_info
def test_cwd_oserror_forwards_none_project_context(self) -> None:
loader = AsyncMock(return_value=([], None, []))
with (
patch(
"deepagents_code.project_utils.ProjectContext.from_user_cwd",
side_effect=OSError("no cwd"),
),
patch("deepagents_code.mcp_tools.resolve_and_load_mcp_tools", new=loader),
):
result = asyncio.run(
_load_mcp_server_info(mcp_config_path=None, trust_project_mcp=None)
)
assert result == []
await_args = loader.await_args
assert await_args is not None
assert await_args.kwargs["project_context"] is None
class TestCollectCatalog:
"""Tests for the combined built-in + MCP assembly."""
def test_built_in_group_first_and_mcp_optional(self) -> None:
with (
patch(
"deepagents_code.tool_catalog.collect_built_in_tools",
return_value=[],
) as built_in,
patch("deepagents_code.tool_catalog.collect_mcp_catalog") as mock_mcp,
):
catalog = collect_catalog(include_mcp=False)
mock_mcp.assert_not_called()
built_in.assert_called_once_with(assistant_id="agent", enable_interpreter=False)
assert len(catalog.groups) == 1
assert catalog.groups[0].label == BUILT_IN_GROUP
assert catalog.groups[0].source == "built-in"
assert catalog.unavailable == ()
assert catalog.mcp_error is None
def test_appends_mcp_groups_and_carries_unavailable(self) -> None:
mcp_groups = [
ToolGroup(
label="docs",
source="mcp",
tools=(ToolEntry(name="search_docs", description="Search"),),
)
]
unavailable = [UnavailableServer(name="broken", status="error", detail="boom")]
with (
patch(
"deepagents_code.tool_catalog.collect_mcp_catalog",
return_value=(mcp_groups, unavailable, None),
),
patch(
"deepagents_code.tool_catalog.collect_built_in_tools",
return_value=[],
) as built_in,
):
catalog = collect_catalog(
assistant_id="custom-agent",
include_mcp=True,
mcp_config_path="/tmp/mcp.json",
)
built_in.assert_called_once_with(
assistant_id="custom-agent", enable_interpreter=False
)
# Built-in group stays first; MCP groups follow.
assert catalog.groups[0].label == BUILT_IN_GROUP
assert catalog.groups[-1].label == "docs"
assert catalog.unavailable == (
UnavailableServer(name="broken", status="error", detail="boom"),
)
@@ -8,16 +8,23 @@ import json
from pathlib import Path
from unittest.mock import patch
import pytest
from rich.console import Console
from deepagents_code import managed_tools
from deepagents_code._env_vars import OFFLINE, RIPGREP_INSTALLER
from deepagents_code.tools_commands import run_tools_command
from deepagents_code.tool_catalog import (
ToolCatalog,
ToolEntry,
ToolGroup,
UnavailableServer,
)
from deepagents_code.tools_commands import _truncate, run_tools_command
def _run_text(args: argparse.Namespace) -> tuple[int, str]:
def _run_text(args: argparse.Namespace, *, width: int = 200) -> tuple[int, str]:
buf = io.StringIO()
test_console = Console(file=buf, highlight=False, width=200)
test_console = Console(file=buf, highlight=False, width=width)
with patch("deepagents_code.config.console", test_console):
code = run_tools_command(args)
return code, buf.getvalue()
@@ -142,3 +149,379 @@ class TestToolsInstall:
code = run_tools_command(args)
assert code == 0
show_help.assert_called_once()
_SAMPLE_GROUPS = (
ToolGroup(
label="Built-in",
source="built-in",
tools=(
ToolEntry(name="read_file", description="Read a file"),
ToolEntry(name="execute", description="Run a shell command"),
),
),
ToolGroup(
label="docs",
source="mcp",
tools=(ToolEntry(name="search_docs", description="Search the docs"),),
),
)
_SAMPLE_CATALOG = ToolCatalog(groups=_SAMPLE_GROUPS, unavailable=(), mcp_error=None)
class TestToolsList:
"""Tests for `dcode tools list` dispatch."""
def test_list_text_output(self) -> None:
args = argparse.Namespace(tools_command="list", output_format="text")
with patch(
"deepagents_code.tool_catalog.collect_catalog",
return_value=_SAMPLE_CATALOG,
):
code, output = _run_text(args)
assert code == 0
assert "3 tools available" in output
assert "Built-in" in output
assert "read_file" in output
assert "Run a shell command" in output
# MCP tools grouped under their server name.
assert "docs" in output
assert "search_docs" in output
def test_list_json_output(self, capsys) -> None:
args = argparse.Namespace(tools_command="list", output_format="json")
with patch(
"deepagents_code.tool_catalog.collect_catalog",
return_value=_SAMPLE_CATALOG,
):
code = run_tools_command(args)
assert code == 0
envelope = json.loads(capsys.readouterr().out)
assert envelope["command"] == "tools list"
data = envelope["data"]
assert data["count"] == 3
assert len(data["tools"]) == 3
first = data["tools"][0]
assert first == {
"name": "read_file",
"description": "Read a file",
"group": "Built-in",
"source": "built-in",
}
assert data["tools"][-1]["source"] == "mcp"
assert data["tools"][-1]["group"] == "docs"
# No discovery problems for this catalog.
assert data["unavailable"] == []
assert data["mcp_error"] is None
def test_list_reports_unavailable_servers_text(self) -> None:
args = argparse.Namespace(tools_command="list", output_format="text")
catalog = ToolCatalog(
groups=(
ToolGroup(
label="Built-in",
source="built-in",
tools=(ToolEntry(name="ls", description="List files"),),
),
),
unavailable=(
UnavailableServer(name="broken", status="error", detail="boom"),
),
)
with patch(
"deepagents_code.tool_catalog.collect_catalog", return_value=catalog
):
code, output = _run_text(args)
assert code == 0
# The broken server is shown with its status and reason, not hidden.
assert "Unavailable MCP servers" in output
assert "broken" in output
assert "error" in output
assert "boom" in output
def test_list_json_includes_unavailable_and_mcp_error(self, capsys) -> None:
args = argparse.Namespace(tools_command="list", output_format="text")
# `output_format` defaults to text on the namespace; force json below.
args.output_format = "json"
catalog = ToolCatalog(
groups=(
ToolGroup(
label="Built-in",
source="built-in",
tools=(ToolEntry(name="ls", description="List files"),),
),
),
unavailable=(
UnavailableServer(
name="needslogin", status="unauthenticated", detail="run login"
),
),
mcp_error="MCP discovery failed; showing built-in tools only.",
)
with patch(
"deepagents_code.tool_catalog.collect_catalog", return_value=catalog
):
code = run_tools_command(args)
# No explicit --mcp-config on this namespace, so degradation is exit 0.
assert code == 0
data = json.loads(capsys.readouterr().out)["data"]
assert data["unavailable"] == [
{"name": "needslogin", "status": "unauthenticated", "detail": "run login"}
]
assert data["mcp_error"] == "MCP discovery failed; showing built-in tools only."
def test_list_explicit_mcp_config_failure_exits_nonzero(self, capsys) -> None:
"""A failed explicit --mcp-config is a failed user request → exit 1."""
args = argparse.Namespace(
tools_command="list",
output_format="json",
mcp_config="/tmp/mcp.json",
)
catalog = ToolCatalog(
groups=(
ToolGroup(
label="Built-in",
source="built-in",
tools=(ToolEntry(name="ls", description="List files"),),
),
),
mcp_error="MCP discovery failed; showing built-in tools only.",
)
with patch(
"deepagents_code.tool_catalog.collect_catalog", return_value=catalog
):
code = run_tools_command(args)
assert code == 1
data = json.loads(capsys.readouterr().out)["data"]
assert data["mcp_error"]
def test_list_discovery_failure_without_explicit_config_exits_zero(self) -> None:
"""Best-effort auto-discovery failure stays exit 0 (built-ins render)."""
args = argparse.Namespace(
tools_command="list", output_format="text", mcp_config=None
)
catalog = ToolCatalog(
groups=(
ToolGroup(
label="Built-in",
source="built-in",
tools=(ToolEntry(name="ls", description="List files"),),
),
),
mcp_error="MCP discovery failed; showing built-in tools only.",
)
with patch(
"deepagents_code.tool_catalog.collect_catalog", return_value=catalog
):
code, output = _run_text(args)
assert code == 0
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."""
args = argparse.Namespace(
tools_command="list",
output_format="json",
interpreter=True,
sandbox="none",
no_mcp=True,
mcp_config="/tmp/mcp.json",
trust_project_mcp=True,
)
with patch(
"deepagents_code.tool_catalog.collect_catalog",
return_value=ToolCatalog(groups=()),
) as collect:
code = run_tools_command(args)
# --no-mcp means no discovery, so mcp_error is None → exit 0 even with
# an explicit --mcp-config on the namespace.
assert code == 0
collect.assert_called_once_with(
assistant_id="agent",
enable_interpreter=True,
include_mcp=False,
mcp_config_path="/tmp/mcp.json",
trust_project_mcp=True,
)
def test_list_consults_persisted_project_mcp_trust_by_default(self) -> None:
"""Absent `--trust-project-mcp` lets MCP discovery use stored trust."""
args = argparse.Namespace(
tools_command="list",
output_format="json",
interpreter=False,
sandbox="none",
no_mcp=False,
mcp_config=None,
)
with patch(
"deepagents_code.tool_catalog.collect_catalog",
return_value=ToolCatalog(groups=()),
) as collect:
code = run_tools_command(args)
assert code == 0
collect.assert_called_once_with(
assistant_id="agent",
enable_interpreter=False,
include_mcp=True,
mcp_config_path=None,
trust_project_mcp=None,
)
def test_list_forwards_selected_agent(self) -> None:
"""`--agent` selects the agent whose built-in subagents are cataloged."""
args = argparse.Namespace(
tools_command="list",
output_format="json",
agent="custom-agent",
resume_thread=None,
interpreter=False,
sandbox="none",
no_mcp=False,
mcp_config=None,
trust_project_mcp=False,
)
with patch(
"deepagents_code.tool_catalog.collect_catalog",
return_value=ToolCatalog(groups=()),
) as collect:
code = run_tools_command(args)
assert code == 0
assert collect.call_args.kwargs["assistant_id"] == "custom-agent"
def test_list_interpreter_defaults_to_settings(self) -> None:
"""With no explicit flag, the resolved local default drives `js_eval`."""
args = argparse.Namespace(
tools_command="list",
output_format="json",
interpreter=None,
sandbox="none",
no_mcp=False,
mcp_config=None,
trust_project_mcp=False,
)
with (
patch(
"deepagents_code._server_config._resolve_enable_interpreter",
return_value=True,
),
patch(
"deepagents_code.tool_catalog.collect_catalog",
return_value=ToolCatalog(groups=()),
) as collect,
):
code = run_tools_command(args)
assert code == 0
assert collect.call_args.kwargs["enable_interpreter"] is True
assert collect.call_args.kwargs["include_mcp"] is True
def test_list_singular_noun_for_one_tool(self) -> None:
args = argparse.Namespace(tools_command="list", output_format="text")
catalog = ToolCatalog(
groups=(
ToolGroup(
label="Built-in",
source="built-in",
tools=(ToolEntry(name="ls", description="List files"),),
),
)
)
with patch(
"deepagents_code.tool_catalog.collect_catalog",
return_value=catalog,
):
code, output = _run_text(args)
assert code == 0
assert "1 tool available" in output
def test_long_description_clipped_to_terminal_width(self) -> None:
"""On a narrow terminal, descriptions clip and no row overflows width."""
long_desc = "abcdefg " * 60 # ~480 chars, far wider than the terminal
catalog = ToolCatalog(
groups=(
ToolGroup(
label="Built-in",
source="built-in",
tools=(ToolEntry(name="read_file", description=long_desc),),
),
)
)
args = argparse.Namespace(tools_command="list", output_format="text")
with patch(
"deepagents_code.tool_catalog.collect_catalog", return_value=catalog
):
code, output = _run_text(args, width=40)
assert code == 0
assert "read_file" in output
# The full description cannot fit; it must have been clipped.
assert long_desc.rstrip() not in output
# Every rendered row stays within the terminal width (no_wrap + crop).
assert all(len(line) <= 40 for line in output.splitlines())
def test_disabled_server_renders_without_detail_suffix(self) -> None:
"""An unavailable server with no detail prints status alone, no `: `."""
catalog = ToolCatalog(
groups=(
ToolGroup(
label="Built-in",
source="built-in",
tools=(ToolEntry(name="ls", description="List files"),),
),
),
unavailable=(
UnavailableServer(name="offsvc", status="disabled", detail=""),
),
)
args = argparse.Namespace(tools_command="list", output_format="text")
with patch(
"deepagents_code.tool_catalog.collect_catalog", return_value=catalog
):
code, output = _run_text(args)
assert code == 0
assert "Unavailable MCP servers" in output
assert "offsvc" in output
assert "disabled" in output
# Empty detail → status stands alone, no trailing `: `.
assert "disabled:" not in output
def test_list_end_to_end_offline_renders_real_built_ins(self) -> None:
"""Real `collect_catalog` compiles the agent offline and renders it."""
args = argparse.Namespace(
tools_command="list",
output_format="text",
interpreter=False,
sandbox="none",
no_mcp=True,
mcp_config=None,
trust_project_mcp=False,
)
code, output = _run_text(args)
assert code == 0
assert "tools available" in output
assert "Built-in" in output
# Representative built-in tools the default agent always binds.
assert "read_file" in output
assert "execute" in output
class TestTruncate:
"""Tests for `_truncate` description clipping."""
@pytest.mark.parametrize(
("text", "width", "ellipsis", "expected"),
[
("hello", 10, "...", "hello"), # fits comfortably
("hello", 5, "...", "hello"), # exact fit, no clip
("hello world", 9, "...", "hello..."), # clip + rstrip before ellipsis
("hello world", 8, "...", "hello..."), # trailing space dropped
("abcdef", 3, "...", "abc"), # width == len(ellipsis)
("abcdef", 2, "...", "ab"), # width < len(ellipsis)
("abcdef", 0, "...", "abcdef"), # zero width → unchanged
("abcdef", -5, "...", "abcdef"), # negative width → unchanged
("hello world", 8, "", "hello w…"), # single-char ellipsis
],
)
def test_truncate(
self, text: str, width: int, ellipsis: str, expected: str
) -> None:
assert _truncate(text, width, ellipsis) == expected